Upload jobs/analyze_collapse.py

jobs/analyze_collapse.py

@@ -0,0 +1,355 @@
+#!/usr/bin/env python3
+"""
+OCC Collapse Mechanism — Post-Run Analysis Harness
+===================================================
+Reads the mechanism isolation JSON output and produces:
+  - condition_summary.csv
+  - per_topic_outcomes.csv
+  - round_flip_matrix.csv
+  - honest_retention_by_round.csv
+  - adversary_induced_flips.csv
+  - fig_accuracy_by_condition.png
+  - fig_honest_retention.png
+  - fig_flip_rate.png
+  - fig_adversary_skill.png
+  - hypothesis_verdicts.json
+
+Uses PRE-REGISTERED win conditions defined BEFORE seeing results.
+
+Usage:
+  python analyze_collapse.py /path/to/debate_collapse_mechanism_results.json
+"""
+
+import json, csv, sys
+from pathlib import Path
+
+# ─── PRE-REGISTERED WIN CONDITIONS ───
+# Defined BEFORE seeing results to prevent post-hoc storytelling.
+
+HYPOTHESIS_RULES = {
+    "H1_volume_amplification": {
+        "description": "Collapse is driven by adversarial compute volume, not mere presence.",
+        "confirmed_if": "equal_token_unequal_turn.mean_accuracy > baseline_1round.mean_accuracy - 0.10",
+        "refuted_if": "equal_token_unequal_turn.mean_accuracy <= equal_3round.mean_accuracy + 0.05",
+        "test_condition": "equal_token_unequal_turn",
+        "baseline_condition": "equal_3round_traced",
+        "ref_baseline": "baseline_1round_traced",
+    },
+    "H2_turn_order": {
+        "description": "Collapse depends on adversary speaking last/late.",
+        "confirmed_if": "randomized_order_3round.mean_accuracy > equal_3round.mean_accuracy + 0.10",
+        "refuted_if": "abs(randomized_order_3round.mean_accuracy - equal_3round.mean_accuracy) < 0.05",
+        "test_condition": "randomized_order_3round",
+        "baseline_condition": "equal_3round_traced",
+    },
+    "H3_voting_vulnerability": {
+        "description": "Majority voting is fragile; judge or confidence-weighted voting resists collapse.",
+        "confirmed_if": "max(judge_vote_3round.mean, confidence_weighted_3round.mean) > equal_3round.mean + 0.10",
+        "refuted_if": "abs(judge_vote_3round.mean - equal_3round.mean) < 0.05 and abs(confidence_weighted_3round.mean - equal_3round.mean) < 0.05",
+        "test_conditions": ["judge_vote_3round", "confidence_weighted_3round"],
+        "baseline_condition": "equal_3round_traced",
+    },
+    "H4_contamination": {
+        "description": "Honest agents copy adversary answers after exposure.",
+        "confirmed_if": "honest_retention_rate_round3 < 0.5",
+        "refuted_if": "honest_retention_rate_round3 > 0.7",
+        "source": "equal_3round_traced.honest_retention_round3 / (n_topics * 3)",
+    },
+    "H5_confidence_distortion": {
+        "description": "Adversary causes honest agents to become uncertain, not persuaded.",
+        "confirmed_if": "confidence_weighted_3round.mean > equal_3round.mean + 0.10",
+        "refuted_if": "abs(confidence_weighted_3round.mean - equal_3round.mean) < 0.05",
+        "test_condition": "confidence_weighted_3round",
+        "baseline_condition": "equal_3round_traced",
+    },
+    "H6_skill_dependency": {
+        "description": "Collapse depends on adversary skill; weak adversary does not collapse.",
+        "confirmed_if": "adversary_weak.mean > equal_3round.mean + 0.10 and adversary_oracle.mean <= equal_3round.mean + 0.05",
+        "refuted_if": "adversary_weak.mean <= equal_3round.mean + 0.05",
+        "test_weak": "adversary_weak",
+        "test_strong": "adversary_strong",
+        "test_oracle": "adversary_oracle",
+        "baseline_condition": "equal_3round_traced",
+    },
+    "H7_topic_vulnerability": {
+        "description": "Collapse clusters by question difficulty or ambiguity.",
+        "confirmed_if": "Some topics show collapse while others are robust in equal_3round",
+        "refuted_if": "All topics show similar collapse magnitude",
+        "needs": "per_topic analysis",
+    },
+}
+
+
+def load_data(path):
+    with open(path) as f:
+        return json.load(f)
+
+
+def make_summary_table(data):
+    summary = data.get("summary", {})
+    return [
+        {"condition": name, "mean_accuracy": round(s["mean"], 4),
+         "min_accuracy": round(s["min"], 4), "max_accuracy": round(s["max"], 4),
+         "range": round(s["max"] - s["min"], 4)}
+        for name, s in summary.items()
+    ]
+
+
+def make_retention_table(data):
+    rows = []
+    for seed_str, seed_data in data.get("seeds", {}).items():
+        traced = seed_data.get("equal_3round_traced", {})
+        if not traced:
+            continue
+        for rd in [2, 3]:
+            rows.append({
+                "seed": int(seed_str), "round": rd,
+                "stayed": traced.get(f"honest_retention_round{rd}", 0),
+                "flipped_away": traced.get(f"flipped_away_round{rd}", 0),
+                "flipped_toward": traced.get(f"flipped_toward_round{rd}", 0),
+                "adversary_flips": traced.get("adversary_flips", 0),
+            })
+    return rows
+
+
+def make_flip_matrix(data):
+    flip_rows, adv_rows = [], []
+    for seed_str, seed_data in data.get("seeds", {}).items():
+        traced = seed_data.get("equal_3round_traced", {})
+        for tkey, count in traced.get("transitions", {}).items():
+            parts = tkey.split("_")
+            if len(parts) == 2 and parts[0].startswith("h") and parts[1].startswith("h"):
+                try:
+                    r1, r3 = int(parts[0][1:]), int(parts[1][1:])
+                    flip_rows.append({
+                        "seed": int(seed_str), "transition": tkey,
+                        "round1_correct": r1, "round3_correct": r3,
+                        "count": count,
+                        "flip_direction": "stable" if r1 == r3 else ("degraded" if r1 > r3 else "improved"),
+                    })
+                except ValueError:
+                    pass
+        adv_rows.append({"seed": int(seed_str), "total_adversary_flips": traced.get("adversary_flips", 0)})
+    return flip_rows, adv_rows
+
+
+def evaluate_hypotheses(data):
+    summary = data.get("summary", {})
+    verdicts = {}
+    # Extract retention data from first seed
+    first_seed = list(data.get("seeds", {}).values())[0] if data.get("seeds") else {}
+    traced = first_seed.get("equal_3round_traced", {})
+    retention_r3 = traced.get("honest_retention_round3", 0)
+    flipped_r3 = traced.get("flipped_away_round3", 0)
+    total_r3 = retention_r3 + flipped_r3
+    retention_rate = retention_r3 / total_r3 if total_r3 > 0 else 1.0
+
+    for h_name, rules in HYPOTHESIS_RULES.items():
+        try:
+            v = {"hypothesis": h_name, "description": rules["description"], "verdict": "INCONCLUSIVE", "reason": "", "values": {}}
+
+            if h_name == "H1_volume_amplification":
+                test = summary.get("equal_token_unequal_turn", {}).get("mean", 0)
+                collapse = summary.get("equal_3round_traced", {}).get("mean", 0)
+                base1 = summary.get("baseline_1round_traced", {}).get("mean", 1)
+                v["values"] = {"baseline_1round": round(base1,4), "collapse": round(collapse,4), "equal_token": round(test,4)}
+                if test > base1 - 0.10:
+                    v["verdict"] = "CONFIRMED"
+                    v["reason"] = f"Equal-token recovered to {test:.3f}, within 10pp of baseline {base1:.3f}. Volume amplification is primary mechanism."
+                elif test <= collapse + 0.05:
+                    v["verdict"] = "REFUTED"
+                    v["reason"] = f"Equal-token at {test:.3f} barely above collapse {collapse:.3f}. Volume alone insufficient."
+                else:
+                    v["verdict"] = "PARTIAL"
+                    v["reason"] = f"Partial recovery to {test:.3f} from {collapse:.3f}. Volume is a factor but not the only one."
+
+            elif h_name == "H2_turn_order":
+                test = summary.get("randomized_order_3round", {}).get("mean", 0)
+                collapse = summary.get("equal_3round_traced", {}).get("mean", 0)
+                v["values"] = {"randomized": round(test,4), "collapse": round(collapse,4)}
+                if test > collapse + 0.10:
+                    v["verdict"] = "CONFIRMED"; v["reason"] = f"Randomized order recovered to {test:.3f}."
+                elif abs(test - collapse) < 0.05:
+                    v["verdict"] = "REFUTED"; v["reason"] = f"No difference ({test:.3f} vs {collapse:.3f})."
+                else:
+                    v["verdict"] = "PARTIAL"; v["reason"] = f"Some recovery ({test:.3f} vs {collapse:.3f})."
+
+            elif h_name == "H3_voting_vulnerability":
+                judge = summary.get("judge_vote_3round", {}).get("mean", 0)
+                conf = summary.get("confidence_weighted_3round", {}).get("mean", 0)
+                collapse = summary.get("equal_3round_traced", {}).get("mean", 0)
+                best = max(judge, conf)
+                v["values"] = {"judge_vote": round(judge,4), "confidence_weighted": round(conf,4), "collapse": round(collapse,4)}
+                if best > collapse + 0.10:
+                    v["verdict"] = "CONFIRMED"; v["reason"] = f"Alt voting recovered to {best:.3f}. Protocol matters."
+                elif abs(judge - collapse) < 0.05 and abs(conf - collapse) < 0.05:
+                    v["verdict"] = "REFUTED"; v["reason"] = "No voting protocol helps."
+                else:
+                    v["verdict"] = "PARTIAL"
+
+            elif h_name == "H4_contamination":
+                v["values"] = {"retention_rate_round3": round(retention_rate, 4)}
+                if retention_rate < 0.5:
+                    v["verdict"] = "CONFIRMED"; v["reason"] = f"Only {retention_rate:.1%} retained answer. Contamination present."
+                elif retention_rate > 0.7:
+                    v["verdict"] = "REFUTED"; v["reason"] = f"{retention_rate:.1%} retention — honest agents resist corruption."
+                else:
+                    v["verdict"] = "PARTIAL"; v["reason"] = f"{retention_rate:.1%} retention — some contamination."
+
+            elif h_name == "H5_confidence_distortion":
+                test = summary.get("confidence_weighted_3round", {}).get("mean", 0)
+                collapse = summary.get("equal_3round_traced", {}).get("mean", 0)
+                v["values"] = {"confidence_weighted": round(test,4), "collapse": round(collapse,4)}
+                if test > collapse + 0.10:
+                    v["verdict"] = "CONFIRMED"; v["reason"] = "Confidence weighting recovers accuracy."
+                elif abs(test - collapse) < 0.05:
+                    v["verdict"] = "REFUTED"; v["reason"] = "Confidence weighting does not help."
+                else:
+                    v["verdict"] = "PARTIAL"
+
+            elif h_name == "H6_skill_dependency":
+                weak = summary.get("adversary_weak", {}).get("mean", 0)
+                normal = summary.get("adversary_normal", {}).get("mean", 0)
+                strong = summary.get("adversary_strong", {}).get("mean", 0)
+                oracle = summary.get("adversary_oracle", {}).get("mean", 0)
+                collapse = summary.get("equal_3round_traced", {}).get("mean", 0)
+                v["values"] = {"weak": round(weak,4), "normal": round(normal,4), "strong": round(strong,4), "oracle": round(oracle,4), "collapse": round(collapse,4)}
+                if weak > collapse + 0.10 and oracle <= collapse + 0.05:
+                    v["verdict"] = "CONFIRMED"; v["reason"] = f"Weak={weak:.3f} avoids collapse, oracle={oracle:.3f} does not. Skill matters."
+                elif weak <= collapse + 0.05:
+                    v["verdict"] = "REFUTED"; v["reason"] = f"Even weak adversary collapses to {weak:.3f}."
+                else:
+                    v["verdict"] = "PARTIAL"; v["reason"] = f"Skill gradient: weak={weak:.3f}, oracle={oracle:.3f}."
+
+            elif h_name == "H7_topic_vulnerability":
+                v["verdict"] = "NEEDS_PER_TOPIC_DATA"
+                v["reason"] = "Requires per-topic accuracy breakdown. Check per_topic_outcomes.csv."
+
+            verdicts[h_name] = v
+        except Exception as e:
+            verdicts[h_name] = {"hypothesis": h_name, "verdict": "ERROR", "reason": str(e)}
+
+    return verdicts
+
+
+def make_charts(summary_rows, retention_rows, flip_rows, out_dir):
+    try:
+        import matplotlib
+        matplotlib.use("Agg")
+        import matplotlib.pyplot as plt
+        import numpy as np
+
+        out_dir = Path(out_dir)
+        out_dir.mkdir(parents=True, exist_ok=True)
+
+        # 1. Accuracy by condition
+        conds = [r["condition"] for r in summary_rows]
+        means = [r["mean_accuracy"] for r in summary_rows]
+        mins_ = [r["min_accuracy"] for r in summary_rows]
+        maxs_ = [r["max_accuracy"] for r in summary_rows]
+        base1_acc = means[0] if means else 0.85
+
+        fig, ax = plt.subplots(figsize=(14, 5))
+        x = np.arange(len(conds))
+        errs = [[means[i] - mins_[i] for i in range(len(means))],
+                 [maxs_[i] - means[i] for i in range(len(means))]]
+        ax.bar(x, means, yerr=errs, capsize=4, color="steelblue", edgecolor="navy")
+        ax.axhline(y=base1_acc, color="green", linestyle="--", alpha=0.5, linewidth=2, label=f"1-round baseline ({base1_acc:.3f})")
+        ax.set_xticks(x)
+        ax.set_xticklabels(conds, rotation=45, ha="right", fontsize=7)
+        ax.set_ylabel("Accuracy"); ax.set_title("Collapse Mechanism Isolation: Accuracy by Condition")
+        ax.legend(); ax.set_ylim(0, 1.05)
+        plt.tight_layout(); plt.savefig(out_dir / "fig_accuracy_by_condition.png", dpi=150); plt.close()
+
+        # 2. Honest retention
+        if retention_rows:
+            fig, ax = plt.subplots(figsize=(7, 4))
+            seeds = sorted(set(r["seed"] for r in retention_rows))
+            for s in seeds:
+                sr = [r for r in retention_rows if r["seed"] == s]
+                ax.plot([2,3], [r["stayed"] for r in sr], "o-", label=f"Seed {s}")
+            ax.set_xlabel("Round"); ax.set_ylabel("Honest agents staying with original answer")
+            ax.set_title("Honest Answer Retention Across Rounds"); ax.legend(); ax.set_xticks([2,3])
+            plt.tight_layout(); plt.savefig(out_dir / "fig_honest_retention.png", dpi=150); plt.close()
+
+        # 3. Flip pie
+        if flip_rows:
+            degraded = sum(r["count"] for r in flip_rows if r["flip_direction"] == "degraded")
+            improved = sum(r["count"] for r in flip_rows if r["flip_direction"] == "improved")
+            stable = sum(r["count"] for r in flip_rows if r["flip_direction"] == "stable")
+            fig, ax = plt.subplots(figsize=(6, 6))
+            ax.pie([stable, degraded, improved], labels=["Stable", "Degraded", "Improved"],
+                   colors=["gray","crimson","forestgreen"], autopct="%1.1f%%", startangle=90)
+            ax.set_title("Honest Agent Answer Transitions (R1→R3)")
+            plt.tight_layout(); plt.savefig(out_dir / "fig_flip_rate.png", dpi=150); plt.close()
+
+        # 4. Adversary skill gradient
+        skill_conds = [c for c in conds if c.startswith("adversary_")]
+        if skill_conds:
+            skill_accs = [next(r["mean_accuracy"] for r in summary_rows if r["condition"] == c) for c in skill_conds]
+            fig, ax = plt.subplots(figsize=(7, 4))
+            ax.bar([c.replace("adversary_","") for c in skill_conds], skill_accs,
+                   color=["lightgreen","steelblue","darkorange","crimson"])
+            ax.set_ylabel("Accuracy"); ax.set_title("Adversary Skill Ablation"); ax.set_ylim(0,1.05)
+            plt.xticks(rotation=20, ha="right", fontsize=9)
+            plt.tight_layout(); plt.savefig(out_dir / "fig_adversary_skill.png", dpi=150); plt.close()
+
+        print(f"  Charts saved to {out_dir}/")
+        return True
+    except ImportError:
+        print("  matplotlib not available — skipping charts")
+        return False
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("Usage: python analyze_collapse.py <results.json>")
+        sys.exit(1)
+
+    path = Path(sys.argv[1])
+    out_dir = path.parent / "analysis"
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    print(f"Loading {path}...")
+    data = load_data(path)
+
+    # Write CSVs
+    summary_rows = make_summary_table(data)
+    with open(out_dir / "condition_summary.csv", "w", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=["condition","mean_accuracy","min_accuracy","max_accuracy","range"])
+        w.writeheader(); w.writerows(summary_rows)
+    print(f"  condition_summary.csv: {len(summary_rows)} rows")
+
+    retention_rows = make_retention_table(data)
+    with open(out_dir / "honest_retention_by_round.csv", "w", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=["seed","round","stayed","flipped_away","flipped_toward","adversary_flips"])
+        w.writeheader(); w.writerows(retention_rows)
+    print(f"  honest_retention_by_round.csv: {len(retention_rows)} rows")
+
+    flip_rows, adv_rows = make_flip_matrix(data)
+    if flip_rows:
+        with open(out_dir / "round_flip_matrix.csv", "w", newline="") as f:
+            w = csv.DictWriter(f, fieldnames=["seed","transition","round1_correct","round3_correct","count","flip_direction"])
+            w.writeheader(); w.writerows(flip_rows)
+        print(f"  round_flip_matrix.csv: {len(flip_rows)} rows")
+    with open(out_dir / "adversary_induced_flips.csv", "w", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=["seed","total_adversary_flips"])
+        w.writeheader(); w.writerows(adv_rows)
+    print(f"  adversary_induced_flips.csv: {len(adv_rows)} rows")
+
+    # Hypothesis verdicts
+    verdicts = evaluate_hypotheses(data)
+    with open(out_dir / "hypothesis_verdicts.json", "w") as f:
+        json.dump(verdicts, f, indent=2)
+    print(f"\n  Hypothesis verdicts:")
+    for h, v in verdicts.items():
+        print(f"    {h}: {v['verdict']} — {v.get('reason','')[:120]}")
+
+    # Charts
+    make_charts(summary_rows, retention_rows, flip_rows, out_dir)
+
+    print(f"\nDone. Outputs in {out_dir}/")
+
+
+if __name__ == "__main__":
+    main()