#!/usr/bin/env python3 """Recompute the GAIP Agent Web Index figures from GAIP's public JSON. Standard library only (Python 3.9+). What it does, for one month (method: agent_economy/docs/agent-web-index-queries.md, served at https://www.gaipagents.com/v1/free/agent-web-index/queries): 1. Each figure (tools changed, terms or prices changed, signed agent cards, AI-preference signals), for all watched agents and for the frozen panel AGENTS-COHORT-2026-10: adds up the published weekly parts (``weekly_totals.numerator_by_week`` and ``denominator_by_week``; the tools figure's denominator is not split and is taken as published), applies the rule that a count is shown only when it and its total are at least 5, and compares the result with the published count, total and share. 2. Failure types: recomputes each share from the published counts and checks the order. The monthly counter row behind it is not public (only its hash is), so this is an arithmetic check, not a recomputation. 3. Registry size: checks (d) read in the last 7 days <= (c) watched <= (b) remote <= (a) entries and that the published ``reconciles`` flag agrees; with ``--snapshot`` files (weekly snapshots from /v1/free/ecosystem/snapshots?week=YYYY-Www), compares (a) and (b) with the latest snapshot's walk record when the index says it took them from a weekly snapshot. Usage: curl -s 'https://www.gaipagents.com/v1/free/agent-web-index?month=2026-10' > index.json curl -s 'https://www.gaipagents.com/v1/free/ecosystem/snapshots?week=2026-W44' > w44.json python3 reproduce_agent_web_index.py index.json --snapshot w44.json Prints a JSON report; exits 1 if any recomputed figure differs from the published one, else 0. A figure whose weekly parts are withheld (a part under 5) is reported as NOT_REPRODUCIBLE, never as a match. """ from __future__ import annotations import argparse import json import sys from typing import Any, Iterable MIN_CELL_SIZE = 5 REGISTRY_SOURCE = "official-mcp-registry" MATCH, MISMATCH, NOT_REPRODUCIBLE = "MATCH", "MISMATCH", "NOT_REPRODUCIBLE" def cell(count: int, total: int) -> dict[str, Any]: """GAIP's cell rule: a count is shown only when it and its total are at least MIN_CELL_SIZE.""" if count < MIN_CELL_SIZE or total < MIN_CELL_SIZE: return {"count": None, "of": total if total >= MIN_CELL_SIZE else None, "share": None, "withheld": f"under {MIN_CELL_SIZE}"} return {"count": count, "of": total, "share": round(count / total, 4)} def _same(recomputed: dict[str, Any], published: Any) -> bool: published = published if isinstance(published, dict) else {} return all(recomputed.get(k) == published.get(k) for k in ("count", "of", "share")) def recompute_population(block: Any) -> dict[str, Any]: """One population of one figure (``all_agents`` or ``panel``) from its weekly parts.""" if not isinstance(block, dict) or "value" not in block: return {"status": NOT_REPRODUCIBLE, "why": "not in this copy"} parts = block.get("weekly_totals") or {} published = block.get("value") or {} if "numerator_by_week" not in parts: return {"status": NOT_REPRODUCIBLE, "why": parts.get("withheld") or "no weekly parts published", "published": published} count = sum(int(v) for v in parts["numerator_by_week"].values()) if "denominator_by_week" in parts: total = sum(int(v) for v in parts["denominator_by_week"].values()) basis = "sum of weekly parts (numerator and denominator)" else: total = published.get("of") if not isinstance(total, int): return {"status": NOT_REPRODUCIBLE, "why": "denominator not published (under 5)", "published": published} basis = "sum of weekly numerator parts; denominator as published (watched agents when the copy was built)" recomputed = cell(count, total) return {"status": MATCH if _same(recomputed, published) else MISMATCH, "basis": basis, "recomputed": recomputed, "published": published} def recompute_failure_types(figure: dict[str, Any]) -> dict[str, Any]: total = figure.get("outside_diagnoses") top = figure.get("top") or [] if not isinstance(total, int): return {"status": NOT_REPRODUCIBLE, "why": "fewer than 5 diagnoses: nothing is published"} problems = [] for item in top: if not _same(cell(int(item.get("count") or 0), total), item): problems.append(item.get("class")) counts = [int(item.get("count") or 0) for item in top] if counts != sorted(counts, reverse=True): problems.append("order") return {"status": MISMATCH if problems else MATCH, "problems": problems, "basis": ("shares and order recomputed from the published counts; the monthly counter row is not " "public, so the counts themselves cannot be recomputed from outside")} def _snapshot_walk(snapshots: Iterable[dict[str, Any]]) -> dict[str, Any] | None: latest = None for snapshot in snapshots: snapshot = snapshot.get("snapshot", snapshot) if isinstance(snapshot, dict) else {} seen = ((snapshot.get("market_coverage") or {}).get("registry_total_seen") or {}).get(REGISTRY_SOURCE) if isinstance(seen, dict) and (latest is None or str(snapshot.get("week")) >= latest[0]): latest = (str(snapshot.get("week")), seen) return latest[1] if latest else None def check_registry(block: Any, snapshots: Iterable[dict[str, Any]] = ()) -> dict[str, Any]: block = block if isinstance(block, dict) else {} a = (block.get("entries") or {}).get("count") b = (block.get("remote") or {}).get("count") c = (block.get("watched") or {}).get("count") d = (block.get("read_last_7_days") or {}).get("count") order = [(d, c), (c, b), (b, a)] known = [(x, y) for x, y in order if isinstance(x, int) and isinstance(y, int)] holds = all(x <= y for x, y in known) if isinstance(a, int) else None out: dict[str, Any] = {"entries": a, "remote": b, "watched": c, "read_last_7_days": d, "d<=c<=b<=a": holds, "published_reconciles": block.get("reconciles")} out["status"] = MATCH if holds == block.get("reconciles") else MISMATCH walk = _snapshot_walk(snapshots) if walk is not None: listed = walk.get("listed_by_registry_total") or walk.get("listed_in_last_full_pass") out["snapshot_entries"] = listed out["snapshot_remote"] = walk.get("public_https_in_last_full_pass") if (block.get("entries") or {}).get("from") == "weekly snapshot": if listed != a: out["status"] = MISMATCH else: out["snapshot_note"] = ("The index took (a) from the walk's own record, which can be newer than the " "latest weekly snapshot; compared for information only.") return out FIGURES = ("tools_changed", "terms_or_prices_changed", "signed_agent_cards", "ai_preference_adoption") def reproduce(index: dict[str, Any], snapshots: Iterable[dict[str, Any]] = ()) -> dict[str, Any]: """The full report for one index document (the /v1/free/agent-web-index JSON or a report holding it).""" index = index.get("agent_web_index", index) if isinstance(index, dict) else {} figures = {f.get("id"): f for f in index.get("figures") or [] if isinstance(f, dict)} report: dict[str, Any] = {"month": index.get("month"), "status_of_month": index.get("status"), "figures": {}} for fid in FIGURES: figure = figures.get(fid) if not isinstance(figure, dict) or figure.get("unavailable"): report["figures"][fid] = {"status": NOT_REPRODUCIBLE, "why": "not available in this copy"} continue report["figures"][fid] = {"all_agents": recompute_population(figure.get("all_agents")), "panel": recompute_population(figure.get("panel"))} failures = figures.get("failure_types") report["figures"]["failure_types"] = (recompute_failure_types(failures) if isinstance(failures, dict) and not failures.get("unavailable") else {"status": NOT_REPRODUCIBLE, "why": "not available in this copy"}) report["registry_size"] = check_registry((index.get("sample_frame") or {}).get("official_mcp_registry"), list(snapshots)) statuses = [report["registry_size"]["status"]] for value in report["figures"].values(): statuses += [v["status"] for v in value.values()] if "status" not in value else [value["status"]] report["mismatches"] = statuses.count(MISMATCH) report["not_reproducible"] = statuses.count(NOT_REPRODUCIBLE) report["matches"] = statuses.count(MATCH) return report def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Recompute the GAIP Agent Web Index from its public JSON.") parser.add_argument("index", help="the saved /v1/free/agent-web-index?month=YYYY-MM JSON") parser.add_argument("--snapshot", action="append", default=[], help="a saved /v1/free/ecosystem/snapshots?week=YYYY-Www JSON (repeatable)") args = parser.parse_args(argv) with open(args.index, encoding="utf-8") as handle: index = json.load(handle) snapshots = [] for path in args.snapshot: with open(path, encoding="utf-8") as handle: snapshots.append(json.load(handle)) report = reproduce(index, snapshots) json.dump(report, sys.stdout, indent=2, sort_keys=True) sys.stdout.write("\n") return 1 if report["mismatches"] else 0 if __name__ == "__main__": raise SystemExit(main())