#!/usr/bin/env python3 """PRE-LIVE machine differential for the seller-facing sold-row A/B. Purpose: prove, before any operator time is spent at a real FIFA client, that the two variants differ in EXACTLY the field under test. A contaminated A/B cannot attribute the client's reaction to anything, so this gate runs first. It performs ONE synthetic settlement, then re-reads every seller-facing surface under each variant by restarting only the host (same Core, same DBs, same sale), and diffs the payloads with explicit classification: MISSING key present in A, absent in B EXTRA key absent in A, present in B TYPE_MISMATCH same key, different JSON type VALUE_MISMATCH same key and type, different value Two orthogonal pairs are checked: PRIMARY bidState highest vs buyNow coinsProcessed held at 0 ORTHOGONAL coinsProcessed 0 vs 1 bidState held at highest Plus the two counts states the live experiment needs to interpret: S1 0 active + 1 sold S2 1 active + 1 sold Evidence is written per variant under docs/evidence/sold-ab-/ with a sha256 of each payload, so variant A evidence can never be confused with variant B. ISOLATION: reuses the verified staging bring-up from sold-wire-check.py, which binds only ephemeral loopback ports, refuses every production port, and opens nothing under /home/alex/openfut-promotion/state/. python3 scripts/sold-ab-differential.py [--out DIR] [--keep] """ import argparse import hashlib import importlib.util import json import os import shutil import sqlite3 import subprocess import sys import tempfile import time REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # Reuse the proven harness helpers rather than re-deriving them. The filename has # hyphens, so it cannot be imported normally. _spec = importlib.util.spec_from_file_location( "sold_wire_check", os.path.join(REPO, "scripts", "sold-wire-check.py")) H = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(H) SURFACES = { "tradePile": "/ut/game/fifa17/tradePile", "counts": "/ut/game/fifa17/tradePile/counts", "trade_status_filtered": f"/ut/game/fifa17/trade/status?tradeIds={H.TRADE_ID}", "trade_status_unfiltered": "/ut/game/fifa17/trade/status", } results = [] def check(label, ok, detail=""): results.append((label, bool(ok), detail)) print(f" [{'PASS' if ok else 'FAIL'}] {label}{(': ' + detail) if detail else ''}") return ok def banner(t): print("\n" + "=" * 74) print(f"== {t}") print("=" * 74) def jtype(v): if isinstance(v, bool): return "bool" if isinstance(v, int): return "int" if isinstance(v, float): return "float" if isinstance(v, str): return "str" if isinstance(v, list): return "list" if isinstance(v, dict): return "dict" return "null" def classify(a, b, path=""): """Recursive classified diff. Returns a list of (kind, path, a, b).""" out = [] if jtype(a) != jtype(b): return [("TYPE_MISMATCH", path or "", jtype(a), jtype(b))] if isinstance(a, dict): for k in a: if k not in b: out.append(("MISSING", f"{path}.{k}".lstrip("."), a[k], None)) for k in b: if k not in a: out.append(("EXTRA", f"{path}.{k}".lstrip("."), None, b[k])) for k in a: if k in b: out += classify(a[k], b[k], f"{path}.{k}".lstrip(".")) return out if isinstance(a, list): if len(a) != len(b): out.append(("VALUE_MISMATCH", f"{path}[len]", len(a), len(b))) for i, (x, y) in enumerate(zip(a, b)): out += classify(x, y, f"{path}[{i}]") return out if a != b: out.append(("VALUE_MISMATCH", path or "", a, b)) return out def capture(port): """Every seller-facing surface, as parsed JSON plus a sha256 of the raw bytes.""" snap = {} for name, path in SURFACES.items(): st, body = H.req(port, "GET", path) raw = json.dumps(body, sort_keys=True, separators=(",", ":")).encode() snap[name] = { "status": st, "body": body, "sha256": hashlib.sha256(raw).hexdigest(), } return snap def sold_row(snap): rows = snap["tradePile"]["body"].get("auctionInfo", []) return rows[0] if rows else {} def report_diff(label, a, b, allowed): """Print the classified diff and assert only `allowed` paths differ.""" diffs = classify(a, b) print(f" classified diff ({len(diffs)} finding(s)):") for kind, path, av, bv in diffs: print(f" {kind:15s} {path:34s} A={av!r} B={bv!r}") kinds = {k for k, _, _, _ in diffs} for bad in ("MISSING", "EXTRA", "TYPE_MISMATCH"): check(f"{label}: no {bad}", bad not in kinds, ", ".join(p for k, p, _, _ in diffs if k == bad) or "none") paths = sorted(p for _, p, _, _ in diffs) check(f"{label}: only {allowed} differ", paths == sorted(allowed), str(paths)) return diffs def restart_host(state, variant, coins_processed): """Same Core, same DBs, same settled sale — only the host's variant changes.""" if state.get("host"): state["host"].terminate() state["host"].wait(timeout=20) state["host"] = H.start_host(state["tmp"], state["host_port"], state["core_port"], state["host_log"], variant, coins_processed=coins_processed) return state["host"] def main(): ap = argparse.ArgumentParser() ap.add_argument("--out", default=os.path.join( REPO, "docs/evidence", f"sold-ab-{time.strftime('%Y-%m-%d')}")) ap.add_argument("--keep", action="store_true") args = ap.parse_args() os.makedirs(args.out, exist_ok=True) tmp = tempfile.mkdtemp(prefix="openfut-sold-ab-") state = {"tmp": tmp, "host": None} evidence = {"generated": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "variants": {}, "diffs": {}, "counts_states": {}} try: state["core_port"] = H.free_port() state["host_port"] = H.free_port() state["core_log"] = os.path.join(tmp, "core.log") state["host_log"] = os.path.join(tmp, "host.log") banner("ISOLATED STAGING") print(f" tmp {tmp}\n core 127.0.0.1:{state['core_port']}" f"\n host 127.0.0.1:{state['host_port']}") core, core_db = H.start_core(tmp, state["core_port"], state["core_log"]) state["core"] = core state["core_db"] = core_db mdb = os.path.join(tmp, "market.db") # ---- one authentic listing + one real settlement ------------------- restart_host(state, "highest", "0") con = sqlite3.connect(mdb) con.execute( "INSERT INTO listings (listing_id, card_id, core_item_id, wire_item_id, " "wire_resource_id, start_price, buy_now_price, owner, state, created_at, " "item_json, duration_secs) VALUES (?,?,?,?,?,?,?,?, 'active', ?, ?, ?)", (H.TRADE_ID, H.CARD, H.ITEM, 100000178, 212188, H.GROSS, H.GROSS, "CAGE", str(int(time.time() * 1000)), json.dumps({ "id": 100000178, "resourceId": 212188, "rating": 75, "preferredPosition": "ST", "itemState": "forSale", "untradeable": False, "assetId": 212188}), 3600)) con.commit(); con.close() banner(f"SETTLE — synthetic Buyer B buys at {H.GROSS}") out = subprocess.run( [os.path.join(REPO, "target/release/staging_sell"), "--market-db", mdb, "--core-url", f"http://127.0.0.1:{state['core_port']}", "--trade-id", H.TRADE_ID, "--item", H.ITEM, "--seller", H.SELLER_CLUB, "--buyer", H.BUYER_CLUB, "--gross", str(H.GROSS)], capture_output=True, text=True, timeout=120) print(" " + "\n ".join((out.stdout + out.stderr).strip().splitlines())) check("settlement succeeded", out.returncode == 0) own, n, coins = H.owner_of(core_db, H.ITEM) fee = H.GROSS * 5 // 100 check("ownership -> buyer", own == H.BUYER_CLUB, str(own)) check("exactly one authoritative instance", n == 1, str(n)) check("seller credited net", coins[H.SELLER_CLUB] == 1000 + H.GROSS - fee, str(coins[H.SELLER_CLUB])) check("buyer debited gross", coins[H.BUYER_CLUB] == 20000 - H.GROSS, str(coins[H.BUYER_CLUB])) economy = {"seller": coins[H.SELLER_CLUB], "buyer": coins[H.BUYER_CLUB], "fee": fee, "owner": own, "instances": n} evidence["settlement"] = economy # ---- PRIMARY A/B: bidState only ------------------------------------ banner("PRIMARY A/B — bidState highest vs buyNow (coinsProcessed held at 0)") snaps = {} for name, variant in (("A_highest", "highest"), ("B_buyNow", "buyNow")): restart_host(state, variant, "0") snaps[name] = capture(state["host_port"]) evidence["variants"][name] = { "env": {"OPENFUT_FIFA17_SOLD_EXPERIMENT": variant, "OPENFUT_FIFA17_SOLD_COINS_PROCESSED": "0"}, "surfaces": snaps[name], } row = sold_row(snaps[name]) print(f" {name}: tradeState={row.get('tradeState')} " f"bidState={row.get('bidState')} currentBid={row.get('currentBid')} " f"coinsProcessed={row.get('coinsProcessed')} atoms={len(row)}") # Every variant must independently be a well-formed sold row. check(f"{name}: tradeState closed", row.get("tradeState") == "closed") check(f"{name}: twelve atoms", len(row) == 12, str(len(row))) check(f"{name}: currentBid == gross", row.get("currentBid") == H.GROSS) check(f"{name}: expires 0", row.get("expires") == 0) check(f"{name}: counts.sold == 1", snaps[name]["counts"]["body"].get("sold") == 1) evidence["diffs"]["primary_bidstate"] = [ {"kind": k, "path": p, "a": av, "b": bv} for k, p, av, bv in report_diff( "PRIMARY", snaps["A_highest"]["tradePile"]["body"], snaps["B_buyNow"]["tradePile"]["body"], ["auctionInfo[0].bidState"]) ] # /trade/status must move in lockstep, or the screen would contradict itself. report_diff("PRIMARY /trade/status", snaps["A_highest"]["trade_status_filtered"]["body"], snaps["B_buyNow"]["trade_status_filtered"]["body"], ["auctionInfo[0].bidState"]) report_diff("PRIMARY counts", snaps["A_highest"]["counts"]["body"], snaps["B_buyNow"]["counts"]["body"], []) # ---- ORTHOGONAL A/B: coinsProcessed only --------------------------- banner("ORTHOGONAL A/B — coinsProcessed 0 vs 1 (bidState held at highest)") for name, cp in (("C_cp0", "0"), ("D_cp1", "1")): restart_host(state, "highest", cp) snaps[name] = capture(state["host_port"]) evidence["variants"][name] = { "env": {"OPENFUT_FIFA17_SOLD_EXPERIMENT": "highest", "OPENFUT_FIFA17_SOLD_COINS_PROCESSED": cp}, "surfaces": snaps[name], } row = sold_row(snaps[name]) print(f" {name}: bidState={row.get('bidState')} " f"coinsProcessed={row.get('coinsProcessed')}") evidence["diffs"]["orthogonal_coinsprocessed"] = [ {"kind": k, "path": p, "a": av, "b": bv} for k, p, av, bv in report_diff( "ORTHOGONAL", snaps["C_cp0"]["tradePile"]["body"], snaps["D_cp1"]["tradePile"]["body"], ["auctionInfo[0].coinsProcessed"]) ] check("C_cp0 is identical to A_highest (same env => same bytes)", snaps["C_cp0"]["tradePile"]["sha256"] == snaps["A_highest"]["tradePile"]["sha256"], "reproducible") # ---- counts states the live run must interpret ---------------------- banner("COUNTS STATES") restart_host(state, "highest", "0") s1 = H.req(state["host_port"], "GET", SURFACES["counts"])[1] print(f" S1 0 active + 1 sold -> {json.dumps(s1)}") con = sqlite3.connect(mdb) con.execute( "INSERT INTO listings (listing_id, card_id, core_item_id, wire_item_id, " "wire_resource_id, start_price, buy_now_price, owner, state, created_at, " "item_json, duration_secs) VALUES (?,?,?,?,?,?,?,?, 'active', ?, ?, ?)", ("900500999", H.CARD, "core-second-x", 100000179, 212188, 300, 400, "CAGE", str(int(time.time() * 1000)), json.dumps({ "id": 100000179, "resourceId": 212188, "rating": 75, "preferredPosition": "ST", "itemState": "forSale", "untradeable": False, "assetId": 212188}), 3600)) con.commit(); con.close() s2_active = H.req(state["host_port"], "GET", SURFACES["counts"])[1] print(f" S2 1 active + 1 sold -> {json.dumps(s2_active)} (count mode: active)") restart_host(state, "highest", "0") os.environ["_"] = "_" # no-op; count mode is a host env, set below state["host"].terminate(); state["host"].wait(timeout=20) state["host"] = H.start_host(tmp, state["host_port"], state["core_port"], state["host_log"], "highest", coins_processed="0", count_mode="active_plus_sold") s2_member = H.req(state["host_port"], "GET", SURFACES["counts"])[1] print(f" S2 1 active + 1 sold -> {json.dumps(s2_member)} (count mode: membership)") evidence["counts_states"] = { "S1_zero_active_one_sold": s1, "S2_one_active_one_sold_count_active": s2_active, "S2_one_active_one_sold_count_membership": s2_member, } check("S1 reports sold 1 / selling 0", s1.get("sold") == 1 and s1.get("selling") == 0) check("S2 reports sold 1 / selling 1", s2_active.get("sold") == 1 and s2_active.get("selling") == 1) check("S2 count differs by mode (1 vs 2) — the open question for the client", s2_active.get("count") == 1 and s2_member.get("count") == 2, f"{s2_active.get('count')} vs {s2_member.get('count')}") # ---- emit evidence -------------------------------------------------- banner("EVIDENCE") for name, v in evidence["variants"].items(): path = os.path.join(args.out, f"{name}.json") with open(path, "w") as f: json.dump(v, f, indent=2, sort_keys=True) print(f" {os.path.relpath(path, REPO)} " f"tradePile sha256={v['surfaces']['tradePile']['sha256'][:16]}…") idx = os.path.join(args.out, "differential.json") with open(idx, "w") as f: json.dump(evidence, f, indent=2, sort_keys=True) print(f" {os.path.relpath(idx, REPO)}") banner("PRODUCTION UNTOUCHED") alive = subprocess.run(["ps", "-o", "pid=", "-p", "3631953"], capture_output=True, text=True).stdout.strip() check("prod-host pid 3631953 alive", alive == "3631953", alive or "gone") banner("RESULT") passed = sum(1 for _, ok, _ in results if ok) failed = [l for l, ok, _ in results if not ok] print(f" {passed}/{len(results)} checks passed") if failed: print(" FAILED: " + "; ".join(failed)) print("\n A/B IS CONTAMINATED — do NOT run the live experiment") return 1 print("\n A/B IS CLEAN — one-field attribution proven; ready for the live client") return 0 finally: for k in ("host", "core"): p = state.get(k) if p: try: p.terminate(); p.wait(timeout=15) except Exception: try: p.kill() except Exception: pass if args.keep: print(f"\n kept {tmp}") else: shutil.rmtree(tmp, ignore_errors=True) if __name__ == "__main__": sys.exit(main())