//! Python-oracle differential economy harness.
//!
//! Boots the REAL Python UTAS oracle (`fifa17-recon/tools/utas_server.py`) as an
//! isolated subprocess (temp `fut_profile.json`, temp account file, chosen
//! loopback port, disposable log — never the production container, port, or
//! save) AND the REAL Rust economy stack (in-process Core seeded via
//! `start_core_seeded` + a real `Server` with `EconomyServices` from
//! `build_econ_server`). It then drives the SAME semantic op sequence against
//! BOTH and compares HTTP status, JSON structure, coin/entitlement/inventory
//! deltas, wire-significant ids, and pile/listing/reveal state — classifying
//! every op PARITY or DIFFERENT-BY-DESIGN.
//!
//! Starting fixtures are semantically aligned (both own one pack-70 entitlement;
//! Python's fresh profile seeds pack 70, Core is granted the `"70"` entitlement
//! via its economy API). The seed BALANCES differ by design — the oracle's
//! `_new_profile` grants 15 000 coins, `start_core_seeded` grants 100 000 — so
//! the differential compares coin DELTAS and structure, never absolute balances.
//! No Python source is modified; the classifier is untouched.
//!
//! ────────────────────────── PARITY / DIFFERENT-BY-DESIGN matrix ─────────────
//!
//! | op                     | class              | notes                                                        |
//! |------------------------|--------------------|--------------------------------------------------------------|
//! | credits                | PARITY             | identical body shape: `credits`+`currencies[coins/points]`   |
//! |                        |                    | (`funds==finalFunds==balance`) + `unopenedPacks.recovered=1`.|
//! | userMassInfo economy   | PARITY             | `userInfo.currencies[coins].funds == credits coins`; both    |
//! |                        |                    | carry `unopenedPacks.recoveredPacks==1`. Coins consistent    |
//! |                        |                    | across the credits & massinfo surfaces on each side.         |
//! | purchasegroup pack70   | PARITY             | owned pack present -> id set {1,5,6,7,70}, NO 65534 sentinel. |
//! | purchasegroup sentinel | PARITY             | empty My Packs + unverified session -> {1,5,6,7,65534},      |
//! |                        |                    | sentinel `state:"active"`.                                   |
//! | purchasegroup clean-v1 | PARITY             | empty My Packs + verified capability session -> {1,5,6,7},   |
//! |                        |                    | sentinel stripped. Rust drives the REAL `SessionStore` state |
//! |                        |                    | machine (register_capability+open_session+freeze) exactly as |
//! |                        |                    | the oracle's launcher/auth handshake does.                   |
//! | Store BUY (pack 1)     | PARITY             | 200; `createPackResponse{itemList(5),numberItems:5,          |
//! |                        |                    | purchasedPackId,duplicateItemIdList}`; coin delta -400.      |
//! | POST /purchased open70 | PARITY             | 200; envelope `{packId:70,firstPartyStoreId,productId:"70",  |
//! |                        |                    | purchasePackType:"GOLD"}`; coin delta 0; entitlement -1.     |
//! | GET /purchased reveal  | PARITY             | Durable single-profile purchased pile on BOTH (Rust reveal   |
//! |                        |                    | = `PileStore.list_by_pile("purchased")` + Core inventory, NO |
//! |                        |                    | per-SID cache). Non-empty after open; idempotent on repeat.  |
//! |                        |                    | The hypothesised per-SID cache does NOT exist — VERIFIED.    |
//! | quick-sell DELETE/item | PARITY             | 200; `{items:[{id}],totalCredits}`; exactly one sold; coin   |
//! |                        |                    | delta > 0; `totalCredits`==absolute post-sale balance.       |
//! | quick-sell POST delete | PARITY             | batch form `{itemData:[{id}]}` -> same body shape/mechanism. |
//! | move-items PUT /item   | PARITY             | 200; `{itemData:[{id,pile,success:true}]}` verdict ack.      |
//! | match reward (WIN)     | PARITY             | 200; `{allCoins,matchCoins:400,gameModeAward{coins:400},     |
//! |                        |                    | seasonCoins,tournamentCoins,boostConis,participationAward,   |
//! |                        |                    | teamOfTournamentWinner}`; coin delta +400. Byte-shape match. |
//! | market list POST /ah   | PARITY             | 200; `{"id":<tradeId>}` (a positive listing id). Id SPACES   |
//! |                        |                    | differ (Rust MarketStore vs oracle 900500000+seq) — a        |
//! |                        |                    | wire-insignificant server-private handle.                    |
//! | market query tradePile | PARITY             | after list -> `auctionInfo` len 1, `tradeState:"active"`.    |
//! | market buy POST /trade | DIFFERENT-BY-DESIGN| first buy debits exactly `buyNowPrice` & closes on BOTH, but |
//! |                        |                    | Rust's MarketStore is STATEFUL single-debit (a second buy of |
//! |                        |                    | a sold listing is a no-op: 0 delta, empty `auctionInfo`)     |
//! |                        |                    | whereas the oracle's buyable market is a STATELESS sample    |
//! |                        |                    | reconstruction (`_auction_by_tradeid`, id base 900000000)    |
//! |                        |                    | that RE-DEBITS on every repeat buy and is a DISTINCT id      |
//! |                        |                    | space from the user's own sale pile (900500000+). See below. |
//! | market cancel DELETE   | PARITY             | 200; a cancelled listing is no longer active/buyable on both |
//! |                        |                    | (Rust: buying it is a 0-delta empty `auctionInfo`; oracle:   |
//! |                        |                    | it drops out of `tradePile`). Response body differs (`{}`    |
//! |                        |                    | oracle vs closed-record Rust) — wire-insignificant.          |
//!
//! DIFFERENT-BY-DESIGN — market buy idempotency & id-space, in full:
//! * Python behaviour: `trade_route` reconstructs a buyable auction from the
//!   SAMPLE pool every call (`_auction_by_tradeid(tid)` = PACK_POOL[tid-900000000]);
//!   it holds no per-listing sold state, so POSTing the same sample tradeId twice
//!   debits `buyNowPrice` twice and grants twice. The user's OWN listings live in
//!   a separate id space (`list_for_sale` -> 900500000+seq) surfaced only by
//!   `/tradePile`, and are NOT buyable via `/trade/{id}` (that returns an empty
//!   auction). Evidence: probed live — buying sample tradeId 900000000 twice each
//!   returned `tradeState:"closed"` with a -buyNow coin delta both times.
//! * Rust behaviour: `handle_market_buy` reserves+sells against a durable
//!   `MarketStore` keyed by the listing's own tradeId (unified with the sale
//!   pile); a second buy of a `sold` listing is a no-op (0 coin delta, empty
//!   `auctionInfo`). Evidence: asserted below (second buy delta 0).
//! * Reason: the oracle is a single-file, single-profile emulator whose sample
//!   market exists only to make the client's Transfer Market browsable; it never
//!   needed durable auction state. The Rust cutover makes the market a real,
//!   crash-consistent, single-debit ledger backed by SQLite. Compatibility
//!   impact: NONE for the client — a buy-now is a one-shot UI action, so the
//!   re-debit path was never reachable in normal play; the Rust behaviour is a
//!   strict correctness improvement and is pinned by assertion so it cannot
//!   regress toward the stateless oracle shape.
//!
//! Safety: temp dir + `127.0.0.1` ephemeral ports only. The oracle subprocess is
//! killed on guard drop. Never touches the production Core DB/ports or `.105`.

use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog;
use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
use openfut_adapter_fifa17::fut::store_session::{SessionStore, StoreMode, SENTINEL_PACK_ID};
use openfut_identity::JsonIdentityStore;
use openfut_utas_host::async_bridge::AsyncBridge;
use openfut_utas_host::market_store::MarketStore;
use openfut_utas_host::pile_store::PileStore;
use openfut_utas_host::{
    build_content_pool, handle_purchasegroup, overlay_massinfo_economy, CoreAccess, CoreEconomy,
    EconomyServices, Fifa17IdentityResolver, HttpCoreClient, PassClient, Server,
};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::process::{Child, Command, Stdio};
use std::sync::Arc;

/// The persona the oracle's `ACCOUNT` defaults to (tier-2 built-in). The Rust
/// `SessionStore` clean-v1 handshake keys on the same value so both sides bind a
/// verified capability to the same `(ip, persona)`.
const PERSONA_ID: i64 = 33068179;

// ─────────────────────────── Python oracle subprocess ───────────────────────

/// A running, isolated Python UTAS oracle; killed on drop.
struct Oracle {
    child: Child,
    base: String,
    http: reqwest::blocking::Client,
}

impl Drop for Oracle {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

impl Oracle {
    /// Spawn the oracle bound to a fresh loopback port with a disposable profile,
    /// account file, and log — fully isolated from any production instance.
    fn spawn(dir: &std::path::Path) -> Oracle {
        // A free loopback port: bind :0, read the assignment, release it, hand the
        // number to Python via FUT_PORT. A brief TOCTOU window is acceptable on a
        // loopback test host.
        let port = {
            let l = std::net::TcpListener::bind("127.0.0.1:0").expect("free port");
            l.local_addr().unwrap().port()
        };
        let tools = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .parent()
            .unwrap()
            .join("fifa17-recon/tools");
        assert!(
            tools.join("utas_server.py").exists(),
            "oracle source not found at {}",
            tools.display()
        );
        let child = Command::new("python3")
            .arg("utas_server.py")
            .current_dir(&tools)
            .env("FUT_PROFILE", dir.join("fut_profile.json"))
            .env("FUT_ACCOUNT_PATH", dir.join("fut_account.json"))
            .env("FUT_LOG", dir.join("oracle.log"))
            .env("FUT_PORT", port.to_string())
            .env("OPENFUT_BIND", "127.0.0.1")
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("spawn python oracle (is python3 on PATH?)");
        Oracle {
            child,
            base: format!("http://127.0.0.1:{port}"),
            http: reqwest::blocking::Client::new(),
        }
    }

    /// Poll `/user/credits` until the oracle answers 200 (or panic after ~15s).
    fn wait_ready(&mut self) {
        for _ in 0..750 {
            if let Some(status) = self.child.try_wait().expect("try_wait") {
                panic!("oracle exited before ready: {status}");
            }
            if let Ok(r) = self
                .http
                .get(format!("{}/ut/game/fifa17/user/credits", self.base))
                .header("X-OpenFUT-Game", "fifa17")
                .send()
            {
                if r.status().is_success() {
                    return;
                }
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        panic!("oracle never became ready at {}", self.base);
    }

    /// One request against the oracle. Returns `(status, json_body)`; a missing or
    /// empty body decodes to `Value::Null`.
    fn req(
        &self,
        method: &str,
        path: &str,
        body: Option<Value>,
        sid: Option<&str>,
    ) -> (u16, Value) {
        let m = reqwest::Method::from_bytes(method.as_bytes()).expect("method");
        let mut rb = self
            .http
            .request(m, format!("{}{path}", self.base))
            .header("X-OpenFUT-Game", "fifa17");
        if let Some(s) = sid {
            rb = rb.header("X-UT-SID", s);
        }
        if let Some(b) = body {
            rb = rb.json(&b);
        }
        let resp = rb
            .send()
            .unwrap_or_else(|e| panic!("oracle {method} {path}: {e}"));
        let status = resp.status().as_u16();
        let text = resp.text().unwrap_or_default();
        let v: Value = if text.is_empty() {
            Value::Null
        } else {
            serde_json::from_str(&text).unwrap_or(Value::Null)
        };
        (status, v)
    }

    fn coins(&self) -> i64 {
        self.req("GET", "/ut/game/fifa17/user/credits", None, None)
            .1["currencies"][0]["funds"]
            .as_i64()
            .expect("oracle coins")
    }
}

// ─────────────────── Core boot + econ server (reference harness) ─────────────
// Replicated from `tests/economy_integration.rs` (separate test binary, no shared
// module) so this file is self-contained per the one-file-per-agent contract.

/// Boot a Core with FIFA17 dev CONTENT loaded and, on first boot, the dev
/// inventory SEEDED (a fifa17 profile + club with 100k coins + one owned
/// instance per definition). Ephemeral loopback port.
async fn start_core_seeded(db_url: &str, seed: bool) -> (tokio::task::JoinHandle<()>, String) {
    use openfut_core::config::Config;
    use openfut_core::seed::{seed_fifa17_dev, FIFA17_GAME};
    use openfut_core::services::card_db::CardDb;

    let data_dir = "../openfut-core/data";
    let pool = openfut_core::db::init_pool(db_url, 5)
        .await
        .expect("core pool");
    openfut_core::db::run_migrations(&pool)
        .await
        .expect("core migrations");
    if seed {
        let mut card_db = CardDb::load(data_dir).expect("card_db load");
        card_db
            .load_game_dev(data_dir, FIFA17_GAME)
            .expect("load fifa17 dev content");
        seed_fifa17_dev(&pool, &card_db)
            .await
            .expect("seed fifa17 dev inventory");
    }
    let cfg = Config {
        listen_addr: "127.0.0.1:0".into(),
        database_url: "sqlite::memory:".into(),
        data_dir: data_dir.into(),
        max_connections: 5,
        dev_content_games: vec![FIFA17_GAME.to_string()],
        content_packs: Vec::new(),
    };
    let app = openfut_core::app::build(pool, cfg)
        .await
        .expect("core app::build");
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("bind ephemeral");
    let addr = listener.local_addr().unwrap();
    let handle = tokio::spawn(async move {
        let _ = axum::serve(listener, app).await;
    });
    (handle, format!("http://{addr}"))
}

fn wait_ready(base: &str) {
    let http = reqwest::blocking::Client::new();
    for _ in 0..1500 {
        if let Ok(r) = http.get(format!("{base}/health")).send() {
            if r.status().is_success() {
                return;
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(20));
    }
    panic!("core did not become ready at {base}");
}

fn core_post(http: &reqwest::blocking::Client, base: &str, path: &str, body: Value) -> Value {
    let resp = http
        .post(format!("{base}{path}"))
        .header("X-OpenFUT-Game", "fifa17")
        .json(&body)
        .send()
        .unwrap_or_else(|e| panic!("POST {path}: {e}"));
    let status = resp.status();
    let v: Value = resp.json().unwrap_or(Value::Null);
    assert!(status.is_success(), "POST {path} -> {status}: {v}");
    v
}

/// Build a real `Server` with economy authority wired against the seeded Core.
/// Returns the server, a direct Core client for balance/entitlement assertions,
/// and a valid wire `resourceId` (20000) that reverse-maps to a real Core card.
fn build_econ_server(base: &str, dir: &std::path::Path) -> (Server, HttpCoreClient, i64) {
    let probe = HttpCoreClient::new(base, "fifa17");
    let owned = probe.all_owned().expect("core collection");
    assert!(!owned.is_empty(), "seed must grant a starter collection");
    let mut entries = String::new();
    let mut seen = std::collections::HashSet::new();
    let mut asset = 20000u32;
    for it in &owned {
        if !seen.insert(it.card_id.clone()) {
            continue;
        }
        if !entries.is_empty() {
            entries.push(',');
        }
        entries.push_str(&format!("\"{}\":{{\"asset_id\":{asset}}}", it.card_id));
        asset += 1;
    }
    let doc = format!("{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{entries}}}}}");
    let catalog = Fifa17CardCatalog::from_json_str(&doc).expect("catalog");
    let store = JsonIdentityStore::open(dir.join("identity.json").to_str().unwrap()).unwrap();
    let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store)));
    let entities = Arc::new(Fifa17Entities::from_maps(
        HashMap::new(),
        HashMap::new(),
        HashMap::new(),
    ));
    let core: Arc<dyn CoreAccess> = Arc::new(HttpCoreClient::new(base, "fifa17"));
    let bridge = Arc::new(AsyncBridge::new().unwrap());
    let market_path = dir.join("market.db").to_string_lossy().into_owned();
    let market = Arc::new(
        bridge
            .block_on(async move { MarketStore::open(&market_path).await })
            .expect("market store"),
    );
    let pile_path = dir.join("pile.db").to_string_lossy().into_owned();
    let piles = Arc::new(
        bridge
            .block_on(async move { PileStore::open(&pile_path).await })
            .expect("pile store"),
    );
    let econ: Arc<dyn CoreEconomy> = Arc::new(HttpCoreClient::new(base, "fifa17"));
    let pool = Arc::new(build_content_pool(core.as_ref(), resolver.as_ref()));
    assert!(
        !pool.is_empty(),
        "content pool derived from real Core content"
    );
    let services = Arc::new(EconomyServices {
        econ,
        market,
        piles,
        bridge,
        pool,
    });
    let server = Server::new(
        core,
        entities,
        resolver.clone(),
        Arc::new(PassClient::new("http://127.0.0.1:9")),
        PERSONA_ID,
    )
    .with_economy(services);
    (server, probe, 20000)
}

// ─────────────────────────── differential comparison ────────────────────────

/// A single `try_handle_economy` call against the Rust dispatch. Panics if the
/// route is unhandled (all economy ops here are wired).
fn rust(server: &Server, method: &str, path: &str, body: &[u8], sid: Option<&str>) -> (u16, Value) {
    let headers: Vec<(String, String)> = match sid {
        Some(s) => vec![("X-UT-SID".into(), s.into())],
        None => vec![],
    };
    let resp = server
        .try_handle_economy(method, path, &headers, body, None)
        .unwrap_or_else(|| panic!("rust dispatch did not route {method} {path}"));
    let v: Value = if resp.body.is_empty() {
        Value::Null
    } else {
        serde_json::from_slice(&resp.body).unwrap_or(Value::Null)
    };
    (resp.status, v)
}

/// Sorted `purchase[].id` set from a purchasegroup body.
fn pack_ids(pg: &Value) -> Vec<u64> {
    let mut v: Vec<u64> = pg["purchase"]
        .as_array()
        .expect("purchase array")
        .iter()
        .map(|p| p["id"].as_u64().expect("pack id"))
        .collect();
    v.sort_unstable();
    v
}

/// Every op runs on a plain OS thread with NO ambient Tokio runtime (the blocking
/// Core client + `reqwest::blocking` require this), exactly like the
/// thread-per-connection server — the bridge takes its direct `block_on` path.
fn run_differential(core_base: &str, oracle: &Oracle, dir: &std::path::Path) {
    wait_ready(core_base);
    let http = reqwest::blocking::Client::new();
    let (server, client, sample_resource) = build_econ_server(core_base, dir);

    // ── Fixture alignment: both sides own exactly one pack-70 entitlement. ──
    // Oracle: fresh profile already owns pack 70. Core: grant the "70" entitlement
    // through its real economy API (cost 0, no debit).
    core_post(
        &http,
        core_base,
        "/economy/purchase-entitlement",
        json!({ "cost": 0, "definition_id": "70" }),
    );
    assert_eq!(
        oracle
            .req("GET", "/ut/game/fifa17/user/credits", None, None)
            .0,
        200
    );

    let mut matrix: Vec<(&str, &str)> = Vec::new();

    // ── OP 1: credits ──────────────────────────────────────────────────────
    let (os, ob) = oracle.req("GET", "/ut/game/fifa17/user/credits", None, None);
    let (rs, rb) = rust(&server, "GET", "/ut/game/fifa17/user/credits", b"", None);
    assert_eq!(os, 200);
    assert_eq!(rs, 200, "credits status parity");
    for (tag, b) in [("oracle", &ob), ("rust", &rb)] {
        assert_eq!(
            b["currencies"][0]["name"], "coins",
            "{tag} currencies[0]=coins"
        );
        assert_eq!(
            b["currencies"][1]["name"], "points",
            "{tag} currencies[1]=points"
        );
        let coins = b["currencies"][0]["funds"].as_i64().unwrap();
        assert_eq!(
            b["currencies"][0]["finalFunds"].as_i64().unwrap(),
            coins,
            "{tag} funds==finalFunds"
        );
        assert_eq!(
            b["credits"].as_i64().unwrap(),
            coins,
            "{tag} credits scalar == coins"
        );
        assert_eq!(
            b["unopenedPacks"]["recoveredPacks"].as_i64().unwrap(),
            1,
            "{tag} recoveredPacks==1"
        );
    }
    matrix.push(("credits", "PARITY"));

    // ── OP 2: userMassInfo economy ─────────────────────────────────────────
    let om = oracle
        .req("GET", "/ut/game/fifa17/userMassInfo", None, None)
        .1;
    let o_mass_coins = om["userInfo"]["currencies"][0]["funds"].as_i64().unwrap();
    assert_eq!(om["userInfo"]["currencies"][0]["name"], "coins");
    assert_eq!(
        o_mass_coins,
        ob["currencies"][0]["funds"].as_i64().unwrap(),
        "oracle massinfo coins == credits coins"
    );
    assert_eq!(
        om["userInfo"]["unopenedPacks"]["recoveredPacks"]
            .as_i64()
            .unwrap(),
        1,
        "oracle massinfo recoveredPacks==1"
    );
    // Rust: overlay the authoritative Core economy onto a massinfo body in place.
    let mut mass =
        json!({ "userInfo": { "currencies": [ {"name":"coins","funds":0,"finalFunds":0} ] } });
    let r_coins = client.balance().unwrap();
    let r_ents = client.entitlements().unwrap().len();
    assert!(overlay_massinfo_economy(&mut mass, r_coins, r_ents));
    assert_eq!(
        mass["userInfo"]["currencies"][0]["funds"].as_i64().unwrap(),
        r_coins,
        "rust overlay coins == Core balance"
    );
    assert_eq!(
        mass["userInfo"]["currencies"][0]["funds"],
        mass["userInfo"]["currencies"][0]["finalFunds"]
    );
    assert_eq!(
        mass["userInfo"]["unopenedPacks"]["recoveredPacks"]
            .as_i64()
            .unwrap(),
        r_ents as i64,
        "rust overlay recoveredPacks == entitlements"
    );
    assert_eq!(
        rb["currencies"][0]["funds"].as_i64().unwrap(),
        r_coins,
        "rust massinfo coins == credits coins"
    );
    matrix.push(("userMassInfo economy", "PARITY"));

    // ── OP 3a: purchasegroup — pack70 owned (no sentinel) ──────────────────
    let opg = oracle
        .req("GET", "/ut/game/fifa17/store/purchasegroup", None, None)
        .1;
    let rpg = rust(
        &server,
        "GET",
        "/ut/game/fifa17/store/purchasegroup",
        b"",
        None,
    )
    .1;
    assert_eq!(
        pack_ids(&opg),
        vec![1, 5, 6, 7, 70],
        "oracle owned pack70 id set"
    );
    assert_eq!(
        pack_ids(&rpg),
        vec![1, 5, 6, 7, 70],
        "rust owned pack70 id set"
    );
    for b in [&opg, &rpg] {
        assert!(
            !pack_ids(b).contains(&SENTINEL_PACK_ID),
            "no 65534 sentinel while a pack is owned"
        );
        // packType parity per id (BRONZE for 1, GOLD for the rest).
        for p in b["purchase"].as_array().unwrap() {
            let id = p["id"].as_u64().unwrap();
            let want = if id == 1 { "BRONZE" } else { "GOLD" };
            assert_eq!(p["packType"], want, "packType parity for pack {id}");
        }
    }
    matrix.push(("purchasegroup pack70", "PARITY"));

    // ── OP 4: Store BUY (pack 1 Bronze, price 400, 5 cards) ────────────────
    let o_bal0 = oracle.coins();
    let (o_bs, o_buy) = oracle.req(
        "PUT",
        "/ut/game/fifa17/store/transaction",
        Some(json!({"packId":1})),
        None,
    );
    let r_bal0 = client.balance().unwrap();
    let (r_bs, r_buy) = rust(
        &server,
        "PUT",
        "/ut/game/fifa17/store/transaction",
        br#"{"packId":1}"#,
        None,
    );
    assert_eq!(o_bs, 200);
    assert_eq!(r_bs, 200, "BUY status parity");
    let o_items = o_buy["createPackResponse"]["itemList"]
        .as_array()
        .expect("oracle itemList")
        .clone();
    let r_items = r_buy["createPackResponse"]["itemList"]
        .as_array()
        .expect("rust itemList")
        .clone();
    assert_eq!(o_items.len(), 5, "oracle pack1 -> 5 cards");
    assert_eq!(r_items.len(), 5, "rust pack1 -> 5 cards");
    for b in [&o_buy, &r_buy] {
        let cpr = &b["createPackResponse"];
        assert_eq!(cpr["numberItems"].as_i64().unwrap(), 5, "numberItems==5");
        assert_eq!(
            cpr["purchasedPackId"].as_i64().unwrap(),
            1,
            "purchasedPackId==1"
        );
        assert!(
            cpr["duplicateItemIdList"].is_array(),
            "duplicateItemIdList is an array"
        );
    }
    assert_eq!(oracle.coins() - o_bal0, -400, "oracle BUY debits 400");
    assert_eq!(
        client.balance().unwrap() - r_bal0,
        -400,
        "rust BUY debits 400"
    );
    matrix.push(("Store BUY (pack1)", "PARITY"));
    // Minted wire ids for the item ops that follow (both sides put them in the
    // pending purchased pile).
    let o_wire: Vec<i64> = o_items.iter().map(|i| i["id"].as_i64().unwrap()).collect();
    let r_wire: Vec<i64> = r_items.iter().map(|i| i["id"].as_i64().unwrap()).collect();
    assert!(
        r_wire[0] >= 100_000_000,
        "rust mints above the FIFA wire floor"
    );

    // ── OP 5: POST /purchased — owned pack-70 open ─────────────────────────
    let o_bal_p = oracle.coins();
    let (o_ps, o_open) = oracle.req(
        "POST",
        "/ut/game/fifa17/purchased",
        Some(json!({"packId":70})),
        None,
    );
    let r_ent_before = client.entitlements().unwrap().len();
    let r_bal_p = client.balance().unwrap();
    let (r_ps, r_open) = rust(
        &server,
        "POST",
        "/ut/game/fifa17/purchased",
        br#"{"packId":70}"#,
        None,
    );
    assert_eq!(o_ps, 200);
    assert_eq!(r_ps, 200, "pack-70 open status parity");
    for b in [&o_open, &r_open] {
        assert_eq!(b["packId"].as_i64().unwrap(), 70, "open echoes pack id 70");
        assert_eq!(b["productId"], "70", "productId echo");
        assert_eq!(b["purchasePackType"], "GOLD", "reward pack is GOLD");
    }
    assert_eq!(oracle.coins() - o_bal_p, 0, "oracle pack-70 open is free");
    assert_eq!(
        client.balance().unwrap() - r_bal_p,
        0,
        "rust pack-70 open is free"
    );
    assert_eq!(
        client.entitlements().unwrap().len(),
        r_ent_before - 1,
        "rust consumes the 70 entitlement"
    );
    matrix.push(("POST /purchased open70", "PARITY"));

    // ── OP 6: GET /purchased reveal (durable single-profile pile; idempotent) ─
    let o_rev1 = oracle.req("GET", "/ut/game/fifa17/purchased", None, None).1["itemData"]
        .as_array()
        .unwrap()
        .len();
    let r_rev1 = rust(&server, "GET", "/ut/game/fifa17/purchased", b"", None).1["itemData"]
        .as_array()
        .unwrap()
        .len();
    assert!(o_rev1 > 0, "oracle reveal shows opened items ({o_rev1})");
    assert!(r_rev1 > 0, "rust reveal shows opened items ({r_rev1})");
    // Idempotent + durable: a repeat GET (with a DIFFERENT SID) returns the same
    // reveal — there is no per-SID cache; the pile is single-profile durable.
    let o_rev2 = oracle
        .req(
            "GET",
            "/ut/game/fifa17/purchased",
            None,
            Some("OPENFUT-SID-OTHER1"),
        )
        .1["itemData"]
        .as_array()
        .unwrap()
        .len();
    let r_rev2 = rust(
        &server,
        "GET",
        "/ut/game/fifa17/purchased",
        b"",
        Some("OPENFUT-SID-OTHER1"),
    )
    .1["itemData"]
        .as_array()
        .unwrap()
        .len();
    assert_eq!(
        o_rev2, o_rev1,
        "oracle reveal idempotent & SID-independent (durable single-profile)"
    );
    assert_eq!(
        r_rev2, r_rev1,
        "rust reveal idempotent & SID-independent (durable single-profile, NO per-SID cache)"
    );
    matrix.push(("GET /purchased reveal", "PARITY"));

    // ── OP 7: quick-sell (single, DELETE /item/<wireId>) ───────────────────
    let o_qbal = oracle.coins();
    let (o_qs, o_q) = oracle.req(
        "DELETE",
        &format!("/ut/game/fifa17/item/{}", o_wire[0]),
        None,
        None,
    );
    let r_qbal = client.balance().unwrap();
    let (r_qs, r_q) = rust(
        &server,
        "DELETE",
        &format!("/ut/game/fifa17/item/{}", r_wire[0]),
        b"",
        None,
    );
    assert_eq!(o_qs, 200);
    assert_eq!(r_qs, 200, "quick-sell status parity");
    let o_qafter = oracle.coins();
    let r_qafter = client.balance().unwrap();
    for (tag, b, bal) in [("oracle", &o_q, o_qafter), ("rust", &r_q, r_qafter)] {
        assert_eq!(
            b["items"].as_array().unwrap().len(),
            1,
            "{tag} sold exactly one"
        );
        assert_eq!(
            b["totalCredits"].as_i64().unwrap(),
            bal,
            "{tag} totalCredits == absolute balance"
        );
    }
    assert!(o_qafter - o_qbal > 0, "oracle quick-sell credits coins");
    assert!(r_qafter - r_qbal > 0, "rust quick-sell credits coins");
    matrix.push(("quick-sell DELETE/item", "PARITY"));

    // ── OP 8: quick-sell (batch, POST /ut/delete/.../item) ─────────────────
    let (o_qbs, o_qb) = oracle.req(
        "POST",
        "/ut/delete/game/fifa17/item",
        Some(json!({"itemData":[{"id":o_wire[3]}]})),
        None,
    );
    let (r_qbs, r_qb) = rust(
        &server,
        "POST",
        "/ut/delete/game/fifa17/item",
        format!(r#"{{"itemData":[{{"id":{}}}]}}"#, r_wire[3]).as_bytes(),
        None,
    );
    assert_eq!(o_qbs, 200);
    assert_eq!(r_qbs, 200, "batch quick-sell status parity");
    for (tag, b, bal) in [
        ("oracle", &o_qb, oracle.coins()),
        ("rust", &r_qb, client.balance().unwrap()),
    ] {
        assert_eq!(
            b["items"].as_array().unwrap().len(),
            1,
            "{tag} batch sold one"
        );
        assert_eq!(
            b["totalCredits"].as_i64().unwrap(),
            bal,
            "{tag} batch totalCredits == balance"
        );
    }
    matrix.push(("quick-sell POST delete", "PARITY"));

    // ── OP 9: move-items (PUT /item, still-owned card -> trade pile) ────────
    let (o_ms, o_mv) = oracle.req(
        "PUT",
        "/ut/game/fifa17/item",
        Some(json!({"itemData":[{"id":o_wire[1],"pile":"trade"}]})),
        None,
    );
    let (r_ms, r_mv) = rust(
        &server,
        "PUT",
        "/ut/game/fifa17/item",
        format!(r#"{{"itemData":[{{"id":{},"pile":"trade"}}]}}"#, r_wire[1]).as_bytes(),
        None,
    );
    assert_eq!(o_ms, 200);
    assert_eq!(r_ms, 200, "move status parity");
    for (tag, b, id) in [("oracle", &o_mv, o_wire[1]), ("rust", &r_mv, r_wire[1])] {
        let v = &b["itemData"][0];
        assert_eq!(v["success"], true, "{tag} move recorded success");
        assert_eq!(v["pile"], "trade", "{tag} move pile echo");
        assert_eq!(v["id"].as_i64().unwrap(), id, "{tag} move id echo");
    }
    matrix.push(("move-items PUT /item", "PARITY"));

    // ── OP 10: match reward (WIN = +400) ───────────────────────────────────
    let o_mbal = oracle.coins();
    let (o_mms, o_mm) = oracle.req(
        "POST",
        "/ut/delete/game/fifa17/match",
        Some(json!({"endReason":"WIN"})),
        None,
    );
    let r_mbal = client.balance().unwrap();
    let (r_mms, r_mm) = rust(
        &server,
        "POST",
        "/ut/delete/game/fifa17/match",
        br#"{"endReason":"WIN"}"#,
        None,
    );
    assert_eq!(o_mms, 200);
    assert_eq!(r_mms, 200, "match status parity");
    for (tag, b, bal) in [
        ("oracle", &o_mm, oracle.coins()),
        ("rust", &r_mm, client.balance().unwrap()),
    ] {
        assert_eq!(
            b["matchCoins"].as_i64().unwrap(),
            400,
            "{tag} WIN matchCoins==400"
        );
        assert_eq!(
            b["gameModeAward"]["coins"].as_i64().unwrap(),
            400,
            "{tag} gameModeAward.coins==400"
        );
        assert_eq!(
            b["allCoins"].as_i64().unwrap(),
            bal,
            "{tag} allCoins == post-credit balance"
        );
        assert_eq!(b["seasonCoins"].as_i64().unwrap(), 0, "{tag} seasonCoins");
        assert_eq!(
            b["tournamentCoins"].as_i64().unwrap(),
            0,
            "{tag} tournamentCoins"
        );
        assert_eq!(
            b["boostConis"].as_i64().unwrap(),
            0,
            "{tag} boostConis (EA typo key)"
        );
        assert_eq!(
            b["teamOfTournamentWinner"], false,
            "{tag} teamOfTournamentWinner"
        );
        assert!(
            b["participationAward"].is_i64(),
            "{tag} participationAward present"
        );
    }
    assert_eq!(oracle.coins() - o_mbal, 400, "oracle WIN +400");
    assert_eq!(client.balance().unwrap() - r_mbal, 400, "rust WIN +400");
    matrix.push(("match reward (WIN)", "PARITY"));

    // ── OP 11: market list (POST /auctionhouse) ────────────────────────────
    // Oracle lists an OWNED CLUB item: the moved card (OP 9) is now in the club,
    // so it is a valid, tradePile-visible listing. Rust lists a valid wire id
    // against its MarketStore.
    let (o_ls, o_list) = oracle.req(
        "POST",
        "/ut/game/fifa17/auctionhouse",
        Some(json!({"itemData":{"id":o_wire[1]},"buyNowPrice":1000,"startingBid":500})),
        None,
    );
    let o_trade_id = o_list["id"].as_i64().expect("oracle trade id");
    let (r_ls, r_list) = rust(&server, "POST", "/ut/game/fifa17/auctionhouse", format!(r#"{{"itemData":{{"id":777,"resourceId":{sample_resource}}},"buyNowPrice":1000,"startingBid":500}}"#).as_bytes(), None);
    let r_trade_id = r_list["id"].as_i64().expect("rust trade id");
    assert_eq!(o_ls, 200);
    assert_eq!(r_ls, 200, "market list status parity");
    assert!(
        o_trade_id > 0 && r_trade_id > 0,
        "both return a positive listing id"
    );
    matrix.push(("market list POST /ah", "PARITY"));

    // ── OP 12: market query (GET /tradePile) ───────────────────────────────
    let o_tp = oracle.req("GET", "/ut/game/fifa17/tradePile", None, None).1;
    let r_tp = rust(&server, "GET", "/ut/game/fifa17/tradePile", b"", None).1;
    assert_eq!(
        o_tp["auctionInfo"].as_array().unwrap().len(),
        1,
        "oracle tradePile shows the listing"
    );
    assert_eq!(
        r_tp["auctionInfo"].as_array().unwrap().len(),
        1,
        "rust tradePile shows the listing"
    );
    assert_eq!(
        o_tp["auctionInfo"][0]["tradeState"], "active",
        "oracle listing active"
    );
    assert_eq!(
        r_tp["auctionInfo"][0]["tradeState"], "active",
        "rust listing active"
    );
    matrix.push(("market query tradePile", "PARITY"));

    // ── OP 13: market buy (POST /trade/<id>) — DIFFERENT-BY-DESIGN ─────────
    // Shared invariant: the first buy debits exactly buyNowPrice and closes the
    // auction on BOTH. Rust buys its own (unified) listing; the oracle's buyable
    // market is the STATELESS sample pool, so we buy a sample auction there.
    let o_sample = oracle
        .req("GET", "/ut/game/fifa17/auctionhouse", None, None)
        .1;
    let o_sample_tid = o_sample["auctionInfo"][0]["tradeId"]
        .as_i64()
        .expect("sample tradeId");
    let o_sample_bn = o_sample["auctionInfo"][0]["buyNowPrice"]
        .as_i64()
        .expect("sample buyNow");
    let o_bbal = oracle.coins();
    let (o_bys, o_buy1) = oracle.req(
        "POST",
        &format!("/ut/game/fifa17/trade/{o_sample_tid}"),
        Some(json!({})),
        None,
    );
    assert_eq!(o_bys, 200);
    assert_eq!(
        o_buy1["auctionInfo"][0]["tradeState"], "closed",
        "oracle buy closes the auction"
    );
    assert_eq!(
        oracle.coins() - o_bbal,
        -o_sample_bn,
        "oracle first buy debits exactly buyNowPrice"
    );

    let r_bbal = client.balance().unwrap();
    let (r_bys, r_buy1) = rust(
        &server,
        "POST",
        &format!("/ut/game/fifa17/trade/{r_trade_id}"),
        b"{}",
        None,
    );
    assert_eq!(r_bys, 200);
    assert_eq!(
        r_buy1["auctionInfo"][0]["tradeState"], "closed",
        "rust buy closes the auction"
    );
    assert_eq!(
        client.balance().unwrap() - r_bbal,
        -1000,
        "rust first buy debits exactly buyNowPrice"
    );

    // The divergence, PINNED. Rust: a second buy of a sold listing is a no-op
    // (0 delta, empty auctionInfo). Oracle: a repeat sample buy RE-DEBITS.
    let r_before2 = client.balance().unwrap();
    let (_, r_buy2) = rust(
        &server,
        "POST",
        &format!("/ut/game/fifa17/trade/{r_trade_id}"),
        b"{}",
        None,
    );
    assert_eq!(
        client.balance().unwrap(),
        r_before2,
        "RUST single-debit: second buy of a sold listing does NOT re-debit"
    );
    assert_eq!(
        r_buy2["auctionInfo"].as_array().unwrap().len(),
        0,
        "rust second buy sees a closed/absent auction"
    );
    // rust tradePile no longer active after sale.
    let r_tp2 = rust(&server, "GET", "/ut/game/fifa17/tradePile", b"", None).1;
    assert_eq!(
        r_tp2["auctionInfo"].as_array().unwrap().len(),
        0,
        "rust listing sold: no longer active"
    );

    let o_before2 = oracle.coins();
    let (_, o_buy2) = oracle.req(
        "POST",
        &format!("/ut/game/fifa17/trade/{o_sample_tid}"),
        Some(json!({})),
        None,
    );
    assert_eq!(
        o_buy2["auctionInfo"][0]["tradeState"], "closed",
        "oracle repeat buy re-closes a fresh reconstruction"
    );
    assert_eq!(
        oracle.coins() - o_before2,
        -o_sample_bn,
        "ORACLE stateless: sample-market buy RE-DEBITS on repeat (by design)"
    );
    matrix.push(("market buy POST /trade", "DIFFERENT-BY-DESIGN"));

    // ── OP 14: market cancel (DELETE /ut/delete/.../trade/<id>) ────────────
    // Oracle: cancel the OWN listing (OP 11) -> it drops out of tradePile.
    let (o_cs, _) = oracle.req(
        "DELETE",
        &format!("/ut/delete/game/fifa17/trade/{o_trade_id}"),
        None,
        None,
    );
    assert_eq!(o_cs, 200, "oracle cancel 200");
    let o_tp_after = oracle.req("GET", "/ut/game/fifa17/tradePile", None, None).1;
    assert_eq!(
        o_tp_after["auctionInfo"].as_array().unwrap().len(),
        0,
        "oracle cancelled listing gone from tradePile"
    );
    // Rust: list a fresh item, cancel it, then a buy is a 0-delta empty auction.
    let clist = rust(&server, "POST", "/ut/game/fifa17/auctionhouse", format!(r#"{{"itemData":{{"id":888,"resourceId":{sample_resource}}},"buyNowPrice":1000,"startingBid":500}}"#).as_bytes(), None).1;
    let r_cancel_id = clist["id"].as_i64().unwrap();
    let (r_cs, _) = rust(
        &server,
        "DELETE",
        &format!("/ut/delete/game/fifa17/trade/{r_cancel_id}"),
        b"",
        None,
    );
    assert_eq!(r_cs, 200, "rust cancel 200");
    let r_bc = client.balance().unwrap();
    let cbuy = rust(
        &server,
        "POST",
        &format!("/ut/game/fifa17/trade/{r_cancel_id}"),
        b"{}",
        None,
    )
    .1;
    assert_eq!(
        cbuy["auctionInfo"].as_array().unwrap().len(),
        0,
        "rust cancelled listing not buyable"
    );
    assert_eq!(
        client.balance().unwrap(),
        r_bc,
        "rust buying a cancelled listing does not debit"
    );
    matrix.push(("market cancel DELETE", "PARITY"));

    // ── OP 3b: purchasegroup — sentinel (empty My Packs, unverified session) ─
    // Both sides now own zero packs (pack 70 was consumed in OP 5). Oracle: clear
    // the opened-pack grace via /hub, then an unverified session gets the sentinel.
    oracle.req("GET", "/ut/game/fifa17/hub", None, None);
    let o_pg_s = oracle
        .req("GET", "/ut/game/fifa17/store/purchasegroup", None, None)
        .1;
    let r_pg_s = rust(
        &server,
        "GET",
        "/ut/game/fifa17/store/purchasegroup",
        b"",
        None,
    )
    .1;
    assert_eq!(
        pack_ids(&o_pg_s),
        vec![1, 5, 6, 7, SENTINEL_PACK_ID],
        "oracle empty -> sentinel"
    );
    assert_eq!(
        pack_ids(&r_pg_s),
        vec![1, 5, 6, 7, SENTINEL_PACK_ID],
        "rust empty (unknown SID) -> sentinel"
    );
    for b in [&o_pg_s, &r_pg_s] {
        let s = b["purchase"]
            .as_array()
            .unwrap()
            .iter()
            .find(|p| p["id"].as_u64() == Some(SENTINEL_PACK_ID))
            .unwrap();
        assert_eq!(s["state"], "active", "sentinel state active");
    }
    matrix.push(("purchasegroup sentinel", "PARITY"));

    // ── OP 3c: purchasegroup — clean-v1 (empty + verified capability session) ─
    // Oracle: register the launcher capability, open a session (auth), present the
    // SID -> clean topology (no sentinel).
    oracle.req(
        "POST",
        "/openfut/fifa17/capability",
        Some(json!({"capability":"empty_mypacks_resolver","version":1,"personaId":PERSONA_ID})),
        None,
    );
    let auth = oracle
        .req(
            "POST",
            "/ut/auth",
            Some(json!({"nucleusPersonaId":PERSONA_ID,"nucleusPersonaDisplayName":"CAGE"})),
            None,
        )
        .1;
    let sid = auth["sid"].as_str().expect("oracle sid").to_string();
    let o_pg_c = oracle
        .req(
            "GET",
            "/ut/game/fifa17/store/purchasegroup",
            None,
            Some(&sid),
        )
        .1;
    assert_eq!(
        pack_ids(&o_pg_c),
        vec![1, 5, 6, 7],
        "oracle clean-v1 strips the sentinel"
    );
    // Rust: drive the REAL SessionStore state machine identically (capability
    // before login -> pending, open the session, freeze the mode), then the real
    // Core-backed builder. Core owns zero entitlements now, so this is the empty
    // clean case.
    let mut sessions = SessionStore::new();
    let now = 10.0;
    sessions.register_capability(Some("127.0.0.1".into()), PERSONA_ID, 1, now);
    let rsid = "OPENFUT-SID-DIFFERENTIAL01";
    sessions.open_session(rsid, Some("127.0.0.1".into()), PERSONA_ID, now);
    let mode = sessions.empty_mypacks_mode(rsid, Some("127.0.0.1"), now);
    assert_eq!(
        mode,
        StoreMode::CleanV1,
        "verified capability -> clean-v1 mode"
    );
    let r_pg_c_resp = handle_purchasegroup(&client, mode);
    let r_pg_c: Value = serde_json::from_slice(&r_pg_c_resp.body).unwrap();
    assert_eq!(
        pack_ids(&r_pg_c),
        vec![1, 5, 6, 7],
        "rust clean-v1 strips the sentinel"
    );
    assert!(!pack_ids(&r_pg_c).contains(&SENTINEL_PACK_ID));
    matrix.push(("purchasegroup clean-v1", "PARITY"));

    // ── Emit the matrix for the run log. ────────────────────────────────────
    eprintln!("\n===== economy_differential PARITY / DIFFERENT-BY-DESIGN matrix =====");
    for (op, class) in &matrix {
        eprintln!("  {op:<26} {class}");
    }
    let dbd = matrix
        .iter()
        .filter(|(_, c)| *c == "DIFFERENT-BY-DESIGN")
        .count();
    eprintln!(
        "  ({} ops compared: {} PARITY, {dbd} DIFFERENT-BY-DESIGN)\n",
        matrix.len(),
        matrix.len() - dbd
    );
    assert_eq!(matrix.len(), 16, "all economy ops classified");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn economy_differential_python_oracle() {
    let dir = std::env::temp_dir().join(format!(
        "openfut-econ-diff-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    let db_url = format!("sqlite://{}/econ.db", dir.display());

    // Boot the seeded in-process Core.
    let (core_handle, core_base) = start_core_seeded(&db_url, true).await;

    // Everything below runs off-runtime on a plain OS thread. This is mandatory:
    // the blocking Core client, `reqwest::blocking` (including the oracle client,
    // whose construction spins up + tears down a temporary runtime), and the
    // AsyncBridge's direct `block_on` all panic if a Tokio runtime is entered on
    // the thread. spawn_blocking joins a std::thread — a truly runtime-free
    // context, exactly like the thread-per-connection server. The oracle guard is
    // created AND owned inside that thread, so the subprocess is killed when the
    // thread finishes or unwinds on panic.
    let core_base_run = core_base.clone();
    let dir_run = dir.clone();
    let outcome = tokio::task::spawn_blocking(move || {
        std::thread::spawn(move || {
            let mut oracle = Oracle::spawn(&dir_run);
            oracle.wait_ready();
            run_differential(&core_base_run, &oracle, &dir_run);
        })
        .join()
    })
    .await
    .expect("join spawn_blocking");

    core_handle.abort();
    std::fs::remove_dir_all(&dir).ok();
    outcome.expect("differential thread panicked");
}