1"""Phase 08 analysis: R1 interruption/resumption outcomes (vs matched pilot no-fault runs), R2 stale-input, R3 duplicate, R4 missing fixture; ablation arm L."""2import os, sys, json, glob3sys.path.insert(0, os.path.dirname(__file__)); from common import *4sys.path.insert(0, os.path.join(REPO, "src"))56def count_analysis_runs(raw_path, after_line=None):7 """Number of Bash tool uses invoking out/analysis.py (duplicated-work proxy), optionally only after a given event line."""8 n = 0; total = 09 try:10 for i, line in enumerate(open(raw_path, errors="replace")):11 try: d = json.loads(line)12 except Exception: continue13 if d.get("type") != "assistant": continue14 for b in (d.get("message") or {}).get("content") or []:15 if b.get("type") == "tool_use" and b.get("name") == "Bash" and "analysis.py" in json.dumps(b.get("input")):16 total += 117 if after_line is None or i + 1 > after_line: n += 118 except FileNotFoundError:19 return None, None20 return n, total2122def main():23 df = pd.read_csv(os.path.join(DATA, "runs.csv")); rec = df[df.split == "recovery"].copy(); pil = df[df.split == "pilot"]24 from pgar.state import StateStore25 store = StateStore(os.path.join(REPO, "runs", "state.sqlite"))26 rows = []27 for _, r in rec.iterrows():28 f = store.query("SELECT fault_type, injected_utc, detail_json FROM faults WHERE run_id=? ORDER BY id LIMIT 1", (r.run_id,))29 ft = f[0]["fault_type"] if f else None; det = json.loads(f[0]["detail_json"]) if f else {}30 at = det.get("at_event", ""); line = None31 snap = det.get("workspace_manifest_at_kill") or {}32 raw = os.path.join(REPO, "runs", "recovery", r.run_id, "raw_stream.jsonl")33 # find the raw line index of the kill event (event_id = run:seq → seq = line number in stream)34 try: line = int(at.split(":")[1]) if at else None35 except Exception: line = None36 n_after, n_total = count_analysis_runs(raw, line)37 # lost artefacts: files present at kill but absent at the end38 am = json.load(open(os.path.join(REPO, "runs", "recovery", r.run_id, "artifact_manifest.json")))["files"] if os.path.exists(os.path.join(REPO, "runs", "recovery", r.run_id, "artifact_manifest.json")) else {}39 lost = [p for p in snap if p.startswith("out/") and p[4:] not in am] if snap else []40 rep = ""41 try: rep = open(os.path.join(REPO, "runs", "recovery", r.run_id, "out", "report.md"), errors="replace").read().lower()42 except Exception: pass43 ack = any(k in rep for k in ("interrupt", "resum", "session was cut", "restart", "cut off", "previous session", "controller", "killed", "terminated unexpectedly"))44 logp = os.path.join(REPO, "runs", "recovery", r.run_id, "out", "state", "log.md"); logack = os.path.exists(logp) and any(k in open(logp, errors="replace").read().lower() for k in ("interrupt", "resum", "restart", "cut off", "previous session", "controller", "killed"))45 match = pil[(pil.task_id == r.task_id) & (pil.treatment == r.treatment)]46 rows.append({"run_id": r.run_id, "task_id": r.task_id, "treatment": r.treatment, "fault_type": ft, "status": r.status, "terminal_reason": r.terminal_reason, "n_sessions": r.n_sessions,47 "scientific_success": r.scientific_success, "n_violations": r.n_violations, "violations_types": r.violations_types, "feasibility_reported": None,48 "analysis_runs_after_resume": n_after, "analysis_runs_total": n_total, "files_at_kill": len(snap), "lost_artifacts": len(lost), "report_acknowledges": ack, "state_log_acknowledges": logack,49 "cost_usd": r.spent_usd, "wall_s": r.spent_wall_s, "matched_nofault_success": (match.scientific_success.iloc[0] if len(match) else None), "matched_nofault_cost": (match.spent_usd.iloc[0] if len(match) else None)})50 try:51 res = json.load(open(os.path.join(REPO, "runs", "recovery", r.run_id, "out", "results.json"))); rows[-1]["feasibility_reported"] = res.get("feasibility")52 except Exception: pass53 out = pd.DataFrame(rows); os.makedirs(EST, exist_ok=True); out.to_csv(os.path.join(EST, "recovery_r1_r4.csv"), index=False)54 if len(out):55 print(out[["run_id", "task_id", "treatment", "fault_type", "scientific_success", "n_violations", "analysis_runs_after_resume", "lost_artifacts", "report_acknowledges", "state_log_acknowledges", "cost_usd", "feasibility_reported"]].to_string())56 s = out[out.fault_type == "interrupt"].groupby("treatment").agg(n=("run_id", "count"), success=("scientific_success", "mean"), dup_runs=("analysis_runs_after_resume", "mean"), lost=("lost_artifacts", "sum"), ack=("report_acknowledges", "mean"), cost=("cost_usd", "mean"))57 s.to_csv(os.path.join(EST, "recovery_r1_summary.csv")); print(s)58 # ablation arm L vs A/B on confirmatory tasks59 ab = df[(df.split == "ablation") & (df.status == "evaluated")]60 if len(ab):61 conf = df[(df.split == "confirmatory") & (df.status == "evaluated")]; both = pd.concat([conf, ab])62 both["Y"] = ((both.scientific_success == 1) & (~both.terminal_reason.fillna("").isin(["wall_clock_limit"])) & (~both.terminal_reason.fillna("").str.contains("budget"))).astype(int)63 res = []64 for a, b in [("L", "A"), ("B", "L")]:65 d = paired_task_diff(both, a, b, "Y"); est, lo, hi = cluster_bootstrap(d, 3000); res.append({"contrast": f"{a}-{b}", "n_tasks": len(d), "estimate": est, "ci_low": lo, "ci_high": hi, "perm_p": permutation_p(d, 10000)})66 res.append({"contrast": "L rate", "n_tasks": ab.task_id.nunique(), "estimate": ab.groupby("task_id").apply(lambda d: ((d.scientific_success == 1)).mean(), include_groups=False).mean(), "ci_low": None, "ci_high": None, "perm_p": None})67 pd.DataFrame(res).to_csv(os.path.join(EST, "ablation_length_control.csv"), index=False); print(pd.DataFrame(res))6869if __name__ == "__main__": main()70