degenbot-simulation 0.6.0-alpha.14

//! `BotStateDb` — a thin `revm::DatabaseRef` wrapper over the RPC fallback
//! with the always-on divergence observer + the code-less membership tripwire.
//!
//! ## What this is today
//!
//! `storage_ref` forwards to the `fallback` (`WrapDatabaseAsync<AlloyDB>` in
//! production) and, between the fallback read and the return, runs the
//! divergence observer (`super::divergence_probe`) on tracked pool slots: pure
//! observation — it compares the engine's packed typed state against the
//! just-fetched RPC value, logs a `[sim-divergence]` line on mismatch, and
//! accumulates the tally. It never changes what the sim reads.
//!
//! The env-gated serving seam that once overrode `storage_ref` with the
//! engine's packed word is DELETED (ADR-056). Its premise ("stale engine state
//! causes `CurrencyNotSettled`") was REFUTED by mainnet data — V3 hops matched
//! the actual swap output exactly (engine state is correct) while only the V4
//! swap diverged by 1-8 units (a solver calc rounding divergence, not stale
//! state). Production always ran gate-off, so deleting it changes nothing:
//! the fallback's value is the sim's value.
//!
//! The membership the `basic_ref` tripwire asks (is this a tracked pool?) is
//! delegated to a [`SimAnchorOracle`] — the boot-snapshot `RouteRegistry` in
//! the hosted backrun boot, the engine's `SimAnchorState` snapshot in the Python path.
//!
//! The wrapper persists because the live `BlockSimHandle` chain
//! (`simulator.rs`) references it as the `CacheDB` backing; collapsing it to
//! bare `WrapDatabaseAsync<AlloyDB>` is the Tier 1 refactor's scope,
//! not this module's cleanup.
//!
//! ## Historical note — the retired slot encoders
//!
//! This module previously held a toolkit of slot encoders
//! (`encode_v2_reserves_slot`, `encode_v3_slot0`, `encode_v3_liquidity_slot`,
//! `encode_v3_tick_info_slot`, `tick_mapping_slot`, `sign_extend_*`) plus
//! `read_v2_slot`/`read_v3_slot`/`read_tracked_storage`/`SnapshotError`. These
//! are deleted (no consumer; the `read_*_slot` paths returned `None` so
//! `storage_ref` always fell through). Re-derivation (if a tracked-slot encode
//! is ever needed again) starts from the Solidity storage layout — the old
//! `parity_diagnostic_encoding.rs` tests that pinned them are also gone.
//!
//! ## Composition
//!
//! ```text
//! EVM transact -> CacheDB (sim-scoped overrides)
//!                 -> BotStateDb (RPC forwarding + divergence observer + membership tripwire)
//!                 -> WrapDatabaseAsync<AlloyDB> (RPC fallback)
//! ```

use alloy::primitives::Address;
use degenbot_core::op_warn;
use degenbot_substrate::SimAnchorOracle;
use revm::database_interface::DatabaseRef;
use revm::primitives::{StorageKey, StorageValue, B256, KECCAK_EMPTY};
use revm::state::AccountInfo;

/// A thin `DatabaseRef` wrapper that forwards every read to the `fallback`.
///
/// Forwards to the fallback, with the divergence observer + membership
/// tripwire layered on the `oracle` (see the module docs). Whether this wrapper persists
/// or collapses to bare `WrapDatabaseAsync<AlloyDB>` is decided by the
/// Tier 1 refactor .
pub struct BotStateDb<'bot, ExtDb>
where
    ExtDb: DatabaseRef,
{
    /// The typed oracle backing the code-less tripwire (membership) and the
    /// divergence observer's engine words. The engine's snapshot is
    /// OWNED, taken under a short read — no `BotState` guard is held across
    /// the RPC fetches this wrapper performs.
    pub oracle: &'bot dyn SimAnchorOracle,
    /// The RPC cold-miss fallback (`WrapDatabaseAsync<AlloyDB>` in production).
    pub fallback: ExtDb,
    /// False-empty provenance probe (DEGENBOT false-empty capture): an
    /// independent UNCACHED `eth_getCode` at the sim block, fired only when
    /// the false-empty path trips and only when these fields are set. The
    /// alloy `with_default_caching()` 100-entry LRU in `AlloyProvider` PERSISTS
    /// `eth_getCode(address, block)` results (incl. a spurious `0x`), so a
    /// single transient node empty at a block is served to every later
    /// `AlloyDB` read in that block → `KECCAK_EMPTY` here. The fresh uncached
    /// probe attributes the empty to the CACHE layer (fresh read returns real
    /// code) vs the NODE layer (fresh read also empty). Default `new()` leaves
    /// these `None` → probe off; `new_with_code_probe` arms it.
    code_probe_rpc: Option<String>,
    code_probe_block: Option<u64>,
    /// Optional PER-CYCLE storage memo (SIMPIPE2 M2): when present, storage
    /// reads consult it before the fallback and real fetches populate it.
    /// Values are the fallback's own answers at one block height - never the
    /// engine's slots - so the refuted partial-serve hazard does not apply.
    pub storage_memo: Option<std::sync::Arc<super::StorageMemo>>,
    /// VERIFY2 T2: on-demand divergence-probe arm (one sim). When true, the
    /// storage observer compares engine-vs-RPC even with the env gate off -
    /// the same-storage-read observation costs no extra RPC.
    pub divergence_probe: bool,
}

impl<'bot, ExtDb> BotStateDb<'bot, ExtDb>
where
    ExtDb: DatabaseRef,
{
    /// Wrap the cold-miss fallback `DatabaseRef`. The `oracle` borrow backs
    /// the code-less tripwire + divergence observer.
    #[must_use]
    pub fn new(oracle: &'bot dyn SimAnchorOracle, fallback: ExtDb) -> Self {
        Self {
            oracle,
            fallback,
            code_probe_rpc: None,
            code_probe_block: None,
            storage_memo: None,
            divergence_probe: false,
        }
    }

    /// Attach the per-cycle storage memo (builder form - the cycle owner
    /// creates the memo and hands the same Arc to every handle built within
    /// the block).
    #[must_use]
    pub fn with_storage_memo(mut self, memo: std::sync::Arc<super::StorageMemo>) -> Self {
        self.storage_memo = Some(memo);
        self
    }

    /// Same but the caller may hold `None` (memo off).
    #[must_use]
    pub fn with_storage_memo_opt(
        mut self,
        memo: Option<&std::sync::Arc<super::StorageMemo>>,
    ) -> Self {
        self.storage_memo = memo.cloned();
        self
    }
    /// VERIFY2 T2: arm the divergence probe for THIS handle's sim (on-demand
    /// verification - a sim-failure re-verify or a random spot-check).
    #[must_use]
    pub fn with_divergence_probe(mut self, on: bool) -> Self {
        self.divergence_probe = on;
        self
    }

    /// Like [`Self::new`] but arms the false-empty provenance probe: on the
    /// code-less tripwire it performs an independent UNCACHED `eth_getCode`
    /// against `rpc_url` at `sim_block` and logs whether the empty is
    /// cache-served (alloy LRU) or node-origin. Called by `BlockSimHandle::
    /// build` with the live provider URL + pumped block when hunting
    /// false-empty reads.
    #[must_use]
    pub fn new_with_code_probe(
        oracle: &'bot dyn SimAnchorOracle,
        fallback: ExtDb,
        rpc_url: &str,
        sim_block: u64,
    ) -> Self {
        Self {
            oracle,
            fallback,
            code_probe_rpc: Some(rpc_url.to_string()),
            code_probe_block: Some(sim_block),
            storage_memo: None,
            divergence_probe: false,
        }
    }

    /// Provenance of a false-empty, enriched to test the stale/bad-block
    /// theory: it surfaces (a) the `AccountInfo` the RPC fallback returned
    /// (balance/nonce/code-hash) so we can see whether the account actually
    /// EXISTED at the block the fallback queried — `balance=0 nonce=0` is the
    /// ACCOUNT-ABSENT signature of a stale/wrong block where the contract
    /// wasn't deployed — and (b) a fresh UNCACHED `eth_getCode` at the same
    /// sim block (raw HTTP/1.1, bypassing the alloy LRU) to split CACHE-SERVED
    /// vs NODE/block-origin. Returns a human verdict string.
    fn code_probe_provenance(&self, address: Address, info: Option<&AccountInfo>) -> String {
        let acct = match info {
            Some(a) => format!(
                "fallback_account: balance={} nonce={} code_hash={}",
                a.balance, a.nonce, a.code_hash
            ),
            None => "fallback_account: None (read as non-existent)".to_string(),
        };
        let absent_hint = match info {
            Some(a) if a.balance.is_zero() && a.nonce == 0 => {
                " [ACCOUNT-ABSENT: balance=0 nonce=0 — consistent with a STALE/BAD block where \
                 the contract was not yet deployed or served as an EOA]"
            }
            _ => "",
        };
        let (Some(rpc), Some(block)) = (&self.code_probe_rpc, self.code_probe_block) else {
            return format!("{acct};; no-probe (code probe unconfigured)");
        };
        let fresh = raw_uncached_eth_get_code(rpc, address, block);
        match fresh {
            Some(code) if code.len() > 2 && code != "0x" => format!(
                "{acct}{absent_hint};; FRESH UNCACHED GET eth_getCode({address},\
                 0x{block:x}) => {:x}-byte code -> CACHE-SERVED EMPTY (the alloy \
                 with_default_caching() LRU persisted a spurious 0x; the node HAS the pool\
                 at THIS block, so the empty is NOT a stale-block issue on the live read)",
                (code.len() - 2) / 2
            ),
            Some(code) => format!(
                "{acct}{absent_hint};; FRESH UNCACHED GET eth_getCode({address},\
                 {block}) => ALSO EMPTY ({code:?}) -> NODE/BLOCK-ORIGIN (the live node returns\
                 empty at this exact block — a stale/bad block WOULD explain this; cross-check\
                 the block against live head)"
            ),
            None => format!(
                "{acct};; PROBE-FAILED — raw eth_getCode HTTP error (cannot attribute layer)"
            ),
        }
    }
}

impl<ExtDb> DatabaseRef for BotStateDb<'_, ExtDb>
where
    ExtDb: DatabaseRef,
{
    type Error = ExtDb::Error;

    /// Forward to the fallback, with a LOUD invalidity tripwire for tracked
    /// pools (ADR-021: detect/classify/stop loudly, never auto-repair).
    ///
    /// # Errors
    ///
    /// Returns the fallback's error if the RPC `basic` fetch fails.
    #[expect(clippy::panic_in_result_fn, clippy::panic)] // code-less tracked pool = invariant break (loud)
    fn basic_ref(&self, address: Address) -> Result<Option<AccountInfo>, Self::Error> {
        let info = self.fallback.basic_ref(address)?;
        // A tracked pool address MUST resolve to a contract with code. If the
        // RPC fallback reports it as non-existent (`None`) or code-less
        // (``KECCAK_EMPTY``), every later per-block sim would execute the pool
        // as an EOA → zero reserves → the mysterious empty-revert family
        // (path-205: `0xb01C29F3` BNB/WETH read `reserve0=0` in sim while
        // healthy on-chain). A registered pool can never legitimately be
        // code-less, so this is an impossibility — fail loudly HERE. Because
        // this is the value-origin point (the warm cache only caches what this
        // returns), the poisoned `None`/empty can never reach the warm cache
        // to be served on a later block's hit.
        if self.oracle.is_registered_pool(&address) {
            let invalid = match &info {
                None => true,
                Some(acc) => acc.code_hash == KECCAK_EMPTY,
            };
            if invalid {
                let state = if info.is_none() {
                    "non-existent (None)"
                } else {
                    "code-less (KECCAK_EMPTY)"
                };
                let provenance = self.code_probe_provenance(address, info.as_ref());
                op_warn!(
                    domain = sim,
                    "tracked pool {address} resolved {state} @ block {:?}; {provenance}",
                    self.code_probe_block,
                );
                panic!(
                    "Sim DB invariant: tracked pool {address} resolved as {state} by the \
                     RPC fallback — refusing to simulate a code-less pool"
                );
            }
        }
        Ok(info)
    }

    /// Forward storage reads to the RPC fallback, with the always-on
    /// divergence observer layered between the fetch and the return.
    ///
    /// - [`super::divergence_probe`]: pure observation — compares the engine's
    ///   packed typed state against the RPC value, logs a `[sim-divergence]`
    ///   line on mismatch, accumulates the tally. Never changes what the sim
    ///   reads. The on-demand arm (`self.divergence_probe`) skips the env gate.
    ///
    /// The env-gated serving seam that once returned the engine's packed word
    /// for tracked slots is DELETED (ADR-056); the fallback's value is the
    /// sim's value, exactly as production ran gate-off.
    ///
    /// # Errors
    ///
    /// Returns the fallback's error if the RPC fetch fails.
    fn storage_ref(
        &self,
        address: Address,
        index: StorageKey,
    ) -> Result<StorageValue, Self::Error> {
        // M2: per-cycle memo FIRST - a hit is the same block's own earlier
        // RPC answer (the fallback remains the authority; the memo never
        // serves engine slots, so the K-invariant / LOK partial-serve hazard
        // does not apply). The owner recreates the memo per block.
        if let Some(memo) = self.storage_memo.as_ref() {
            if let Some(v) = memo.get(address, index) {
                return Ok(v);
            }
        }
        let rpc_value = self.fallback.storage_ref(address, index)?;
        if let Some(memo) = self.storage_memo.as_ref() {
            memo.put(address, index, rpc_value);
        }
        // Observation (compares engine vs RPC, logs divergence, never changes
        // what the sim reads). The per-sim on-demand arm (VERIFY2 T2 -
        // sim-failure re-verify or a random spot-check) skips the env gate.
        if self.divergence_probe {
            super::divergence_probe::observe_storage_read_forced(
                self.oracle,
                address,
                index,
                rpc_value,
            );
        } else {
            super::divergence_probe::observe_storage_read(self.oracle, address, index, rpc_value);
        }
        // Serving retired (ADR-056): the fallback's own answer is the sim's.
        Ok(rpc_value)
    }

    /// Forward to the fallback. `code_by_hash` is **never invoked** if
    /// `basic` eagerly loads code (the spike-verified
    /// `code_by_hash` panic-safety invariant).
    fn code_by_hash_ref(&self, code_hash: B256) -> Result<revm::bytecode::Bytecode, Self::Error> {
        self.fallback.code_by_hash_ref(code_hash)
    }

    /// Block hashes are not in `Bot`'s domain — always fall through to
    /// `AlloyDB` (the live-network axis).
    fn block_hash_ref(&self, number: u64) -> Result<B256, Self::Error> {
        self.fallback.block_hash_ref(number)
    }
}

/// Fresh, UNCACHED `eth_getCode` via a raw HTTP/1.1 POST (bypassing the alloy
/// `with_default_caching()` response cache entirely). Returns `Some(hex)` on
/// success, `None` on any transport/parse failure. Only `http://` (plaintext)
/// is supported — the probe is a diagnostic and the bot's RPC is plain HTTP.
fn raw_uncached_eth_get_code(rpc_url: &str, address: Address, block: u64) -> Option<String> {
    use std::io::{Read, Write};
    use std::net::TcpStream;
    use std::time::Duration;

    // Strip a leading scheme; only plaintext http is supported (a TLS/https
    // URL needs a real client, not a raw TCP socket).
    let no_scheme = rpc_url.strip_prefix("http://")?;
    let (host, port) = match no_scheme.split_once(':') {
        Some((h, p)) => (h.to_string(), p.parse::<u16>().ok()?),
        None => (no_scheme.to_string(), 80u16),
    };
    let body = format!(
        r#"{{"jsonrpc":"2.0","id":1,"method":"eth_getCode","params":["{address}","0x{block:x}"]}}"#
    );
    let req = format!(
        "POST / HTTP/1.1\r\nHost: {host}:{port}\r\nContent-Type: application/json\r\n\
         Content-Length: {}\r\nConnection: close\r\n\r\n{}",
        body.len(),
        body
    );
    let mut stream = TcpStream::connect((host.as_str(), port)).ok()?;
    stream.set_read_timeout(Some(Duration::from_secs(5))).ok()?;
    stream.write_all(req.as_bytes()).ok()?;
    let mut resp = String::new();
    stream.read_to_string(&mut resp).ok()?;
    let body_json = resp.split("\r\n\r\n").nth(1)?;
    let v: serde_json::Value = serde_json::from_str(body_json).ok()?;
    v.get("result")?.as_str().map(str::to_string)
}

#[expect(clippy::unwrap_used, clippy::expect_used)]
#[cfg(test)]
mod tests {
    use super::*;
    use alloy::primitives::aliases::U112;
    use alloy::primitives::{Address, U256};
    use degenbot_bot::bot_core::RegisterV2PoolParams;
    use degenbot_bot::connector_index::{V2ConnectorIndex, V2Edge};
    use degenbot_substrate::{BotState, RouteRegistry, SimAnchorState};
    use degenbot_uniswap::dex_identity::DexVariant;
    use revm::bytecode::Bytecode;
    use revm::primitives::B256;
    use std::cell::Cell;

    /// The tracked-pool address (matches the bot-core test fixture shape).
    const POOL: Address = Address::new([0xaa; 20]);
    /// A NON-pool address — an EOA / warmup-victim basic read must stay exempt.
    const EOA: Address = Address::new([0xee; 20]);

    /// A `BotState` with one tracked V2 pool at [`POOL`], so
    /// `pool_id_by_address(POOL)` resolves (and `EOA` does not).
    fn bot_state_with_pool() -> BotState {
        let mut core = BotState::new();
        core.register_v2_pool(&RegisterV2PoolParams {
            address: POOL,
            token0: Address::from([0xbb; 20]),
            token1: Address::from([0xcc; 20]),
            reserve0: U112::from(1000),
            reserve1: U112::from(2000),
            fee_token0: (997, 1000),
            fee_token1: (997, 1000),
            factory: Address::from([0xdd; 20]),
            update_block: 0,
            variant: DexVariant::UniswapV2,
            stable_swap: false,
            fee_denominator: None,
            ..Default::default()
        })
        .expect("test setup: V2 registration");
        core
    }

    /// A boot registry whose frozen set contains exactly [`POOL`].
    fn registry_with_pool() -> RouteRegistry {
        let mut index = V2ConnectorIndex::default();
        index.push_edge(V2Edge {
            pool_id: 1,
            token0_id: 10,
            token1_id: 20,
            address: POOL,
            fees: degenbot_substrate::executor_hop::V2FeePair::from_discovered(
                Some(3),
                Some(3),
                Some(1_000),
            ),
        });
        RouteRegistry::new(index)
    }

    /// A minimal `DatabaseRef` fallback whose `basic_ref` returns the test's
    /// scripted result and counts calls.
    struct ScriptedDb {
        result: Option<AccountInfo>,
        calls: Cell<u64>,
    }
    impl DatabaseRef for ScriptedDb {
        type Error = core::convert::Infallible;
        fn basic_ref(&self, _address: Address) -> Result<Option<AccountInfo>, Self::Error> {
            self.calls.set(self.calls.get() + 1);
            Ok(self.result.clone())
        }
        fn code_by_hash_ref(&self, _code_hash: B256) -> Result<Bytecode, Self::Error> {
            Ok(Bytecode::new_legacy(alloy::primitives::Bytes::new()))
        }
        fn storage_ref(
            &self,
            _address: Address,
            _index: StorageKey,
        ) -> Result<StorageValue, Self::Error> {
            Ok(StorageValue::ZERO)
        }
        fn block_hash_ref(&self, _number: u64) -> Result<B256, Self::Error> {
            Ok(B256::ZERO)
        }
    }

    fn code_info() -> AccountInfo {
        let code = Bytecode::new_legacy(alloy::primitives::Bytes::from_static(&[0x60, 0x00]));
        AccountInfo::new(U256::from(1), 0, code.hash_slow(), code)
    }
    fn codeless_info() -> AccountInfo {
        AccountInfo::new(U256::from(1), 0, KECCAK_EMPTY, Bytecode::default())
    }

    // ── Green 1: a tracked pool resolving to `None` (non-existent) PANICS ──

    #[test]
    #[should_panic(expected = "Sim DB invariant")]
    fn tracked_pool_resolving_none_panics() {
        let core = bot_state_with_pool();
        let db = ScriptedDb {
            result: None,
            calls: Cell::new(0),
        };
        let anchor = SimAnchorState::snapshot(&core);
        let bsd = BotStateDb::new(&anchor, db);
        let _ = bsd.basic_ref(POOL).unwrap();
    }

    // ── Green 2: a tracked pool resolving to CODE-LESS PANICS ──

    #[test]
    #[should_panic(expected = "Sim DB invariant")]
    fn tracked_pool_resolving_codeless_panics() {
        let core = bot_state_with_pool();
        let db = ScriptedDb {
            result: Some(codeless_info()),
            calls: Cell::new(0),
        };
        let anchor = SimAnchorState::snapshot(&core);
        let bsd = BotStateDb::new(&anchor, db);
        let _ = bsd.basic_ref(POOL).unwrap();
    }

    // ── Green 3: a tracked pool resolving WITH code passes through ──

    #[test]
    fn tracked_pool_with_code_passes_through() {
        let core = bot_state_with_pool();
        let db = ScriptedDb {
            result: Some(code_info()),
            calls: Cell::new(0),
        };
        let anchor = SimAnchorState::snapshot(&core);
        let bsd = BotStateDb::new(&anchor, db);
        let got = bsd.basic_ref(POOL).unwrap();
        assert!(got.is_some(), "tracked pool with code is forwarded");
        assert_ne!(got.unwrap().code_hash, KECCAK_EMPTY);
    }

    // ── Green 4: NON-pool addresses (EOA/warmup) are EXEMPT from the tripwire ──

    #[test]
    fn non_pool_none_is_exempt() {
        let core = bot_state_with_pool();
        let db = ScriptedDb {
            result: None,
            calls: Cell::new(0),
        };
        let anchor = SimAnchorState::snapshot(&core);
        let bsd = BotStateDb::new(&anchor, db);
        // EOA is not a pool: `None` is a legitimate EOA read, forwarded no-panic.
        assert!(bsd.basic_ref(EOA).unwrap().is_none());
    }

    /// The serving-retirement equivalence pin: the deleted serving seam is gone, so a tracked
    /// slot's read is the fallback's own value — the gate-off behavior
    /// production always had. The observer still compares, unchanged.
    #[test]
    fn serving_retired_storage_ref_forwards_rpc_for_tracked_slot() {
        let _tally = crate::sim::evm::divergence_probe::TALLY_TEST_GUARD
            .lock()
            .unwrap();
        let core = bot_state_with_pool();
        let anchor = SimAnchorState::snapshot(&core);
        let db = ScriptedDb {
            result: None,
            calls: Cell::new(0),
        };
        let bsd = BotStateDb::new(&anchor, db);
        let got = bsd.storage_ref(POOL, U256::from(8u64)).unwrap();
        assert_eq!(
            got,
            StorageValue::ZERO,
            "retired serving: the fallback's value is the sim's value"
        );
    }

    /// The serving-retirement equivalence pin: swapping the engine snapshot for the boot registry
    /// oracle changes no sim read — the serving path that consumed the words
    /// is gone, and the observer is inert on a state-less view.
    #[test]
    fn oracle_choice_does_not_change_sim_storage_reads() {
        let _tally = crate::sim::evm::divergence_probe::TALLY_TEST_GUARD
            .lock()
            .unwrap();
        let core = bot_state_with_pool();
        let anchor = SimAnchorState::snapshot(&core);
        let registry = registry_with_pool();
        let anchor_db = BotStateDb::new(
            &anchor,
            ScriptedDb {
                result: None,
                calls: Cell::new(0),
            },
        );
        let registry_db = BotStateDb::new(
            &registry,
            ScriptedDb {
                result: None,
                calls: Cell::new(0),
            },
        );
        for slot in [U256::from(8u64), U256::ZERO, U256::from(6u64)] {
            assert_eq!(
                anchor_db.storage_ref(POOL, slot).unwrap(),
                registry_db.storage_ref(POOL, slot).unwrap(),
                "oracle choice must not change the read at {slot}"
            );
        }
    }

    /// The membership question now routes through the registry oracle: a
    /// registry member resolving code-less still trips the invariant.
    #[test]
    #[should_panic(expected = "Sim DB invariant")]
    fn code_less_registry_member_panics() {
        let registry = registry_with_pool();
        let db = ScriptedDb {
            result: Some(codeless_info()),
            calls: Cell::new(0),
        };
        let bsd = BotStateDb::new(&registry, db);
        let _ = bsd.basic_ref(POOL).unwrap();
    }

    // ── The false-empty provenance probe (alloy LRU vs node) ────────────

    #[test]
    fn raw_uncached_eth_get_code_reads_fresh_code_past_the_cache() {
        // A local HTTP server that answers `eth_getCode` with a real (non-empty)
        // contract-body string. The probe must bypass the alloy provider's
        // `with_default_caching()` LRU and see the real code — the `CACHE`
        // provenance path.
        use std::io::{Read, Write};
        use std::net::TcpListener;
        use std::thread;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let serve = thread::spawn(move || {
            let (mut sock, _) = listener.accept().unwrap();
            let mut buf = [0u8; 4096];
            let _ = sock.read(&mut buf);
            let body = r#"{"jsonrpc":"2.0","id":1,"result":"0x608060405234"}"#;
            let resp = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
                 Content-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            );
            sock.write_all(resp.as_bytes()).unwrap();
        });

        let got = raw_uncached_eth_get_code(
            &format!("http://127.0.0.1:{port}"),
            parse_addr(),
            1_234_567u64,
        );
        serve.join().unwrap();
        assert_eq!(
            got.as_deref(),
            Some("0x608060405234"),
            "probe reads fresh code straight off the wire, uncached"
        );
    }

    fn parse_addr() -> Address {
        "0x36D2b521d708537B98F01Ab8d5207BD8E42b2806"
            .parse()
            .unwrap()
    }

    #[test]
    fn stale_block_signature_reports_node_origin_and_account_absent() {
        // The user's stale/bad-block theory path: the fallback ACCOUNT reads as
        // zero-balance + zero-nonce (the contract never existed at the queried
        // block — the ACCOUNT-ABSENT signature) AND the fresh uncached probe at
        // the SAME block also returns empty (0x). The enriched verdict must say
        // so, so the operator can cross-check the block against live head.
        use std::io::{Read, Write};
        use std::net::TcpListener;
        use std::thread;

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let serve = thread::spawn(move || {
            let (mut sock, _) = listener.accept().unwrap();
            let mut buf = [0u8; 4096];
            let _ = sock.read(&mut buf);
            let body = r#"{"jsonrpc":"2.0","id":1,"result":"0x"}"#;
            let resp = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
                 Content-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            );
            sock.write_all(resp.as_bytes()).unwrap();
        });

        let core = bot_state_with_pool();
        let anchor = SimAnchorState::snapshot(&core);
        let db = ScriptedDb {
            result: Some(codeless_info()),
            calls: Cell::new(0),
        };
        let bsd = BotStateDb::new_with_code_probe(
            &anchor,
            db,
            &format!("http://127.0.0.1:{port}"),
            1_234_567u64,
        );
        // Force balance=0 to exercise the ACCOUNT-ABSENT / stale-block hint.
        let mut empty_acct = codeless_info();
        empty_acct.balance = U256::ZERO;
        let verdict = bsd.code_probe_provenance(POOL, Some(&empty_acct));
        serve.join().unwrap();
        assert!(
            verdict.contains("ACCOUNT-ABSENT"),
            "hint missing: {verdict}"
        );
        assert!(
            verdict.contains("NODE/BLOCK-ORIGIN"),
            "node/block verdict missing: {verdict}"
        );
    }
}