#!/usr/bin/env python3 """Build and verify MEGA EVIDENCE 150 from public Business Arena replays. The collector reads the public replay bundles published by Accio's Business Arena, but writes only compact hashes, counts, and ledger roots. It deliberately excludes model reasoning/conversation text. Source bundles remain at the publisher URL. Standard-library only; a full 150-run collection is about 48 MB compressed input. """ from __future__ import annotations import argparse import concurrent.futures import gzip import hashlib import json import sys import urllib.request from collections import Counter from dataclasses import dataclass from pathlib import Path from typing import Any, Iterable SCHEMA = "neoalibaba-mega-evidence-150-v1" BASE_URL = "https://business-arena.site.accio.ai/" INDEX_PATH = "shop-runs/index.json" USER_AGENT = "neoalibaba-proof/1.0 (+https://cotrugli.tech/)" # Reads are evidence too, but only these calls change marketplace state. WRITE_ACTIONS = frozenset( { "buy_supplier", "suppliers_buy", "request_quote", "supplier_quotes_request", "respond_quote", "list_product_on", "list_product", "update_listing", "update_listings", "set_price_tiers", "upgrade_listing_content", "set_default_incoterm", "liquidate_inventory", "liquidation", "set_store_focus", "apply_certification", "reply_inquiry", "inquiries_reply", "respond_rfq", "rfqs_respond", "resolve_dispute", "respond_to_return", "ad_budget", "set_ad_budget", "launch_marketing_campaign", "launch_promotion", "create_promotion", "cancel_promotion", "buy_fx_hedge", "factor_ar", "repay_loan", "loans_borrow", "counter", "accept", "decline", } ) VALUE_FIELDS = ("delta_cash", "delta_escrow", "delta_receivable", "delta_payable") DOMAIN_LEAF = b"NEOALIBABA/LEAF/V1\0" DOMAIN_NODE = b"NEOALIBABA/NODE/V1\0" DOMAIN_EMPTY = b"NEOALIBABA/EMPTY/V1" DOMAIN_MEGA = b"NEOALIBABA/MEGA/V1\0" def canonical_bytes(value: Any) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode("utf-8") def sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def leaf_hash(domain: str, value: Any) -> bytes: return hashlib.sha256(DOMAIN_LEAF + domain.encode("utf-8") + b"\0" + canonical_bytes(value)).digest() def merkle_root(leaves: Iterable[bytes]) -> str: level = list(leaves) if not level: return sha256_hex(DOMAIN_EMPTY) while len(level) > 1: if len(level) % 2: level.append(level[-1]) level = [ hashlib.sha256(DOMAIN_NODE + level[i] + level[i + 1]).digest() for i in range(0, len(level), 2) ] return level[0].hex() def root_rows(domain: str, rows: Iterable[Any]) -> str: return merkle_root(leaf_hash(domain, row) for row in rows) def fetch_bytes(base_url: str, path: str, timeout: int = 90) -> bytes: request = urllib.request.Request(base_url + path, headers={"User-Agent": USER_AGENT}) with urllib.request.urlopen(request, timeout=timeout) as response: return response.read() def endpoint_rows(api: dict[str, Any], needle: str, field: str) -> list[dict[str, Any]]: matches: list[tuple[str, list[dict[str, Any]]]] = [] for key, value in api.items(): rows = value.get(field) if isinstance(value, dict) else None if needle in key and isinstance(rows, list): matches.append((key, rows)) if not matches: return [] # The public bundle can carry multiple views. Prefer the largest complete view. return max(matches, key=lambda pair: len(pair[1]))[1] def safe_action(row: dict[str, Any]) -> dict[str, Any]: """Keep tool I/O only; never serialize reasoning/conversation content.""" return { "action": row.get("action"), "day": row.get("day"), "payload": row.get("payload", {}), "result": row.get("result", {}), "ts": row.get("ts"), } def operation_projection(episode_id: str, actions: list[dict[str, Any]]) -> list[dict[str, Any]]: projected = [] for ordinal, original in enumerate(actions): row = safe_action(original) if row["action"] not in WRITE_ACTIONS: continue projected.append( { "schema": "neo_trade_operation_v1", "source": "business-arena-public-replay", "episode_id": episode_id, "ordinal": ordinal, "action": row["action"], "day": row["day"], "ts": row["ts"], "payload_sha256": sha256_hex(canonical_bytes(row["payload"])), "result_sha256": sha256_hex(canonical_bytes(row["result"])), "synthetic": True, } ) return projected def entity_projection( episode_id: str, kind: str, rows: list[dict[str, Any]] ) -> list[dict[str, Any]]: return [ { "schema": "neo_trade_entity_v1", "source": "business-arena-public-replay", "episode_id": episode_id, "kind": kind, "ordinal": ordinal, "source_row_sha256": sha256_hex(canonical_bytes(row)), "synthetic": True, } for ordinal, row in enumerate(rows) ] def value_projection(episode_id: str, transactions: list[dict[str, Any]]) -> list[dict[str, Any]]: """Project source balance deltas into balanced synthetic journal pairs. The source unit is intentionally named BUSINESS_ARENA_SIM. A future CLU policy may transform it, but this proof never relabels source money silently. """ entries: list[dict[str, Any]] = [] for ordinal, row in enumerate(transactions): source_hash = sha256_hex(canonical_bytes(row)) for account in VALUE_FIELDS: amount = row.get(account, 0) if not isinstance(amount, (int, float)) or amount == 0: continue common = { "schema": "neo_trade_value_entry_v1", "source": "business-arena-public-replay", "episode_id": episode_id, "transaction_ordinal": ordinal, "day": row.get("day"), "kind": row.get("kind"), "ref_id": row.get("ref_id"), "unit": "BUSINESS_ARENA_SIM", "source_row_sha256": source_hash, "synthetic": True, } entries.append({**common, "account": account.removeprefix("delta_"), "amount": amount}) entries.append({**common, "account": "arena_world:" + account.removeprefix("delta_"), "amount": -amount}) return entries def projection_balance(entries: list[dict[str, Any]]) -> float: return round(sum(float(entry["amount"]) for entry in entries), 8) @dataclass(frozen=True) class EpisodeProof: value: dict[str, Any] def build_episode_proof(run: dict[str, Any], bundle_bytes: bytes) -> EpisodeProof: episode_id = str(run["episode_id"]) raw = gzip.decompress(bundle_bytes) bundle = json.loads(raw) if bundle.get("episode_id") != episode_id: raise ValueError(f"episode mismatch: expected {episode_id!r}") api = bundle.get("api") if not isinstance(api, dict): raise ValueError(f"missing api map in {episode_id}") actions = endpoint_rows(api, "/api/agent/action_log/", "actions") transactions = endpoint_rows(api, "/api/transactions/", "transactions") orders = endpoint_rows(api, "/api/orders/", "orders") inquiries = endpoint_rows(api, "/api/inquiries/", "inquiries") rfqs = endpoint_rows(api, "/api/rfqs/", "rfqs") safe_actions = [safe_action(row) for row in actions] operations = operation_projection(episode_id, actions) entities = [] for kind, rows in (("ORDER", orders), ("INQUIRY", inquiries), ("RFQ", rfqs)): entities.extend(entity_projection(episode_id, kind, rows)) values = value_projection(episode_id, transactions) balance = projection_balance(values) if balance != 0: raise ValueError(f"unbalanced value projection in {episode_id}: {balance}") return EpisodeProof( { "episode_id": episode_id, "run_name": run.get("run_name") or run.get("name"), "family": run.get("family"), "completion_status": run.get("completion_status"), "source_bundle": f"shop-runs/{episode_id}.json.gz", "source_bundle_sha256": sha256_hex(bundle_bytes), "source_compressed_bytes": len(bundle_bytes), "source_uncompressed_bytes": len(raw), "counts": { "action_records": len(actions), "state_changing_actions": len(operations), "orders": len(orders), "financial_transactions": len(transactions), "customer_inquiries": len(inquiries), "rfqs": len(rfqs), "balanced_value_entries": len(values), }, "roots": { "evidence_actions_sha256": root_rows("ACTION", safe_actions), "operations_sha256": root_rows("OPERATION", operations + entities), "source_transactions_sha256": root_rows("SOURCE_TRANSACTION", transactions), "value_entries_sha256": root_rows("VALUE_ENTRY", values), }, "value_projection_balance": balance, } ) def aggregate_episode_proofs( episodes: list[dict[str, Any]], index_bytes: bytes, base_url: str ) -> dict[str, Any]: episodes = sorted(episodes, key=lambda item: item["episode_id"]) totals: Counter[str] = Counter() for episode in episodes: totals.update(episode["counts"]) source_root = root_rows( "SOURCE_BUNDLE", [ { "episode_id": episode["episode_id"], "sha256": episode["source_bundle_sha256"], "bytes": episode["source_compressed_bytes"], } for episode in episodes ], ) evidence_root = root_rows( "EPISODE_EVIDENCE_ROOT", [{"episode_id": e["episode_id"], "root": e["roots"]["evidence_actions_sha256"]} for e in episodes], ) operations_root = root_rows( "EPISODE_OPERATIONS_ROOT", [{"episode_id": e["episode_id"], "root": e["roots"]["operations_sha256"]} for e in episodes], ) value_root = root_rows( "EPISODE_VALUE_ROOT", [{"episode_id": e["episode_id"], "root": e["roots"]["value_entries_sha256"]} for e in episodes], ) root_payload = { "schema": SCHEMA, "source_index_sha256": sha256_hex(index_bytes), "source_root_sha256": source_root, "evidence_root_sha256": evidence_root, "operations_root_sha256": operations_root, "value_root_sha256": value_root, "episodes": len(episodes), } mega_root = sha256_hex(DOMAIN_MEGA + canonical_bytes(root_payload)) return { "schema": SCHEMA, "status": "OBSERVED_PUBLIC_REPLAY_PROJECTION", "source": { "publisher": "Accio Team, Alibaba Group", "project": "Business Arena", "base_url": base_url, "index_path": INDEX_PATH, "index_sha256": sha256_hex(index_bytes), "access": "PUBLIC_READ_ONLY_REPLAY", }, "scope": { "episodes": len(episodes), "simulated_days_per_episode": 30, "simulated_business_days": len(episodes) * 30, "reasoning_or_conversation_exported": False, }, "totals": dict(sorted(totals.items())), "roots": {**root_payload, "mega_root_sha256": mega_root}, "episodes": episodes, } def collect(base_url: str = BASE_URL, limit: int | None = None, workers: int = 4) -> dict[str, Any]: index_bytes = fetch_bytes(base_url, INDEX_PATH) index = json.loads(index_bytes) runs = index.get("runs") if not isinstance(runs, list): raise ValueError("Business Arena index has no runs list") if limit is not None: runs = runs[:limit] def fetch_run(run: dict[str, Any]) -> dict[str, Any]: episode_id = str(run["episode_id"]) bundle = fetch_bytes(base_url, f"shop-runs/{episode_id}.json.gz") return build_episode_proof(run, bundle).value episodes: list[dict[str, Any]] = [] with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, workers)) as executor: futures = [executor.submit(fetch_run, run) for run in runs] for index_no, future in enumerate(concurrent.futures.as_completed(futures), 1): episodes.append(future.result()) if index_no % 10 == 0 or index_no == len(futures): print(f"[neoalibaba] collected {index_no}/{len(futures)}", file=sys.stderr, flush=True) return aggregate_episode_proofs(episodes, index_bytes, base_url) def verify(proof: dict[str, Any]) -> list[str]: errors: list[str] = [] if proof.get("schema") != SCHEMA: errors.append("wrong schema") return errors episodes = proof.get("episodes") if not isinstance(episodes, list): return ["episodes is not a list"] if len({episode.get("episode_id") for episode in episodes}) != len(episodes): errors.append("duplicate episode_id") for episode in episodes: if episode.get("value_projection_balance") != 0: errors.append(f"unbalanced episode {episode.get('episode_id')}") source = proof.get("source", {}) try: index_hash = source["index_sha256"] source_root = root_rows( "SOURCE_BUNDLE", [ { "episode_id": e["episode_id"], "sha256": e["source_bundle_sha256"], "bytes": e["source_compressed_bytes"], } for e in sorted(episodes, key=lambda item: item["episode_id"]) ], ) evidence_root = root_rows( "EPISODE_EVIDENCE_ROOT", [ {"episode_id": e["episode_id"], "root": e["roots"]["evidence_actions_sha256"]} for e in sorted(episodes, key=lambda item: item["episode_id"]) ], ) operations_root = root_rows( "EPISODE_OPERATIONS_ROOT", [ {"episode_id": e["episode_id"], "root": e["roots"]["operations_sha256"]} for e in sorted(episodes, key=lambda item: item["episode_id"]) ], ) value_root = root_rows( "EPISODE_VALUE_ROOT", [ {"episode_id": e["episode_id"], "root": e["roots"]["value_entries_sha256"]} for e in sorted(episodes, key=lambda item: item["episode_id"]) ], ) except (KeyError, TypeError) as exc: return errors + [f"malformed proof: {exc}"] expected_totals: Counter[str] = Counter() for episode in episodes: expected_totals.update(episode["counts"]) if dict(sorted(expected_totals.items())) != proof.get("totals"): errors.append("totals mismatch") roots = proof.get("roots", {}) expected = { "schema": SCHEMA, "source_index_sha256": index_hash, "source_root_sha256": source_root, "evidence_root_sha256": evidence_root, "operations_root_sha256": operations_root, "value_root_sha256": value_root, "episodes": len(episodes), } for key, value in expected.items(): if roots.get(key) != value: errors.append(f"root field mismatch: {key}") expected_mega = sha256_hex(DOMAIN_MEGA + canonical_bytes(expected)) if roots.get("mega_root_sha256") != expected_mega: errors.append("mega root mismatch") if proof.get("scope", {}).get("episodes") != len(episodes): errors.append("scope episode count mismatch") if proof.get("scope", {}).get("simulated_business_days") != len(episodes) * 30: errors.append("simulated business days mismatch") if proof.get("scope", {}).get("reasoning_or_conversation_exported") is not False: errors.append("reasoning export boundary missing") return errors def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="command", required=True) collect_parser = sub.add_parser("collect", help="fetch public bundles and build compact proof") collect_parser.add_argument("--base-url", default=BASE_URL) collect_parser.add_argument("--limit", type=int) collect_parser.add_argument("--workers", type=int, default=4) collect_parser.add_argument("--output", type=Path, required=True) verify_parser = sub.add_parser("verify", help="verify a compact proof without network") verify_parser.add_argument("proof", type=Path) args = parser.parse_args(argv) if args.command == "collect": proof = collect(args.base_url, args.limit, args.workers) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(proof, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") errors = verify(proof) if errors: print("INVALID: " + "; ".join(errors), file=sys.stderr) return 1 print(json.dumps({"status": "VERIFIED", "output": str(args.output), "mega_root_sha256": proof["roots"]["mega_root_sha256"], "totals": proof["totals"]}, sort_keys=True)) return 0 proof = json.loads(args.proof.read_text(encoding="utf-8")) errors = verify(proof) if errors: print("INVALID: " + "; ".join(errors)) return 1 print(json.dumps({"status": "VERIFIED", "mega_root_sha256": proof["roots"]["mega_root_sha256"], "episodes": len(proof["episodes"])}, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())