freenet/node/

testing_impl.rs

//! Simulation network implementation for testing Freenet nodes.
//!
//! This module provides `SimNetwork`, a fully in-memory simulation framework for testing
//! Freenet's peer-to-peer network behavior. It enables deterministic testing of complex
//! distributed scenarios without real network I/O.
//!
//! # Key Components
//!
//! - [`SimNetwork`]: The main simulation controller that manages virtual nodes
//! - [`NodeLabel`]: Identifies nodes in the simulation (gateways and regular nodes)
//! - [`FaultConfig`]: Configures fault injection (message loss, partitions, latency)
//! - [`VirtualTime`]: Controls deterministic time progression
//!
//! # Features
//!
//! - **Deterministic execution**: All randomness is seeded for reproducible tests
//! - **Fault injection**: Simulate message loss, network partitions, and node crashes
//! - **Virtual time**: Control time progression for testing time-dependent behavior
//! - **Node lifecycle**: Crash and restart nodes while preserving identity
//! - **Network-scoped state**: Each simulation is isolated from others
//!
//! # Example
//!
//! ```ignore
//! use freenet::dev_tool::SimNetwork;
//! use freenet::simulation::FaultConfig;
//!
//! // Create a 3-node network with 1 gateway
//! let mut sim = SimNetwork::new(
//!     "my-test",
//!     1,  // gateways
//!     2,  // regular nodes
//!     7,  // ring_max_htl
//!     3,  // rnd_if_htl_above
//!     10, // max_connections
//!     2,  // min_connections
//!     42, // seed for determinism
//! ).await;
//!
//! // Start the network
//! let handles = sim.start_with_rand_gen::<rand::rngs::SmallRng>(42, 1, 1).await;
//!
//! // Inject 10% message loss
//! sim.with_fault_injection(FaultConfig::builder()
//!     .message_loss_rate(0.1)
//!     .build());
//!
//! // Advance virtual time to trigger pending message deliveries
//! sim.advance_time(Duration::from_millis(100));
//! ```
//!
//! # Network Isolation
//!
//! Each `SimNetwork` instance is identified by its name and maintains isolated state.
//! This allows running multiple independent simulations in parallel tests without
//! interference. The fault injection state is automatically cleaned up when the
//! simulation is dropped.
//!
//! # Thread Safety
//!
//! The simulation is thread-safe and can be used with `#[tokio::test]` multi-threaded
//! test configurations. Internal state is protected by appropriate synchronization
//! primitives.

use freenet_stdlib::prelude::*;
use futures::Future;
use rand::prelude::IndexedRandom;
use std::{
    collections::{BTreeMap, HashMap, HashSet},
    net::{Ipv6Addr, SocketAddr},
    num::NonZeroUsize,
    pin::Pin,
    sync::Arc,
    time::Duration,
};
use tokio::sync::{broadcast, mpsc, watch};
use tracing::info;

#[cfg(feature = "trace-ot")]
use crate::tracing::CombinedRegister;
use crate::{
    client_events::test::{MemoryEventsGen, RandomEventGenerator},
    config::{ConfigArgs, GlobalExecutor, GlobalRng},
    dev_tool::TransportKeypair,
    node::{InitPeerNode, NetEventRegister, NodeConfig},
    ring::{ConnectionManager, Distance, Location, PeerKeyLocation},
    simulation::{FaultConfig, VirtualTime},
    tracing::TestEventListener,
    transport::{
        TransportPublicKey,
        in_memory_socket::{register_network_time_source, unregister_network_time_source},
    },
};

mod in_memory;
mod network;
pub mod turmoil_runner;

pub use self::network::{NetworkPeer, PeerMessage, PeerStatus};
pub use self::turmoil_runner::{TurmoilConfig, TurmoilResult, run_turmoil_simulation};

pub(crate) type EventId = u32;

/// A controlled operation to execute on a specific node during simulation.
///
/// `SimOperation` allows tests to specify exact operations instead of relying
/// on random event generation. This enables precise testing of specific scenarios
/// like subscription topology formation.
///
/// # Usage
///
/// ```ignore
/// use freenet::dev_tool::{SimNetwork, SimOperation, NodeLabel};
///
/// let mut sim = SimNetwork::new("test", 1, 3, ...).await;
///
/// // Create a contract and have specific nodes subscribe
/// let contract = SimOperation::create_test_contract(42);
/// let operations = vec![
///     (NodeLabel::gateway("test", 0), SimOperation::Put {
///         contract: contract.clone(),
///         state: vec![1, 2, 3],
///         subscribe: true,
///     }),
///     (NodeLabel::node("test", 0), SimOperation::Subscribe {
///         contract_id: *contract.key().id(),
///     }),
/// ];
///
/// let handles = sim.start_with_controlled_events(operations).await;
/// ```
#[derive(Clone, Debug)]
pub enum SimOperation {
    /// PUT a new contract with initial state.
    Put {
        /// The contract container (code + parameters)
        contract: ContractContainer,
        /// Initial state bytes
        state: Vec<u8>,
        /// Whether to subscribe to updates after PUT
        subscribe: bool,
    },
    /// GET a contract's current state.
    Get {
        /// The contract instance ID to retrieve
        contract_id: ContractInstanceId,
        /// Whether to return the contract code
        return_contract_code: bool,
        /// Whether to subscribe to updates after GET
        subscribe: bool,
    },
    /// SUBSCRIBE to a contract's updates.
    Subscribe {
        /// The contract instance ID to subscribe to
        contract_id: ContractInstanceId,
    },
    /// UPDATE a contract's state.
    Update {
        /// The contract key to update
        key: ContractKey,
        /// New state data
        data: Vec<u8>,
    },
    /// Seed a contract into a node's local store WITHOUT network propagation.
    ///
    /// Unlike `Put`, this does not trigger a network PUT operation — the
    /// contract is only stored in the target node's `MockStateStorage`.
    /// This is useful for setting up test scenarios where specific nodes
    /// must have (or not have) a contract before subscribe/get operations.
    ///
    /// Example: seed only on gateway, then subscribe from other nodes.
    /// The subscribe will go through the network because other nodes lack
    /// the contract, and relay peers will forward (generating ForwardingAck).
    SeedContract {
        /// The contract container (code + parameters)
        contract: ContractContainer,
        /// Initial state bytes
        state: Vec<u8>,
    },
    /// Seed a contract into a node's local store AND register the node as
    /// genuinely HOSTING it (state + `host_contract` + active subscription),
    /// without any network propagation.
    ///
    /// Unlike [`SeedContract`](Self::SeedContract) — which only writes the raw
    /// state/params into `MockStateStorage` — this routes the contract through
    /// the node's startup `append_contracts` path with `subscription = true`,
    /// so `ring.is_hosting_contract` / `ring.hosting_contract_keys` report it.
    /// That is what makes the node a valid migration source: the
    /// placement-migration trigger only nudges contracts the node actually
    /// hosts. Use this when a test needs a node to be the authoritative holder
    /// of a contract that was never PUT through the network (e.g. reproducing
    /// the GET-dead-end placement gap, #4404).
    SeedHostedContract {
        /// The contract container (code + parameters)
        contract: ContractContainer,
        /// Initial state bytes
        state: Vec<u8>,
    },
    /// Seed a **demandless every-hop copy** into a node's local store: state
    /// present, `is_hosting_contract` true, `has_local_interest` true, but NO
    /// subscription (`is_receiving_updates` false) and NO prior local client
    /// access.
    ///
    /// This is the state a production every-hop GET/PUT store leaves on a RELAY
    /// hop (`get`/`put` `op_ctx_task` call `register_local_hosting` without
    /// `ring.subscribe` and, on a non-originator hop, without
    /// `mark_local_client_access`). Unlike
    /// [`SeedHostedContract`](Self::SeedHostedContract) — which registers an
    /// active subscription, so the pre-serve-DURING gate already served it — a
    /// demandless copy is exactly what the pre-serve-DURING originator gate
    /// (`is_hosting && has_local_client_access`) would REFUSE to serve, forcing
    /// a whole GET through the network for a copy the node already held fresh.
    /// Use this to exercise serve-DURING (#4642 R3 piece C): a connected node
    /// holding a demandless copy must answer a GET from its local copy instead
    /// of going dark on the network.
    SeedDemandlessCopy {
        /// The contract container (code + parameters)
        contract: ContractContainer,
        /// Initial state bytes
        state: Vec<u8>,
    },
    /// Simulate a client disconnecting from this node.
    ///
    /// Emits `ClientRequest::Disconnect` through the same
    /// `MemoryEventsGen` → `client_event_handling` path the real
    /// WebSocket client takes, so the node runs
    /// `remove_client_from_all_subscriptions`,
    /// `should_unsubscribe_upstream`, and `send_unsubscribe_upstream`
    /// — producing `UnsubscribeSent` / `UnsubscribeReceived` telemetry
    /// that can be asserted on.
    ///
    /// This targets `ClientId::FIRST`, which is the client id used by
    /// every other `SimOperation` in the same node, so subscriptions
    /// issued earlier by this operation's node are the ones cleaned up.
    Disconnect,
    /// Advance the sim's controllable hosting clock by `duration`, in-order
    /// with the surrounding events.
    ///
    /// Requires [`SimNetwork::enable_hosting_time_control`]; the clock is shared
    /// by every node, so the `node` on the [`ScheduledOperation`] is only a
    /// sequencing placeholder (advancement is global). Use this to jump past the
    /// hosting-cache TTL gate deterministically so a subsequent access triggers
    /// demand-driven eviction — without running minutes of virtual time
    /// (#4642 piece A). Handled specially by `run_controlled_simulation`
    /// (it is not a client request).
    AdvanceHostingClock {
        /// How far to advance the shared hosting clock.
        duration: Duration,
    },
    /// Crash the target node mid-run by blocking all its inbound/outbound
    /// messages via the per-network fault injector, in-order with the
    /// surrounding events.
    ///
    /// This is a SILENT crash (message-blocking, like the churn driver's
    /// soft-crash): the node's transport is not torn down, so peers detect the
    /// loss on their own routing-health cadence rather than an immediate
    /// transport-close. It lets a test remove a specific peer at a scripted
    /// point (e.g. to observe re-routing / re-subscription). Handled specially
    /// by `run_controlled_simulation` (it is not a client request).
    CrashNode,
    /// Recover a previously [`CrashNode`](Self::CrashNode)-crashed node mid-run,
    /// un-blocking its inbound/outbound messages via the per-network fault
    /// injector, in-order with the surrounding events.
    ///
    /// The inverse of [`CrashNode`](Self::CrashNode): the node was never torn
    /// down (message-blocking crash), so recovering simply lets its traffic flow
    /// again. Use it to reconnect a node after it has deliberately missed some
    /// traffic (e.g. an update fan-out), so a test can observe it re-converge via
    /// anti-entropy. Handled specially by `run_controlled_simulation` (it is not
    /// a client request).
    RecoverNode,
}

impl SimOperation {
    /// Creates a deterministic test contract from a seed.
    ///
    /// The contract code and parameters are derived from the seed,
    /// making it reproducible across test runs.
    pub fn create_test_contract(seed: u8) -> ContractContainer {
        let mut code_bytes = vec![0u8; 32];
        let mut params_bytes = vec![0u8; 16];

        // Fill with deterministic bytes based on seed
        for (i, byte) in code_bytes.iter_mut().enumerate() {
            *byte = seed.wrapping_add(i as u8);
        }
        for (i, byte) in params_bytes.iter_mut().enumerate() {
            *byte = seed.wrapping_add(i as u8).wrapping_mul(2);
        }

        let code = ContractCode::from(code_bytes);
        let params = Parameters::from(params_bytes);
        ContractWasmAPIVersion::V1(WrappedContract::new(code.into(), params)).into()
    }

    /// Creates a large deterministic state for streaming tests.
    ///
    /// `size_bytes` controls total state size (use > streaming_threshold to trigger streaming).
    /// The content is deterministic based on `seed` for reproducibility.
    pub fn create_large_state(size_bytes: usize, seed: u8) -> Vec<u8> {
        let mut state = Vec::with_capacity(size_bytes);
        for i in 0..size_bytes {
            state.push(seed.wrapping_add((i % 256) as u8).wrapping_mul(3));
        }
        state
    }

    /// Creates a deterministic test state from a seed.
    pub fn create_test_state(seed: u8) -> Vec<u8> {
        let mut state_bytes = vec![0u8; 64];
        for (i, byte) in state_bytes.iter_mut().enumerate() {
            *byte = seed.wrapping_add(i as u8).wrapping_mul(3);
        }
        state_bytes
    }

    /// Creates a CRDT-mode test state with version prefix.
    ///
    /// Format: [version: u64 LE][64 bytes of data]
    ///
    /// Use this with `register_crdt_contract()` to test version-aware delta handling.
    /// The CRDT mode enables testing of PR #2763's summary caching fix by:
    /// - Using version-aware summaries (version + hash)
    /// - Computing version-specific deltas
    /// - Failing delta application when versions don't match
    pub fn create_crdt_state(version: u64, seed: u8) -> Vec<u8> {
        // State must be > 80 bytes for delta to be "efficient"
        // (efficiency check: summary_size * 2 < state_size, where summary = 40 bytes)
        // Using 128 bytes of data for total state size of 136 bytes
        let mut state_bytes = Vec::with_capacity(8 + 128);
        // Add version prefix
        state_bytes.extend_from_slice(&version.to_le_bytes());
        // Add data (128 bytes to make delta efficient)
        for i in 0..128u8 {
            state_bytes.push(seed.wrapping_add(i).wrapping_mul(3));
        }
        state_bytes
    }

    /// Converts this operation to a ClientRequest.
    #[cfg(any(test, feature = "testing"))]
    pub(crate) fn into_client_request(self) -> freenet_stdlib::client_api::ClientRequest<'static> {
        use freenet_stdlib::client_api::{ClientRequest, ContractRequest};

        match self {
            SimOperation::Put {
                contract,
                state,
                subscribe,
            } => ClientRequest::ContractOp(ContractRequest::Put {
                contract,
                state: WrappedState::new(state),
                related_contracts: RelatedContracts::new(),
                subscribe,
                blocking_subscribe: false,
            }),
            SimOperation::Get {
                contract_id,
                return_contract_code,
                subscribe,
            } => ClientRequest::ContractOp(ContractRequest::Get {
                key: contract_id,
                return_contract_code,
                subscribe,
                blocking_subscribe: false,
            }),
            SimOperation::Subscribe { contract_id } => {
                ClientRequest::ContractOp(ContractRequest::Subscribe {
                    key: contract_id,
                    summary: None,
                })
            }
            SimOperation::Update { key, data } => {
                ClientRequest::ContractOp(ContractRequest::Update {
                    key,
                    data: UpdateData::State(State::from(data)),
                })
            }
            SimOperation::SeedContract { .. } => {
                panic!(
                    "SeedContract is not a client request — it must be handled \
                     by run_controlled_simulation before event dispatch"
                )
            }
            SimOperation::SeedHostedContract { .. } => {
                panic!(
                    "SeedHostedContract is not a client request — it must be handled \
                     by run_controlled_simulation before event dispatch"
                )
            }
            SimOperation::SeedDemandlessCopy { .. } => {
                panic!(
                    "SeedDemandlessCopy is not a client request — it must be handled \
                     by run_controlled_simulation before event dispatch"
                )
            }
            SimOperation::AdvanceHostingClock { .. } => {
                panic!(
                    "AdvanceHostingClock is not a client request — it must be handled \
                     by run_controlled_simulation during event dispatch"
                )
            }
            SimOperation::CrashNode => {
                panic!(
                    "CrashNode is not a client request — it must be handled \
                     by run_controlled_simulation during event dispatch"
                )
            }
            SimOperation::RecoverNode => {
                panic!(
                    "RecoverNode is not a client request — it must be handled \
                     by run_controlled_simulation during event dispatch"
                )
            }
            SimOperation::Disconnect => ClientRequest::Disconnect { cause: None },
        }
    }
}

/// A scheduled operation for controlled event simulation.
#[derive(Clone, Debug)]
pub struct ScheduledOperation {
    /// Which node should execute this operation
    pub node: NodeLabel,
    /// The operation to execute
    pub operation: SimOperation,
}

impl ScheduledOperation {
    /// Create a new scheduled operation.
    pub fn new(node: NodeLabel, operation: SimOperation) -> Self {
        Self { node, operation }
    }
}

/// Packet-delivery decision consulted by every `SimulationSocket` (via the
/// global delivery callback installed by `run_controlled_simulation` and
/// `run_simulation_direct`) so an injected node crash or partition actually
/// drops packets instead of being a silent no-op.
///
/// Per-network: it looks up the fault injector for the socket's own network and,
/// only if that network opted in (`enforce_fault_drops`), DROPS any packet whose
/// source or destination is a crashed node (counted in
/// `stats.messages_dropped_crash`) or that is blocked by an active partition
/// (counted in `stats.messages_dropped_partition`). Every other packet (and
/// every network that did not opt in) is delivered unchanged, so installing this
/// callback cannot alter fault behavior for any path that did not opt in.
/// See #4642 piece F (scripted crashes) and #4694 (direct-runner churn).
#[cfg(any(test, feature = "testing"))]
fn fault_injection_delivery_decision(
    network_name: &str,
    from: SocketAddr,
    to: SocketAddr,
) -> crate::transport::in_memory_socket::PacketDeliveryDecision {
    use crate::transport::in_memory_socket::PacketDeliveryDecision;
    if let Some(injector) = crate::node::network_bridge::get_fault_injector(network_name) {
        let mut inj = injector.lock().unwrap();
        if inj.enforce_fault_drops {
            if inj.config.is_crashed(&from) || inj.config.is_crashed(&to) {
                inj.stats.messages_dropped_crash += 1;
                return PacketDeliveryDecision::Drop;
            }
            // Partitions are time-scoped (start/heal), so use this network's
            // VirtualTime as the current time; without VirtualTime a partition
            // cannot be evaluated deterministically, so treat it as inactive.
            use crate::simulation::TimeSource;
            if let Some(now) = inj.virtual_time.as_ref().map(|vt| vt.now_nanos()) {
                if inj.config.is_partitioned(&from, &to, now) {
                    inj.stats.messages_dropped_partition += 1;
                    return PacketDeliveryDecision::Drop;
                }
            }
        }
    }
    PacketDeliveryDecision::Deliver
}

/// Result of a controlled simulation, including topology snapshots.
///
/// This struct captures the simulation result along with topology snapshots
/// taken at the end of the simulation (before cleanup). This allows tests
/// to validate subscription topology after the simulation completes.
pub struct ControlledSimulationResult {
    /// The Turmoil simulation result
    pub turmoil_result: turmoil::Result,
    /// Topology snapshots captured at the end of simulation
    pub topology_snapshots: Vec<crate::ring::topology_registry::TopologySnapshot>,
    /// Shared storage handles for each node, keyed by NodeLabel.
    /// These are clones of the Arc-backed storages passed into Turmoil,
    /// so they reflect all state stored during the simulation.
    pub node_storages: HashMap<NodeLabel, crate::wasm_runtime::MockStateStorage>,
    /// Live `Arc<Ring>` for each node that started, keyed by NodeLabel.
    /// Captured via the `shared_ring` slots so governance sim tests can
    /// read `ring.contract_ban_list.is_banned(..)` after the simulation
    /// (the `SimNetwork` itself is consumed by `run_controlled_simulation`).
    /// Only present under `cfg(test)`/`testing`.
    ///
    /// `allow(dead_code)`: read only by the `cfg(test)` governance e2e
    /// module, so non-test builds with the `testing` feature see it unused.
    #[cfg(any(test, feature = "testing"))]
    #[allow(dead_code)]
    pub(crate) node_rings: HashMap<NodeLabel, Arc<crate::ring::Ring>>,
    /// Per-peer subscription-renewal metrics captured at the end of the
    /// simulation, keyed by socket address. Captured here (before the
    /// `SimNetwork` is dropped, which clears the registry) so renewal-storm
    /// tests can read `wire_attempts` / `terminus_satisfied` after the run
    /// returns. See `crate::ring::topology_registry::RenewalMetrics` (#4440).
    pub renewal_metrics:
        HashMap<std::net::SocketAddr, crate::ring::topology_registry::RenewalMetrics>,
    /// Number of packets DROPPED because their source or destination was a
    /// crashed node (`SimOperation::CrashNode`), captured from the fault
    /// injector's `stats.messages_dropped_crash` before the `SimNetwork` is
    /// dropped. A discriminating signal that a scripted crash actually took
    /// effect: it is `> 0` only if a crash was scheduled AND real traffic
    /// to/from the crashed node was subsequently blocked. `0` when no crash was
    /// scheduled. See #4642 piece F.
    pub crash_packets_dropped: u64,
    /// Per-peer count of WASM `summarize_state` invocations (summary-cache
    /// SLOW-path misses) captured at the end of the run, keyed by socket
    /// address. Captured here (before the `SimNetwork` is dropped, which clears
    /// the registry) so the every-hop-placement summarize-storm falsifier can
    /// assert this stays flat as the hosted set / neighbor overlap grows. A
    /// cache hit does NOT increment, so a working cache keeps this proportional
    /// to the state-change rate, not to hosted-set size. See
    /// `crate::ring::topology_registry::record_summarize_wasm_call` (#4440, spec
    /// step 8).
    pub summarize_wasm_calls: HashMap<std::net::SocketAddr, u64>,
}

#[cfg(any(test, feature = "testing"))]
impl ControlledSimulationResult {
    /// Whether `label`'s node was hosting `key` in its live Ring at the end of
    /// the simulation.
    ///
    /// Reads the per-node `Ring` captured via the `shared_ring` slots, so it
    /// reflects `ring.is_hosting_contract` exactly (not a storage proxy). Used
    /// by placement/migration tests to assert a contract has migrated onto a
    /// given peer. Returns `false` if the node never started or never published
    /// its Ring.
    pub fn is_node_hosting(&self, label: &NodeLabel, key: &ContractKey) -> bool {
        self.node_rings
            .get(label)
            .is_some_and(|ring| ring.is_hosting_contract(key))
    }

    /// The protocol version `label`'s node had recorded for the peer at `addr`
    /// at the end of the run, or `None` if it never learned one (#5161).
    ///
    /// Reads the same `ConnectionManager` mirror the production emission gates
    /// read, so it answers the question those gates ask rather than a proxy for
    /// it. `None` when the node never published its Ring.
    pub fn node_recorded_remote_version(
        &self,
        label: &NodeLabel,
        addr: SocketAddr,
    ) -> Option<(u8, u8, u16)> {
        self.node_rings
            .get(label)
            .and_then(|ring| ring.connection_manager.remote_version(addr))
    }

    /// Whether `label`'s node would emit the hash-first summary encoding to the
    /// peer at `addr` — i.e. whether that link passes the version gate (#4965).
    ///
    /// The end-to-end consequence of #5161 on a node->gateway link: before the
    /// version-carrying ack this could never be true from the node's side, so
    /// the node answered every anti-entropy exchange with full summary bytes.
    /// `false` when the node never published its Ring.
    pub fn node_supports_hash_first_summaries(&self, label: &NodeLabel, addr: SocketAddr) -> bool {
        self.node_rings
            .get(label)
            .is_some_and(|ring| ring.connection_manager.supports_hash_first_summaries(addr))
    }

    /// Number of contracts `label`'s node held in its hosting cache at the end
    /// of the run (its "cache size"). Returns 0 if the node never published its
    /// Ring. See [`Ring::hosting_contracts_count`](crate::ring::Ring::hosting_contracts_count).
    pub fn node_hosting_count(&self, label: &NodeLabel) -> usize {
        self.node_rings
            .get(label)
            .map(|ring| ring.hosting_contracts_count())
            .unwrap_or(0)
    }

    /// Number of active network subscription leases `label`'s node held at the
    /// end of the run. Returns 0 if the node never published its Ring. See
    /// [`Ring::active_subscription_count`](crate::ring::Ring::active_subscription_count).
    pub fn node_subscription_count(&self, label: &NodeLabel) -> usize {
        self.node_rings
            .get(label)
            .map(|ring| ring.active_subscription_count())
            .unwrap_or(0)
    }

    /// Number of contracts `label`'s node had *real demand* for (local client
    /// or downstream subscriber, EXCLUDING cache-only hosting) at the end of the
    /// run. This is the denominator for the #3763 no-storm assertion — see
    /// [`Ring::active_demand_count`](crate::ring::Ring::active_demand_count).
    ///
    /// Returns `None` when the count is UNMEASURABLE — the node never published
    /// its Ring, or its `OpManager` was detached (startup/shutdown). A no-storm
    /// assertion MUST distinguish `None` (unmeasurable) from `Some(0)`
    /// ("attached, genuinely no demand"): treating a torn-down node's `None` as
    /// `0` would let a real storm read as "no demand" and pass falsely. Callers
    /// that only log can `{:?}` it; callers that assert should handle `None`
    /// explicitly (e.g. skip the node, or fail if a measurement was required).
    pub fn node_active_demand_count(&self, label: &NodeLabel) -> Option<usize> {
        self.node_rings
            .get(label)
            .and_then(|ring| ring.active_demand_count())
    }

    /// Number of upstream peers `label`'s node recorded for `key` (its parents
    /// in the subscription tree) at the end of the run. Used by re-subscribe
    /// tests (#4642 piece F) to assert a KNOWN subscriber→upstream edge. See
    /// [`Ring::upstream_interest_count`](crate::ring::Ring::upstream_interest_count).
    ///
    /// Returns `None` when unmeasurable (node never published its Ring, or its
    /// `OpManager` was detached) — distinct from `Some(0)` ("attached, no
    /// upstream edge recorded"). A piece-F re-root assertion must not read a
    /// detached node's `None` as "edge gone".
    pub fn node_upstream_count(&self, label: &NodeLabel, key: &ContractKey) -> Option<usize> {
        self.node_rings
            .get(label)
            .and_then(|ring| ring.upstream_interest_count(key))
    }

    /// Labels of every node whose live Ring was captured (started and published
    /// its Ring). Handy for iterating per-node assertions.
    pub fn captured_node_labels(&self) -> Vec<NodeLabel> {
        self.node_rings.keys().cloned().collect()
    }

    /// Number of open ring connections `label`'s node held at the end of the run.
    /// Returns 0 if the node never published its Ring. Used by the serve-DURING
    /// falsifier to confirm the serving node was CONNECTED (so a local serve is
    /// attributable to serve-DURING interest, not the isolated-node fallback).
    pub fn node_open_connections(&self, label: &NodeLabel) -> usize {
        self.node_rings
            .get(label)
            .map(|ring| ring.open_connections())
            .unwrap_or(0)
    }

    /// Ring locations (as `f64`) of `label`'s connected neighbors at the end of
    /// the run, read from the captured live `Ring`'s connection manager. Empty
    /// if the node never published its Ring. Used by the nearest-neighbor
    /// findability experiment to premise-check that the guaranteed
    /// successor/predecessor edges actually formed. Only present under
    /// `cfg(test)`/`testing`.
    #[cfg(any(test, feature = "testing"))]
    pub fn node_neighbor_locations(&self, label: &NodeLabel) -> Vec<f64> {
        self.node_rings
            .get(label)
            .map(|ring| {
                ring.connection_manager
                    .get_connections_by_location()
                    .keys()
                    .map(|loc| loc.as_f64())
                    .collect()
            })
            .unwrap_or_default()
    }

    /// Whether `label`'s node was actively receiving updates for `key` (has a
    /// live network/client subscription keeping the copy fresh) at the end of the
    /// run. Returns `false` if the node never published its Ring. The serve-DURING
    /// falsifier asserts this is `false` for a demandless copy, so the local serve
    /// is attributable to `has_local_interest`, not an active subscription.
    pub fn node_is_receiving_updates(&self, label: &NodeLabel, key: &ContractKey) -> bool {
        self.node_rings
            .get(label)
            .is_some_and(|ring| ring.is_receiving_updates(key))
    }

    /// Number of client GETs `label`'s node answered from local hosted state (A3
    /// serve-DURING hit counter). Returns 0 if the node never published its Ring.
    /// See [`Ring::local_get_serves`](crate::ring::Ring::local_get_serves).
    pub fn node_local_get_serves(&self, label: &NodeLabel) -> u64 {
        self.node_rings
            .get(label)
            .map(|ring| ring.local_get_serves())
            .unwrap_or(0)
    }

    /// Number of client GETs `label`'s node routed to the network (A3 forward/miss
    /// counter). Returns 0 if the node never published its Ring. The serve-DURING
    /// falsifier asserts this stays 0 for a demandless-copy GET (never went dark).
    /// See [`Ring::local_get_forwards`](crate::ring::Ring::local_get_forwards).
    pub fn node_local_get_forwards(&self, label: &NodeLabel) -> u64 {
        self.node_rings
            .get(label)
            .map(|ring| ring.local_get_forwards())
            .unwrap_or(0)
    }

    /// Subscription-renewal metrics for the peer at `addr` (captured before the
    /// `SimNetwork` was dropped). Returns `RenewalMetrics::default()` (all zero)
    /// if the peer recorded no renewal activity. See #4440.
    pub fn renewal_metrics_for(
        &self,
        addr: &std::net::SocketAddr,
    ) -> crate::ring::topology_registry::RenewalMetrics {
        self.renewal_metrics.get(addr).copied().unwrap_or_default()
    }

    /// Sum of the renewal metrics across every peer in the simulation. See #4440.
    pub fn aggregate_renewal_metrics(&self) -> crate::ring::topology_registry::RenewalMetrics {
        self.renewal_metrics.values().fold(
            crate::ring::topology_registry::RenewalMetrics::default(),
            |mut acc, m| {
                acc.wire_attempts += m.wire_attempts;
                acc.terminus_satisfied += m.terminus_satisfied;
                // Per-node peak → aggregate as the max across peers.
                acc.max_cycle_batch = acc.max_cycle_batch.max(m.max_cycle_batch);
                acc.max_reroot_batch = acc.max_reroot_batch.max(m.max_reroot_batch);
                acc
            },
        )
    }

    /// Number of packets dropped because a `SimOperation::CrashNode` marked
    /// their source or destination crashed. `> 0` proves a scripted crash
    /// actually blocked real traffic (discriminating signal for piece-F crash
    /// tests); `0` when no crash was scheduled. See #4642 piece F.
    pub fn crash_packets_dropped(&self) -> u64 {
        self.crash_packets_dropped
    }

    /// WASM `summarize_state` invocations (summary-cache misses) recorded for
    /// the peer at `addr`. `0` if the peer ran none. See #4440 / spec step 8.
    pub fn summarize_wasm_calls_for(&self, addr: &std::net::SocketAddr) -> u64 {
        self.summarize_wasm_calls.get(addr).copied().unwrap_or(0)
    }

    /// Total WASM `summarize_state` invocations across every peer in the run.
    ///
    /// This is the every-hop summarize-storm falsifier's primary signal: it must
    /// stay proportional to the contract STATE-CHANGE count, NOT to hosted-set
    /// size or neighbor overlap. If it grows with the number of shared hosted
    /// contracts × neighbors, the state-hash summary cache is not covering
    /// every-hop load and the #4440 storm has re-armed.
    pub fn total_summarize_wasm_calls(&self) -> u64 {
        self.summarize_wasm_calls.values().copied().sum()
    }

    /// The single peer's peak WASM-summarize count — the worst per-node
    /// summarize load any peer bore during the run.
    pub fn max_summarize_wasm_calls(&self) -> u64 {
        self.summarize_wasm_calls
            .values()
            .copied()
            .max()
            .unwrap_or(0)
    }
}

#[derive(PartialEq, Eq, Hash, Clone, PartialOrd, Ord, Debug)]
pub struct NodeLabel(Arc<str>);

impl NodeLabel {
    /// Creates a gateway label for a network.
    pub fn gateway(network_name: &str, id: usize) -> Self {
        Self(format!("{network_name}-gateway-{id}").into())
    }

    /// Creates a regular node label for a network.
    pub fn node(network_name: &str, id: usize) -> Self {
        Self(format!("{network_name}-node-{id}").into())
    }

    /// Returns true if this is a gateway label.
    pub fn is_gateway(&self) -> bool {
        self.0.contains("-gateway-")
    }

    /// Returns true if this is a regular node label.
    pub fn is_node(&self) -> bool {
        self.0.contains("-node-")
    }

    pub fn number(&self) -> usize {
        // Label format is "{network_name}-{gateway|node}-{id}"
        // The number is always the last part after the final '-'
        self.0
            .rsplit('-')
            .next()
            .expect("should have a number part")
            .parse::<usize>()
            .expect("last part should be a number")
    }
}

impl std::fmt::Display for NodeLabel {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

impl std::ops::Deref for NodeLabel {
    type Target = str;

    fn deref(&self) -> &Self::Target {
        self.0.deref()
    }
}

impl<'a> From<&'a str> for NodeLabel {
    fn from(value: &'a str) -> Self {
        assert!(value.starts_with("gateway-") || value.starts_with("node-"));
        let mut parts = value.split('-');
        assert!(parts.next().is_some());
        assert!(
            parts
                .next()
                .map(|s| s.parse::<u16>())
                .transpose()
                .expect("should be an u16")
                .is_some()
        );
        assert!(parts.next().is_none());
        Self(value.to_string().into())
    }
}

#[derive(Clone)]
pub(crate) struct GatewayConfig {
    #[allow(dead_code)]
    label: NodeLabel,
    peer_key_location: PeerKeyLocation,
    location: Location,
}

/// Summary of a network event for deterministic comparison.
///
/// Excludes timestamps which vary between runs.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct EventSummary {
    pub tx: crate::message::Transaction,
    pub peer_addr: std::net::SocketAddr,
    /// String representation of the event kind for sorting
    pub event_kind_name: String,
    /// Contract key if this event involves a contract operation
    pub contract_key: Option<String>,
    /// State hash if this event includes state (Put/Update success/broadcast)
    pub state_hash: Option<String>,
    /// Full debug representation of the event (for backwards compatibility)
    pub event_detail: String,
}

/// A stream of events for simulation testing.
///
/// `EventChain` provides a `Stream` of events that drive user interactions in
/// the simulated network.
///
/// # Ownership and Cleanup
///
/// There are two ways to obtain an `EventChain`:
///
/// 1. **`event_chain(&mut sim)`** - Borrows from `SimNetwork`. The `SimNetwork`
///    retains ownership of labels and handles cleanup on drop. The `EventChain`
///    is created with `clean_up_tmp_dirs = false`.
///
/// 2. **`into_event_chain(sim)`** - Consumes `SimNetwork`. The `EventChain` takes
///    ownership of labels and handles cleanup on drop. The `EventChain` is created
///    with `clean_up_tmp_dirs` set to whatever `SimNetwork` had.
///
/// This design allows tests to either retain access to `SimNetwork` for verification
/// methods, or to release it for simpler cleanup.
pub struct EventChain<S = watch::Sender<(EventId, TransportPublicKey)>> {
    labels: Vec<(NodeLabel, TransportPublicKey)>,
    user_ev_controller: S,
    total_events: u32,
    count: u32,
    rng: rand::rngs::SmallRng,
    /// Whether this EventChain is responsible for cleaning up temp directories.
    /// Set to `true` when created via `into_event_chain()` (consuming SimNetwork),
    /// `false` when created via `event_chain()` (borrowing from SimNetwork).
    clean_up_tmp_dirs: bool,
    choice: Option<TransportPublicKey>,
}

impl<S> EventChain<S> {
    pub fn new(
        labels: Vec<(NodeLabel, TransportPublicKey)>,
        user_ev_controller: S,
        total_events: u32,
        clean_up_tmp_dirs: bool,
    ) -> Self {
        const SEED: u64 = 0xdeadbeef;
        EventChain {
            labels,
            user_ev_controller,
            total_events,
            count: 0,
            rng: rand::rngs::SmallRng::seed_from_u64(SEED),
            clean_up_tmp_dirs,
            choice: None,
        }
    }

    fn increment_count(self: Pin<&mut Self>) {
        // SAFETY: We only modify `count` (a non-address-sensitive `usize` field)
        // through the pinned reference; the EventChain itself is not moved.
        unsafe {
            let this = self.get_unchecked_mut();
            this.count += 1;
        }
    }

    fn choose_peer(self: Pin<&mut Self>) -> TransportPublicKey {
        // SAFETY: We access `choice`, `rng`, and `labels` by mutable reference
        // without moving the EventChain out of its pinned location.
        let this = unsafe { self.get_unchecked_mut() };
        if let Some(id) = this.choice.take() {
            return id;
        }
        let rng = &mut this.rng;
        let labels = &mut this.labels;
        let (_, id) = labels.choose(rng).expect("not empty");
        id.clone()
    }

    fn set_choice(self: Pin<&mut Self>, id: TransportPublicKey) {
        // SAFETY: We only write to the `choice` field without moving the
        // pinned EventChain itself.
        let this = unsafe { self.get_unchecked_mut() };
        this.choice = Some(id);
    }
}

trait EventSender {
    fn send(
        &self,
        cx: &mut std::task::Context<'_>,
        value: (EventId, TransportPublicKey),
    ) -> std::task::Poll<Result<(), ()>>;
}

impl EventSender for mpsc::Sender<(EventId, TransportPublicKey)> {
    fn send(
        &self,
        cx: &mut std::task::Context<'_>,
        value: (EventId, TransportPublicKey),
    ) -> std::task::Poll<Result<(), ()>> {
        let f = self.send(value);
        futures::pin_mut!(f);
        f.poll(cx).map(|r| r.map_err(|_| ()))
    }
}

impl EventSender for watch::Sender<(EventId, TransportPublicKey)> {
    fn send(
        &self,
        _cx: &mut std::task::Context<'_>,
        value: (EventId, TransportPublicKey),
    ) -> std::task::Poll<Result<(), ()>> {
        match self.send(value) {
            Ok(_) => std::task::Poll::Ready(Ok(())),
            Err(_) => std::task::Poll::Ready(Err(())),
        }
    }
}

impl EventSender for broadcast::Sender<(EventId, TransportPublicKey)> {
    fn send(
        &self,
        _cx: &mut std::task::Context<'_>,
        value: (EventId, TransportPublicKey),
    ) -> std::task::Poll<Result<(), ()>> {
        match self.send(value) {
            Ok(_) => std::task::Poll::Ready(Ok(())),
            Err(_) => std::task::Poll::Ready(Err(())),
        }
    }
}

impl<S: EventSender> futures::stream::Stream for EventChain<S> {
    type Item = EventId;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        if self.count < self.total_events {
            let id = self.as_mut().choose_peer();
            match self
                .user_ev_controller
                .send(cx, (self.count, id.clone()))
                .map_err(|_| {
                    tracing::error!("peer controller should be alive, finishing event chain")
                }) {
                std::task::Poll::Ready(_) => {}
                std::task::Poll::Pending => {
                    self.as_mut().set_choice(id);
                    return std::task::Poll::Pending;
                }
            }
            self.as_mut().increment_count();
            std::task::Poll::Ready(Some(self.count))
        } else {
            std::task::Poll::Ready(None)
        }
    }
}

impl<S> Drop for EventChain<S> {
    fn drop(&mut self) {
        if self.clean_up_tmp_dirs {
            clean_up_tmp_dirs(self.labels.iter().map(|(l, _)| l));
        }
    }
}

/// A controlled event chain that triggers events in a specific sequence.
///
/// Unlike `EventChain` which randomly selects peers, `ControlledEventChain`
/// triggers events in the exact order specified, allowing deterministic testing.
pub struct ControlledEventChain {
    user_ev_controller: watch::Sender<(EventId, TransportPublicKey)>,
    /// Sequence of (EventId, NodeLabel) to trigger, in order
    event_sequence: Vec<(EventId, NodeLabel)>,
    /// Map from NodeLabel to TransportPublicKey
    label_to_key: HashMap<NodeLabel, TransportPublicKey>,
    /// Current position in the event sequence
    current_index: usize,
}

impl ControlledEventChain {
    /// Create a new controlled event chain.
    pub fn new(
        user_ev_controller: watch::Sender<(EventId, TransportPublicKey)>,
        event_sequence: Vec<(EventId, NodeLabel)>,
        label_to_key: HashMap<NodeLabel, TransportPublicKey>,
    ) -> Self {
        Self {
            user_ev_controller,
            event_sequence,
            label_to_key,
            current_index: 0,
        }
    }

    /// Trigger the next event in the sequence.
    ///
    /// Returns the EventId that was triggered, or None if all events have been triggered.
    pub fn trigger_next(&mut self) -> Option<EventId> {
        if self.current_index >= self.event_sequence.len() {
            return None;
        }

        let (event_id, label) = &self.event_sequence[self.current_index];
        let key = self
            .label_to_key
            .get(label)
            .expect("Label should exist in mapping");

        match self.user_ev_controller.send((*event_id, key.clone())) {
            Ok(()) => {
                self.current_index += 1;
                Some(*event_id)
            }
            Err(e) => {
                tracing::error!("Failed to send event {}: {:?}", event_id, e);
                None
            }
        }
    }

    /// Trigger all remaining events in sequence.
    ///
    /// Returns the number of events triggered.
    pub fn trigger_all(&mut self) -> usize {
        let mut count = 0;
        while self.trigger_next().is_some() {
            count += 1;
        }
        count
    }

    /// Returns true if all events have been triggered.
    pub fn is_complete(&self) -> bool {
        self.current_index >= self.event_sequence.len()
    }

    /// Returns the number of events remaining.
    pub fn remaining(&self) -> usize {
        self.event_sequence.len().saturating_sub(self.current_index)
    }
}

impl futures::stream::Stream for ControlledEventChain {
    type Item = EventId;

    fn poll_next(
        mut self: std::pin::Pin<&mut Self>,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Self::Item>> {
        match self.trigger_next() {
            Some(event_id) => std::task::Poll::Ready(Some(event_id)),
            None => std::task::Poll::Ready(None),
        }
    }
}

#[cfg(feature = "trace-ot")]
type DefaultRegistry = CombinedRegister<2>;

#[cfg(not(feature = "trace-ot"))]
type DefaultRegistry = TestEventListener;

/// How a startup-seeded contract is registered on its owning node.
///
/// Threads the distinction between the two seed primitives from
/// `run_controlled_simulation` down into [`in_memory::append_contracts`], which
/// runs at node startup. Both store the state locally with no network
/// propagation; they differ ONLY in the hosting/demand bookkeeping they install,
/// which is exactly what the serve-DURING gate keys on.
///
/// The variants are only ever constructed by `run_controlled_simulation`
/// (`#[cfg(any(test, feature = "testing"))]`), but `testing_impl` itself compiles
/// in every build (e.g. `trace-ot`), where `append_contracts` still matches on
/// this enum. So in a non-testing build the variants are legitimately never
/// constructed — allow dead_code there, keeping the lint live under testing.
#[cfg_attr(not(any(test, feature = "testing")), allow(dead_code))]
#[derive(Clone, Copy, Debug)]
pub(super) enum SeedMode {
    /// Genuine host WITH an active subscription (`SeedHostedContract`):
    /// `host_contract` + `ring.subscribe`, so `is_receiving_updates` is true.
    Subscribed,
    /// Demandless every-hop copy (`SeedDemandlessCopy`): `host_contract` +
    /// `register_local_hosting` (interest `hosting = true`), with NO subscription
    /// and NO prior local client access. This is the exact state a production
    /// every-hop GET/PUT store leaves on a RELAY hop (`is_hosting_contract` and
    /// `has_local_interest` true, `is_receiving_updates` false), which the
    /// serve-DURING originator gate (#4642 R3 piece C) must serve locally.
    Demandless,
}

pub(super) struct Builder<ER> {
    pub config: NodeConfig,
    contract_handler_name: String,
    event_register: ER,
    contracts: Vec<(ContractContainer, WrappedState, SeedMode)>,
    contract_subscribers: HashMap<ContractKey, Vec<PeerKeyLocation>>,
    /// Pre-formed ring connections injected at node startup, bypassing the
    /// organic CONNECT handshake. Each entry is a peer this node should already
    /// hold a direct ring connection to when its event loop starts. Drained by
    /// `run_node_with_{shared_storage,mock_wasm}` into
    /// `apply_preseeded_connections` (see that helper for the rationale).
    /// Populated by [`SimNetwork::preseed_direct_star`]; empty for every node by
    /// default, so this is a zero-cost no-op for all existing simulations.
    preseed_connections: Vec<PeerKeyLocation>,
    /// Seed for deterministic RNG in this node's transport layer
    pub rng_seed: u64,
    /// Network name for scoped fault injection
    pub network_name: String,
    /// Shared handle for capturing the live `ConnectionManager` after node start.
    pub shared_cm: Option<Arc<parking_lot::Mutex<Option<crate::ring::ConnectionManager>>>>,
    /// Shared handle for capturing the live `Arc<Ring>` after node start.
    /// Used by governance sim tests to observe per-node ban-list state
    /// (`ring.contract_ban_list.is_banned(..)`) without exposing the Ring
    /// on the public `RunningNode` surface. Mirrors `shared_cm`.
    pub shared_ring: Option<Arc<parking_lot::Mutex<Option<Arc<crate::ring::Ring>>>>>,
}

impl<ER: NetEventRegister> Builder<ER> {
    /// Builds an in-memory node. Does nothing upon construction.
    pub fn build(
        builder: NodeConfig,
        event_register: ER,
        contract_handler_name: String,
        rng_seed: u64,
        network_name: String,
    ) -> Builder<ER> {
        Builder {
            config: builder.clone(),
            contract_handler_name,
            event_register,
            contracts: Vec::new(),
            contract_subscribers: HashMap::new(),
            preseed_connections: Vec::new(),
            rng_seed,
            network_name,
            shared_cm: None,
            shared_ring: None,
        }
    }
}

/// Information about a running node, used for crash/restart operations.
#[derive(Debug)]
pub struct RunningNode {
    /// The label identifying this node
    pub label: NodeLabel,
    /// Socket address for fault injection
    pub addr: SocketAddr,
    /// Handle to abort the running task (AbortHandle can be cloned and used independently)
    pub abort_handle: tokio::task::AbortHandle,
}

/// Configuration saved for node restart.
///
/// When a node is started, its configuration is saved here so it can be
/// restarted with the same identity (keypair), location, and data directory.
#[derive(Clone)]
pub struct RestartableNodeConfig {
    /// The node's configuration (contains keypair, location, data dir, etc.)
    pub config: NodeConfig,
    /// The node label (reserved for future use in restart scenarios)
    #[allow(dead_code)]
    pub label: NodeLabel,
    /// Whether this is a gateway node
    pub is_gateway: bool,
    /// Gateway addresses to connect to (for non-gateway nodes)
    #[allow(dead_code)]
    pub gateway_configs: Vec<GatewayConfig>,
    /// Seed for deterministic RNG in this node's transport layer
    pub rng_seed: u64,
    /// Shared in-memory storage for contract state (persists across restarts)
    pub shared_storage: crate::wasm_runtime::MockStateStorage,
}

/// State for a node in the direct simulation runner, shared with the chaos driver.
#[cfg(any(test, feature = "testing"))]
struct DirectNodeState {
    label: NodeLabel,
    addr: SocketAddr,
    is_gateway: bool,
    permanently_dropped: bool,
}

/// Configuration for deterministic node churn (crash/restart) during simulation.
///
/// When enabled, a chaos driver task periodically crashes and restarts nodes
/// to test network resilience under churn conditions.
#[derive(Debug, Clone)]
pub struct ChurnConfig {
    /// Probability (0.0–1.0) that a non-gateway node is crashed each tick.
    pub crash_probability: f64,
    /// How often the chaos driver evaluates crashes.
    pub tick_interval: Duration,
    /// How long a crashed node stays down before restarting.
    pub recovery_delay: Duration,
    /// Maximum number of nodes crashed simultaneously.
    /// Defaults to 25% of non-gateway node count.
    pub max_simultaneous_crashes: Option<usize>,
    /// Fraction (0.0–1.0) of crashes that are permanent (no restart).
    pub permanent_crash_rate: f64,
    /// Time to wait before the chaos driver starts crashing nodes.
    pub warmup_delay: Duration,
}

impl Default for ChurnConfig {
    fn default() -> Self {
        Self {
            crash_probability: 0.1,
            tick_interval: Duration::from_secs(5),
            recovery_delay: Duration::from_secs(3),
            max_simultaneous_crashes: None,
            permanent_crash_rate: 0.05,
            warmup_delay: Duration::from_secs(5),
        }
    }
}

/// For each target ring location, find a distinct IPv6-loopback port whose
/// `Location::from_address((::1, port))` is closest (in ring distance) to that
/// target. Used by [`SimNetwork::new_with_node_locations`] to place nodes at
/// chosen locations without decoupling a peer's advertised location from its
/// address-derived one.
///
/// Ports are unique across the returned vector (a port already chosen for an
/// earlier target is skipped for later targets), so the resulting peers occupy
/// distinct addresses. The achieved location for each target is the closest
/// available port's location — approximate (ports are discrete) but fully
/// deterministic. Read the achieved locations back with
/// [`SimNetwork::get_peer_locations`].
///
/// The search range excludes the `50000..=59999` band used by
/// `derive_deterministic_port`, so explicit ports never collide with the
/// default-path derived ports.
pub fn loopback_ports_for_locations(targets: &[f64]) -> Vec<u16> {
    use std::collections::HashSet;
    const SEARCH_RANGE: std::ops::RangeInclusive<u16> = 1024..=49999;
    let mut used: HashSet<u16> = HashSet::new();
    let mut out = Vec::with_capacity(targets.len());
    for &target in targets {
        let mut best_port = *SEARCH_RANGE.start();
        let mut best_dist = f64::INFINITY;
        for port in SEARCH_RANGE {
            if used.contains(&port) {
                continue;
            }
            let addr: SocketAddr = (Ipv6Addr::LOCALHOST, port).into();
            let loc = Location::from_address(&addr).as_f64();
            let raw = (loc - target).abs();
            let dist = raw.min(1.0 - raw); // ring distance
            if dist < best_dist {
                best_dist = dist;
                best_port = port;
            }
        }
        used.insert(best_port);
        out.push(best_port);
    }
    out
}

/// A simulated in-memory network topology.
pub struct SimNetwork {
    name: String,
    clean_up_tmp_dirs: bool,
    labels: Vec<(NodeLabel, TransportPublicKey)>,
    pub(crate) event_listener: TestEventListener,
    user_ev_controller: Option<watch::Sender<(EventId, TransportPublicKey)>>,
    receiver_ch: watch::Receiver<(EventId, TransportPublicKey)>,
    number_of_gateways: usize,
    gateways: Vec<(Builder<DefaultRegistry>, GatewayConfig)>,
    number_of_nodes: usize,
    nodes: Vec<(Builder<DefaultRegistry>, NodeLabel)>,
    ring_max_htl: usize,
    rnd_if_htl_above: usize,
    max_connections: usize,
    min_connections: usize,
    start_backoff: Duration,
    /// Master seed for deterministic RNG - used to derive per-peer seeds
    seed: u64,
    /// VirtualTime for deterministic simulation - always enabled
    virtual_time: VirtualTime,
    /// Running nodes indexed by label for crash/restart operations
    running_nodes: HashMap<NodeLabel, RunningNode>,
    /// Map from label to socket address for quick lookup
    node_addresses: HashMap<NodeLabel, SocketAddr>,
    /// Saved configurations for node restart (preserved after crash)
    restartable_configs: HashMap<NodeLabel, RestartableNodeConfig>,
    /// All gateway configs (needed for restarting non-gateway nodes)
    all_gateway_configs: Vec<GatewayConfig>,
    /// Size threshold (bytes) above which streaming is used.
    /// Default: `None` → uses `usize::MAX` so tests don't stream unless explicitly opted in
    /// via `with_streaming_threshold()`. This preserves the pre-streaming-always-on behavior
    /// where simulation tests used inline messages for all payloads.
    pub streaming_threshold: Option<usize>,
    connection_managers: HashMap<NodeLabel, ConnectionManager>,
    /// When true, use `MockWasmRuntime` (production `ContractExecutor` code path)
    /// instead of `MockRuntime` (simplified hash-based merge).
    pub use_mock_wasm: bool,
    /// When true, the direct runner skips the long post-event convergence-polling
    /// loop and does only a brief fixed propagation settle instead.
    ///
    /// The direct runner normally polls for full state convergence for up to
    /// ~1800s of virtual time after the event phase (30 rounds × 60s, early-exit
    /// on convergence). That tail is advisory — the runner only logs a warning if
    /// convergence is never reached, it does not fail the simulation. For tests
    /// that analyze events produced *during* the event phase and never assert
    /// convergence (e.g. `test_interest_renewal`), running the full polling tail
    /// when the network happens not to converge adds up to 1800s of virtual time
    /// for no benefit, blowing the wall-clock budget and making the test time out
    /// in CI. The test crate sets this via `TestConfig::no_convergence_wait()` so
    /// such tests stay bounded regardless of whether the network converges.
    /// See #3792.
    pub skip_convergence_wait: bool,
    /// Optional churn (crash/restart) configuration for the chaos driver.
    churn_config: Option<ChurnConfig>,
    /// Optional pre-operation join-convergence barrier for
    /// [`run_controlled_simulation`](Self::run_controlled_simulation). When
    /// `Some((min_fraction, max_wait))`, the controlled-event client waits —
    /// before firing any scheduled operation — until at least `min_fraction`
    /// of the network's peers have completed their network join, or `max_wait`
    /// virtual time elapses, whichever comes first. Join is detected via the
    /// topology-snapshot registry: a peer only registers a snapshot once its
    /// own address/location is established (`peer_ready`), so the count of
    /// distinct snapshot peers is a direct join tally (see
    /// `ring::register_topology_snapshots_periodically`, which `continue`s
    /// while `get_own_addr()` is `None`).
    ///
    /// `None` (default) preserves the historical behavior of firing operations
    /// after a fixed 3s warmup, racing topology formation. This barrier exists
    /// because a cold-start GET-reliability measurement must run against a
    /// FORMED network: with only the fixed warmup, higher-index nodes' one-shot,
    /// no-retry GETs fire before those nodes finish joining and are rejected
    /// with `PeerNotJoined`, so the metric conflates join speed with GET
    /// reliability. See `test_get_reliability_diagnostic`. Opt in via
    /// [`wait_for_join_convergence_before_ops`](Self::wait_for_join_convergence_before_ops).
    wait_for_join_before_ops: Option<(f64, Duration)>,
    /// Optional override for the delay the controlled-event client waits after
    /// triggering each scheduled operation before triggering the next one.
    /// `None` (default) uses the historical fixed 3s settle. A test whose
    /// operations complete quickly against an already-formed network (e.g. a
    /// read-only GET sweep behind `wait_for_join_convergence_before_ops`) can
    /// shrink this to avoid spending ~3s of virtual time — and the wall-clock
    /// to simulate it — per operation for work that finishes in milliseconds.
    /// Only affects regular client-request operations; special in-client ops
    /// (clock advance, crash/recover) keep their own fixed settle. See
    /// [`with_controlled_op_interval`](Self::with_controlled_op_interval).
    controlled_op_interval: Option<Duration>,
    /// Optional governance-manager config override applied to every node
    /// this network builds. Lets governance sim tests compress the
    /// production minute-to-hour timescales and lower `min_samples` so the
    /// rate-limit → MAD → evict → ban chain fires within a paused-time sim.
    /// Pair with `use_mock_wasm = true` so the production cost-reporting
    /// path actually feeds the detector. See #4301.
    governance_config_override: Option<crate::contract::governance::GovernanceConfig>,
    /// Per-node placement-migration version-floor override (see
    /// [`SimNetwork::enable_placement_migration`]). Defaults to
    /// `Some(SIM_MIGRATION_DISABLED_FLOOR)` (an unreachable floor) so the
    /// `SubscribeHint` cascade is FAIL-CLOSED — OFF — in every sim regardless of
    /// build version, and only a test that explicitly calls
    /// `enable_placement_migration` gets the cascade. This keeps migration from
    /// perturbing unrelated simulations now that the crate version is past the
    /// real production floor (#4601). Production is untouched: `NodeConfig::new`
    /// sets this to `None`, which resolves to the real `SUBSCRIBE_HINT_MIN_VERSION`.
    subscribe_hint_floor_override: Option<(u8, u8, u16)>,
    /// Per-node summary-first PUT probe version-floor override (see
    /// [`SimNetwork::enable_summary_first_put`]). Mirrors
    /// `subscribe_hint_floor_override` exactly: defaults to
    /// `Some(SIM_MIGRATION_DISABLED_FLOOR)` (an unreachable floor) so the
    /// probe/dispatch cascade is FAIL-CLOSED — OFF — in every sim regardless
    /// of build version, and only a test that explicitly calls
    /// `enable_summary_first_put` gets summary-first PUT. Production is
    /// untouched: `NodeConfig::new` sets this to `None`, which resolves to
    /// the real `SUMMARY_FIRST_PUT_MIN_VERSION`.
    summary_first_put_floor_override: Option<(u8, u8, u16)>,
    /// Per-node hash-first summary version-floor override (#4965, see
    /// [`SimNetwork::enable_hash_first_summaries`]).
    ///
    /// **Defaults to `Some(SIM_MIGRATION_ENABLED_FLOOR)` — ON — which is the
    /// OPPOSITE of the two overrides above.** They gate behavioural cascades
    /// that pile load onto unrelated sims, so they fail closed. Hash-first
    /// only changes the ENCODING of the `Summaries` exchange that already
    /// runs in every sim, with identical convergence semantics, so ON by
    /// default costs unrelated sims nothing — and it is the only way the
    /// suite exercises the new wire path before the release that lifts the
    /// crate version over the production floor. Without it the first
    /// integration-level run of a Full-tier protocol change would be the
    /// release PR, where a red sim could not be attributed between the
    /// feature and the version bump.
    hash_first_summaries_floor_override: Option<(u8, u8, u16)>,
    /// Per-node version-carrying-ack version-floor override (#5161, see
    /// [`SimNetwork::enable_gateway_ack_version`]).
    ///
    /// **Defaults to `Some(SIM_MIGRATION_DISABLED_FLOOR)` — OFF**, like the two
    /// cascade gates and unlike `hash_first_summaries_floor_override`. The gate
    /// is encoding-only where it fires, but the version it teaches is the INPUT
    /// to every other `version_supports_*` gate, so enabling it network-wide
    /// makes node->gateway links newly eligible for those features. Measured,
    /// not assumed: defaulting it ON changed the outcome of unrelated sims.
    ///
    /// The cost of OFF, stated plainly: sims keep the pre-0.2.120 topology
    /// asymmetry (a regular node never learns its gateway's version) even after
    /// production stops having it. `enable_gateway_ack_version` is how a test
    /// that cares opts in.
    ack_version_floor_override: Option<(u8, u8, u16)>,
    /// Per-node override for the originator-target-list version floor (#5147).
    ///
    /// Defaults to `SIM_MIGRATION_DISABLED_FLOOR` — OFF — grouping with
    /// `subscribe_hint_floor_override` and `summary_first_put_floor_override`
    /// rather than with `hash_first_summaries_floor_override`. Hash-first is ON
    /// by default because it changes only the ENCODING of an exchange with
    /// identical semantics. This one changes fan-out BEHAVIOUR: it removes
    /// peers from a broadcast. Defaulting it ON would silently alter the
    /// delivery graph under every sim that asserts on convergence or delivery
    /// counts, and a failure could not be attributed between that sim's own
    /// subject and this suppression. Tests that want it call
    /// [`enable_broadcast_target_list`](Self::enable_broadcast_target_list).
    broadcast_target_list_floor_override: Option<(u8, u8, u16)>,
    /// Optional controllable hosting clock injected into every node's
    /// `HostingManager` (via `NodeConfig::hosting_time_source_override`). When
    /// set, hosting-cache TTL and subscription-lease eviction advance ONLY when
    /// the sim advances this clock — either directly (`hosting_clock()`) or, for
    /// mid-run control under the turmoil runner, via
    /// `SimOperation::AdvanceHostingClock`. Enables deterministic eviction sims
    /// (#4642 piece A). `None` → nodes use the default wall-clock time source.
    hosting_clock: Option<crate::util::time_source::SharedMockTimeSource>,
    /// Optional per-node hosting-cache byte budget (`max_hosting_storage`).
    /// When set, every node this network builds is given this budget instead of
    /// the capability-derived default, so a test can force cache pressure with a
    /// tiny budget and observe demand-driven eviction (#4642 piece A).
    hosting_budget_override: Option<u64>,
    /// Per-node capture slots for the live `Arc<Ring>`, populated by
    /// `run_controlled_simulation` so governance sim tests can read each
    /// node's `contract_ban_list` after the simulation completes. Keyed by
    /// label; the inner `Option` is filled once the node's run loop reaches
    /// the `shared_ring` write in `run_node_with_{shared_storage,mock_wasm}`.
    ///
    /// `allow(dead_code)`: populated unconditionally in
    /// `run_controlled_simulation`, but only *read* in the
    /// `cfg(any(test, feature = "testing"))` `node_rings` extraction, so a
    /// production build (where `testing_impl` is still compiled) sees it
    /// as write-only.
    #[allow(dead_code)]
    shared_rings: HashMap<NodeLabel, Arc<parking_lot::Mutex<Option<Arc<crate::ring::Ring>>>>>,
    /// Optional explicit loopback ports for regular nodes, indexed by
    /// regular-node order (i.e. `node_no - number_of_gateways`). When present,
    /// `config_nodes` uses these instead of `derive_deterministic_port`, which
    /// lets a test place nodes at chosen ring locations: a peer's location is
    /// `Location::from_address(&addr)`, a pure function of the loopback port, so
    /// choosing the port chooses the location *consistently* (the address-derived
    /// location and the advertised `own_location` stay equal, exactly as the
    /// default port path keeps them). Build via `new_with_node_locations`.
    node_port_override: Option<Vec<u16>>,
}

impl SimNetwork {
    /// Default seed for deterministic simulation
    pub const DEFAULT_SEED: u64 = 0xDEADBEEF_CAFEBABE;

    /// Placement-migration version floor that pins the `SubscribeHint` cascade
    /// OFF for a simulation: an unreachable value above any real or simulated
    /// crate version, so `version_supports_subscribe_hint` always returns false.
    /// This is the per-node default for every `SimNetwork` (see `new_inner`),
    /// making migration genuinely OPT-IN in sims; `enable_placement_migration`
    /// lowers it to [`Self::SIM_MIGRATION_ENABLED_FLOOR`]. See #4601.
    pub(crate) const SIM_MIGRATION_DISABLED_FLOOR: (u8, u8, u16) = (255, 255, 65535);

    /// Placement-migration version floor that forces the `SubscribeHint` cascade
    /// ON for a simulation regardless of build version (every peer's version is
    /// `>= (0,0,0)`). Set by [`Self::enable_placement_migration`].
    pub(crate) const SIM_MIGRATION_ENABLED_FLOOR: (u8, u8, u16) = (0, 0, 0);

    #[allow(clippy::too_many_arguments)]
    pub async fn new(
        name: &str,
        gateways: usize,
        nodes: usize,
        ring_max_htl: usize,
        rnd_if_htl_above: usize,
        max_connections: usize,
        min_connections: usize,
        seed: u64,
    ) -> Self {
        Self::new_inner(
            name,
            gateways,
            nodes,
            ring_max_htl,
            rnd_if_htl_above,
            max_connections,
            min_connections,
            seed,
            None,
        )
        .await
    }

    /// Like [`new`](Self::new), but places each regular node at (approximately)
    /// a chosen ring location. `node_target_locations` is matched to regular
    /// nodes in order and must have exactly `nodes` entries.
    ///
    /// Locations are realised by choosing, for each target, a distinct loopback
    /// port whose `Location::from_address` is closest to the target (see
    /// [`loopback_ports_for_locations`]). The achieved locations are therefore
    /// approximate (discrete ports); read them back with [`get_peer_locations`]
    /// (gateways first, then regular nodes in order) and build the scenario
    /// around the actual values. Gateways keep their derived locations.
    #[allow(clippy::too_many_arguments)]
    pub async fn new_with_node_locations(
        name: &str,
        gateways: usize,
        nodes: usize,
        ring_max_htl: usize,
        rnd_if_htl_above: usize,
        max_connections: usize,
        min_connections: usize,
        seed: u64,
        node_target_locations: &[f64],
    ) -> Self {
        assert_eq!(
            node_target_locations.len(),
            nodes,
            "node_target_locations must have exactly `nodes` entries"
        );
        let ports = loopback_ports_for_locations(node_target_locations);
        Self::new_inner(
            name,
            gateways,
            nodes,
            ring_max_htl,
            rnd_if_htl_above,
            max_connections,
            min_connections,
            seed,
            Some(ports),
        )
        .await
    }

    #[allow(clippy::too_many_arguments)]
    async fn new_inner(
        name: &str,
        gateways: usize,
        nodes: usize,
        ring_max_htl: usize,
        rnd_if_htl_above: usize,
        max_connections: usize,
        min_connections: usize,
        seed: u64,
        node_port_override: Option<Vec<u16>>,
    ) -> Self {
        assert!(nodes > 0);

        // Seed GlobalRng for deterministic location generation
        // This ensures Location::random() calls in config_gateways/config_nodes are deterministic
        GlobalRng::set_seed(seed);

        let (user_ev_controller, mut receiver_ch) =
            watch::channel((0, TransportKeypair::new().public().clone()));
        receiver_ch.borrow_and_update();

        // VirtualTime is always enabled for deterministic simulation
        let virtual_time = VirtualTime::new();

        // Register the VirtualTime for this network so SimulationSocket can use it
        register_network_time_source(name, virtual_time.clone());

        let mut net = Self {
            name: name.into(),
            clean_up_tmp_dirs: true,
            event_listener: TestEventListener::new().await,
            labels: Vec::with_capacity(nodes + gateways),
            user_ev_controller: Some(user_ev_controller),
            receiver_ch,
            number_of_gateways: gateways,
            gateways: Vec::with_capacity(gateways),
            number_of_nodes: nodes,
            nodes: Vec::with_capacity(nodes),
            ring_max_htl,
            rnd_if_htl_above,
            max_connections,
            min_connections,
            start_backoff: Duration::from_millis(1),
            seed,
            virtual_time,
            running_nodes: HashMap::new(),
            node_addresses: HashMap::new(),
            restartable_configs: HashMap::new(),
            all_gateway_configs: Vec::new(),
            streaming_threshold: None,
            connection_managers: HashMap::new(),
            use_mock_wasm: false,
            skip_convergence_wait: false,
            churn_config: None,
            wait_for_join_before_ops: None,
            controlled_op_interval: None,
            governance_config_override: None,
            // Fail-closed by default: pin the placement-migration (`SubscribeHint`)
            // cascade OFF for every simulation unless a test explicitly opts in
            // via `enable_placement_migration`. The cascade gates on the build
            // version being `>= SUBSCRIBE_HINT_MIN_VERSION` (0.2.80); since this
            // crate is now past that floor, a `None` override (real floor) would
            // run migration in EVERY sim and pile directed-subscribe + renewal
            // load onto unrelated simulations — the #4601 regression that turned
            // the 500-node nightly red. Defaulting to an unreachable floor keeps
            // migration genuinely opt-in, as the docs have always claimed.
            subscribe_hint_floor_override: Some(Self::SIM_MIGRATION_DISABLED_FLOOR),
            // Fail-closed by default, same rationale as
            // `subscribe_hint_floor_override` above: summary-first PUT is
            // genuinely opt-in per sim via `enable_summary_first_put`.
            summary_first_put_floor_override: Some(Self::SIM_MIGRATION_DISABLED_FLOOR),
            // Fail-OPEN by default, deliberately unlike the two above (#4965).
            // See the field's rustdoc: hash-first is an encoding change, not a
            // load-bearing cascade, and defaulting it ON is what gives a
            // version-gated wire change simulation coverage BEFORE the release
            // that lifts the crate version past its floor.
            hash_first_summaries_floor_override: Some(Self::SIM_MIGRATION_ENABLED_FLOOR),
            // Fail-CLOSED by default, like the two cascade gates and unlike
            // hash-first. This gate is encoding-only where it fires, but what
            // it teaches — the remote's version — is the INPUT to every other
            // `version_supports_*` gate, so switching it on network-wide is a
            // cascade in effect. Opt in per sim via `enable_gateway_ack_version`.
            ack_version_floor_override: Some(Self::SIM_MIGRATION_DISABLED_FLOOR),
            // Fail-CLOSED by default, like the two behavioural gates above and
            // unlike hash-first (#5147). See the field's rustdoc.
            broadcast_target_list_floor_override: Some(Self::SIM_MIGRATION_DISABLED_FLOOR),
            hosting_clock: None,
            hosting_budget_override: None,
            shared_rings: HashMap::new(),
            node_port_override,
        };
        net.config_gateways(
            gateways
                .try_into()
                .expect("should have at least one gateway"),
        )
        .await;
        net.config_nodes(nodes).await;

        // Auto-configure fault injection with VirtualTime enabled
        // Users can call with_fault_injection() to customize
        net.init_default_fault_injection();

        net
    }

    /// Initializes the default fault injection with VirtualTime but no faults.
    /// Users can call with_fault_injection() to add message loss, latency, etc.
    fn init_default_fault_injection(&mut self) {
        use crate::node::network_bridge::{FaultInjectorState, set_fault_injector};
        let fault_seed = self.seed.wrapping_add(0xFA01_7777);
        let state = FaultInjectorState::new(FaultConfig::default(), fault_seed)
            .with_virtual_time(self.virtual_time.clone());
        set_fault_injector(
            &self.name,
            Some(std::sync::Arc::new(std::sync::Mutex::new(state))),
        );
    }

    /// Sets the streaming threshold for operations.
    ///
    /// Payloads larger than `threshold` bytes will use streaming instead of
    /// inline messages. This retroactively updates all already-built gateway
    /// and node configs.
    pub fn with_streaming_threshold(&mut self, threshold: usize) {
        self.streaming_threshold = Some(threshold);

        // Retroactively update already-built node configs (since config_gateways/config_nodes
        // ran inside new() before this method could be called).
        for (builder, _) in &mut self.gateways {
            let old_config = &*builder.config.config;
            let mut new_network_api = old_config.network_api.clone();
            new_network_api.streaming_threshold = threshold;
            let mut new_config = old_config.clone();
            new_config.network_api = new_network_api;
            builder.config.config = Arc::new(new_config);
        }
        for (builder, _) in &mut self.nodes {
            let old_config = &*builder.config.config;
            let mut new_network_api = old_config.network_api.clone();
            new_network_api.streaming_threshold = threshold;
            let mut new_config = old_config.clone();
            new_config.network_api = new_network_api;
            builder.config.config = Arc::new(new_config);
        }
    }

    /// Sets the readiness gating threshold for all nodes.
    ///
    /// Nodes must have at least `min` ring connections before they advertise
    /// readiness for non-CONNECT operations. Retroactively updates all
    /// already-built gateway and node configs.
    pub fn with_readiness_gating(&mut self, min: usize) {
        for (builder, _) in &mut self.gateways {
            builder.config.relay_ready_connections = Some(min);
        }
        for (builder, _) in &mut self.nodes {
            builder.config.relay_ready_connections = Some(min);
        }
    }

    /// Derives a deterministic per-peer seed from the master seed and peer index.
    fn derive_peer_seed(&self, peer_index: usize) -> u64 {
        // Use a simple but effective mixing function
        let mut seed = self.seed;
        seed = seed.wrapping_add(peer_index as u64);
        seed ^= seed >> 33;
        seed = seed.wrapping_mul(0xff51afd7ed558ccd);
        seed ^= seed >> 33;
        seed = seed.wrapping_mul(0xc4ceb9fe1a85ec53);
        seed ^= seed >> 33;
        seed
    }
}

impl SimNetwork {
    pub fn with_start_backoff(&mut self, value: Duration) {
        self.start_backoff = value;
    }

    /// Enables the chaos driver which periodically crashes and restarts nodes.
    pub fn with_churn(&mut self, config: ChurnConfig) -> &mut Self {
        self.churn_config = Some(config);
        self
    }

    /// Inject a controllable hosting clock into every node this network builds,
    /// so hosting-cache TTL and subscription-lease eviction advance only when
    /// the sim advances the clock. Returns a handle the test can advance
    /// directly (e.g. before/after the run); for mid-run advancement under the
    /// turmoil-based `run_controlled_simulation`, schedule
    /// [`SimOperation::AdvanceHostingClock`] instead (the runner advances THIS
    /// same clock in-order with the other events).
    ///
    /// Enables deterministic eviction/TTL simulations (#4642 piece A) that were
    /// previously impossible because `HostingManager` hardcoded the wall clock.
    /// Pair with [`with_hosting_budget`](Self::with_hosting_budget) to force
    /// cache pressure. Idempotent: repeated calls reuse the existing clock.
    pub fn enable_hosting_time_control(
        &mut self,
    ) -> crate::util::time_source::SharedMockTimeSource {
        use crate::util::time_source::{DynTimeSource, SharedMockTimeSource};
        let clock = self
            .hosting_clock
            .get_or_insert_with(SharedMockTimeSource::new)
            .clone();
        // `config_gateways` / `config_nodes` already ran during
        // `SimNetwork::new()` with the override still `None`, so patch the
        // already-built node/gateway configs here (mirrors
        // `with_governance_config`). Without this the builders keep the
        // production wall clock and the injection is silently a no-op.
        let dyn_clock: DynTimeSource = std::sync::Arc::new(clock.clone());
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.hosting_time_source_override = Some(dyn_clock.clone());
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.hosting_time_source_override = Some(dyn_clock.clone());
        }
        clock
    }

    /// The controllable hosting clock, if [`enable_hosting_time_control`] was
    /// called. Advancing it moves every node's hosting TTL/eviction clock
    /// forward deterministically.
    ///
    /// [`enable_hosting_time_control`]: Self::enable_hosting_time_control
    pub fn hosting_clock(&self) -> Option<crate::util::time_source::SharedMockTimeSource> {
        self.hosting_clock.clone()
    }

    /// Set the per-node hosting-cache byte budget (`max_hosting_storage`) for
    /// every node this network builds. A tiny budget forces cache pressure so a
    /// test can observe demand-driven eviction of low-demand contracts
    /// (#4642 piece A). Pair with [`enable_hosting_time_control`] so aged
    /// entries actually cross the TTL gate.
    ///
    /// [`enable_hosting_time_control`]: Self::enable_hosting_time_control
    pub fn with_hosting_budget(&mut self, budget_bytes: u64) -> &mut Self {
        self.hosting_budget_override = Some(budget_bytes);
        // Patch already-built configs (see `with_governance_config`).
        for (builder, _) in self.gateways.iter_mut() {
            std::sync::Arc::make_mut(&mut builder.config.config).max_hosting_storage = budget_bytes;
        }
        for (builder, _) in self.nodes.iter_mut() {
            std::sync::Arc::make_mut(&mut builder.config.config).max_hosting_storage = budget_bytes;
        }
        self
    }

    /// Inject a governance-manager config override into every node this
    /// network builds. Compresses the production minute-to-hour governance
    /// timescales and lowers `min_samples` so the rate-limit → MAD → evict
    /// → ban chain can be exercised within a paused-time sim. Pair with
    /// `use_mock_wasm = true` so the production cost-reporting path feeds
    /// the detector. Test-only; see #4301.
    ///
    /// `allow(dead_code)`: consumed only by the `cfg(test)` governance
    /// e2e module, so the lib-only build sees it as unused.
    #[allow(dead_code)]
    pub(crate) fn with_governance_config(
        &mut self,
        config: crate::contract::governance::GovernanceConfig,
    ) -> &mut Self {
        self.governance_config_override = Some(config.clone());
        // `config_gateways` / `config_nodes` already ran during
        // `SimNetwork::new()`, when `governance_config_override` was still
        // `None`, so the builders captured `None`. Patch the already-built
        // node/gateway configs here so the override actually reaches each
        // node's `Ring::new` → `GovernanceManager`. Without this, the
        // builders keep the production default (e.g. `min_samples = 30`)
        // and the override is silently a no-op.
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.governance_config_override = Some(config.clone());
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.governance_config_override = Some(config.clone());
        }
        self
    }

    /// Enable the placement-migration (`SubscribeHint`) cascade for this
    /// simulation by lowering the version floor to
    /// [`Self::SIM_MIGRATION_ENABLED_FLOOR`] (`(0,0,0)`) on every node.
    ///
    /// Migration is OFF by default in every `SimNetwork` (the per-node floor
    /// defaults to the unreachable [`Self::SIM_MIGRATION_DISABLED_FLOOR`], so the
    /// `SubscribeHint` gate never fires — see `new_inner` and #4601). This makes
    /// the cascade genuinely opt-in: only a test that specifically exercises
    /// migration calls this to force it ON regardless of build version; every
    /// other sim is left untouched and cannot be perturbed by migration load.
    /// Like `with_governance_config`, this patches the already-built node/gateway
    /// configs (config_* ran during construction with the disabled default).
    /// Make [`run_controlled_simulation`](Self::run_controlled_simulation) wait
    /// for the ring to form before firing any scheduled operation: the
    /// controlled-event client blocks until at least `min_fraction` of this
    /// network's peers (gateways + nodes) have joined — registered a topology
    /// snapshot, which a peer only does once its own address is established
    /// (`peer_ready`) — or `max_wait` virtual time elapses, whichever comes
    /// first.
    ///
    /// Without this, operations fire after a fixed 3s warmup and race topology
    /// formation; a node whose one-shot, no-retry GET fires before it finishes
    /// joining is rejected with `PeerNotJoined` and never dispatches, so a
    /// cold-start reliability metric ends up measuring join speed rather than
    /// GET reliability. Opt-in; the default is the historical fixed-warmup
    /// behavior. See the `wait_for_join_before_ops` field.
    #[allow(dead_code)]
    pub fn wait_for_join_convergence_before_ops(
        &mut self,
        min_fraction: f64,
        max_wait: Duration,
    ) -> &mut Self {
        // Clamp to [0, 1]: a fraction > 1.0 would make the target unreachable
        // and always burn the full `max_wait` before proceeding.
        self.wait_for_join_before_ops = Some((min_fraction.clamp(0.0, 1.0), max_wait));
        self
    }

    /// Override the delay the controlled-event client waits after triggering
    /// each scheduled *regular* operation (GET/PUT/etc.) before triggering the
    /// next. Default is a fixed 3s settle. Lower it for a test whose operations
    /// finish quickly against a formed network to avoid burning ~3s of virtual
    /// time — and the wall-clock to simulate it — per operation. Does not affect
    /// special in-client ops (clock advance, crash/recover). See the
    /// `controlled_op_interval` field.
    #[allow(dead_code)]
    pub fn with_controlled_op_interval(&mut self, interval: Duration) -> &mut Self {
        self.controlled_op_interval = Some(interval);
        self
    }

    #[allow(dead_code)]
    pub fn enable_placement_migration(&mut self) -> &mut Self {
        let floor = Some(Self::SIM_MIGRATION_ENABLED_FLOOR);
        self.subscribe_hint_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.subscribe_hint_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.subscribe_hint_floor_override = floor;
        }
        self
    }

    /// Opt this simulation into advertising startup-hosted contracts
    /// (`SeedHostedContract`) through the neighbor-hosting mesh, so the
    /// terminal advertisement consult (hosting redesign piece C, invariant 5)
    /// can find a seeded off-path host.
    ///
    /// OFF by default: a seeded host does NOT advertise (the harness's
    /// historical behavior), so a key-routed GET to a non-hosting region still
    /// dead-ends — which the migration dead-end controls rely on. A test that
    /// exercises the consult opts in here. Patches the already-built
    /// node/gateway configs, mirroring
    /// [`enable_placement_migration`](Self::enable_placement_migration).
    #[allow(dead_code)]
    pub fn enable_seeded_host_advertisements(&mut self) -> &mut Self {
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.advertise_seeded_hosts = true;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.advertise_seeded_hosts = true;
        }
        self
    }

    /// Force the placement-migration (`SubscribeHint`) cascade OFF for this
    /// simulation by pinning the per-node version floor to the unreachable
    /// [`Self::SIM_MIGRATION_DISABLED_FLOOR`].
    ///
    /// The mirror of [`enable_placement_migration`](Self::enable_placement_migration).
    /// Since #4601 the cascade is already OFF by default in every sim (the
    /// per-node floor defaults to this same unreachable value), so this is now
    /// belt-and-suspenders: it makes a test's "migration must stay off" premise
    /// explicit at the call site. Calling it is harmless and recommended for
    /// tests whose correctness depends on migration NOT running. Like
    /// `enable_placement_migration`, this patches the already-built node/gateway
    /// configs.
    #[allow(dead_code)]
    pub fn disable_placement_migration(&mut self) -> &mut Self {
        // Unreachable floor: no real or simulated peer version reaches it, so
        // `version_supports_subscribe_hint` always returns false → cascade off.
        let floor = Some(Self::SIM_MIGRATION_DISABLED_FLOOR);
        self.subscribe_hint_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.subscribe_hint_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.subscribe_hint_floor_override = floor;
        }
        self
    }

    /// Enable summary-first PUT (#4642, step 3-bis) for this simulation by
    /// lowering the probe version floor to [`Self::SIM_MIGRATION_ENABLED_FLOOR`]
    /// (`(0,0,0)`) on every node.
    ///
    /// Summary-first PUT is OFF by default in every `SimNetwork` (the
    /// per-node floor defaults to the unreachable
    /// [`Self::SIM_MIGRATION_DISABLED_FLOOR`] — see `new_inner`), so this
    /// makes the probe/dispatch cascade genuinely opt-in: only a test that
    /// specifically exercises it calls this to force it ON regardless of
    /// build version; every other sim is left untouched. Mirrors
    /// [`enable_placement_migration`](Self::enable_placement_migration) —
    /// same sentinel floors, same patch-already-built-configs shape, distinct
    /// override field (`summary_first_put_floor_override`).
    #[allow(dead_code)]
    pub fn enable_summary_first_put(&mut self) -> &mut Self {
        let floor = Some(Self::SIM_MIGRATION_ENABLED_FLOOR);
        self.summary_first_put_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.summary_first_put_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.summary_first_put_floor_override = floor;
        }
        self
    }

    /// State explicitly that this simulation exercises the hash-first summary
    /// exchange (#4965) by pinning the per-node floor to the always-passing
    /// [`Self::SIM_MIGRATION_ENABLED_FLOOR`].
    ///
    /// This is ALREADY the default for every `SimNetwork` (see `new_inner`),
    /// so calling it changes nothing — it exists so a test whose premise
    /// depends on hash-first being on says so at the call site instead of
    /// relying on a default that a future edit could flip out from under it.
    /// Mirrors [`disable_summary_first_put`](Self::disable_summary_first_put)'s
    /// belt-and-suspenders rationale, in the opposite direction.
    #[allow(dead_code)]
    pub fn enable_hash_first_summaries(&mut self) -> &mut Self {
        let floor = Some(Self::SIM_MIGRATION_ENABLED_FLOOR);
        self.hash_first_summaries_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.hash_first_summaries_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.hash_first_summaries_floor_override = floor;
        }
        self
    }

    /// Opt this simulation into the version-carrying connection ack (#5161) by
    /// pinning the per-node floor to the always-passing
    /// [`Self::SIM_MIGRATION_ENABLED_FLOOR`], so a joiner learns the version of
    /// the gateway (or ack-racing peer) it connects to.
    ///
    /// Genuinely opt-in, unlike
    /// [`enable_hash_first_summaries`](Self::enable_hash_first_summaries) which
    /// only restates a default. The reasoning is the fail-closed criterion this
    /// codebase already applies: hash-first is opt-OUT because it changes only
    /// the ENCODING of an exchange every sim already runs. This gate looks like
    /// that from the inside — it changes only how one ack is encoded — but its
    /// EFFECT is a cascade, because the version it teaches is the input to
    /// every other `version_supports_*` gate. Turning it on network-wide makes
    /// node->gateway links newly eligible for summary-first PUT probes and
    /// hash-first digests in whatever sims have those enabled, which is exactly
    /// the "piles load onto unrelated simulations" shape that
    /// `subscribe_hint_floor_override` and `summary_first_put_floor_override`
    /// default OFF to avoid. Measured, not assumed: defaulting it ON changed
    /// the outcome of several unrelated sims.
    ///
    /// Note this leaves the sim suite pinned OFF even after the production
    /// floor is reached, so sims and production diverge on this from 0.2.120
    /// onwards. That is the deliberate trade — the alternative is every sim
    /// silently changing behaviour at a release — and the coverage it costs is
    /// bought back by `test_joiner_records_gateway_version_through_sim_handshake`
    /// plus the transport-level tests in `connection_handler`.
    #[allow(dead_code)]
    pub fn enable_gateway_ack_version(&mut self) -> &mut Self {
        let floor = Some(Self::SIM_MIGRATION_ENABLED_FLOOR);
        self.ack_version_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.ack_version_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.ack_version_floor_override = floor;
        }
        self
    }

    /// Turn the originator target list (#5147) ON for this simulation by
    /// pinning the per-node version floor to the always-passing
    /// [`Self::SIM_MIGRATION_ENABLED_FLOOR`].
    ///
    /// Unlike [`enable_hash_first_summaries`](Self::enable_hash_first_summaries),
    /// this is NOT the default and the call is load-bearing: without it every
    /// peer falls back to the legacy `BroadcastTo` and the sim measures
    /// today's unsuppressed fan-out. That is exactly what the control arm of a
    /// suppression measurement wants, so the two arms differ by this one call.
    #[allow(dead_code)]
    pub fn enable_broadcast_target_list(&mut self) -> &mut Self {
        let floor = Some(Self::SIM_MIGRATION_ENABLED_FLOOR);
        self.broadcast_target_list_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.broadcast_target_list_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.broadcast_target_list_floor_override = floor;
        }
        self
    }

    /// Force the originator target list OFF for this simulation.
    ///
    /// Already the default (see `new_inner`); exists so the CONTROL arm of a
    /// suppression measurement states its premise at the call site rather than
    /// depending on a default a future edit could flip.
    #[allow(dead_code)]
    pub fn disable_broadcast_target_list(&mut self) -> &mut Self {
        let floor = Some(Self::SIM_MIGRATION_DISABLED_FLOOR);
        self.broadcast_target_list_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.broadcast_target_list_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.broadcast_target_list_floor_override = floor;
        }
        self
    }

    /// Force the hash-first summary exchange OFF for this simulation by
    /// pinning the per-node floor to the unreachable
    /// [`Self::SIM_MIGRATION_DISABLED_FLOOR`], so every peer falls back to the
    /// full-bytes `InterestMessage::Summaries`.
    ///
    /// This is the PRE-0.2.116 fleet: it reproduces what an un-upgraded peer
    /// does.
    ///
    /// Note the override is per-node (`builder.config`), so a genuinely MIXED
    /// simulation — one peer at the floor, one below — is constructible by
    /// setting the two builders differently rather than using this helper,
    /// which sets every node uniformly. No such test exists yet; the mixed
    /// case that IS covered today is incidental rather than constructed: unless
    /// a sim calls [`enable_gateway_ack_version`](Self::enable_gateway_ack_version),
    /// a regular node never learns its gateway's version, so every gateway link
    /// already runs digests one way and full bytes the other.
    #[allow(dead_code)]
    pub fn disable_hash_first_summaries(&mut self) -> &mut Self {
        let floor = Some(Self::SIM_MIGRATION_DISABLED_FLOOR);
        self.hash_first_summaries_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.hash_first_summaries_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.hash_first_summaries_floor_override = floor;
        }
        self
    }

    /// Force summary-first PUT OFF for this simulation by pinning the
    /// per-node probe version floor to the unreachable
    /// [`Self::SIM_MIGRATION_DISABLED_FLOOR`].
    ///
    /// The mirror of [`enable_summary_first_put`](Self::enable_summary_first_put),
    /// following [`disable_placement_migration`](Self::disable_placement_migration)'s
    /// belt-and-suspenders rationale: the cascade is already OFF by default in
    /// every sim, so this just makes a test's "summary-first PUT must stay
    /// off" premise explicit at the call site.
    #[allow(dead_code)]
    pub fn disable_summary_first_put(&mut self) -> &mut Self {
        let floor = Some(Self::SIM_MIGRATION_DISABLED_FLOOR);
        self.summary_first_put_floor_override = floor;
        for (builder, _) in self.gateways.iter_mut() {
            builder.config.summary_first_put_floor_override = floor;
        }
        for (builder, _) in self.nodes.iter_mut() {
            builder.config.summary_first_put_floor_override = floor;
        }
        self
    }

    /// Resolve a label to its full [`PeerKeyLocation`] (pub_key + address) from
    /// the already-built node/gateway configs. Returns `None` if the label is
    /// not found among this network's nodes. Must be called before the builders
    /// are drained into the run loop (i.e. before `run_controlled_simulation`).
    fn peer_key_location_for(&self, label: &NodeLabel) -> Option<PeerKeyLocation> {
        let builder = self
            .gateways
            .iter()
            .map(|(b, cfg)| (b, &cfg.label))
            .chain(self.nodes.iter().map(|(b, l)| (b, l)))
            .find(|(_, l)| *l == label)
            .map(|(b, _)| b)?;
        let addr = std::net::SocketAddr::new(
            builder.config.network_listener_ip,
            builder.config.network_listener_port,
        );
        Some(PeerKeyLocation::new(
            builder.config.key_pair.public().clone(),
            addr,
        ))
    }

    /// Preseed a **direct star** topology: inject a pre-formed ring connection
    /// between `host` and every label in `subscribers`, in BOTH directions, so
    /// that when the simulation starts the host already holds direct connections
    /// to all subscribers (and each subscriber to the host) — bypassing the
    /// organic ring-routed CONNECT handshake entirely.
    ///
    /// ## Why this exists (#4233)
    ///
    /// A faithful 40-80-subscriber direct star (the production-incident
    /// fan-out topology) cannot be bootstrapped organically in this Turmoil
    /// harness: ring-routed CONNECT cannot form that many direct connections to
    /// a single 1-gateway hub within the virtual-time budget — subscribers hit
    /// "at terminus, no uphill peers available — rejecting" and the simulation
    /// exhausts its whole schedule in topology formation before any UPDATE
    /// fires (N=8/16 form; N>=24 never finish). This primitive is the
    /// connection-side analogue of the `SeedHostedContract` contract preseed: it
    /// writes the ring connection state directly so the star exists at t=0 and
    /// the test can reach the sustained-fan-out broadcast phase it actually
    /// targets.
    ///
    /// ## What it injects — and what it leaves to the organic path
    ///
    /// Only the ring `Connection`/`Location` state is injected (via
    /// [`Ring::add_connection`](crate::ring::Ring::add_connection) at node
    /// startup). The underlying transport connection is materialized lazily on
    /// first send, and each subscriber's interest/subscription is registered
    /// organically when it runs its real `Subscribe` op (which now routes
    /// directly to the host over the injected connection). So the genuine
    /// subscribe + interest + broadcast machinery is exercised unchanged; only
    /// the CONNECT handshake is replaced. See
    /// `apply_preseeded_connections` for the full mechanism.
    ///
    /// ## Caller responsibilities
    ///
    /// - The host's `max_connections` must be `>= subscribers.len()` (plus slack
    ///   for its own gateway-mesh connections), or the [`ConnectionManager`] cap
    ///   will reject the overflow connections at injection time (logged + skipped,
    ///   surfacing as a connectivity shortfall in the test's assertions).
    /// - Call this AFTER `SimNetwork::new*` and BEFORE
    ///   `run_controlled_simulation` (it mutates the builders in place; once they
    ///   are drained into the run loop the preseed list is consumed at startup).
    ///
    /// Panics if `host` or any subscriber label is not found in this network.
    pub fn preseed_direct_star(&mut self, host: &NodeLabel, subscribers: &[NodeLabel]) {
        let host_pkl = self
            .peer_key_location_for(host)
            .unwrap_or_else(|| panic!("preseed_direct_star: host {host:?} not found in network"));

        // Resolve every subscriber up front so we fail fast on a bad label and
        // do not partially mutate the builders.
        let sub_pkls: Vec<(NodeLabel, PeerKeyLocation)> = subscribers
            .iter()
            .map(|label| {
                let pkl = self.peer_key_location_for(label).unwrap_or_else(|| {
                    panic!("preseed_direct_star: subscriber {label:?} not found in network")
                });
                (label.clone(), pkl)
            })
            .collect();

        for (sub_label, sub_pkl) in sub_pkls {
            // host -> subscriber
            self.push_preseed_connection(host, sub_pkl);
            // subscriber -> host
            self.push_preseed_connection(&sub_label, host_pkl.clone());
        }
    }

    /// Append a single pre-formed connection to `label`'s builder. Internal
    /// helper for [`preseed_direct_star`](Self::preseed_direct_star).
    fn push_preseed_connection(&mut self, label: &NodeLabel, peer: PeerKeyLocation) {
        let builder = self
            .gateways
            .iter_mut()
            .find(|(_, cfg)| cfg.label == *label)
            .map(|(b, _)| b)
            .or_else(|| {
                self.nodes
                    .iter_mut()
                    .find(|(_, l)| *l == *label)
                    .map(|(b, _)| b)
            })
            .unwrap_or_else(|| {
                panic!("push_preseed_connection: label {label:?} not found in network")
            });
        builder.preseed_connections.push(peer);
    }

    /// Returns the VirtualTime instance for this simulation.
    ///
    /// VirtualTime is always enabled. Use this to advance time and control
    /// message delivery timing.
    ///
    /// # Example
    /// ```ignore
    /// let mut sim = SimNetwork::new(...).await;
    ///
    /// // Advance virtual time by 100ms
    /// sim.virtual_time().advance(Duration::from_millis(100));
    ///
    /// // Deliver pending messages
    /// let delivered = sim.advance_virtual_time();
    /// ```
    pub fn virtual_time(&self) -> &VirtualTime {
        &self.virtual_time
    }

    /// Configures fault injection for the network simulation.
    ///
    /// This enables deterministic fault injection using the simulation framework's
    /// `FaultConfig`. Faults include:
    /// - Message drops (via `message_loss_rate`) - deterministic with seeded RNG
    /// - Network partitions
    /// - Node crashes
    /// - Latency injection (via `latency_range`)
    ///
    /// VirtualTime is automatically used for deterministic latency injection.
    ///
    /// # Example
    /// ```ignore
    /// use freenet::simulation::FaultConfig;
    /// use std::time::Duration;
    ///
    /// let mut sim = SimNetwork::new(...).await;
    /// sim.with_fault_injection(FaultConfig::builder()
    ///     .message_loss_rate(0.1)
    ///     .latency_range(Duration::from_millis(10)..Duration::from_millis(50))
    ///     .build());
    ///
    /// // Advance time and deliver pending messages
    /// sim.virtual_time().advance(Duration::from_millis(100));
    /// sim.advance_virtual_time();
    /// ```
    pub fn with_fault_injection(&mut self, config: FaultConfig) {
        use crate::node::network_bridge::{FaultInjectorState, set_fault_injector};
        // Use a derived seed for fault injection to maintain determinism
        let fault_seed = self.seed.wrapping_add(0xFA01_7777);
        // Always use VirtualTime for deterministic behavior
        let state = FaultInjectorState::new(config, fault_seed)
            .with_virtual_time(self.virtual_time.clone());
        set_fault_injector(
            &self.name,
            Some(std::sync::Arc::new(std::sync::Mutex::new(state))),
        );
    }

    /// Clears any configured fault injection and resets to default (VirtualTime only).
    pub fn clear_fault_injection(&mut self) {
        self.init_default_fault_injection();
    }

    /// Configures fault injection with a custom VirtualTime instance.
    ///
    /// This is useful if you want to use a shared VirtualTime across multiple
    /// simulations or have more control over time advancement.
    ///
    /// For most cases, use [`with_fault_injection`](Self::with_fault_injection) instead,
    /// which uses the built-in VirtualTime.
    #[deprecated(
        since = "0.1.0",
        note = "VirtualTime is now always enabled. Use with_fault_injection() and virtual_time() instead."
    )]
    pub fn with_fault_injection_virtual_time(
        &mut self,
        config: FaultConfig,
        virtual_time: VirtualTime,
    ) {
        use crate::node::network_bridge::{FaultInjectorState, set_fault_injector};
        let fault_seed = self.seed.wrapping_add(0xFA01_7777);
        let state = FaultInjectorState::new(config, fault_seed).with_virtual_time(virtual_time);
        set_fault_injector(
            &self.name,
            Some(std::sync::Arc::new(std::sync::Mutex::new(state))),
        );
    }

    /// Advances virtual time and delivers pending messages.
    ///
    /// This should be called after advancing the VirtualTime instance to deliver
    /// messages whose deadlines have passed.
    ///
    /// Returns the number of messages delivered.
    pub fn advance_virtual_time(&mut self) -> usize {
        use crate::node::network_bridge::get_fault_injector;
        if let Some(injector) = get_fault_injector(&self.name) {
            let mut state = injector.lock().unwrap();
            state.advance_time()
        } else {
            0
        }
    }

    /// Advances virtual time by the given duration and delivers pending messages.
    ///
    /// This is a convenience method combining `virtual_time().advance()` and
    /// `advance_virtual_time()`.
    ///
    /// Returns the number of messages delivered.
    pub fn advance_time(&mut self, duration: Duration) -> usize {
        self.virtual_time.advance(duration);
        self.advance_virtual_time()
    }

    /// Returns the current network statistics from fault injection.
    ///
    /// These statistics track:
    /// - Messages sent, delivered, dropped
    /// - Drop reasons (loss rate, partition, crash)
    /// - Latency injection metrics
    pub fn get_network_stats(&self) -> Option<crate::node::network_bridge::NetworkStats> {
        use crate::node::network_bridge::get_fault_injector;
        get_fault_injector(&self.name).map(|injector| {
            let state = injector.lock().unwrap();
            state.stats().clone()
        })
    }

    /// Resets the network statistics.
    pub fn reset_network_stats(&mut self) {
        use crate::node::network_bridge::get_fault_injector;
        if let Some(injector) = get_fault_injector(&self.name) {
            let mut state = injector.lock().unwrap();
            state.reset_stats();
        }
    }

    // =========================================================================
    // Node Lifecycle Management (Crash/Restart)
    // =========================================================================

    /// Crashes a node, stopping its execution and blocking all messages to/from it.
    ///
    /// The node's task is aborted and all messages to/from its address are dropped.
    /// In-flight operations will fail. Use [`restart_node`](Self::restart_node) to
    /// bring the node back (note: restart is not yet implemented).
    ///
    /// # Example
    /// ```ignore
    /// let mut sim = SimNetwork::new(...).await;
    /// let handles = sim.start_with_rand_gen::<SmallRng>(seed, 10, 5).await;
    ///
    /// // Crash node "node-0"
    /// sim.crash_node(&NodeLabel::node(0));
    ///
    /// // Messages to/from this node will be dropped
    /// // Other nodes should handle the failure gracefully
    /// ```
    pub fn crash_node(&mut self, label: &NodeLabel) -> bool {
        use crate::node::network_bridge::get_fault_injector;

        // Get the node's address
        let addr = match self.node_addresses.get(label) {
            Some(addr) => *addr,
            None => {
                tracing::warn!(?label, "Cannot crash node: address not found");
                return false;
            }
        };

        // Mark as crashed in fault injector
        if let Some(injector) = get_fault_injector(&self.name) {
            let mut state = injector.lock().unwrap();
            state.config.crash_node(addr);
            tracing::info!(?label, ?addr, "Node marked as crashed in fault injector");
        }

        // Abort the running task if we have a handle
        if let Some(running) = self.running_nodes.remove(label) {
            running.abort_handle.abort();
            tracing::info!(?label, "Node task aborted");
            true
        } else {
            tracing::warn!(
                ?label,
                "Node not found in running_nodes (may not be started yet)"
            );
            false
        }
    }

    /// Recovers a crashed node, allowing messages to flow again.
    ///
    /// This removes the node from the crashed set, but does NOT restart the node's
    /// execution. The node will not process messages, but other nodes can attempt
    /// to reach it (messages won't be dropped due to crash status).
    ///
    /// For full restart, see the roadmap in `docs/architecture/simulation-testing-design.md`.
    pub fn recover_node(&mut self, label: &NodeLabel) -> bool {
        use crate::node::network_bridge::get_fault_injector;

        let addr = match self.node_addresses.get(label) {
            Some(addr) => addr,
            None => {
                tracing::warn!(?label, "Cannot recover node: address not found");
                return false;
            }
        };

        if let Some(injector) = get_fault_injector(&self.name) {
            let mut state = injector.lock().unwrap();
            state.config.recover_node(addr);
            tracing::info!(
                ?label,
                ?addr,
                "Node recovered (no longer marked as crashed)"
            );
            true
        } else {
            false
        }
    }

    /// Checks if a node is currently marked as crashed.
    pub fn is_node_crashed(&self, label: &NodeLabel) -> bool {
        use crate::node::network_bridge::get_fault_injector;

        let addr = match self.node_addresses.get(label) {
            Some(addr) => addr,
            None => return false,
        };

        if let Some(injector) = get_fault_injector(&self.name) {
            let state = injector.lock().unwrap();
            state.config.is_crashed(addr)
        } else {
            false
        }
    }

    /// Returns the socket address for a node label, if known.
    pub fn node_address(&self, label: &NodeLabel) -> Option<SocketAddr> {
        self.node_addresses.get(label).copied()
    }

    /// Returns all node labels and their addresses.
    pub fn all_node_addresses(&self) -> &HashMap<NodeLabel, SocketAddr> {
        &self.node_addresses
    }

    /// Restarts a crashed node, preserving its identity.
    ///
    /// This method:
    /// 1. Retrieves the saved configuration (keypair, data directory, location)
    /// 2. Unregisters the old peer from the transport layer
    /// 3. Creates a new node with the same identity
    /// 4. Starts the node task
    ///
    /// # What's Preserved
    /// - **Keypair/identity**: Same public key and address
    /// - **Data directory path**: Same location for any disk-based storage
    /// - **Network location**: Same ring location
    /// - **Gateway configs**: Same gateway connections
    ///
    /// # What's NOT Preserved (Current Limitation)
    /// - **In-memory state**: Memory caches are lost on crash
    /// - **In-flight transactions**: Any pending operations are lost
    /// - **Contract state**: Currently stored on disk (SQLite), which may or may not
    ///   survive the abrupt task abort depending on flush timing
    ///
    /// # Future Work
    /// For truly deterministic state persistence, we need to implement shared
    /// in-memory storage using `MockStateStorage` instead of SQLite. This would
    /// require making `Executor` generic over the storage type.
    ///
    /// # Arguments
    /// * `label` - The label of the node to restart
    /// * `seed` - Seed for random event generation
    /// * `max_contract_num` - Maximum number of contracts for event generation
    /// * `iterations` - Number of iterations for event generation
    ///
    /// # Returns
    /// * `Some(JoinHandle)` if restart was successful
    /// * `None` if the node config was not found or restart failed
    ///
    /// # Example
    /// ```ignore
    /// // Crash a node
    /// sim.crash_node(&label);
    /// assert!(sim.is_node_crashed(&label));
    ///
    /// // Restart it with same identity
    /// let handle = sim.restart_node::<SmallRng>(&label, 0x5678, 10, 5).await;
    /// assert!(handle.is_some());
    /// assert!(!sim.is_node_crashed(&label));
    /// ```
    pub async fn restart_node<R>(
        &mut self,
        label: &NodeLabel,
        seed: u64,
        max_contract_num: usize,
        iterations: usize,
    ) -> Option<tokio::task::JoinHandle<anyhow::Result<()>>>
    where
        R: crate::client_events::test::RandomEventGenerator + Send + 'static,
    {
        use crate::node::network_bridge::get_fault_injector;
        use crate::transport::in_memory_socket::unregister_socket;

        // Get the saved restartable config
        let restart_config = match self.restartable_configs.get(label) {
            Some(config) => config.clone(),
            None => {
                tracing::warn!(?label, "Cannot restart node: config not found");
                return None;
            }
        };

        // Get the node address
        let node_addr = match self.node_addresses.get(label) {
            Some(addr) => *addr,
            None => {
                tracing::warn!(?label, "Cannot restart node: address not found");
                return None;
            }
        };

        tracing::info!(?label, ?node_addr, "Restarting node with persisted state");

        // Unregister the old socket from transport (the new node will re-register)
        unregister_socket(&self.name, &node_addr);

        // Clear crash status in fault injector
        if let Some(injector) = get_fault_injector(&self.name) {
            let mut state = injector.lock().unwrap();
            state.config.recover_node(&node_addr);
        }

        // Create a new Builder with the saved configuration
        let event_listener = {
            #[cfg(feature = "trace-ot")]
            {
                use crate::tracing::OTEventRegister;
                CombinedRegister::new([
                    self.event_listener.trait_clone(),
                    Box::new(OTEventRegister::new()),
                ])
            }
            #[cfg(not(feature = "trace-ot"))]
            {
                self.event_listener.clone()
            }
        };

        let builder = Builder::build(
            restart_config.config.clone(),
            event_listener,
            format!("{}-{}", self.name, label),
            restart_config.rng_seed,
            self.name.clone(),
        );

        // Calculate total peers for event generation params
        let total_peer_num = self.labels.len();

        // Create user events for the restarted node
        let mut user_events = MemoryEventsGen::<R>::new_with_seed(
            self.receiver_ch.clone(),
            restart_config.config.key_pair.public().clone(),
            seed,
        );
        user_events.rng_params(label.number(), total_peer_num, max_contract_num, iterations);

        // Create the appropriate span
        let span = if restart_config.is_gateway {
            tracing::info_span!("in_mem_gateway_restart", %label)
        } else {
            tracing::info_span!("in_mem_node_restart", %label)
        };

        // Start the node task with shared storage for state persistence
        let shared_storage = restart_config.shared_storage.clone();
        let node_task = async move {
            builder
                .run_node_with_shared_storage(user_events, span, shared_storage)
                .await
        };
        let handle = GlobalExecutor::spawn(node_task);

        // Track the new running node
        self.running_nodes.insert(
            label.clone(),
            RunningNode {
                label: label.clone(),
                addr: node_addr,
                abort_handle: handle.abort_handle(),
            },
        );

        tracing::info!(?label, "Node restarted successfully with persisted state");
        Some(handle)
    }

    #[allow(dead_code)]
    pub(crate) fn connection_manager(&self, label: &NodeLabel) -> Option<&ConnectionManager> {
        self.connection_managers.get(label)
    }

    pub fn has_connection_or_pending(&self, label: &NodeLabel, addr: SocketAddr) -> Option<bool> {
        self.connection_managers
            .get(label)
            .map(|cm| cm.has_connection_or_pending(addr))
    }

    /// Injects a pending reservation backdated by `age`, simulating a failed ConnectOp.
    pub fn inject_stale_reservation(
        &self,
        label: &NodeLabel,
        addr: SocketAddr,
        location: Location,
        age: Duration,
    ) -> bool {
        if let Some(cm) = self.connection_managers.get(label) {
            let created = tokio::time::Instant::now() - age;
            cm.inject_reservation(addr, location, created);
            true
        } else {
            false
        }
    }

    pub fn connection_count(&self, label: &NodeLabel) -> Option<usize> {
        self.connection_managers
            .get(label)
            .map(|cm| cm.connection_count())
    }

    /// Returns the number of pending reservations (including stale ones) for a node.
    pub fn reserved_connections_count(&self, label: &NodeLabel) -> Option<usize> {
        self.connection_managers
            .get(label)
            .map(|cm| cm.get_reserved_connections())
    }

    /// Runs cleanup on stale pending reservations for a node.
    /// Returns the number of stale entries removed.
    pub fn cleanup_stale_reservations(&self, label: &NodeLabel) -> Option<usize> {
        self.connection_managers
            .get(label)
            .map(|cm| cm.cleanup_stale_reservations())
    }

    /// Tests whether the node's connection manager would accept a new connection
    /// from the given address and location.
    pub fn should_accept(
        &self,
        label: &NodeLabel,
        location: Location,
        addr: SocketAddr,
    ) -> Option<bool> {
        self.connection_managers
            .get(label)
            .map(|cm| cm.should_accept(location, addr))
    }

    /// Checks if a node has a saved configuration for restart.
    pub fn can_restart(&self, label: &NodeLabel) -> bool {
        self.restartable_configs.contains_key(label)
    }

    #[allow(unused)]
    pub fn debug(&mut self) {
        self.clean_up_tmp_dirs = false;
    }

    /// Derives a deterministic port from seed and peer index for simulation.
    /// Uses ports in the dynamic range (49152-65535) to avoid conflicts.
    fn derive_deterministic_port(&self, peer_index: usize) -> u16 {
        const BASE_PORT: u16 = 50000;
        const PORT_RANGE: u16 = 10000;
        // Use peer seed to get a deterministic offset
        let peer_seed = self.derive_peer_seed(peer_index);
        BASE_PORT + ((peer_seed % PORT_RANGE as u64) as u16)
    }

    async fn config_gateways(&mut self, num: NonZeroUsize) {
        info!("Building {} gateways", num);
        let mut configs = Vec::with_capacity(num.into());
        for node_no in 0..num.into() {
            let label = NodeLabel::gateway(&self.name, node_no);
            // Use deterministic port for simulation instead of querying system
            let port = self.derive_deterministic_port(node_no);
            let keypair = crate::transport::TransportKeypair::new();
            let addr: SocketAddr = (Ipv6Addr::LOCALHOST, port).into();
            let peer_key_location = PeerKeyLocation::new(keypair.public().clone(), addr);
            // Use location computed from address for consistency with PeerKeyLocation::location()
            let location = Location::from_address(&addr);

            // Track address for crash/restart operations
            self.node_addresses.insert(label.clone(), addr);

            let config_args = ConfigArgs {
                id: Some(format!("{label}")),
                mode: Some(OperationMode::Local),
                network_api: crate::config::NetworkArgs {
                    // Default to usize::MAX so tests don't trigger streaming unless
                    // explicitly opted in via with_streaming_threshold().
                    streaming_threshold: Some(self.streaming_threshold.unwrap_or(usize::MAX)),
                    ..Default::default()
                },
                ..Default::default()
            };
            // TODO: it may be unnecessary use config_args.build() for the simulation. Related with the TODO in Config line 238
            let mut config = NodeConfig::new(config_args.build().await.unwrap())
                .await
                .unwrap();
            config.governance_config_override = self.governance_config_override.clone();
            config.subscribe_hint_floor_override = self.subscribe_hint_floor_override;
            config.summary_first_put_floor_override = self.summary_first_put_floor_override;
            config.hash_first_summaries_floor_override = self.hash_first_summaries_floor_override;
            config.ack_version_floor_override = self.ack_version_floor_override;
            config.broadcast_target_list_floor_override = self.broadcast_target_list_floor_override;
            config.hosting_time_source_override = self
                .hosting_clock
                .clone()
                .map(|c| std::sync::Arc::new(c) as crate::util::time_source::DynTimeSource);
            if let Some(budget) = self.hosting_budget_override {
                std::sync::Arc::make_mut(&mut config.config).max_hosting_storage = budget;
            }
            config.key_pair = keypair;
            config.network_listener_ip = Ipv6Addr::LOCALHOST.into();
            config.network_listener_port = port;
            config.with_own_addr(addr);
            config
                .with_location(location)
                .max_hops_to_live(self.ring_max_htl)
                .max_number_of_connections(self.max_connections)
                .min_number_of_connections(self.min_connections)
                .is_gateway()
                .rnd_if_htl_above(self.rnd_if_htl_above);
            // Disable readiness gating in SimNetwork by default
            config.relay_ready_connections = Some(0);
            self.event_listener
                .add_node(label.clone(), config.key_pair.public().clone());
            configs.push((
                config,
                GatewayConfig {
                    label,
                    peer_key_location,
                    location,
                },
            ));
        }
        // All gateways are passive - they don't initiate connections to other gateways.
        // This ensures deterministic startup order where only regular nodes initiate connections.
        for config in &mut configs {
            config.0.should_connect = false;
        }

        let gateways: Vec<_> = configs.iter().map(|(_, gw)| gw.clone()).collect();
        // Store all gateway configs for use when restarting non-gateway nodes
        self.all_gateway_configs = gateways.clone();

        // Note: Gateways don't need to know about each other - they are passive entry points.
        // Only regular nodes need to know about gateways. This simplifies the topology and
        // improves determinism by avoiding gateway-to-gateway connection attempts.
        for (this_node, this_config) in configs {
            let event_listener = {
                #[cfg(feature = "trace-ot")]
                {
                    use crate::tracing::OTEventRegister;
                    CombinedRegister::new([
                        self.event_listener.trait_clone(),
                        Box::new(OTEventRegister::new()),
                    ])
                }
                #[cfg(not(feature = "trace-ot"))]
                {
                    self.event_listener.clone()
                }
            };
            let peer_seed = self.derive_peer_seed(this_config.label.number());
            let gateway = Builder::build(
                this_node,
                event_listener,
                format!("{}-{label}", self.name, label = this_config.label),
                peer_seed,
                self.name.clone(),
            );
            self.gateways.push((gateway, this_config));
        }
    }

    async fn config_nodes(&mut self, num: usize) {
        info!("Building {} regular nodes", num);
        let gateways: Vec<_> = self
            .gateways
            .iter()
            .map(|(_node, config)| config)
            .cloned()
            .collect();

        for node_no in self.number_of_gateways..num + self.number_of_gateways {
            let label = NodeLabel::node(&self.name, node_no);

            let config_args = ConfigArgs {
                id: Some(format!("{label}")),
                mode: Some(OperationMode::Local),
                network_api: crate::config::NetworkArgs {
                    // Default to usize::MAX so tests don't trigger streaming unless
                    // explicitly opted in via with_streaming_threshold().
                    streaming_threshold: Some(self.streaming_threshold.unwrap_or(usize::MAX)),
                    ..Default::default()
                },
                ..Default::default()
            };
            let mut config = NodeConfig::new(config_args.build().await.unwrap())
                .await
                .unwrap();
            config.governance_config_override = self.governance_config_override.clone();
            config.subscribe_hint_floor_override = self.subscribe_hint_floor_override;
            config.summary_first_put_floor_override = self.summary_first_put_floor_override;
            config.hash_first_summaries_floor_override = self.hash_first_summaries_floor_override;
            config.ack_version_floor_override = self.ack_version_floor_override;
            config.broadcast_target_list_floor_override = self.broadcast_target_list_floor_override;
            config.hosting_time_source_override = self
                .hosting_clock
                .clone()
                .map(|c| std::sync::Arc::new(c) as crate::util::time_source::DynTimeSource);
            if let Some(budget) = self.hosting_budget_override {
                std::sync::Arc::make_mut(&mut config.config).max_hosting_storage = budget;
            }
            for GatewayConfig {
                peer_key_location,
                location,
                ..
            } in &gateways
            {
                config.add_gateway(InitPeerNode::new(peer_key_location.clone(), *location));
            }
            // Use an explicit per-node port when the test requested specific
            // ring locations (see `new_with_node_locations`); otherwise the
            // deterministic derived port. Either way location is computed from
            // the address below, so it stays consistent with what other peers
            // compute from this peer's address.
            let port = match &self.node_port_override {
                Some(ports) => ports[node_no - self.number_of_gateways],
                None => self.derive_deterministic_port(node_no),
            };
            let addr: SocketAddr = (Ipv6Addr::LOCALHOST, port).into();
            // Use location computed from address for consistency with PeerKeyLocation::location().
            // This ensures the stored location matches what other peers compute when looking at our address.
            let location = Location::from_address(&addr);
            config.network_listener_port = port;
            config.network_listener_ip = Ipv6Addr::LOCALHOST.into();
            config.key_pair = crate::transport::TransportKeypair::new();
            config.with_own_addr(addr);
            config
                .with_location(location)
                .max_hops_to_live(self.ring_max_htl)
                .rnd_if_htl_above(self.rnd_if_htl_above)
                .max_number_of_connections(self.max_connections);
            // Disable readiness gating in SimNetwork by default
            config.relay_ready_connections = Some(0);

            // Track address for crash/restart operations
            self.node_addresses.insert(label.clone(), addr);

            self.event_listener
                .add_node(label.clone(), config.key_pair.public().clone());

            let event_listener = {
                #[cfg(feature = "trace-ot")]
                {
                    use crate::tracing::OTEventRegister;
                    CombinedRegister::new([
                        self.event_listener.trait_clone(),
                        Box::new(OTEventRegister::new()),
                    ])
                }
                #[cfg(not(feature = "trace-ot"))]
                {
                    self.event_listener.clone()
                }
            };
            let peer_seed = self.derive_peer_seed(node_no);
            let node = Builder::build(
                config,
                event_listener,
                format!("{}-{label}", self.name),
                peer_seed,
                self.name.clone(),
            );
            self.nodes.push((node, label));
        }
    }

    pub async fn start_with_rand_gen<R>(
        &mut self,
        seed: u64,
        max_contract_num: usize,
        iterations: usize,
    ) -> Vec<tokio::task::JoinHandle<anyhow::Result<()>>>
    where
        R: RandomEventGenerator + Send + 'static,
    {
        use crate::ring::topology_registry::set_current_network_name;
        use crate::transport::in_memory_socket::is_socket_registered;

        // Set the current network name so Ring can register topology snapshots
        set_current_network_name(&self.name);

        let total_peer_num = self.gateways.len() + self.nodes.len();
        let mut peers = vec![];

        // Phase 1: Start all gateways first and collect their addresses
        let gateways: Vec<_> = self.gateways.drain(..).collect();
        let mut gateway_addrs = Vec::with_capacity(gateways.len());

        for (mut node, config) in gateways {
            let label = config.label.clone();
            // Use the peer_key_location address from GatewayConfig - this is the address
            // that will be registered in the peer registry
            let gateway_addr = *config
                .peer_key_location
                .peer_addr
                .as_known()
                .expect("Gateway should have known address");
            gateway_addrs.push(gateway_addr);

            tracing::debug!(peer = %label, addr = %gateway_addr, "starting gateway");

            // Create shared in-memory storage for this node (persists across restarts)
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();

            // Save restartable config BEFORE starting (NodeConfig gets consumed)
            self.restartable_configs.insert(
                label.clone(),
                RestartableNodeConfig {
                    config: node.config.clone(),
                    label: label.clone(),
                    is_gateway: true,
                    gateway_configs: self.all_gateway_configs.clone(),
                    rng_seed: node.rng_seed,
                    shared_storage: shared_storage.clone(),
                },
            );

            let mut user_events = MemoryEventsGen::<R>::new_with_seed(
                self.receiver_ch.clone(),
                node.config.key_pair.public().clone(),
                seed,
            );
            user_events.rng_params(label.number(), total_peer_num, max_contract_num, iterations);
            let span = tracing::info_span!("in_mem_gateway", %label);
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            let shared_cm: Arc<parking_lot::Mutex<Option<ConnectionManager>>> =
                Arc::new(parking_lot::Mutex::new(None));
            node.shared_cm = Some(shared_cm.clone());

            // Use shared in-memory storage for state persistence across restarts
            let node_task = async move {
                node.run_node_with_shared_storage(user_events, span, shared_storage)
                    .await
            };
            let handle = GlobalExecutor::spawn(node_task);

            // Track running node for crash/restart
            self.running_nodes.insert(
                label.clone(),
                RunningNode {
                    label: label.clone(),
                    addr: gateway_addr,
                    abort_handle: handle.abort_handle(),
                },
            );

            peers.push(handle);

            let captured_cm = capture_shared_slot(&shared_cm, self.start_backoff, &label)
                .await
                .unwrap_or_else(|| {
                    panic!(
                        "SimNetwork: node {label} failed to publish ConnectionManager \
                         within the capture budget. This means the spawned `run_node` \
                         task panicked before reaching the `shared_cm` write at the top \
                         of run_node_with_shared_storage, or the task was never scheduled \
                         at all. Check preceding logs for the real failure."
                    )
                });
            self.connection_managers.insert(label, captured_cm);
        }

        // Phase 2: Wait for all gateways to be registered in the peer registry
        // This prevents the race condition where regular nodes try to connect
        // before gateways are ready to receive connections
        let registration_timeout = Duration::from_secs(10);
        let poll_interval = Duration::from_millis(10);
        // Use tokio::time::Instant instead of std::time::Instant to work correctly
        // with start_paused = true in tests. std::time::Instant uses wall-clock time
        // which doesn't advance when tokio's time is paused.
        let start = tokio::time::Instant::now();

        'wait_loop: loop {
            let mut all_registered = true;
            for addr in &gateway_addrs {
                if !is_socket_registered(&self.name, addr) {
                    all_registered = false;
                    break;
                }
            }

            if all_registered {
                tracing::debug!("All {} gateways registered", gateway_addrs.len());
                // Give gateways additional time to fully initialize their event loops
                // before regular nodes start connecting
                tokio::time::sleep(Duration::from_millis(100)).await;
                tracing::debug!("Starting regular nodes");
                break 'wait_loop;
            }

            if start.elapsed() > registration_timeout {
                tracing::warn!(
                    "Timeout waiting for gateway registration, some may not be ready. \
                     Registered: {}/{}",
                    gateway_addrs
                        .iter()
                        .filter(|a| is_socket_registered(&self.name, a))
                        .count(),
                    gateway_addrs.len()
                );
                break 'wait_loop;
            }

            tokio::time::sleep(poll_interval).await;
        }

        // Phase 3: Start all regular nodes
        let nodes: Vec<_> = self.nodes.drain(..).collect();
        for (mut node, label) in nodes {
            // Get node address from tracked addresses
            let node_addr = self
                .node_addresses
                .get(&label)
                .copied()
                .expect("Node address should be tracked");

            tracing::debug!(peer = %label, addr = %node_addr, "starting regular node");

            // Create shared in-memory storage for this node (persists across restarts)
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();

            // Save restartable config BEFORE starting (NodeConfig gets consumed)
            self.restartable_configs.insert(
                label.clone(),
                RestartableNodeConfig {
                    config: node.config.clone(),
                    label: label.clone(),
                    is_gateway: false,
                    gateway_configs: self.all_gateway_configs.clone(),
                    rng_seed: node.rng_seed,
                    shared_storage: shared_storage.clone(),
                },
            );

            let mut user_events = MemoryEventsGen::<R>::new_with_seed(
                self.receiver_ch.clone(),
                node.config.key_pair.public().clone(),
                seed,
            );
            user_events.rng_params(label.number(), total_peer_num, max_contract_num, iterations);
            let span = tracing::info_span!("in_mem_node", %label);
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            let shared_cm: Arc<parking_lot::Mutex<Option<ConnectionManager>>> =
                Arc::new(parking_lot::Mutex::new(None));
            node.shared_cm = Some(shared_cm.clone());

            // Use shared in-memory storage for state persistence across restarts
            let node_task = async move {
                node.run_node_with_shared_storage(user_events, span, shared_storage)
                    .await
            };
            let handle = GlobalExecutor::spawn(node_task);

            // Track running node for crash/restart
            self.running_nodes.insert(
                label.clone(),
                RunningNode {
                    label: label.clone(),
                    addr: node_addr,
                    abort_handle: handle.abort_handle(),
                },
            );

            peers.push(handle);

            let captured_cm = capture_shared_slot(&shared_cm, self.start_backoff, &label)
                .await
                .unwrap_or_else(|| {
                    panic!(
                        "SimNetwork: node {label} failed to publish ConnectionManager \
                         within the capture budget. This means the spawned `run_node` \
                         task panicked before reaching the `shared_cm` write at the top \
                         of run_node_with_shared_storage, or the task was never scheduled \
                         at all. Check preceding logs for the real failure."
                    )
                });
            self.connection_managers.insert(label, captured_cm);
        }

        self.labels.sort_by(|(a, _), (b, _)| a.cmp(b));
        peers
    }

    /// Starts the network with controlled events instead of random event generation.
    ///
    /// This method allows tests to specify exact operations to execute on specific nodes,
    /// enabling precise testing of scenarios like subscription topology formation.
    ///
    /// # Arguments
    ///
    /// * `operations` - A list of `(NodeLabel, SimOperation)` pairs specifying which
    ///   operations to execute on which nodes.
    ///
    /// # Returns
    ///
    /// A vector of join handles for the node tasks and the number of operations scheduled.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let contract = SimOperation::create_test_contract(42);
    /// let operations = vec![
    ///     ScheduledOperation::new(
    ///         NodeLabel::gateway("test", 0),
    ///         SimOperation::Put {
    ///             contract: contract.clone(),
    ///             state: vec![1, 2, 3],
    ///             subscribe: true,
    ///         }
    ///     ),
    ///     ScheduledOperation::new(
    ///         NodeLabel::node("test", 0),
    ///         SimOperation::Subscribe {
    ///             contract_id: *contract.key().id(),
    ///         }
    ///     ),
    /// ];
    ///
    /// let (handles, num_ops) = sim.start_with_controlled_events(operations).await;
    /// ```
    #[cfg(any(test, feature = "testing"))]
    pub async fn start_with_controlled_events(
        &mut self,
        operations: Vec<ScheduledOperation>,
    ) -> (Vec<tokio::task::JoinHandle<anyhow::Result<()>>>, usize) {
        use crate::ring::topology_registry::set_current_network_name;
        use crate::transport::in_memory_socket::is_socket_registered;

        // Set the current network name so Ring can register topology snapshots
        set_current_network_name(&self.name);

        let num_operations = operations.len();
        let mut peers = vec![];

        // Build a map of label -> list of (event_id, operation)
        let mut operations_by_node: HashMap<NodeLabel, Vec<(EventId, SimOperation)>> =
            HashMap::new();
        for (event_id, scheduled_op) in operations.into_iter().enumerate() {
            operations_by_node
                .entry(scheduled_op.node)
                .or_default()
                .push((event_id as EventId, scheduled_op.operation));
        }

        // Phase 1: Start all gateways first and collect their addresses
        let gateways: Vec<_> = self.gateways.drain(..).collect();
        let mut gateway_addrs = Vec::with_capacity(gateways.len());

        for (node, config) in gateways {
            let label = config.label.clone();
            let gateway_addr = *config
                .peer_key_location
                .peer_addr
                .as_known()
                .expect("Gateway should have known address");
            gateway_addrs.push(gateway_addr);

            tracing::debug!(peer = %label, addr = %gateway_addr, "starting gateway with controlled events");

            // Create shared in-memory storage for this node (persists across restarts)
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();

            // Save restartable config BEFORE starting (NodeConfig gets consumed)
            self.restartable_configs.insert(
                label.clone(),
                RestartableNodeConfig {
                    config: node.config.clone(),
                    label: label.clone(),
                    is_gateway: true,
                    gateway_configs: self.all_gateway_configs.clone(),
                    rng_seed: node.rng_seed,
                    shared_storage: shared_storage.clone(),
                },
            );

            // Create MemoryEventsGen without RNG (deterministic mode)
            let mut user_events = MemoryEventsGen::new(
                self.receiver_ch.clone(),
                node.config.key_pair.public().clone(),
            );

            // Populate events for this node
            if let Some(node_ops) = operations_by_node.remove(&label) {
                let events: Vec<_> = node_ops
                    .into_iter()
                    .map(|(id, op)| (id, op.into_client_request()))
                    .collect();
                user_events.generate_events(events);
            }

            let span = tracing::info_span!("in_mem_gateway_controlled", %label);
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            // Use shared in-memory storage for state persistence across restarts
            let node_task = async move {
                node.run_node_with_shared_storage(user_events, span, shared_storage)
                    .await
            };
            let handle = GlobalExecutor::spawn(node_task);

            // Track running node for crash/restart
            self.running_nodes.insert(
                label.clone(),
                RunningNode {
                    label: label.clone(),
                    addr: gateway_addr,
                    abort_handle: handle.abort_handle(),
                },
            );

            peers.push(handle);

            tokio::time::sleep(self.start_backoff).await;
        }

        // Phase 2: Wait for all gateways to be registered
        let registration_timeout = Duration::from_secs(10);
        let poll_interval = Duration::from_millis(10);
        let start = tokio::time::Instant::now();

        'wait_loop: loop {
            let mut all_registered = true;
            for addr in &gateway_addrs {
                if !is_socket_registered(&self.name, addr) {
                    all_registered = false;
                    break;
                }
            }

            if all_registered {
                tracing::debug!(
                    "All {} gateways registered in {:?}",
                    gateway_addrs.len(),
                    start.elapsed()
                );
                break 'wait_loop;
            }

            if start.elapsed() > registration_timeout {
                tracing::warn!(
                    "Timeout waiting for gateway registration. {} of {} registered.",
                    gateway_addrs
                        .iter()
                        .filter(|a| is_socket_registered(&self.name, a))
                        .count(),
                    gateway_addrs.len()
                );
                break 'wait_loop;
            }

            tokio::time::sleep(poll_interval).await;
        }

        // Phase 3: Start regular nodes
        for (node, label) in std::mem::take(&mut self.nodes) {
            let node_addr = *self
                .node_addresses
                .get(&label)
                .expect("Node address should be tracked");

            tracing::debug!(peer = %label, addr = %node_addr, "starting regular node with controlled events");

            // Create shared in-memory storage for this node (persists across restarts)
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();

            // Save restartable config BEFORE starting (NodeConfig gets consumed)
            self.restartable_configs.insert(
                label.clone(),
                RestartableNodeConfig {
                    config: node.config.clone(),
                    label: label.clone(),
                    is_gateway: false,
                    gateway_configs: self.all_gateway_configs.clone(),
                    rng_seed: node.rng_seed,
                    shared_storage: shared_storage.clone(),
                },
            );

            // Create MemoryEventsGen without RNG (deterministic mode)
            let mut user_events = MemoryEventsGen::new(
                self.receiver_ch.clone(),
                node.config.key_pair.public().clone(),
            );

            // Populate events for this node
            if let Some(node_ops) = operations_by_node.remove(&label) {
                let events: Vec<_> = node_ops
                    .into_iter()
                    .map(|(id, op)| (id, op.into_client_request()))
                    .collect();
                user_events.generate_events(events);
            }

            let span = tracing::info_span!("in_mem_node_controlled", %label);
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            // Use shared in-memory storage for state persistence across restarts
            let node_task = async move {
                node.run_node_with_shared_storage(user_events, span, shared_storage)
                    .await
            };
            let handle = GlobalExecutor::spawn(node_task);

            // Track running node for crash/restart
            self.running_nodes.insert(
                label.clone(),
                RunningNode {
                    label: label.clone(),
                    addr: node_addr,
                    abort_handle: handle.abort_handle(),
                },
            );

            peers.push(handle);

            tokio::time::sleep(self.start_backoff).await;
        }

        // Warn if there were operations for unknown nodes
        if !operations_by_node.is_empty() {
            let unknown_nodes: Vec<_> = operations_by_node.keys().collect();
            tracing::warn!(
                "Operations scheduled for unknown nodes: {:?}",
                unknown_nodes
            );
        }

        self.labels.sort_by(|(a, _), (b, _)| a.cmp(b));
        (peers, num_operations)
    }

    /// Creates an event chain for controlled events.
    ///
    /// Unlike the standard `event_chain()`, this allows specifying the exact
    /// sequence of events to trigger, enabling deterministic testing.
    ///
    /// # Arguments
    ///
    /// * `event_sequence` - Pairs of `(EventId, NodeLabel)` specifying which events
    ///   to trigger on which nodes, in order.
    ///
    /// # Returns
    ///
    /// A ControlledEventChain that can be used as a Stream.
    pub fn controlled_event_chain(
        &mut self,
        event_sequence: Vec<(EventId, NodeLabel)>,
    ) -> ControlledEventChain {
        let user_ev_controller = self
            .user_ev_controller
            .take()
            .expect("controller should be set");
        let label_to_key: HashMap<_, _> = self.labels.iter().cloned().collect();
        ControlledEventChain::new(user_ev_controller, event_sequence, label_to_key)
    }

    /// Returns the locations of all peers (gateways + nodes) without consuming them.
    pub fn get_peer_locations(&self) -> Vec<f64> {
        let mut locations = Vec::new();
        for (builder, _) in &self.gateways {
            if let Some(loc) = &builder.config.location {
                locations.push(loc.as_f64());
            }
        }
        for (builder, _) in &self.nodes {
            if let Some(loc) = &builder.config.location {
                locations.push(loc.as_f64());
            }
        }
        locations
    }

    /// Builds peer nodes and returns the controller to trigger events.
    pub fn build_peers(&mut self) -> Vec<(NodeLabel, NodeConfig)> {
        let gw = self.gateways.drain(..).map(|(n, c)| (n, c.label));
        let mut peers = vec![];
        for (builder, label) in gw.chain(self.nodes.drain(..)).collect::<Vec<_>>() {
            let pub_key = builder.config.key_pair.public();
            self.labels.push((label.clone(), pub_key.clone()));
            peers.push((label, builder.config));
        }
        self.labels.sort_by(|(a, _), (b, _)| a.cmp(b));
        peers.sort_by(|(a, _), (b, _)| a.cmp(b));
        peers
    }

    /// Returns the connectivity in the network per peer (that is all the connections
    /// this peers has registered).
    pub fn node_connectivity(
        &self,
    ) -> HashMap<NodeLabel, (TransportPublicKey, HashMap<NodeLabel, Distance>)> {
        let mut peers_connections = HashMap::with_capacity(self.labels.len());
        let key_to_label: HashMap<_, _> = self.labels.iter().map(|(k, v)| (v, k)).collect();
        for (label, key) in &self.labels {
            let conns = self
                .event_listener
                .connections(key)
                .map(|(k, d)| (key_to_label[k.pub_key()].clone(), d))
                .collect::<HashMap<_, _>>();
            peers_connections.insert(label.clone(), (key.clone(), conns));
        }
        peers_connections
    }

    /// Returns the set of neighbor public keys for a given node label.
    ///
    /// Uses the event listener's `connections()` to get the current peer set,
    /// returning just the `TransportPublicKey`s (without distances).
    pub fn neighbor_peer_keys(&self, label: &NodeLabel) -> Option<HashSet<TransportPublicKey>> {
        let key = self
            .labels
            .iter()
            .find(|(l, _)| l == label)
            .map(|(_, k)| k)?;
        Some(
            self.event_listener
                .connections(key)
                .map(|(k, _)| k.pub_key().clone())
                .collect(),
        )
    }

    /// Start an event chain for this simulation. Allows passing a different controller for the peers.
    ///
    /// This method borrows the SimNetwork, allowing you to call verification methods
    /// (like `check_convergence()`, `get_operation_summary()`) after events complete.
    ///
    /// If done make sure you set the proper receiving side for the controller. For example in the
    /// nodes built through the [`build_peers`](`Self::build_peers`) method.
    ///
    /// # Example
    /// ```ignore
    /// let mut stream = sim.event_chain(100, None);
    /// while stream.next().await.is_some() {
    ///     tokio::time::sleep(Duration::from_millis(100)).await;
    /// }
    /// drop(stream); // Drop the stream before verification
    /// let result = sim.check_convergence().await; // sim is still usable!
    /// ```
    pub fn event_chain(
        &mut self,
        total_events: u32,
        controller: Option<watch::Sender<(EventId, TransportPublicKey)>>,
    ) -> EventChain {
        let user_ev_controller = controller.unwrap_or_else(|| {
            self.user_ev_controller
                .take()
                .expect("controller should be set")
        });
        // Clone labels - SimNetwork retains ownership for verification methods
        let labels = self.labels.clone();
        // EventChain no longer handles cleanup - SimNetwork's Drop does
        EventChain::new(labels, user_ev_controller, total_events, false)
    }

    /// Consumes the SimNetwork and returns an event chain.
    ///
    /// Use this when you don't need to access SimNetwork after events complete.
    /// For post-event verification, use [`event_chain`](Self::event_chain) instead.
    #[deprecated(
        since = "0.1.0",
        note = "Use event_chain(&mut self) instead to retain access to SimNetwork for verification"
    )]
    pub fn into_event_chain(
        mut self,
        total_events: u32,
        controller: Option<watch::Sender<(EventId, TransportPublicKey)>>,
    ) -> EventChain {
        let user_ev_controller = controller.unwrap_or_else(|| {
            self.user_ev_controller
                .take()
                .expect("controller should be set")
        });
        let labels = std::mem::take(&mut self.labels);
        let debug_val = self.clean_up_tmp_dirs;
        self.clean_up_tmp_dirs = false; // EventChain handles cleanup
        EventChain::new(labels, user_ev_controller, total_events, debug_val)
    }

    /// Checks that all peers in the network have acquired at least one connection to any
    /// other peers.
    ///
    /// This function advances VirtualTime to allow connection messages to be delivered.
    pub async fn check_connectivity(&mut self, time_out: Duration) -> anyhow::Result<()> {
        self.connectivity(time_out, 1.0).await
    }

    /// Checks that a percentage (given as a float between 0 and 1) of the nodes has at least
    /// one connection to any other peers.
    ///
    /// This function advances VirtualTime to allow connection messages to be delivered.
    pub async fn check_partial_connectivity(
        &mut self,
        time_out: Duration,
        percent: f64,
    ) -> anyhow::Result<()> {
        self.connectivity(time_out, percent).await
    }

    /// Internal connectivity check that advances VirtualTime.
    ///
    /// Issue #2725: The previous implementation used std::time::Instant (real wall-clock time)
    /// but never advanced VirtualTime. Since simulations use VirtualTime for deterministic
    /// message delivery, connection handshake messages were never delivered, causing all
    /// connectivity checks to fail with "found disconnected nodes".
    ///
    /// This implementation advances VirtualTime in small steps while checking connectivity,
    /// allowing connection messages to be delivered and processed.
    async fn connectivity(&mut self, time_out: Duration, percent: f64) -> anyhow::Result<()> {
        let num_nodes = self.number_of_nodes;
        let mut connected = HashSet::new();
        let elapsed = std::time::Instant::now();

        // Time step for advancing VirtualTime - small enough for responsiveness,
        // large enough to batch message delivery efficiently
        let time_step = Duration::from_millis(100);

        while elapsed.elapsed() < time_out && (connected.len() as f64 / num_nodes as f64) < percent
        {
            // Advance VirtualTime to trigger message delivery
            // This is critical for VirtualTime-based simulations where messages
            // are scheduled for delivery at VirtualTime + latency
            self.advance_time(time_step);

            // Yield to tokio so tasks can process delivered messages
            tokio::task::yield_now().await;

            // Small real-time sleep to allow task scheduling without spinning CPU
            tokio::time::sleep(Duration::from_millis(10)).await;

            // Check which nodes have connected
            for node in self.number_of_gateways..num_nodes + self.number_of_gateways {
                if !connected.contains(&node)
                    && self.is_connected(&NodeLabel::node(&self.name, node))
                {
                    connected.insert(node);
                }
            }
        }

        let expected =
            HashSet::from_iter(self.number_of_gateways..num_nodes + self.number_of_gateways);
        let mut missing: Vec<_> = expected
            .difference(&connected)
            .map(|n| format!("{}-node-{n}", self.name))
            .collect();

        tracing::info!("Number of simulated nodes: {num_nodes}");

        // Calculate the percentage of nodes that are connected
        let connected_percent = connected.len() as f64 / num_nodes as f64;
        // Fail if fewer nodes are connected than the required percentage (with tolerance)
        if connected_percent < (percent - 0.01/* 1% error tolerance */) {
            missing.sort();
            let show_max = missing.len().min(100);
            tracing::error!("Nodes without connection: {:?}(..)", &missing[..show_max],);
            tracing::error!(
                "Total nodes without connection: {:?},  ({:.1}% connected < {:.1}% required)",
                missing.len(),
                connected_percent * 100.0,
                percent * 100.0
            );
            anyhow::bail!("found disconnected nodes");
        }

        tracing::info!(
            "Required time for connecting all peers: {} secs",
            elapsed.elapsed().as_secs()
        );

        Ok(())
    }

    pub fn is_connected(&self, peer: &NodeLabel) -> bool {
        let pos = self
            .labels
            .binary_search_by(|(label, _)| label.cmp(peer))
            .expect("peer not found");
        self.event_listener.is_connected(&self.labels[pos].1)
    }

    /// Returns a copy of all network event logs captured during the simulation.
    ///
    /// These logs can be used to verify deterministic behavior by comparing
    /// event sequences across runs with the same seed.
    pub async fn get_event_logs(&self) -> Vec<crate::tracing::NetLogMessage> {
        self.event_listener.logs.lock().await.clone()
    }

    /// Returns event logs filtered and sorted for deterministic comparison.
    ///
    /// Events are sorted by (transaction, peer_addr, event_kind_name).
    /// Timestamps are ignored since they may vary between runs.
    pub async fn get_deterministic_event_summary(&self) -> Vec<EventSummary> {
        let logs = self.event_listener.logs.lock().await;
        let mut summaries: Vec<EventSummary> = logs
            .iter()
            .map(|log| {
                let event_detail = format!("{:?}", log.kind);
                // Use the structured variant_name() method instead of parsing debug output
                let event_kind_name = log.kind.variant_name().to_string();
                // Extract contract key using the structured accessor
                let contract_key = log.kind.contract_key().map(|k| format!("{:?}", k));
                // Extract state hash using the structured accessor
                let state_hash = log.kind.state_hash().map(String::from);
                EventSummary {
                    tx: log.tx,
                    peer_addr: log.peer_id.socket_addr(),
                    event_kind_name,
                    contract_key,
                    state_hash,
                    event_detail,
                }
            })
            .collect();
        // Sort for deterministic comparison
        summaries.sort();
        summaries
    }

    /// Counts events by type for quick comparison.
    pub async fn get_event_counts(&self) -> std::collections::HashMap<String, usize> {
        let logs = self.event_listener.logs.lock().await;
        let mut counts = std::collections::HashMap::new();
        for log in logs.iter() {
            // Use the structured variant_name() method instead of parsing debug output
            let key = log.kind.variant_name();
            *counts.entry(key.to_string()).or_insert(0) += 1;
        }
        counts
    }

    /// Returns a handle to the event logs that can be accessed after `run_simulation` consumes `self`.
    ///
    /// This is useful for tests that need to compare event logs between runs when using
    /// Turmoil's deterministic scheduler via `run_simulation()`.
    ///
    /// # Example
    /// ```ignore
    /// let sim = SimNetwork::new(...).await;
    /// let logs_handle = sim.event_logs_handle();
    ///
    /// sim.run_simulation::<SmallRng, _, _>(...)?;
    ///
    /// // Access logs after simulation completes
    /// let logs = logs_handle.lock().await;
    /// ```
    pub fn event_logs_handle(&self) -> Arc<tokio::sync::Mutex<Vec<crate::tracing::NetLogMessage>>> {
        self.event_listener.logs.clone()
    }

    /// Recommended to calling after `check_connectivity` to ensure enough time
    /// elapsed for all peers to become connected.
    ///
    /// Checks that there is a good connectivity over the simulated network,
    /// meaning that:
    ///
    /// - at least 50% of the peers have more than the minimum connections
    /// - the average number of connections per peer is above the mean between max and min connections
    pub fn network_connectivity_quality(&self) -> anyhow::Result<()> {
        const HIGHER_THAN_MIN_THRESHOLD: f64 = 0.5;
        let num_nodes = self.number_of_nodes;

        // Guard against division by zero
        if num_nodes == 0 {
            anyhow::bail!("cannot check connectivity quality with zero nodes");
        }

        let min_connections_threshold = (num_nodes as f64 * HIGHER_THAN_MIN_THRESHOLD) as usize;
        let node_connectivity = self.node_connectivity();

        let mut connections_per_peer: Vec<_> = node_connectivity
            .iter()
            .map(|(k, v)| (k, v.1.len()))
            .filter(|&(k, _)| !k.is_gateway())
            .map(|(_, v)| v)
            .collect();

        // Guard against empty connections list
        if connections_per_peer.is_empty() {
            anyhow::bail!("no non-gateway nodes found for connectivity check");
        }

        // ensure at least "most" normal nodes have more than one connection
        connections_per_peer.sort_unstable_by_key(|num_conn| *num_conn);
        if connections_per_peer
            .get(min_connections_threshold)
            .copied()
            .unwrap_or(0)
            < self.min_connections
        {
            tracing::error!(
                "Low connectivity; more than {:.0}% of the nodes don't have more than minimum connections",
                HIGHER_THAN_MIN_THRESHOLD * 100.0
            );
            anyhow::bail!("low connectivity");
        } else {
            let idx = connections_per_peer[min_connections_threshold..]
                .iter()
                .position(|num_conn| *num_conn < self.min_connections)
                .unwrap_or_else(|| connections_per_peer[min_connections_threshold..].len() - 1)
                + (min_connections_threshold - 1);
            let percentile = idx as f64 / connections_per_peer.len() as f64 * 100.0;
            tracing::info!("{percentile:.0}% nodes have higher than required minimum connections");
        }

        // ensure the average number of connections per peer is above the mean between max and min connections
        let expected_avg_connections =
            ((self.max_connections - self.min_connections) / 2) + self.min_connections;
        let avg_connections: usize = connections_per_peer.iter().sum::<usize>() / num_nodes;
        if avg_connections < expected_avg_connections {
            tracing::warn!(
                "Average number of connections ({avg_connections}) is low (< {expected_avg_connections})"
            );
        }
        Ok(())
    }

    /// Checks convergence of contract states across all peers.
    ///
    /// Returns a `ConvergenceResult` containing:
    /// - Which contracts have converged (all replicas have the same state hash)
    /// - Which contracts have not converged (different state hashes across replicas)
    /// - Per-contract details of state hashes per peer
    ///
    /// This is useful for testing eventual consistency properties.
    ///
    /// # Implementation Note
    ///
    /// This method iterates through logs in insertion order (chronological order)
    /// rather than sorting by transaction ID. This is critical because broadcast
    /// events are logged with the sender's original transaction ID, not a new
    /// local timestamp. Sorting by transaction ID would cause delayed broadcasts
    /// with older transaction IDs to appear before newer local updates in the
    /// sorted order, resulting in incorrect "latest state" detection.
    ///
    /// For example, if Peer A updates to state S3 (tx=T15) and then receives a
    /// delayed broadcast of state S1 (tx=T1 < T15), the actual state is S1 (the
    /// broadcast was applied), but tx-sorted order would show S3 as "latest".
    pub async fn check_convergence(&self) -> ConvergenceResult {
        // Get logs in insertion order (chronological order within the simulation).
        // DO NOT use get_deterministic_event_summary() here - it sorts by transaction ID,
        // which is incorrect for determining latest state because broadcasts use the
        // sender's original transaction ID rather than a local timestamp.
        let logs = self.event_listener.logs.lock().await;

        // Group (contract_key -> peer_addr -> latest_state_hash)
        // Use BTreeMap for deterministic iteration order in DST
        let mut contract_states: BTreeMap<String, BTreeMap<SocketAddr, String>> = BTreeMap::new();

        // Iterate in insertion order - the last event for each (contract, peer) pair
        // is the actual current state.
        //
        // IMPORTANT: Use stored_state_hash() instead of state_hash() to only consider
        // events that represent actual stored state (PutSuccess, UpdateSuccess, BroadcastApplied).
        // Using state_hash() would incorrectly include BroadcastReceived events which record
        // the incoming state hash BEFORE it's applied, not the actual stored state.
        for log in logs.iter() {
            let contract_key = log.kind.contract_key().map(|k| format!("{:?}", k));
            let state_hash = log.kind.stored_state_hash().map(String::from);

            if let (Some(contract_key), Some(state_hash)) = (contract_key, state_hash) {
                // Keep the latest state for each peer/contract pair
                contract_states
                    .entry(contract_key)
                    .or_default()
                    .insert(log.peer_id.socket_addr(), state_hash);
            }
        }

        let mut converged = Vec::new();
        let mut diverged = Vec::new();

        for (contract_key, peer_states) in contract_states {
            if peer_states.len() < 2 {
                // Need at least 2 peers to check convergence
                continue;
            }

            let unique_states: HashSet<&String> = peer_states.values().collect();

            if unique_states.len() == 1 {
                let state = unique_states.into_iter().next().unwrap().clone();
                converged.push(ConvergedContract {
                    contract_key,
                    state_hash: state,
                    replica_count: peer_states.len(),
                });
            } else {
                diverged.push(DivergedContract {
                    contract_key,
                    peer_states: peer_states.into_iter().collect(),
                });
            }
        }

        ConvergenceResult {
            converged,
            diverged,
        }
    }

    /// Waits for convergence of contract states, polling at regular intervals.
    ///
    /// # Arguments
    /// * `timeout` - Maximum time to wait for convergence
    /// * `poll_interval` - How often to check convergence
    /// * `min_contracts` - Minimum number of contracts that must be replicated (default: 1)
    ///
    /// # Returns
    /// * `Ok(ConvergenceResult)` - If convergence achieved (no diverged contracts)
    /// * `Err(ConvergenceResult)` - If timeout reached with diverged contracts
    ///
    /// # Example
    /// ```ignore
    /// let result = sim.await_convergence(
    ///     Duration::from_secs(30),
    ///     Duration::from_millis(500),
    ///     1,
    /// ).await;
    ///
    /// match result {
    ///     Ok(r) => println!("{} contracts converged", r.converged.len()),
    ///     Err(r) => panic!("{} contracts still diverged", r.diverged.len()),
    /// }
    /// ```
    pub async fn await_convergence(
        &self,
        timeout: Duration,
        poll_interval: Duration,
        min_contracts: usize,
    ) -> Result<ConvergenceResult, ConvergenceResult> {
        // Use tokio::time::Instant for deterministic behavior in simulation
        let start = tokio::time::Instant::now();

        loop {
            let result = self.check_convergence().await;

            // Check if we have enough contracts and all have converged
            let total_replicated = result.converged.len() + result.diverged.len();
            if total_replicated >= min_contracts && result.diverged.is_empty() {
                tracing::info!(
                    "Convergence achieved: {} contracts converged in {:?}",
                    result.converged.len(),
                    start.elapsed()
                );
                return Ok(result);
            }

            // Check timeout
            if start.elapsed() >= timeout {
                tracing::warn!(
                    "Convergence timeout after {:?}: {} converged, {} diverged",
                    timeout,
                    result.converged.len(),
                    result.diverged.len()
                );
                return Err(result);
            }

            tokio::time::sleep(poll_interval).await;
        }
    }

    /// Asserts that all replicated contracts have converged.
    ///
    /// This is a convenience method that combines `await_convergence` with
    /// an assertion. It panics if convergence is not achieved.
    ///
    /// # Panics
    /// Panics if convergence is not achieved within the timeout.
    pub async fn assert_convergence(&self, timeout: Duration, poll_interval: Duration) {
        match self.await_convergence(timeout, poll_interval, 1).await {
            Ok(result) => {
                tracing::info!(
                    "Convergence assertion passed: {} contracts",
                    result.converged.len()
                );
            }
            Err(result) => {
                let diverged_details: Vec<String> = result
                    .diverged
                    .iter()
                    .map(|d| {
                        format!(
                            "Contract {}: {} different states across {} peers",
                            d.contract_key,
                            d.unique_state_count(),
                            d.peer_states.len()
                        )
                    })
                    .collect();

                panic!(
                    "Convergence assertion failed after {:?}: {} contracts diverged\n{}",
                    timeout,
                    result.diverged.len(),
                    diverged_details.join("\n")
                );
            }
        }
    }

    /// Returns the convergence rate as a ratio (0.0 to 1.0).
    ///
    /// Convergence rate = converged_contracts / total_replicated_contracts
    pub async fn convergence_rate(&self) -> f64 {
        let result = self.check_convergence().await;
        let total = result.converged.len() + result.diverged.len();
        if total == 0 {
            return 1.0; // No contracts replicated yet
        }
        result.converged.len() as f64 / total as f64
    }

    /// Run the state verifier on all collected telemetry events.
    ///
    /// This linearizes the state transitions for every contract across all peers
    /// and detects anomalies such as:
    /// - Missing broadcasts (emitted but never received)
    /// - Unapplied broadcasts (received but never applied)
    /// - Unexpected state changes (merge produced an unknown hash)
    /// - Final divergence (peers disagree after all events) with root-cause analysis
    ///
    /// Unlike `check_convergence()`, which only checks the final state snapshot,
    /// this traces the full causal history to pinpoint WHERE divergence originated.
    ///
    /// # Example
    /// ```ignore
    /// let report = sim.verify_state().await;
    /// if !report.is_clean() {
    ///     eprintln!("{}", report.display());
    ///     for anomaly in &report.anomalies {
    ///         eprintln!("  {}", anomaly);
    ///     }
    /// }
    /// ```
    pub async fn verify_state(&self) -> crate::tracing::VerificationReport {
        let logs = self.event_listener.logs.lock().await;
        let verifier = crate::tracing::StateVerifier::from_events(logs.clone());
        verifier.verify()
    }

    /// Assert that state verification passes with no anomalies.
    ///
    /// Panics with a detailed report if any anomalies are detected.
    pub async fn assert_state_verified(&self) {
        let report = self.verify_state().await;
        if !report.is_clean() {
            panic!(
                "State verification failed with {} anomalies:\n{}",
                report.anomalies.len(),
                report.display()
            );
        }
    }

    /// Returns the number of unique contracts that have been subscribed to.
    ///
    /// Counts distinct contracts from SubscribeSuccess events. Use this to verify
    /// that subscribed contracts are actually getting replicated and converged.
    ///
    /// # Example
    /// ```ignore
    /// let subscribed_count = sim.count_subscribed_contracts().await;
    /// let converged = sim.check_convergence().await;
    /// assert_eq!(subscribed_count, converged.total_contracts(),
    ///     "All subscribed contracts should be replicated and checked for convergence");
    /// ```
    pub async fn count_subscribed_contracts(&self) -> usize {
        use std::collections::HashSet;
        let logs = self.event_listener.logs.lock().await;

        let mut subscribed_contracts: HashSet<String> = HashSet::new();

        for log in logs.iter() {
            if let crate::tracing::EventKind::Subscribe(
                crate::tracing::SubscribeEvent::SubscribeSuccess { key, .. },
            ) = &log.kind
            {
                subscribed_contracts.insert(format!("{:?}", key));
            }
        }

        subscribed_contracts.len()
    }

    // =========================================================================
    // Gap 4: Direct State Query API (event-based)
    // =========================================================================

    /// Returns the latest known state hash for each contract on each peer.
    ///
    /// This provides a view of contract state distribution across the network
    /// by examining event logs. Returns a map of contract_key -> (peer_addr -> state_hash).
    ///
    /// # Example
    /// ```ignore
    /// let states = sim.get_contract_state_hashes().await;
    /// for (contract, peer_states) in states {
    ///     println!("Contract {}: {} replicas", contract, peer_states.len());
    ///     for (peer, hash) in peer_states {
    ///         println!("  {}: {}", peer, hash);
    ///     }
    /// }
    /// ```
    pub async fn get_contract_state_hashes(
        &self,
    ) -> BTreeMap<String, BTreeMap<SocketAddr, String>> {
        let summary = self.get_deterministic_event_summary().await;

        // Use BTreeMap for deterministic iteration order in DST
        let mut contract_states: BTreeMap<String, BTreeMap<SocketAddr, String>> = BTreeMap::new();

        for event in &summary {
            if let (Some(contract_key), Some(state_hash)) = (&event.contract_key, &event.state_hash)
            {
                // Keep the latest state for each peer/contract pair
                contract_states
                    .entry(contract_key.clone())
                    .or_default()
                    .insert(event.peer_addr, state_hash.clone());
            }
        }

        contract_states
    }

    /// Returns a list of contracts and the peers that have them.
    ///
    /// This is useful for checking contract distribution across the network.
    pub async fn get_contract_distribution(&self) -> Vec<ContractDistribution> {
        let states = self.get_contract_state_hashes().await;
        states
            .into_iter()
            .map(|(contract_key, peer_states)| ContractDistribution {
                contract_key,
                replica_count: peer_states.len(),
                peers: peer_states.keys().cloned().collect(),
            })
            .collect()
    }

    // =========================================================================
    // Gap T4: Operation Completion Tracking
    // =========================================================================

    /// Returns a summary of operation completion status.
    ///
    /// This tracks Put, Get, Subscribe, and Update operations from request to
    /// completion (success or failure).
    ///
    /// # Example
    /// ```ignore
    /// let summary = sim.get_operation_summary().await;
    /// println!("Put: {}/{} completed ({:.1}% success)",
    ///     summary.put.completed(),
    ///     summary.put.requested,
    ///     summary.put.success_rate() * 100.0);
    /// ```
    pub async fn get_operation_summary(&self) -> OperationSummary {
        let logs = self.event_listener.logs.lock().await;

        let mut summary = OperationSummary::default();

        for log in logs.iter() {
            match &log.kind {
                // Put operations
                crate::tracing::EventKind::Put(put_event) => {
                    use crate::tracing::PutEvent;
                    match put_event {
                        PutEvent::Request { .. } => summary.put.requested += 1,
                        PutEvent::PutSuccess { .. } => summary.put.succeeded += 1,
                        PutEvent::PutFailure { .. } => summary.put.failed += 1,
                        PutEvent::ResponseSent { .. } => {} // Response tracking, doesn't affect summary
                        PutEvent::BroadcastEmitted { .. } => summary.put.broadcasts_emitted += 1,
                        PutEvent::BroadcastReceived { .. } => summary.put.broadcasts_received += 1,
                    }
                }
                // Get operations
                crate::tracing::EventKind::Get(get_event) => {
                    use crate::tracing::GetEvent;
                    match get_event {
                        GetEvent::Request { .. } => summary.get.requested += 1,
                        GetEvent::GetSuccess { .. } => summary.get.succeeded += 1,
                        GetEvent::GetFailure { .. } => summary.get.failed += 1,
                        // ClientTerminal is emitted IN ADDITION to
                        // GetSuccess / GetNotFound (once per client op from
                        // the driver), so counting it here would double-count
                        // the sim summary; the GetSuccess arm above already
                        // tallies succeeded.
                        GetEvent::GetNotFound { .. }
                        | GetEvent::ResponseSent { .. }
                        | GetEvent::ForwardingAckSent { .. }
                        | GetEvent::ForwardingAckReceived { .. }
                        | GetEvent::ClientTerminal { .. } => {}
                    }
                }
                // Subscribe operations
                crate::tracing::EventKind::Subscribe(sub_event) => {
                    use crate::tracing::SubscribeEvent;
                    match sub_event {
                        SubscribeEvent::Request { .. } => summary.subscribe.requested += 1,
                        SubscribeEvent::SubscribeSuccess { .. } => summary.subscribe.succeeded += 1,
                        SubscribeEvent::SubscribeNotFound { .. }
                        | SubscribeEvent::SubscribeTimeout { .. } => summary.subscribe.failed += 1,
                        SubscribeEvent::ResponseSent { .. }
                        | SubscribeEvent::HostingStarted { .. }
                        | SubscribeEvent::HostingStopped { .. }
                        | SubscribeEvent::_Reserved6
                        | SubscribeEvent::_Reserved7
                        | SubscribeEvent::_Reserved8
                        | SubscribeEvent::_Reserved9
                        | SubscribeEvent::_Reserved10
                        | SubscribeEvent::UnsubscribeSent { .. }
                        | SubscribeEvent::UnsubscribeReceived { .. } => {}
                    }
                }
                crate::tracing::EventKind::Update(update_event) => {
                    use crate::tracing::UpdateEvent;
                    match update_event {
                        UpdateEvent::Request { .. } => summary.update.requested += 1,
                        UpdateEvent::UpdateSuccess { .. } => summary.update.succeeded += 1,
                        UpdateEvent::BroadcastReceived { .. } => {
                            summary.update.broadcasts_received += 1
                        }
                        UpdateEvent::BroadcastEmitted { .. } => {
                            summary.update.broadcasts_emitted += 1
                        }
                        UpdateEvent::UpdateFailure { .. }
                        | UpdateEvent::BroadcastComplete { .. }
                        | UpdateEvent::BroadcastApplied { .. }
                        | UpdateEvent::BroadcastDeliverySummary { .. } => {}
                    }
                }
                // Timeouts
                crate::tracing::EventKind::Timeout { .. } => {
                    summary.timeouts += 1;
                }
                crate::tracing::EventKind::Connect(_)
                | crate::tracing::EventKind::Route(_)
                | crate::tracing::EventKind::Transfer(_)
                | crate::tracing::EventKind::Lifecycle(_)
                | crate::tracing::EventKind::Ignored
                | crate::tracing::EventKind::Disconnected { .. }
                | crate::tracing::EventKind::TransportSnapshot(_)
                | crate::tracing::EventKind::InterestSync(_)
                | crate::tracing::EventKind::RoutingDecision(_)
                | crate::tracing::EventKind::RouterSnapshot(_) => {}
            }
        }

        summary
    }

    /// Checks if all requested operations have completed (either succeeded or failed).
    ///
    /// Returns (completed, pending) counts.
    pub async fn operation_completion_status(&self) -> (usize, usize) {
        let summary = self.get_operation_summary().await;
        let completed = summary.total_completed();
        let requested = summary.total_requested();
        let pending = requested.saturating_sub(completed);
        (completed, pending)
    }

    /// Returns the overall operation success rate (0.0 to 1.0).
    pub async fn operation_success_rate(&self) -> f64 {
        let summary = self.get_operation_summary().await;
        summary.overall_success_rate()
    }

    /// Waits for all operations to complete, up to a timeout.
    ///
    /// # Returns
    /// * `Ok(OperationSummary)` - If all operations completed
    /// * `Err(OperationSummary)` - If timeout reached with pending operations
    pub async fn await_operation_completion(
        &self,
        timeout: Duration,
        poll_interval: Duration,
    ) -> Result<OperationSummary, OperationSummary> {
        // Use tokio::time::Instant for deterministic behavior in simulation
        let start = tokio::time::Instant::now();

        loop {
            let summary = self.get_operation_summary().await;
            let (completed, pending) = (summary.total_completed(), summary.total_requested());

            // All operations completed (or none requested)
            if pending == 0 || completed >= pending {
                return Ok(summary);
            }

            if start.elapsed() >= timeout {
                tracing::warn!(
                    "Operation completion timeout: {} completed, {} pending",
                    completed,
                    pending.saturating_sub(completed)
                );
                return Err(summary);
            }

            tokio::time::sleep(poll_interval).await;
        }
    }

    /// Waits for network quiescence - when no new network activity is detected.
    ///
    /// This is useful for determining when broadcasts have finished propagating.
    /// It works by monitoring the event log and waiting until no new entries are
    /// added for `quiescence_duration`.
    ///
    /// # Arguments
    /// * `timeout` - Maximum time to wait for quiescence
    /// * `quiescence_duration` - How long activity must be quiet to consider quiesced
    /// * `poll_interval` - How often to check for new activity
    ///
    /// # Returns
    /// * `Ok(usize)` - Total log entries when quiesced
    /// * `Err(usize)` - Log entries at timeout (still active)
    pub async fn await_network_quiescence(
        &self,
        timeout: Duration,
        quiescence_duration: Duration,
        poll_interval: Duration,
    ) -> Result<usize, usize> {
        let start = tokio::time::Instant::now();
        let mut last_log_count = 0usize;
        let mut quiet_since = tokio::time::Instant::now();

        loop {
            let current_count = self.event_listener.logs.lock().await.len();

            if current_count != last_log_count {
                // Activity detected, reset quiet timer
                last_log_count = current_count;
                quiet_since = tokio::time::Instant::now();
            } else if quiet_since.elapsed() >= quiescence_duration {
                // Been quiet long enough
                tracing::info!(
                    "Network quiesced after {:?} with {} log entries",
                    start.elapsed(),
                    current_count
                );
                return Ok(current_count);
            }

            if start.elapsed() >= timeout {
                tracing::warn!(
                    "Network quiescence timeout after {:?}: {} log entries, still active",
                    timeout,
                    current_count
                );
                return Err(current_count);
            }

            tokio::time::sleep(poll_interval).await;
        }
    }

    /// Asserts that operation success rate meets the threshold.
    ///
    /// # Panics
    /// Panics if success rate is below the threshold.
    pub async fn assert_operation_success_rate(&self, min_rate: f64) {
        let summary = self.get_operation_summary().await;
        let rate = summary.overall_success_rate();

        if rate < min_rate {
            panic!(
                "Operation success rate {:.1}% is below threshold {:.1}%\n\
                 Put: {}/{} ({:.1}%), Get: {}/{} ({:.1}%), \
                 Subscribe: {}/{} ({:.1}%), Update: {}/{} ({:.1}%)",
                rate * 100.0,
                min_rate * 100.0,
                summary.put.succeeded,
                summary.put.completed(),
                summary.put.success_rate() * 100.0,
                summary.get.succeeded,
                summary.get.completed(),
                summary.get.success_rate() * 100.0,
                summary.subscribe.succeeded,
                summary.subscribe.completed(),
                summary.subscribe.success_rate() * 100.0,
                summary.update.succeeded,
                summary.update.completed(),
                summary.update.success_rate() * 100.0,
            );
        }
    }

    // =========================================================================
    // Turmoil-based Deterministic Simulation
    // =========================================================================

    /// Runs the simulation with deterministic scheduling using Turmoil.
    ///
    /// This method sets up all nodes as Turmoil hosts and runs the simulation
    /// deterministically. All tokio async operations are scheduled by Turmoil's
    /// deterministic scheduler, making tests reproducible.
    ///
    /// # Arguments
    ///
    /// * `seed` - Seed for random event generation
    /// * `max_contract_num` - Maximum number of contracts in the simulation
    /// * `iterations` - Number of iterations/events to run
    /// * `simulation_duration` - Maximum duration for the simulation
    /// * `test_fn` - A closure containing the test logic to run inside Turmoil
    ///
    /// # Example
    ///
    /// ```ignore
    /// use freenet::dev_tool::SimNetwork;
    /// use std::time::Duration;
    ///
    /// let sim = SimNetwork::new("test", 1, 5, 7, 3, 10, 2, 42).await;
    ///
    /// sim.run_simulation::<rand::rngs::SmallRng, _, _>(
    ///     42,
    ///     10,
    ///     100,
    ///     Duration::from_secs(60),
    ///     || async {
    ///         // Test assertions here
    ///         tokio::time::sleep(Duration::from_secs(5)).await;
    ///         Ok(())
    ///     },
    /// )?;
    /// ```
    ///
    /// # Determinism
    ///
    /// Running with the same seed will produce the same execution order and
    /// results. This is essential for:
    /// - Reproducing bugs
    /// - Property-based testing
    /// - CI reliability
    pub fn run_simulation<R, F, Fut>(
        mut self,
        seed: u64,
        max_contract_num: usize,
        iterations: usize,
        simulation_duration: Duration,
        event_wait: Duration,
        test_fn: F,
    ) -> turmoil::Result
    where
        R: RandomEventGenerator + Send + 'static,
        F: FnOnce() -> Fut + Send + 'static,
        Fut: std::future::Future<Output = turmoil::Result> + 'static,
    {
        use crate::config::{GlobalRng, GlobalSimulationTime};
        use std::sync::Mutex;

        // Set up deterministic RNG and time for reproducible simulation
        GlobalRng::set_seed(seed);

        // Derive simulation epoch from seed for deterministic ULID generation
        // Base: 2020-01-01 00:00:00 UTC, Range: ~5 years (keeps dates sensible: 2020-2025)
        const BASE_EPOCH_MS: u64 = 1577836800000; // 2020-01-01 00:00:00 UTC
        const RANGE_MS: u64 = 5 * 365 * 24 * 60 * 60 * 1000; // ~5 years in ms
        let epoch_offset = seed % RANGE_MS;
        GlobalSimulationTime::set_time_ms(BASE_EPOCH_MS + epoch_offset);

        // Build Turmoil simulation with seeded RNG for deterministic execution
        let mut sim = turmoil::Builder::new()
            .simulation_duration(simulation_duration)
            .rng_seed(seed)
            .build();

        // Get total peer count for event generation
        let total_peer_num = self.gateways.len() + self.nodes.len();

        // Register all gateways as Turmoil hosts
        // Turmoil's sim.host requires Fn (can be called multiple times),
        // so we wrap non-Clone values in Arc<Mutex<Option<T>>> and take them on first call
        let gateways: Vec<_> = self.gateways.drain(..).collect();
        for (node, config) in gateways {
            let label = config.label.clone();
            let host_name = label.to_string();
            let receiver_ch = self.receiver_ch.clone();

            // Create shared in-memory storage for this node
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();

            let mut user_events = MemoryEventsGen::<R>::new_with_seed(
                receiver_ch,
                node.config.key_pair.public().clone(),
                seed,
            );
            user_events.rng_params(label.number(), total_peer_num, max_contract_num, iterations);

            let span = tracing::info_span!("turmoil_gateway", %label);

            // Store label for later reference
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            // Wrap in Option so we can take() on first call
            let node = Arc::new(Mutex::new(Some(node)));
            let user_events = Arc::new(Mutex::new(Some(user_events)));
            let span = Arc::new(Mutex::new(Some(span)));
            let shared_storage = Arc::new(Mutex::new(Some(shared_storage)));

            sim.host(host_name, move || {
                let node = node.clone();
                let user_events = user_events.clone();
                let span = span.clone();
                let shared_storage = shared_storage.clone();

                async move {
                    let node = node
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let user_events = user_events
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let span = span
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let shared_storage = shared_storage
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");

                    node.run_node_with_shared_storage(user_events, span, shared_storage)
                        .await
                        .map_err(|e| {
                            Box::new(std::io::Error::other(e.to_string()))
                                as Box<dyn std::error::Error>
                        })
                }
            });
        }

        // Register all regular nodes as Turmoil hosts
        let nodes: Vec<_> = self.nodes.drain(..).collect();
        for (node, label) in nodes {
            let host_name = label.to_string();
            let receiver_ch = self.receiver_ch.clone();

            // Create shared in-memory storage for this node
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();

            let mut user_events = MemoryEventsGen::<R>::new_with_seed(
                receiver_ch,
                node.config.key_pair.public().clone(),
                seed,
            );
            user_events.rng_params(label.number(), total_peer_num, max_contract_num, iterations);

            let span = tracing::info_span!("turmoil_node", %label);

            // Store label for later reference
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            // Wrap in Option so we can take() on first call
            let node = Arc::new(Mutex::new(Some(node)));
            let user_events = Arc::new(Mutex::new(Some(user_events)));
            let span = Arc::new(Mutex::new(Some(span)));
            let shared_storage = Arc::new(Mutex::new(Some(shared_storage)));

            sim.host(host_name, move || {
                let node = node.clone();
                let user_events = user_events.clone();
                let span = span.clone();
                let shared_storage = shared_storage.clone();

                async move {
                    let node = node
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let user_events = user_events
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let span = span
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let shared_storage = shared_storage
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");

                    node.run_node_with_shared_storage(user_events, span, shared_storage)
                        .await
                        .map_err(|e| {
                            Box::new(std::io::Error::other(e.to_string()))
                                as Box<dyn std::error::Error>
                        })
                }
            });
        }

        // Take the event controller and labels for triggering events
        let user_ev_controller = self
            .user_ev_controller
            .take()
            .expect("user_ev_controller should be set");
        let labels: Vec<_> = self.labels.clone();

        // Register the test function as a Turmoil client
        sim.client("test", async move {
            // Give nodes time to start and establish connections
            tokio::time::sleep(Duration::from_secs(2)).await;

            // Use a seeded RNG for deterministic peer selection
            use rand::SeedableRng;
            use rand::prelude::*;
            let mut event_rng = <rand::rngs::SmallRng as SeedableRng>::seed_from_u64(seed);

            // Trigger events by sending signals to peer event generators
            // Each signal tells one peer to generate its next random event
            // Use longer delays to ensure each event completes before the next starts
            for event_id in 0..iterations as u32 {
                // Pick a random peer to generate an event
                if let Some((_, peer_key)) = labels.choose(&mut event_rng) {
                    if user_ev_controller
                        .send((event_id, peer_key.clone()))
                        .is_err()
                    {
                        tracing::warn!(event_id, "Failed to send event signal - receivers dropped");
                        break;
                    }

                    // Delay between events to allow processing and control virtual time pacing
                    tokio::time::sleep(event_wait).await;
                }
            }

            // Wait for events to fully propagate through the network
            tokio::time::sleep(Duration::from_secs(2)).await;

            // Run the user's test function
            test_fn().await
        });

        // Run the simulation
        sim.run()
    }

    /// Run a deterministic simulation with controlled events using Turmoil.
    ///
    /// Unlike `run_simulation` which uses random events, this method takes
    /// a predefined sequence of operations that will be executed in order.
    /// This is useful for testing specific scenarios like subscription topology.
    ///
    /// The simulation runs under Turmoil's deterministic scheduler, so
    /// `tokio::time::sleep` and other time-dependent operations are controlled.
    ///
    /// # Arguments
    /// * `seed` - Random seed for deterministic simulation
    /// * `operations` - Sequence of operations to execute in order
    /// * `simulation_duration` - Maximum duration for the simulation
    /// * `post_operations_wait` - Time to wait after operations complete (for recovery, etc.)
    ///
    /// # Returns
    /// A `turmoil::Result` indicating success or failure.
    ///
    /// # Example
    /// ```ignore
    /// let operations = vec![
    ///     ScheduledOperation::new(NodeLabel::gateway("test", 0), SimOperation::Put { ... }),
    ///     ScheduledOperation::new(NodeLabel::node("test", 1), SimOperation::Subscribe { ... }),
    /// ];
    /// let result = sim.run_controlled_simulation(
    ///     SEED,
    ///     operations,
    ///     Duration::from_secs(120),
    ///     Duration::from_secs(60), // Wait for orphan recovery
    /// );
    /// // After simulation, check result.topology_snapshots
    /// ```
    #[cfg(any(test, feature = "testing"))]
    pub fn run_controlled_simulation(
        mut self,
        seed: u64,
        operations: Vec<ScheduledOperation>,
        simulation_duration: Duration,
        post_operations_wait: Duration,
    ) -> ControlledSimulationResult {
        use crate::config::{GlobalRng, GlobalSimulationTime};
        use crate::ring::topology_registry::{
            get_all_topology_snapshots, set_current_network_name,
        };
        use std::collections::HashMap;
        use std::sync::Mutex;

        // Set up deterministic RNG and time for reproducible simulation
        GlobalRng::set_seed(seed);

        // Derive simulation epoch from seed for deterministic ULID generation
        const BASE_EPOCH_MS: u64 = 1577836800000; // 2020-01-01 00:00:00 UTC
        const RANGE_MS: u64 = 5 * 365 * 24 * 60 * 60 * 1000; // ~5 years in ms
        let epoch_offset = seed % RANGE_MS;
        GlobalSimulationTime::set_time_ms(BASE_EPOCH_MS + epoch_offset);

        // Set the current network name for topology registration
        set_current_network_name(&self.name);

        // Save network name for topology retrieval after simulation
        let network_name = self.name.clone();

        // Make `SimOperation::CrashNode` a REAL crash: install the global
        // packet-delivery callback that consults each network's fault injector
        // and DROPS every packet to/from a crashed node, then opt THIS network
        // in via `enforce_fault_drops`. The callback is per-network aware (keyed
        // on the network name it is handed) and inert for any network that did
        // not opt in, so the direct-runner churn driver's crash semantics stay
        // unchanged. `SimNetwork::Drop` clears the callback. Without this, a
        // "crashed" node kept exchanging packets and piece-F crash tests were
        // false-green (#4642 piece F).
        crate::transport::in_memory_socket::set_packet_delivery_callback(Some(
            std::sync::Arc::new(fault_injection_delivery_decision),
        ));
        if let Some(injector) = crate::node::network_bridge::get_fault_injector(&network_name) {
            injector.lock().unwrap().enforce_fault_drops = true;
        }

        // Build Turmoil simulation with seeded RNG for deterministic execution
        let mut sim = turmoil::Builder::new()
            .simulation_duration(simulation_duration)
            .rng_seed(seed)
            .build();

        // Separate SeedContract operations (pre-simulation storage seeding)
        // from event operations (dispatched during simulation).
        let mut seed_ops: Vec<(NodeLabel, ContractContainer, Vec<u8>)> = Vec::new();
        // SeedHostedContract operations: like seed_ops, but the contract is
        // injected into the owning node's `contracts` list (below, before the
        // nodes are drained into Turmoil host closures) so startup
        // `append_contracts` registers genuine hosting + subscription.
        let mut seed_hosted_ops: Vec<(NodeLabel, ContractContainer, Vec<u8>)> = Vec::new();
        // SeedDemandlessCopy operations: like seed_hosted_ops, but injected with
        // `SeedMode::Demandless` so `append_contracts` registers a demandless
        // every-hop copy (hosting + local interest, NO subscription) rather than
        // a subscribed host. See serve-DURING (#4642 R3 piece C).
        let mut seed_demandless_ops: Vec<(NodeLabel, ContractContainer, Vec<u8>)> = Vec::new();
        let mut event_ops: Vec<ScheduledOperation> = Vec::new();
        for scheduled_op in operations {
            match scheduled_op.operation {
                SimOperation::SeedContract {
                    ref contract,
                    ref state,
                } => {
                    seed_ops.push((scheduled_op.node, contract.clone(), state.clone()));
                }
                SimOperation::SeedHostedContract {
                    ref contract,
                    ref state,
                } => {
                    seed_hosted_ops.push((scheduled_op.node, contract.clone(), state.clone()));
                }
                SimOperation::SeedDemandlessCopy {
                    ref contract,
                    ref state,
                } => {
                    seed_demandless_ops.push((scheduled_op.node, contract.clone(), state.clone()));
                }
                SimOperation::Put { .. }
                | SimOperation::Get { .. }
                | SimOperation::Subscribe { .. }
                | SimOperation::Update { .. }
                | SimOperation::Disconnect
                // AdvanceHostingClock / CrashNode also flow through the ordered
                // event stream (so they fire in-sequence), but are routed to
                // `special_ops` at numbering time rather than to a node's
                // client-request generator.
                | SimOperation::AdvanceHostingClock { .. }
                | SimOperation::CrashNode
                | SimOperation::RecoverNode => event_ops.push(scheduled_op),
            }
        }

        // Inject SeedHostedContract contracts into the owning node's `contracts`
        // list BEFORE the gateway/node Vecs are drained into Turmoil host
        // closures. At startup `append_contracts` runs a local PutQuery (no
        // network propagation) and, for `subscription = true`, registers the
        // contract in the Ring hosting manager (`host_contract` + `subscribe`)
        // so the node genuinely HOSTS it and the placement-migration trigger
        // (`ring.hosting_contract_keys`) sees it.
        for (target_label, contract, state) in seed_hosted_ops {
            let wrapped = WrappedState::new(state);
            let injected = self
                .nodes
                .iter_mut()
                .find(|(_, label)| *label == target_label)
                .map(|(node, _)| &mut node.contracts)
                .or_else(|| {
                    self.gateways
                        .iter_mut()
                        .find(|(_, cfg)| cfg.label == target_label)
                        .map(|(node, _)| &mut node.contracts)
                })
                .map(|contracts| contracts.push((contract, wrapped, SeedMode::Subscribed)))
                .is_some();
            assert!(
                injected,
                "SeedHostedContract: node {target_label:?} not found among gateways/nodes"
            );
        }

        // Inject SeedDemandlessCopy contracts with `SeedMode::Demandless`. Same
        // injection point as the hosted seed above, but startup
        // `append_contracts` installs a demandless every-hop copy (hosting +
        // local interest, no subscription, no local client access) — the state a
        // production every-hop store leaves on a relay hop, which serve-DURING
        // (#4642 R3 piece C) must serve locally.
        for (target_label, contract, state) in seed_demandless_ops {
            let wrapped = WrappedState::new(state);
            let injected = self
                .nodes
                .iter_mut()
                .find(|(_, label)| *label == target_label)
                .map(|(node, _)| &mut node.contracts)
                .or_else(|| {
                    self.gateways
                        .iter_mut()
                        .find(|(_, cfg)| cfg.label == target_label)
                        .map(|(node, _)| &mut node.contracts)
                })
                .map(|contracts| contracts.push((contract, wrapped, SeedMode::Demandless)))
                .is_some();
            assert!(
                injected,
                "SeedDemandlessCopy: node {target_label:?} not found among gateways/nodes"
            );
        }

        // In-order special dispatch: an event that the controlled-event client
        // performs itself (advance the shared hosting clock, or crash a node via
        // the fault injector) instead of turning into a client request.
        enum SpecialDispatch {
            AdvanceHostingClock(Duration),
            CrashNode(SocketAddr),
            RecoverNode(SocketAddr),
        }

        // Build a map of label -> list of (event_id, operation)
        let mut operations_by_node: HashMap<NodeLabel, Vec<(EventId, SimOperation)>> =
            HashMap::new();
        let mut operation_sequence: Vec<(EventId, NodeLabel)> = Vec::new();
        // Special ops (clock advance / crash) are dispatched in-order by the
        // controlled-event client itself rather than being turned into a client
        // request routed to a node. Keyed by their sequence event_id.
        let mut special_ops: HashMap<EventId, SpecialDispatch> = HashMap::new();

        for (event_id, scheduled_op) in event_ops.into_iter().enumerate() {
            let event_id = event_id as EventId;
            operation_sequence.push((event_id, scheduled_op.node.clone()));
            match scheduled_op.operation {
                SimOperation::AdvanceHostingClock { duration } => {
                    special_ops.insert(event_id, SpecialDispatch::AdvanceHostingClock(duration));
                }
                SimOperation::CrashNode => {
                    let addr = *self
                        .node_addresses
                        .get(&scheduled_op.node)
                        .unwrap_or_else(|| {
                            panic!(
                                "CrashNode: node {:?} has no known address",
                                scheduled_op.node
                            )
                        });
                    special_ops.insert(event_id, SpecialDispatch::CrashNode(addr));
                }
                SimOperation::RecoverNode => {
                    let addr = *self
                        .node_addresses
                        .get(&scheduled_op.node)
                        .unwrap_or_else(|| {
                            panic!(
                                "RecoverNode: node {:?} has no known address",
                                scheduled_op.node
                            )
                        });
                    special_ops.insert(event_id, SpecialDispatch::RecoverNode(addr));
                }
                op @ (SimOperation::Put { .. }
                | SimOperation::Get { .. }
                | SimOperation::Subscribe { .. }
                | SimOperation::Update { .. }
                | SimOperation::Disconnect) => {
                    operations_by_node
                        .entry(scheduled_op.node)
                        .or_default()
                        .push((event_id, op));
                }
                // Seed ops were split out before the numbering loop (see the
                // separation match above), so they never reach here.
                SimOperation::SeedContract { .. }
                | SimOperation::SeedHostedContract { .. }
                | SimOperation::SeedDemandlessCopy { .. } => {
                    unreachable!(
                        "SeedContract/SeedHostedContract/SeedDemandlessCopy are separated out \
                         before numbering"
                    )
                }
            }
        }

        // Handle for the controllable hosting clock (if any), captured into the
        // controlled-event client so `AdvanceHostingClock` can advance it
        // in-order with the rest of the sequence.
        let hosting_clock = self.hosting_clock.clone();

        // Collect storage handles for the result — cloning Arc-backed storage
        // gives tests access to the same data written during the simulation.
        let mut node_storages: HashMap<NodeLabel, crate::wasm_runtime::MockStateStorage> =
            HashMap::new();
        // Pre-seeded contract stores for nodes with SeedContract operations.
        // Wrapped in Arc<Mutex> so closures can look up their store at
        // startup (after seed_ops populates the map, before sim.run()).
        let contract_stores: Arc<
            Mutex<HashMap<NodeLabel, crate::wasm_runtime::InMemoryContractStore>>,
        > = Arc::new(Mutex::new(HashMap::new()));

        let use_mock_wasm = self.use_mock_wasm;

        // Register all gateways as Turmoil hosts
        let gateways: Vec<_> = self.gateways.drain(..).collect();
        for (mut node, config) in gateways {
            let label = config.label.clone();
            let host_name = label.to_string();
            let receiver_ch = self.receiver_ch.clone();

            // Capture slot for this node's live Ring (governance ban-list
            // observation post-simulation). See `shared_rings`.
            let ring_slot: Arc<parking_lot::Mutex<Option<Arc<crate::ring::Ring>>>> =
                Arc::new(parking_lot::Mutex::new(None));
            node.shared_ring = Some(ring_slot.clone());
            self.shared_rings.insert(label.clone(), ring_slot);

            // Create shared in-memory storage for this node
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();
            node_storages.insert(label.clone(), shared_storage.clone());

            // Create MemoryEventsGen without RNG (deterministic mode)
            // Clone receiver_ch so each node gets its own subscription
            let mut user_events =
                MemoryEventsGen::new(receiver_ch.clone(), node.config.key_pair.public().clone());

            // Populate events for this node
            if let Some(node_ops) = operations_by_node.remove(&label) {
                let events: Vec<_> = node_ops
                    .into_iter()
                    .map(|(id, op)| (id, op.into_client_request()))
                    .collect();
                user_events.generate_events(events);
            }

            let span = tracing::info_span!("turmoil_gateway_controlled", %label);

            // Store label for later reference
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            // Wrap in Option so we can take() on first call
            let node = Arc::new(Mutex::new(Some(node)));
            let user_events = Arc::new(Mutex::new(Some(user_events)));
            let span = Arc::new(Mutex::new(Some(span)));
            let shared_storage = Arc::new(Mutex::new(Some(shared_storage)));
            let contract_stores = contract_stores.clone();
            let label_for_closure = label.clone();

            sim.host(host_name, move || {
                let node = node.clone();
                let user_events = user_events.clone();
                let span = span.clone();
                let shared_storage = shared_storage.clone();
                let contract_stores = contract_stores.clone();
                let label = label_for_closure.clone();

                async move {
                    let node = node
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let user_events = user_events
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let span = span
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let shared_storage = shared_storage
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let cs = contract_stores.lock().unwrap().remove(&label);

                    if use_mock_wasm {
                        node.run_node_with_mock_wasm(user_events, span, shared_storage, cs)
                            .await
                            .map_err(|e| {
                                Box::new(std::io::Error::other(e.to_string()))
                                    as Box<dyn std::error::Error>
                            })
                    } else {
                        node.run_node_with_shared_storage(user_events, span, shared_storage)
                            .await
                            .map_err(|e| {
                                Box::new(std::io::Error::other(e.to_string()))
                                    as Box<dyn std::error::Error>
                            })
                    }
                }
            });
        }

        // Register all regular nodes as Turmoil hosts
        let nodes: Vec<_> = self.nodes.drain(..).collect();
        for (mut node, label) in nodes {
            let host_name = label.to_string();
            let receiver_ch = self.receiver_ch.clone();

            // Capture slot for this node's live Ring (governance ban-list
            // observation post-simulation). See `shared_rings`.
            let ring_slot: Arc<parking_lot::Mutex<Option<Arc<crate::ring::Ring>>>> =
                Arc::new(parking_lot::Mutex::new(None));
            node.shared_ring = Some(ring_slot.clone());
            self.shared_rings.insert(label.clone(), ring_slot);

            // Create shared in-memory storage for this node
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();
            node_storages.insert(label.clone(), shared_storage.clone());

            // Create MemoryEventsGen without RNG (deterministic mode)
            // Clone receiver_ch so each node gets its own subscription
            let mut user_events =
                MemoryEventsGen::new(receiver_ch.clone(), node.config.key_pair.public().clone());

            // Populate events for this node
            if let Some(node_ops) = operations_by_node.remove(&label) {
                let events: Vec<_> = node_ops
                    .into_iter()
                    .map(|(id, op)| (id, op.into_client_request()))
                    .collect();
                user_events.generate_events(events);
            }

            let span = tracing::info_span!("turmoil_node_controlled", %label);

            // Store label for later reference
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            // Wrap in Option so we can take() on first call
            let node = Arc::new(Mutex::new(Some(node)));
            let user_events = Arc::new(Mutex::new(Some(user_events)));
            let span = Arc::new(Mutex::new(Some(span)));
            let shared_storage = Arc::new(Mutex::new(Some(shared_storage)));
            let contract_stores = contract_stores.clone();
            let label_for_closure = label.clone();

            sim.host(host_name, move || {
                let node = node.clone();
                let user_events = user_events.clone();
                let span = span.clone();
                let shared_storage = shared_storage.clone();
                let contract_stores = contract_stores.clone();
                let label = label_for_closure.clone();

                async move {
                    let node = node
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let user_events = user_events
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let span = span
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let shared_storage = shared_storage
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let cs = contract_stores.lock().unwrap().remove(&label);

                    if use_mock_wasm {
                        node.run_node_with_mock_wasm(user_events, span, shared_storage, cs)
                            .await
                            .map_err(|e| {
                                Box::new(std::io::Error::other(e.to_string()))
                                    as Box<dyn std::error::Error>
                            })
                    } else {
                        node.run_node_with_shared_storage(user_events, span, shared_storage)
                            .await
                            .map_err(|e| {
                                Box::new(std::io::Error::other(e.to_string()))
                                    as Box<dyn std::error::Error>
                            })
                    }
                }
            });
        }

        // Apply SeedContract operations: pre-populate specific nodes'
        // contract stores and state storage. This bypasses network PUT
        // so the contract exists only on the seeded nodes.
        for (label, contract, state) in seed_ops {
            let storage = node_storages
                .get(&label)
                .unwrap_or_else(|| panic!("SeedContract: node {label:?} not found in storages"));
            let key = contract.key();
            storage.seed_state(key, WrappedState::new(state));
            storage.seed_params(key, contract.params().clone());
            // Also seed into the contract store (WASM code + params).
            // Without this, the WASM cache check in client_events.rs
            // rejects subscribes with "contract WASM not cached locally".
            let mut stores = contract_stores.lock().unwrap();
            let contract_store = stores.entry(label.clone()).or_default();
            contract_store
                .store_contract(contract.clone())
                .unwrap_or_else(|e| {
                    panic!("SeedContract: failed to store contract for {label:?}: {e}")
                });
            tracing::debug!(
                node = %label,
                contract = %key,
                "SeedContract: pre-populated contract in node storage"
            );
        }

        // Take the event controller and labels for triggering events
        let user_ev_controller = self
            .user_ev_controller
            .take()
            .expect("user_ev_controller should be set");
        let labels: Vec<_> = self.labels.clone();

        // Build a map from NodeLabel to peer key for event triggering
        let label_to_key: HashMap<NodeLabel, _> = labels.into_iter().collect();

        // Clone the network name for the client closure (the outer
        // `network_name` is still needed after `sim.run()` for topology capture).
        let network_name_for_client = network_name.clone();

        // Optional pre-operation join-convergence barrier (see
        // `wait_for_join_convergence_before_ops`). Captured by value into the
        // client closure.
        let wait_for_join_before_ops = self.wait_for_join_before_ops;
        let total_peers_expected = self.number_of_nodes + self.number_of_gateways;
        // Delay between triggering successive regular operations (default 3s).
        let controlled_op_interval = self
            .controlled_op_interval
            .unwrap_or_else(|| Duration::from_secs(3));
        // A peer counts as "joined" for the barrier once it has >= 1 established
        // ring connection. This is a slightly STRONGER condition than the
        // `peer_ready` bar `ensure_peer_ready` checks before letting a peer
        // originate an operation (`peer_ready` is set at handshake completion,
        // just before the connection is added to the ring), so any peer the
        // barrier counts as joined is definitely `peer_ready` — the barrier can
        // only over-wait, never under-wait, and every scheduled GET dispatches
        // rather than being rejected with `PeerNotJoined`. We deliberately do
        // NOT wait for the stronger min_connections/full-ring criterion: a
        // single connection is all a node needs to originate a GET (routing
        // stays correct from a lightly-connected node via the gateway-fallback
        // path), and waiting for a fully-formed ring roughly doubles the
        // barrier's wall-clock for no change in what the metric measures.
        const PEER_READY_MIN_CONNECTIONS: usize = 1;
        let join_conn_threshold = PEER_READY_MIN_CONNECTIONS;

        // Register the test client that triggers controlled events
        sim.client("test", async move {
            // Give nodes time to start and establish connections
            tokio::time::sleep(Duration::from_secs(3)).await;

            // Optional join-convergence barrier: wait until enough peers have
            // finished joining before firing any scheduled operation, so the
            // operations run against a FORMED network instead of racing topology
            // formation. A peer registers a topology snapshot only after its own
            // address is established (`peer_ready`), so the distinct snapshot
            // count is a live join tally. Poll it, advancing virtual time via
            // short sleeps, until the target fraction joins or the cap elapses.
            // Default (`None`) skips this entirely, preserving historical timing.
            if let Some((min_fraction, max_wait)) = wait_for_join_before_ops {
                let target = ((total_peers_expected as f64) * min_fraction).ceil() as usize;
                let poll_interval = Duration::from_secs(5);
                let mut waited = Duration::ZERO;
                loop {
                    // Count peers that have reached the join threshold. Snapshot
                    // *presence* is gated only on the (early) bind address, so it
                    // is NOT a join signal; the stamped `connection_count` is.
                    let joined = get_all_topology_snapshots(&network_name_for_client)
                        .iter()
                        .filter(|s| s.connection_count >= join_conn_threshold)
                        .count();
                    if joined >= target {
                        tracing::info!(
                            joined,
                            target,
                            total = total_peers_expected,
                            waited_secs = waited.as_secs(),
                            "controlled sim: ring join-convergence reached — firing operations"
                        );
                        break;
                    }
                    if waited >= max_wait {
                        tracing::warn!(
                            joined,
                            target,
                            total = total_peers_expected,
                            waited_secs = waited.as_secs(),
                            "controlled sim: join-convergence cap reached before target — \
                             firing operations against a partially-formed network"
                        );
                        break;
                    }
                    tokio::time::sleep(poll_interval).await;
                    waited += poll_interval;
                }
            }

            // Trigger events in the specified order
            for (event_id, node_label) in operation_sequence {
                // Special in-order ops the client performs itself.
                if let Some(special) = special_ops.get(&event_id) {
                    match special {
                        SpecialDispatch::AdvanceHostingClock(duration) => {
                            match hosting_clock.as_ref() {
                                Some(clock) => {
                                    clock.advance_time(*duration);
                                    tracing::info!(
                                        event_id,
                                        advance_ms = duration.as_millis() as u64,
                                        "Advanced controlled hosting clock"
                                    );
                                }
                                None => tracing::warn!(
                                    event_id,
                                    "AdvanceHostingClock scheduled but no controllable \
                                     hosting clock was enabled (call \
                                     enable_hosting_time_control)"
                                ),
                            }
                        }
                        SpecialDispatch::CrashNode(addr) => {
                            if let Some(injector) = crate::node::network_bridge::get_fault_injector(
                                &network_name_for_client,
                            ) {
                                injector.lock().unwrap().config.crash_node(*addr);
                                tracing::info!(
                                    event_id,
                                    node = %node_label,
                                    ?addr,
                                    "Crashed node via fault injector (message-blocking)"
                                );
                            } else {
                                tracing::warn!(
                                    event_id,
                                    node = %node_label,
                                    "CrashNode scheduled but no fault injector registered"
                                );
                            }
                        }
                        SpecialDispatch::RecoverNode(addr) => {
                            if let Some(injector) = crate::node::network_bridge::get_fault_injector(
                                &network_name_for_client,
                            ) {
                                injector.lock().unwrap().config.recover_node(addr);
                                tracing::info!(
                                    event_id,
                                    node = %node_label,
                                    ?addr,
                                    "Recovered node via fault injector (messages flow again)"
                                );
                            } else {
                                tracing::warn!(
                                    event_id,
                                    node = %node_label,
                                    "RecoverNode scheduled but no fault injector registered"
                                );
                            }
                        }
                    }
                    // Let the effect settle before the next event.
                    tokio::time::sleep(Duration::from_secs(3)).await;
                    continue;
                }

                if let Some(peer_key) = label_to_key.get(&node_label) {
                    tracing::info!(
                        event_id,
                        node = %node_label,
                        "Triggering controlled event"
                    );

                    if user_ev_controller
                        .send((event_id, peer_key.clone()))
                        .is_err()
                    {
                        tracing::warn!(
                            event_id,
                            node = %node_label,
                            "Failed to send event signal - receivers dropped"
                        );
                        break;
                    }

                    // Wait for operation to complete before triggering next
                    // (overridable via `with_controlled_op_interval`; default 3s).
                    tokio::time::sleep(controlled_op_interval).await;
                } else {
                    tracing::warn!(
                        event_id,
                        node = %node_label,
                        "No peer key found for node label"
                    );
                }
            }

            // Wait for post-operation processing (orphan recovery, topology stabilization, etc.)
            tracing::info!(
                wait_secs = post_operations_wait.as_secs(),
                "Waiting for post-operation processing"
            );
            tokio::time::sleep(post_operations_wait).await;

            Ok(())
        });

        // Run the simulation
        let turmoil_result = sim.run();

        // Capture topology snapshots BEFORE self is dropped (which clears them)
        let topology_snapshots = get_all_topology_snapshots(&network_name);

        // Capture renewal metrics BEFORE self is dropped — `SimNetwork::Drop`
        // calls `clear_renewal_metrics(&self.name)`, so reading the registry
        // after this function returns would always see an empty map (#4440).
        let renewal_metrics =
            crate::ring::topology_registry::get_all_renewal_metrics(&network_name);

        // Capture WASM-summarize counts BEFORE self drops (Drop clears the
        // registry), same lifetime constraint as the renewal metrics above. The
        // every-hop summarize-storm falsifier reads these after the run returns.
        let summarize_wasm_calls =
            crate::ring::topology_registry::get_all_summarize_wasm_calls(&network_name);

        // Capture the crash-drop count BEFORE self drops (Drop clears the fault
        // injector via `set_fault_injector(None)`). `> 0` proves a scripted
        // `CrashNode` actually blocked traffic — the discriminating signal for
        // piece-F crash tests.
        let crash_packets_dropped = crate::node::network_bridge::get_fault_injector(&network_name)
            .map(|inj| inj.lock().unwrap().stats.messages_dropped_crash)
            .unwrap_or(0);

        // Extract the live Rings captured during the run BEFORE self drops.
        let node_rings: HashMap<NodeLabel, Arc<crate::ring::Ring>> = self
            .shared_rings
            .iter()
            .filter_map(|(label, slot)| slot.lock().clone().map(|ring| (label.clone(), ring)))
            .collect();

        ControlledSimulationResult {
            turmoil_result,
            topology_snapshots,
            node_storages,
            node_rings,
            renewal_metrics,
            crash_packets_dropped,
            summarize_wasm_calls,
        }
    }

    /// Run an fdev-style test using Turmoil's deterministic executor.
    ///
    /// This method provides the same functionality as the old `start_with_rand_gen`
    /// approach but runs everything inside Turmoil for full determinism.
    ///
    /// Returns `Ok(())` on success, or an error if the test failed.
    #[cfg(any(test, feature = "testing"))]
    pub fn run_fdev_test<R>(
        mut self,
        seed: u64,
        max_contract_num: usize,
        iterations: usize,
        simulation_duration: Duration,
        event_wait: Duration,
    ) -> anyhow::Result<()>
    where
        R: RandomEventGenerator + Send + 'static,
    {
        use crate::config::{GlobalRng, GlobalSimulationTime};
        use crate::ring::topology_registry::set_current_network_name;
        use std::sync::Mutex;

        // Set up deterministic RNG and time for reproducible simulation
        GlobalRng::set_seed(seed);

        // Derive simulation epoch from seed for deterministic ULID generation
        const BASE_EPOCH_MS: u64 = 1577836800000; // 2020-01-01 00:00:00 UTC
        const RANGE_MS: u64 = 5 * 365 * 24 * 60 * 60 * 1000; // ~5 years in ms
        let epoch_offset = seed % RANGE_MS;
        GlobalSimulationTime::set_time_ms(BASE_EPOCH_MS + epoch_offset);

        // Set the current network name for topology registration
        set_current_network_name(&self.name);

        // Build Turmoil simulation with seeded RNG for deterministic execution
        let mut sim = turmoil::Builder::new()
            .simulation_duration(simulation_duration)
            .rng_seed(seed)
            .build();

        // Get total peer count for event generation
        let total_peer_num = self.gateways.len() + self.nodes.len();

        // Register all gateways as Turmoil hosts
        let gateways: Vec<_> = self.gateways.drain(..).collect();
        for (node, config) in gateways {
            let label = config.label.clone();
            let host_name = label.to_string();
            let receiver_ch = self.receiver_ch.clone();

            // Create shared in-memory storage for this node
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();

            let mut user_events = MemoryEventsGen::<R>::new_with_seed(
                receiver_ch,
                node.config.key_pair.public().clone(),
                seed,
            );
            user_events.rng_params(label.number(), total_peer_num, max_contract_num, iterations);

            let span = tracing::info_span!("turmoil_gateway", %label);

            // Store label for later reference
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            // Wrap in Option so we can take() on first call
            let node = Arc::new(Mutex::new(Some(node)));
            let user_events = Arc::new(Mutex::new(Some(user_events)));
            let span = Arc::new(Mutex::new(Some(span)));
            let shared_storage = Arc::new(Mutex::new(Some(shared_storage)));

            sim.host(host_name, move || {
                let node = node.clone();
                let user_events = user_events.clone();
                let span = span.clone();
                let shared_storage = shared_storage.clone();

                async move {
                    let node = node
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let user_events = user_events
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let span = span
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let shared_storage = shared_storage
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");

                    node.run_node_with_shared_storage(user_events, span, shared_storage)
                        .await
                        .map_err(|e| {
                            Box::new(std::io::Error::other(e.to_string()))
                                as Box<dyn std::error::Error>
                        })
                }
            });
        }

        // Register all regular nodes as Turmoil hosts
        let nodes: Vec<_> = self.nodes.drain(..).collect();
        for (node, label) in nodes {
            let host_name = label.to_string();
            let receiver_ch = self.receiver_ch.clone();

            // Create shared in-memory storage for this node
            let shared_storage = crate::wasm_runtime::MockStateStorage::new();

            let mut user_events = MemoryEventsGen::<R>::new_with_seed(
                receiver_ch,
                node.config.key_pair.public().clone(),
                seed,
            );
            user_events.rng_params(label.number(), total_peer_num, max_contract_num, iterations);

            let span = tracing::info_span!("turmoil_node", %label);

            // Store label for later reference
            self.labels
                .push((label.clone(), node.config.key_pair.public().clone()));

            // Wrap in Option so we can take() on first call
            let node = Arc::new(Mutex::new(Some(node)));
            let user_events = Arc::new(Mutex::new(Some(user_events)));
            let span = Arc::new(Mutex::new(Some(span)));
            let shared_storage = Arc::new(Mutex::new(Some(shared_storage)));

            sim.host(host_name, move || {
                let node = node.clone();
                let user_events = user_events.clone();
                let span = span.clone();
                let shared_storage = shared_storage.clone();

                async move {
                    let node = node
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let user_events = user_events
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let span = span
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");
                    let shared_storage = shared_storage
                        .lock()
                        .unwrap()
                        .take()
                        .expect("Turmoil host should only be called once");

                    node.run_node_with_shared_storage(user_events, span, shared_storage)
                        .await
                        .map_err(|e| {
                            Box::new(std::io::Error::other(e.to_string()))
                                as Box<dyn std::error::Error>
                        })
                }
            });
        }

        // Take the event controller and labels for triggering events
        let user_ev_controller = self
            .user_ev_controller
            .take()
            .expect("user_ev_controller should be set");
        let labels: Vec<_> = self.labels.clone();

        // Clone event logs handle for convergence checking
        let event_logs = self.event_listener.logs.clone();

        // Register the test function as a Turmoil client
        sim.client("test", async move {
            // Give nodes time to start and establish connections
            tokio::time::sleep(Duration::from_secs(2)).await;

            // Use a seeded RNG for deterministic peer selection
            use rand::SeedableRng;
            use rand::prelude::*;
            let mut event_rng = <rand::rngs::SmallRng as SeedableRng>::seed_from_u64(seed);

            // Trigger events by sending signals to peer event generators
            for event_id in 0..iterations as u32 {
                // Pick a random peer to generate an event
                if let Some((_, peer_key)) = labels.choose(&mut event_rng) {
                    if user_ev_controller
                        .send((event_id, peer_key.clone()))
                        .is_err()
                    {
                        tracing::warn!(event_id, "Failed to send event signal - receivers dropped");
                        break;
                    }
                }

                // Wait between events (Turmoil handles this deterministically)
                tokio::time::sleep(event_wait).await;
            }

            // Wait for events to fully propagate through the network
            tokio::time::sleep(Duration::from_secs(2)).await;

            // Convergence checking happens here, inside Turmoil
            // Count subscribed contracts from event logs
            let subscribed_count = {
                use std::collections::HashSet;
                let logs = event_logs.lock().await;
                let mut subscribed_contracts: HashSet<String> = HashSet::new();
                for log in logs.iter() {
                    if let crate::tracing::EventKind::Subscribe(
                        crate::tracing::SubscribeEvent::SubscribeSuccess { key, .. },
                    ) = &log.kind
                    {
                        subscribed_contracts.insert(format!("{:?}", key));
                    }
                }
                subscribed_contracts.len()
            };

            if subscribed_count > 0 {
                tracing::info!(
                    "Found {} subscribed contracts, checking convergence...",
                    subscribed_count
                );

                // Simple convergence check within Turmoil
                // We'll just wait a bit and then check final state
                tokio::time::sleep(Duration::from_secs(10)).await;

                // TODO: Add actual convergence checking here if needed
                // For now, we just let the test complete
            }

            Ok(())
        });

        // Run the simulation
        sim.run()
            .map_err(|e| anyhow::anyhow!("Turmoil simulation failed: {:?}", e))
    }

    /// Run a simulation using a single `current_thread` tokio runtime with paused time.
    ///
    /// This avoids Turmoil's O(n²) link overhead and O(n³) per-tick cost, making it
    /// feasible to simulate hundreds of nodes. Determinism is achieved via:
    /// - `current_thread` runtime (single-threaded, no scheduling races)
    /// - `start_paused(true)` (tokio auto-advances time deterministically)
    /// - `GlobalRng::set_seed` (seeded randomness)
    /// - `deterministic_select!` (ordered select branches)
    /// - `SimulationSocket` with BTreeMap (deterministic address ordering)
    #[cfg(any(test, feature = "testing"))]
    pub fn run_simulation_direct<R>(
        mut self,
        seed: u64,
        max_contract_num: usize,
        iterations: usize,
        event_wait: Duration,
    ) -> anyhow::Result<()>
    where
        R: RandomEventGenerator + Send + 'static,
    {
        use crate::config::{GlobalRng, GlobalSimulationTime, SimulationTransportOpt};
        use crate::ring::topology_registry::set_current_network_name;

        // Set up deterministic RNG and time for reproducible simulation
        GlobalRng::set_seed(seed);

        const BASE_EPOCH_MS: u64 = 1577836800000; // 2020-01-01 00:00:00 UTC
        const RANGE_MS: u64 = 5 * 365 * 24 * 60 * 60 * 1000; // ~5 years in ms
        let epoch_offset = seed % RANGE_MS;
        GlobalSimulationTime::set_time_ms(BASE_EPOCH_MS + epoch_offset);

        set_current_network_name(&self.name);

        // Make direct-runner ChurnConfig crashes REAL packet drops (#4694).
        //
        // The chaos driver below marks nodes crashed in this network's fault
        // injector, but a crash only drops packets if the global packet-delivery
        // callback is installed AND this network opted in via
        // `enforce_fault_drops`. Neither happened on the direct runner, so churn
        // faults were inert: a "crashed" node kept exchanging packets and every
        // near-K churn / partition validation on this runner was false-green
        // (#4694; blocker for the demand-driven-hosting release-2 gate, #4642).
        //
        // Gated on `churn_config.is_some()` so non-churn direct-runner tests are
        // byte-for-byte unchanged (the callback is inert unless a node is
        // actually crashed/partitioned, but gating keeps the change surgical and
        // makes the wiring impossible to miss when churn IS configured).
        // `SimNetwork::Drop` clears the global callback.
        if self.churn_config.is_some() {
            crate::transport::in_memory_socket::set_packet_delivery_callback(Some(
                std::sync::Arc::new(fault_injection_delivery_decision),
            ));
            if let Some(injector) = crate::node::network_bridge::get_fault_injector(&self.name) {
                injector.lock().unwrap().enforce_fault_drops = true;
            }
        }

        // Single-threaded runtime with paused time for deterministic execution
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .start_paused(true)
            .build()?;

        let total_peer_num = self.gateways.len() + self.nodes.len();

        // Extend connection idle timeout for ALL simulations. In start_paused(true)
        // mode, virtual time jumps past 120s during spawn_blocking (WASM execution),
        // causing spurious connection drops even with keepalive enabled.
        crate::config::SimulationIdleTimeout::enable();

        // Relax transport timers for ALL direct-runner simulations. Disables keepalive
        // and uses 5x slower ACK/resend/rate-update intervals. In start_paused(true)
        // mode, production-rate timer firings create excessive virtual time advances
        // that interfere with heartbeat and broadcast scheduling, causing
        // non-deterministic convergence failures even in small networks.
        SimulationTransportOpt::enable();
        let use_mock_wasm = self.use_mock_wasm;
        let skip_convergence_wait = self.skip_convergence_wait;

        let result: anyhow::Result<()> = rt.block_on(async {
            // Time driver: bridges tokio's paused time → VirtualTime
            // When all tasks are idle, tokio auto-advances past the 1ms sleep,
            // which wakes this driver, which advances VirtualTime, which wakes
            // VirtualSleep futures in node tasks.
            let vt = self.virtual_time.clone();
            let time_driver = tokio::spawn(async move {
                let start = tokio::time::Instant::now();
                loop {
                    tokio::time::sleep(Duration::from_millis(1)).await;
                    vt.advance_to(start.elapsed().as_nanos() as u64);
                }
            });

            // Shared state for chaos driver to crash/restart nodes
            let direct_nodes: Arc<tokio::sync::Mutex<HashMap<NodeLabel, DirectNodeState>>> =
                Arc::new(tokio::sync::Mutex::new(HashMap::new()));

            // Spawn gateway nodes
            let mut node_handles = Vec::new();

            let gateways: Vec<_> = self.gateways.drain(..).collect();
            for (node, config) in gateways {
                let label = config.label.clone();
                let receiver_ch = self.receiver_ch.clone();
                let addr = *self.node_addresses.get(&label).expect("gateway address");

                let shared_storage = crate::wasm_runtime::MockStateStorage::new();

                let mut user_events = MemoryEventsGen::<R>::new_with_seed(
                    receiver_ch,
                    node.config.key_pair.public().clone(),
                    seed,
                );
                user_events.rng_params(
                    label.number(),
                    total_peer_num,
                    max_contract_num,
                    iterations,
                );

                let span = tracing::info_span!("direct_gateway", %label);

                self.labels
                    .push((label.clone(), node.config.key_pair.public().clone()));

                let handle = if use_mock_wasm {
                    tokio::spawn(async move {
                        node.run_node_with_mock_wasm(user_events, span, shared_storage, None)
                            .await
                    })
                } else {
                    tokio::spawn(async move {
                        node.run_node_with_shared_storage(user_events, span, shared_storage)
                            .await
                    })
                };

                {
                    let mut nodes_map = direct_nodes.lock().await;
                    nodes_map.insert(
                        label.clone(),
                        DirectNodeState {
                            label: label.clone(),
                            addr,
                            is_gateway: true,
                            permanently_dropped: false,
                        },
                    );
                }

                node_handles.push(handle);
            }

            // Spawn regular nodes with staggered start
            let nodes: Vec<_> = self.nodes.drain(..).collect();
            for (i, (node, label)) in nodes.into_iter().enumerate() {
                let receiver_ch = self.receiver_ch.clone();
                let addr = *self.node_addresses.get(&label).expect("node address");

                let shared_storage = crate::wasm_runtime::MockStateStorage::new();

                let mut user_events = MemoryEventsGen::<R>::new_with_seed(
                    receiver_ch,
                    node.config.key_pair.public().clone(),
                    seed,
                );
                user_events.rng_params(
                    label.number(),
                    total_peer_num,
                    max_contract_num,
                    iterations,
                );

                let span = tracing::info_span!("direct_node", %label);

                self.labels
                    .push((label.clone(), node.config.key_pair.public().clone()));

                let backoff = self.start_backoff * (i as u32 + 1);
                let handle = if use_mock_wasm {
                    tokio::spawn(async move {
                        tokio::time::sleep(backoff).await;
                        node.run_node_with_mock_wasm(user_events, span, shared_storage, None)
                            .await
                    })
                } else {
                    tokio::spawn(async move {
                        tokio::time::sleep(backoff).await;
                        node.run_node_with_shared_storage(user_events, span, shared_storage)
                            .await
                    })
                };

                {
                    let mut nodes_map = direct_nodes.lock().await;
                    nodes_map.insert(
                        label.clone(),
                        DirectNodeState {
                            label: label.clone(),
                            addr,
                            is_gateway: false,
                            permanently_dropped: false,
                        },
                    );
                }

                node_handles.push(handle);
            }

            // Chaos driver: periodically crash/recover nodes via fault injection.
            //
            // Uses "soft crash" via the fault injector (blocks all messages to/from
            // a node) rather than aborting tasks. This avoids orphaned sub-tasks
            // that would stall tokio's time auto-advance in start_paused(true).
            // The node's event loop keeps running but is functionally dead since
            // all its messages are dropped.
            let chaos_driver = if let Some(churn_config) = self.churn_config.clone() {
                use crate::node::network_bridge::get_fault_injector;
                use rand::SeedableRng;
                use rand::prelude::*;

                let chaos_nodes = direct_nodes.clone();
                let network_name = self.name.clone();
                let non_gateway_count = self.number_of_nodes;
                let max_crashes = churn_config
                    .max_simultaneous_crashes
                    .unwrap_or((non_gateway_count / 4).max(1));

                Some(tokio::spawn(async move {
                    let mut chaos_rng = <rand::rngs::SmallRng as SeedableRng>::seed_from_u64(
                        seed.wrapping_add(0xC1_0055_DEAD),
                    );

                    // Wait for warmup before starting churn
                    tokio::time::sleep(churn_config.warmup_delay).await;

                    // Track which nodes are currently crashed (pending recovery)
                    let mut pending_recovery: Vec<(NodeLabel, tokio::time::Instant)> = Vec::new();
                    let mut total_crashes: usize = 0;
                    let mut total_recoveries: usize = 0;
                    let mut total_permanent: usize = 0;

                    loop {
                        tokio::time::sleep(churn_config.tick_interval).await;

                        // Phase 1: Recover nodes whose recovery delay has elapsed
                        let now = tokio::time::Instant::now();
                        let mut recovered = Vec::new();
                        pending_recovery.retain(|(label, crash_time)| {
                            if now.duration_since(*crash_time) >= churn_config.recovery_delay {
                                recovered.push(label.clone());
                                false
                            } else {
                                true
                            }
                        });

                        for label in recovered {
                            let nodes_map = chaos_nodes.lock().await;
                            let state = match nodes_map.get(&label) {
                                Some(s) if !s.permanently_dropped => s,
                                _ => continue,
                            };
                            let addr = state.addr;
                            drop(nodes_map);

                            // Unblock messages — node resumes normal operation
                            if let Some(injector) = get_fault_injector(&network_name) {
                                let mut inj = injector.lock().unwrap();
                                inj.config.recover_node(&addr);
                            }

                            total_recoveries += 1;
                            tracing::info!(
                                ?label,
                                ?addr,
                                total_recoveries,
                                "Chaos driver: node recovered"
                            );
                        }

                        // Phase 2: Crash eligible nodes
                        let mut currently_crashed = pending_recovery.len();
                        if currently_crashed >= max_crashes {
                            continue;
                        }

                        let nodes_map = chaos_nodes.lock().await;
                        let eligible: Vec<(NodeLabel, SocketAddr)> = nodes_map
                            .values()
                            .filter(|s| {
                                !s.is_gateway
                                    && !s.permanently_dropped
                                    && !pending_recovery.iter().any(|(l, _)| l == &s.label)
                            })
                            .map(|s| (s.label.clone(), s.addr))
                            .collect();
                        drop(nodes_map);

                        for (label, addr) in eligible {
                            if currently_crashed >= max_crashes {
                                break;
                            }
                            if !chaos_rng.random_bool(churn_config.crash_probability) {
                                continue;
                            }

                            let is_permanent =
                                chaos_rng.random_bool(churn_config.permanent_crash_rate);

                            // Block all messages to/from this node
                            if let Some(injector) = get_fault_injector(&network_name) {
                                let mut inj = injector.lock().unwrap();
                                inj.config.crash_node(addr);
                            }

                            total_crashes += 1;

                            if is_permanent {
                                let mut nodes_map = chaos_nodes.lock().await;
                                if let Some(state) = nodes_map.get_mut(&label) {
                                    state.permanently_dropped = true;
                                }
                                total_permanent += 1;
                                tracing::info!(
                                    ?label,
                                    ?addr,
                                    total_permanent,
                                    "Chaos driver: node permanently crashed"
                                );
                            } else {
                                pending_recovery.push((label.clone(), tokio::time::Instant::now()));
                                currently_crashed += 1;
                                tracing::info!(
                                    ?label,
                                    ?addr,
                                    total_crashes,
                                    "Chaos driver: node crashed (will recover)"
                                );
                            }
                        }
                    }
                }))
            } else {
                None
            };

            // Event driver: identical logic to run_fdev_test's client
            let user_ev_controller = self
                .user_ev_controller
                .take()
                .expect("user_ev_controller should be set");
            let labels: Vec<_> = self.labels.clone();
            let event_logs = self.event_listener.logs.clone();

            // Give nodes time to start and establish connections
            tokio::time::sleep(Duration::from_secs(2)).await;

            // Use a seeded RNG for deterministic peer selection
            use rand::SeedableRng;
            use rand::prelude::*;
            let mut event_rng = <rand::rngs::SmallRng as SeedableRng>::seed_from_u64(seed);

            for event_id in 0..iterations as u32 {
                if let Some((_, peer_key)) = labels.choose(&mut event_rng) {
                    if user_ev_controller
                        .send((event_id, peer_key.clone()))
                        .is_err()
                    {
                        tracing::warn!(event_id, "Failed to send event signal - receivers dropped");
                        break;
                    }
                }
                tokio::time::sleep(event_wait).await;
            }

            // Wait for convergence with interest-sync repair.
            //
            // Poll convergence periodically while advancing virtual time.
            // Each poll interval lets heartbeat timers fire and broadcast
            // processing complete. The heartbeat interval is 300s, so
            // 1800s covers six full cycles — enough for cascading repair
            // broadcasts to propagate even when multiple state versions
            // compete across multi-hop topologies.
            //
            // Early exit: if all contracts converge, we break immediately.
            //
            // When `skip_convergence_wait` is set (TestConfig::no_convergence_wait()),
            // we do only a brief fixed settle and skip the polling loop entirely. The
            // polling tail is advisory — it only logs on failure, never fails the
            // simulation — so a test that never asserts convergence gains nothing from
            // it but pays up to 1800s of virtual time when the network happens not to
            // converge, which is exactly how test_interest_renewal timed out in CI (#3792).
            tokio::time::sleep(Duration::from_secs(15)).await;

            if skip_convergence_wait {
                tracing::info!(
                    "Skipping convergence-polling tail (require_convergence = false); \
                     brief settle only"
                );
            } else {
                let converged = 'convergence: {
                    for round in 0..30u32 {
                        let result = check_convergence_from_logs(&event_logs).await;
                        let total = result.converged.len() + result.diverged.len();
                        if total > 0 && result.diverged.is_empty() {
                            tracing::info!(
                                converged = result.converged.len(),
                                round,
                                "Convergence achieved during propagation"
                            );
                            break 'convergence true;
                        }
                        tracing::debug!(
                            converged = result.converged.len(),
                            diverged = result.diverged.len(),
                            round,
                            "Convergence not yet achieved, waiting..."
                        );
                        tokio::time::sleep(Duration::from_secs(60)).await;
                    }
                    false
                };
                if !converged {
                    tracing::warn!(
                        "Propagation timeout — convergence not achieved after 30 rounds (1800s)"
                    );
                }
            }

            // Shutdown: abort chaos driver, time driver, then check node tasks
            if let Some(chaos) = chaos_driver {
                chaos.abort();
            }
            time_driver.abort();

            let mut first_error: Option<anyhow::Error> = None;
            for handle in node_handles {
                handle.abort();
                match handle.await {
                    // Node was aborted (normal shutdown) — ignore
                    Err(e) if e.is_cancelled() => {}
                    // Node panicked
                    Err(e) => {
                        let msg = format!("Node task panicked: {e}");
                        tracing::error!("{}", msg);
                        if first_error.is_none() {
                            first_error = Some(anyhow::anyhow!("{}", msg));
                        }
                    }
                    // Node returned an error before we aborted it
                    Ok(Err(e)) => {
                        tracing::error!("Node task failed: {e}");
                        if first_error.is_none() {
                            first_error = Some(e);
                        }
                    }
                    // Node completed successfully (unlikely — event loops run forever)
                    Ok(Ok(())) => {}
                }
            }

            if let Some(e) = first_error {
                return Err(e);
            }

            info!("Direct simulation completed successfully");
            Ok(())
        });

        result
    }

    // =========================================================================
    // Subscription Topology Validation
    // =========================================================================

    /// Get the network name for this simulation.
    pub fn network_name(&self) -> &str {
        &self.name
    }

    /// Get all subscription topology snapshots registered for this network.
    ///
    /// Returns snapshots registered by nodes via the topology registry.
    /// Call this after nodes have been running to see their subscription state.
    pub fn get_topology_snapshots(&self) -> Vec<crate::ring::topology_registry::TopologySnapshot> {
        crate::ring::topology_registry::get_all_topology_snapshots(&self.name)
    }

    /// Validate subscription topology for a specific contract.
    ///
    /// Checks for:
    /// - Bidirectional cycles that create isolated islands
    /// - Orphan hosters without recovery paths
    /// - Unreachable hosters
    /// - Proximity violations in upstream selection
    ///
    /// Returns a validation result with any issues found.
    pub fn validate_subscription_topology(
        &self,
        contract_id: &freenet_stdlib::prelude::ContractInstanceId,
        contract_location: f64,
    ) -> crate::ring::topology_registry::TopologyValidationResult {
        crate::ring::topology_registry::validate_topology(
            &self.name,
            contract_id,
            contract_location,
        )
    }

    /// Clear all topology snapshots for this network.
    ///
    /// Call this at the start of a test or after topology validation
    /// to reset the state.
    pub fn clear_topology_snapshots(&self) {
        crate::ring::topology_registry::clear_topology_snapshots(&self.name);
    }

    /// Assert that subscription topology is healthy for a contract.
    ///
    /// # Panics
    /// Panics if any topology issues are detected.
    pub fn assert_topology_healthy(
        &self,
        contract_id: &freenet_stdlib::prelude::ContractInstanceId,
        contract_location: f64,
    ) {
        let result = self.validate_subscription_topology(contract_id, contract_location);

        if !result.is_healthy() {
            let mut issues = Vec::new();

            if !result.bidirectional_cycles.is_empty() {
                issues.push(format!(
                    "ISSUE #2720: {} bidirectional cycles found: {:?}",
                    result.bidirectional_cycles.len(),
                    result.bidirectional_cycles
                ));
            }

            if !result.orphan_hosters.is_empty() {
                issues.push(format!(
                    "ISSUE #2719: {} orphan hosters found: {:?}",
                    result.orphan_hosters.len(),
                    result.orphan_hosters
                ));
            }

            if !result.unreachable_hosters.is_empty() {
                issues.push(format!(
                    "ISSUE #2720: {} unreachable hosters found: {:?}",
                    result.unreachable_hosters.len(),
                    result.unreachable_hosters
                ));
            }

            if !result.proximity_violations.is_empty() {
                issues.push(format!(
                    "ISSUE #2721: {} proximity violations found",
                    result.proximity_violations.len()
                ));
            }

            panic!(
                "Subscription topology has {} issues:\n{}",
                result.issue_count,
                issues.join("\n")
            );
        }
    }
}

/// Result of a convergence check.
#[derive(Debug, Clone)]
pub struct ConvergenceResult {
    /// Contracts where all replicas have the same state hash.
    pub converged: Vec<ConvergedContract>,
    /// Contracts where replicas have different state hashes.
    pub diverged: Vec<DivergedContract>,
}

impl ConvergenceResult {
    /// Returns the total number of contracts checked.
    pub fn total_contracts(&self) -> usize {
        self.converged.len() + self.diverged.len()
    }

    /// Returns the convergence rate as a ratio (0.0 to 1.0).
    pub fn rate(&self) -> f64 {
        if self.total_contracts() == 0 {
            return 1.0;
        }
        self.converged.len() as f64 / self.total_contracts() as f64
    }

    /// Returns true if all replicated contracts have converged.
    pub fn is_converged(&self) -> bool {
        self.diverged.is_empty()
    }
}

/// A contract that has converged (all replicas have the same state).
#[derive(Debug, Clone)]
pub struct ConvergedContract {
    /// The contract key
    pub contract_key: String,
    /// The state hash that all replicas have
    pub state_hash: String,
    /// Number of replicas that have this state
    pub replica_count: usize,
}

/// A contract that has diverged (replicas have different states).
#[derive(Debug, Clone)]
pub struct DivergedContract {
    /// The contract key
    pub contract_key: String,
    /// Map of peer address to their state hash
    pub peer_states: Vec<(SocketAddr, String)>,
}

impl DivergedContract {
    /// Returns the number of unique states across replicas.
    pub fn unique_state_count(&self) -> usize {
        let unique: HashSet<&String> = self.peer_states.iter().map(|(_, h)| h).collect();
        unique.len()
    }
}

// =============================================================================
// Gap 4: Contract Distribution Types
// =============================================================================

/// Information about how a contract is distributed across the network.
#[derive(Debug, Clone)]
pub struct ContractDistribution {
    /// The contract key
    pub contract_key: String,
    /// Number of replicas
    pub replica_count: usize,
    /// Peers that have this contract
    pub peers: Vec<SocketAddr>,
}

// =============================================================================
// Gap T4: Operation Tracking Types
// =============================================================================

/// Summary of operation completion status across the network.
#[derive(Debug, Clone, Default)]
pub struct OperationSummary {
    /// Put operation statistics
    pub put: PutOperationStats,
    /// Get operation statistics
    pub get: OperationStats,
    /// Subscribe operation statistics
    pub subscribe: OperationStats,
    /// Update operation statistics
    pub update: UpdateOperationStats,
    /// Total number of timed-out operations
    pub timeouts: usize,
}

impl OperationSummary {
    /// Returns total number of operations requested.
    pub fn total_requested(&self) -> usize {
        self.put.requested + self.get.requested + self.subscribe.requested + self.update.requested
    }

    /// Returns total number of operations completed (succeeded + failed).
    pub fn total_completed(&self) -> usize {
        self.put.completed()
            + self.get.completed()
            + self.subscribe.completed()
            + self.update.completed()
    }

    /// Returns total number of successful operations.
    pub fn total_succeeded(&self) -> usize {
        self.put.succeeded + self.get.succeeded + self.subscribe.succeeded + self.update.succeeded
    }

    /// Returns total number of failed operations.
    pub fn total_failed(&self) -> usize {
        self.put.failed + self.get.failed + self.subscribe.failed + self.update.failed
    }

    /// Returns overall success rate (0.0 to 1.0).
    /// Includes timeouts as failed operations.
    pub fn overall_success_rate(&self) -> f64 {
        let completed = self.total_completed() + self.timeouts;
        if completed == 0 {
            return 1.0; // No operations completed yet
        }
        self.total_succeeded() as f64 / completed as f64
    }

    /// Returns true if all requested operations have completed.
    pub fn all_completed(&self) -> bool {
        self.total_completed() >= self.total_requested()
    }
}

/// Statistics for a basic operation type (Get, Subscribe).
#[derive(Debug, Clone, Default)]
pub struct OperationStats {
    /// Number of operations requested
    pub requested: usize,
    /// Number of operations that succeeded
    pub succeeded: usize,
    /// Number of operations that failed
    pub failed: usize,
}

impl OperationStats {
    /// Returns total completed operations (succeeded + failed).
    pub fn completed(&self) -> usize {
        self.succeeded + self.failed
    }

    /// Returns success rate (0.0 to 1.0).
    pub fn success_rate(&self) -> f64 {
        let completed = self.completed();
        if completed == 0 {
            return 1.0;
        }
        self.succeeded as f64 / completed as f64
    }
}

/// Statistics for Put operations (includes broadcast tracking).
#[derive(Debug, Clone, Default)]
pub struct PutOperationStats {
    /// Number of Put operations requested
    pub requested: usize,
    /// Number of Put operations that succeeded
    pub succeeded: usize,
    /// Number of Put operations that failed
    pub failed: usize,
    /// Number of broadcasts emitted
    pub broadcasts_emitted: usize,
    /// Number of broadcasts received by peers
    pub broadcasts_received: usize,
}

impl PutOperationStats {
    /// Returns total completed operations (succeeded + failed).
    pub fn completed(&self) -> usize {
        self.succeeded + self.failed
    }

    /// Returns success rate (0.0 to 1.0).
    pub fn success_rate(&self) -> f64 {
        let completed = self.completed();
        if completed == 0 {
            return 1.0;
        }
        self.succeeded as f64 / completed as f64
    }

    /// Returns broadcast propagation rate (received / emitted).
    pub fn broadcast_propagation_rate(&self) -> f64 {
        if self.broadcasts_emitted == 0 {
            return 1.0;
        }
        self.broadcasts_received as f64 / self.broadcasts_emitted as f64
    }
}

/// Statistics for Update operations (includes broadcast tracking).
#[derive(Debug, Clone, Default)]
pub struct UpdateOperationStats {
    /// Number of Update operations requested
    pub requested: usize,
    /// Number of Update operations that succeeded
    pub succeeded: usize,
    /// Number of Update operations that failed
    pub failed: usize,
    /// Number of update broadcasts emitted
    pub broadcasts_emitted: usize,
    /// Number of update broadcasts received by subscribers
    pub broadcasts_received: usize,
}

impl UpdateOperationStats {
    /// Returns total completed operations (succeeded + failed).
    pub fn completed(&self) -> usize {
        self.succeeded + self.failed
    }

    /// Returns success rate (0.0 to 1.0).
    pub fn success_rate(&self) -> f64 {
        let completed = self.completed();
        if completed == 0 {
            return 1.0;
        }
        self.succeeded as f64 / completed as f64
    }
}

#[cfg(any(debug_assertions, test))]
impl std::fmt::Debug for SimNetwork {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SimNetwork")
            .field("name", &self.name)
            .field("labels", &self.labels)
            .field("number_of_gateways", &self.number_of_gateways)
            .field("number_of_nodes", &self.number_of_nodes)
            .field("ring_max_htl", &self.ring_max_htl)
            .field("rnd_if_htl_above", &self.rnd_if_htl_above)
            .field("max_connections", &self.max_connections)
            .field("min_connections", &self.min_connections)
            .field("init_backoff", &self.start_backoff)
            .finish()
    }
}

impl Drop for SimNetwork {
    fn drop(&mut self) {
        use crate::node::network_bridge::set_fault_injector;
        use crate::ring::topology_registry::{
            clear_current_network_name, clear_renewal_metrics, clear_summarize_metrics,
            clear_topology_snapshots,
        };
        use crate::transport::in_memory_socket::{
            clear_network_address_mappings, remove_network_socket_registry,
            set_packet_delivery_callback, set_queue_packet_callback,
        };

        // Per-network cleanup (safe for parallel tests)
        set_fault_injector(&self.name, None);
        unregister_network_time_source(&self.name);
        clear_topology_snapshots(&self.name);
        clear_renewal_metrics(&self.name);
        clear_summarize_metrics(&self.name);
        remove_network_socket_registry(&self.name);
        clear_network_address_mappings(&self.name);

        // Clear global callbacks to prevent stale references between
        // sequential simulation runs (e.g., determinism tests).
        set_packet_delivery_callback(None);
        set_queue_packet_callback(None);

        // Thread-local cleanup
        clear_current_network_name();

        if self.clean_up_tmp_dirs {
            clean_up_tmp_dirs(self.labels.iter().map(|(l, _)| l));
        }
    }
}

/// Wait for a spawned node task to publish its `ConnectionManager` (or any
/// other value) to the shared slot, then take it.
///
/// Returns `None` only if the spawned task fails to publish within
/// `MAX_PUBLISH_POLLS` yield passes, which in practice means it either
/// panicked before reaching its publish point or is wedged in a loop before
/// its first `.await`. The caller (`SimNetwork::start_*`) treats that as a
/// hard failure and panics with the node label, because downstream sim
/// helpers like `sim.connection_count(label)` would otherwise silently
/// return `None` for that node and confuse every assertion that depends on
/// the full node set.
///
/// ## Rationale
///
/// The previous implementation slept for `start_backoff` (50ms in the
/// nightly fault-recovery test) and then called `take()` once. On a busy
/// runner the spawned node task had not yet been polled far enough to reach
/// the `shared_cm` write at the top of `run_node_with_shared_storage`, so
/// `take()` returned `None` and the CM was silently dropped. That was the
/// deterministic "46/50 connection managers" failure in
/// `test_nightly_fault_recovery_speed`.
///
/// ## Mechanism
///
/// 1. Honor the configured `start_backoff` first so staggered-startup timing
///    is preserved for tests that depend on it.
/// 2. Try `take()` once — this covers the fast path where the node task was
///    already polled during the backoff sleep.
/// 3. If the slot is still empty, yield to the scheduler via
///    `tokio::task::yield_now()` and re-check, bounded by an iteration
///    budget. The publish happens synchronously before any `.await` in
///    `run_node_with_shared_storage`, so once the spawned task is polled
///    once past the entry point the slot is populated; we only need enough
///    yields to give the scheduler a chance to choose it. The bound is an
///    iteration count rather than a wall-clock deadline so the helper
///    behaves identically under real and paused tokio runtimes (the direct
///    simulation runner uses `tokio::time::start_paused(true)`, which makes
///    any `tokio::time::Instant`-based deadline auto-advance virtual time
///    while real scheduling has not progressed). It also keeps us out of
///    the DST rule that bans `std::time::Instant::now()` in `crates/core/`.
///
/// Generic over the slot payload so unit tests can drive the helper without
/// constructing a full `ConnectionManager`.
async fn capture_shared_slot<T>(
    slot: &Arc<parking_lot::Mutex<Option<T>>>,
    start_backoff: Duration,
    label: &NodeLabel,
) -> Option<T> {
    tokio::time::sleep(start_backoff).await;

    if let Some(value) = slot.lock().take() {
        return Some(value);
    }

    // Publication race: the spawned task hasn't been polled to its publish
    // point yet. Yield-and-recheck with a generous but bounded budget.
    //
    // On a current_thread runtime each `yield_now().await` hands control to
    // the scheduler, which then polls any other ready task (including our
    // target spawned task) before polling us again. A single yield is
    // usually enough, but a busy sim with hundreds of spawned tasks queued
    // ahead of ours may need many more. 1024 is comfortably above every
    // observed case and still returns in microseconds on the panic path.
    const MAX_PUBLISH_POLLS: usize = 1024;
    for _ in 0..MAX_PUBLISH_POLLS {
        tokio::task::yield_now().await;
        if let Some(value) = slot.lock().take() {
            return Some(value);
        }
    }

    tracing::warn!(
        %label,
        max_polls = MAX_PUBLISH_POLLS,
        "SimNetwork: node did not publish its shared slot within the \
         yield-poll budget — the spawned `run_node` task likely panicked \
         before reaching its publish point, or is wedged before its first \
         `.await`."
    );
    None
}

#[cfg(test)]
mod capture_shared_slot_tests {
    use super::{NodeLabel, capture_shared_slot};
    use std::{sync::Arc, time::Duration};

    fn label() -> NodeLabel {
        NodeLabel::gateway("test", 0)
    }

    /// Fast path: the spawned task already published its value during the
    /// initial `start_backoff` sleep (the old code's `take()` would have seen
    /// it too). `capture_shared_slot` must return the published value
    /// unchanged.
    #[tokio::test(flavor = "current_thread")]
    async fn fast_path_returns_published_value() {
        let slot = Arc::new(parking_lot::Mutex::new(Some(42u32)));
        let result = capture_shared_slot(&slot, Duration::from_millis(1), &label()).await;
        assert_eq!(result, Some(42));
        assert!(slot.lock().is_none(), "slot must be drained by take()");
    }

    /// The pre-fix race: the spawned task publishes *after* the initial
    /// `start_backoff` sleep, during the yield-poll phase. The old code's
    /// single `take()` would miss this and return `None`. The new helper's
    /// yield loop must observe the publication and return `Some`.
    #[tokio::test(flavor = "current_thread")]
    async fn publication_race_resolves_via_yield_loop() {
        let slot = Arc::new(parking_lot::Mutex::new(None::<u32>));
        let publisher_slot = Arc::clone(&slot);

        // Publish after several yield_now() passes — simulates a spawned
        // node task that hadn't been polled to its publish point yet when
        // `capture_shared_slot` entered the yield loop.
        tokio::spawn(async move {
            for _ in 0..3 {
                tokio::task::yield_now().await;
            }
            *publisher_slot.lock() = Some(7);
        });

        let result = capture_shared_slot(&slot, Duration::from_millis(1), &label()).await;
        assert_eq!(result, Some(7));
    }

    /// Timeout path: nothing ever publishes. `capture_shared_slot` must
    /// exhaust its yield-poll budget and return `None` so the caller can
    /// panic loudly with the node label. With the iteration-bounded loop
    /// this completes in microseconds (no wall-clock wait), so unlike the
    /// earlier draft that used a 5-second `std::time::Instant` deadline
    /// the test runs on every `cargo test` pass without `#[ignore]`.
    #[tokio::test(flavor = "current_thread")]
    async fn timeout_returns_none_when_never_published() {
        let slot: Arc<parking_lot::Mutex<Option<u32>>> = Arc::new(parking_lot::Mutex::new(None));
        let result = capture_shared_slot(&slot, Duration::from_millis(1), &label()).await;
        assert_eq!(result, None);
    }
}

fn clean_up_tmp_dirs<'a>(labels: impl Iterator<Item = &'a NodeLabel>) {
    for label in labels {
        let p = std::env::temp_dir().join(format!(
            "freenet-executor-{sim}-{label}",
            sim = "sim",
            label = label
        ));
        // Best-effort cleanup of temp dirs; failure is not critical
        let _removed = std::fs::remove_dir_all(p);
    }
}

/// Check convergence from event logs.
///
/// This function can be used after `run_simulation` completes to check
/// if all contracts have converged to the same state across replicas.
///
/// # Arguments
/// * `logs` - Event logs obtained via `sim.event_logs_handle()` before calling `run_simulation`
///
/// # Example
/// ```ignore
/// let logs_handle = sim.event_logs_handle();
/// let result = sim.run_simulation::<...>(...);
/// let convergence = check_convergence_from_logs(&logs_handle).await;
/// ```
pub async fn check_convergence_from_logs(
    logs: &Arc<tokio::sync::Mutex<Vec<crate::tracing::NetLogMessage>>>,
) -> ConvergenceResult {
    let logs = logs.lock().await;

    // Group (contract_key -> peer_addr -> latest_state_hash)
    // Use BTreeMap for deterministic iteration order
    let mut contract_states: BTreeMap<String, BTreeMap<SocketAddr, String>> = BTreeMap::new();

    // Iterate in insertion order - the last event for each (contract, peer) pair
    // is the actual current state.
    for log in logs.iter() {
        let contract_key = log.kind.contract_key().map(|k| format!("{:?}", k));
        let state_hash = log.kind.stored_state_hash().map(String::from);

        if let (Some(contract_key), Some(state_hash)) = (contract_key, state_hash) {
            contract_states
                .entry(contract_key)
                .or_default()
                .insert(log.peer_id.socket_addr(), state_hash);
        }
    }

    let mut converged = Vec::new();
    let mut diverged = Vec::new();

    for (contract_key, peer_states) in contract_states {
        if peer_states.len() < 2 {
            continue;
        }

        let unique_states: HashSet<&String> = peer_states.values().collect();

        if unique_states.len() == 1 {
            let state = unique_states.into_iter().next().unwrap().clone();
            converged.push(ConvergedContract {
                contract_key,
                state_hash: state,
                replica_count: peer_states.len(),
            });
        } else {
            diverged.push(DivergedContract {
                contract_key,
                peer_states: peer_states.into_iter().collect(),
            });
        }
    }

    ConvergenceResult {
        converged,
        diverged,
    }
}

use crate::contract::OperationMode;

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

    /// Unit test for the fault-injection delivery decision (#4694 / #4642 piece F).
    ///
    /// Directly exercises `fault_injection_delivery_decision` — the callback the
    /// direct/controlled runners install so injected crashes and partitions
    /// become real packet drops. Verifies:
    ///   1. Without `enforce_fault_drops`, every packet is delivered (the inert
    ///      state that made fault injection false-green before #4694).
    ///   2. With `enforce_fault_drops`, packets to/from a crashed node are
    ///      dropped (either direction) and counted in `messages_dropped_crash`.
    ///   3. With `enforce_fault_drops`, partitioned pairs are dropped and counted
    ///      in `messages_dropped_partition`.
    ///   4. Unaffected pairs are still delivered.
    #[test]
    fn test_fault_injection_delivery_decision_drops_crash_and_partition() {
        use crate::node::network_bridge::{FaultInjectorState, set_fault_injector};
        use crate::simulation::{FaultConfig, Partition, VirtualTime};
        use crate::transport::in_memory_socket::PacketDeliveryDecision;
        use std::collections::HashSet;
        use std::sync::{Arc, Mutex};

        const NETWORK: &str = "fault-decision-unit-test";

        let crashed: SocketAddr = "127.0.0.1:20001".parse().unwrap();
        let live_a: SocketAddr = "127.0.0.1:20002".parse().unwrap();
        let part_x: SocketAddr = "127.0.0.1:20003".parse().unwrap();
        let part_y: SocketAddr = "127.0.0.1:20004".parse().unwrap();

        let mut config = FaultConfig::default();
        config.crash_node(crashed);
        config.add_partition(
            Partition::new(HashSet::from([part_x]), HashSet::from([part_y])).permanent(0),
        );

        let state = FaultInjectorState::new(config, 42).with_virtual_time(VirtualTime::new());
        let injector = Arc::new(Mutex::new(state));
        set_fault_injector(NETWORK, Some(injector.clone()));

        let is_drop = |d: PacketDeliveryDecision| matches!(d, PacketDeliveryDecision::Drop);

        // (1) Not opted in: faults are inert, everything is delivered.
        assert!(
            !is_drop(fault_injection_delivery_decision(NETWORK, crashed, live_a)),
            "crash must be inert without enforce_fault_drops (pre-#4694 behavior)"
        );
        assert!(
            !is_drop(fault_injection_delivery_decision(NETWORK, part_x, part_y)),
            "partition must be inert without enforce_fault_drops"
        );

        injector.lock().unwrap().enforce_fault_drops = true;

        // (2) Crashed node: dropped in both directions.
        assert!(
            is_drop(fault_injection_delivery_decision(NETWORK, crashed, live_a)),
            "packet FROM a crashed node must be dropped"
        );
        assert!(
            is_drop(fault_injection_delivery_decision(NETWORK, live_a, crashed)),
            "packet TO a crashed node must be dropped"
        );

        // (3) Partitioned pair: dropped in both directions.
        assert!(
            is_drop(fault_injection_delivery_decision(NETWORK, part_x, part_y)),
            "packet across an active partition must be dropped (x->y)"
        );
        assert!(
            is_drop(fault_injection_delivery_decision(NETWORK, part_y, part_x)),
            "packet across an active partition must be dropped (y->x)"
        );

        // (4) Unaffected pair: delivered.
        assert!(
            !is_drop(fault_injection_delivery_decision(NETWORK, live_a, part_x)),
            "healthy, non-partitioned pair must be delivered"
        );

        let stats = injector.lock().unwrap().stats.clone();
        assert_eq!(stats.messages_dropped_crash, 2, "two crash drops expected");
        assert_eq!(
            stats.messages_dropped_partition, 2,
            "two partition drops expected"
        );

        set_fault_injector(NETWORK, None);
    }

    /// Test that peer locations are deterministic with the same seed.
    ///
    /// Regression test for issue #2759 - SimNetwork should produce identical
    /// peer locations across multiple runs with the same seed.
    #[tokio::test]
    async fn test_deterministic_peer_locations() {
        const SEED: u64 = 0xDEADBEEF_CAFEBABE;

        // Create first network
        let sim1 = SimNetwork::new(
            "determinism-test-1",
            2,  // 2 gateways
            3,  // 3 regular nodes
            7,  // ring_max_htl
            3,  // rnd_if_htl_above
            10, // max_connections
            2,  // min_connections
            SEED,
        )
        .await;

        let locations1 = sim1.get_peer_locations();

        // Create second network with same seed
        let sim2 = SimNetwork::new(
            "determinism-test-2",
            2, // same config
            3,
            7,
            3,
            10,
            2,
            SEED, // same seed
        )
        .await;

        let locations2 = sim2.get_peer_locations();

        // Verify locations are identical
        assert_eq!(
            locations1, locations2,
            "Peer locations should be identical with the same seed.\n\
             Run 1: {:?}\n\
             Run 2: {:?}",
            locations1, locations2
        );
    }
}