"""Phase 08 analysis: R1 interruption/resumption outcomes (vs matched pilot no-fault runs), R2 stale-input, R3 duplicate, R4 missing fixture; ablation arm L."""
import os, sys, json, glob
sys.path.insert(0, os.path.dirname(__file__)); from common import *
sys.path.insert(0, os.path.join(REPO, "src"))

def count_analysis_runs(raw_path, after_line=None):
    """Number of Bash tool uses invoking out/analysis.py (duplicated-work proxy), optionally only after a given event line."""
    n = 0; total = 0
    try:
        for i, line in enumerate(open(raw_path, errors="replace")):
            try: d = json.loads(line)
            except Exception: continue
            if d.get("type") != "assistant": continue
            for b in (d.get("message") or {}).get("content") or []:
                if b.get("type") == "tool_use" and b.get("name") == "Bash" and "analysis.py" in json.dumps(b.get("input")):
                    total += 1
                    if after_line is None or i + 1 > after_line: n += 1
    except FileNotFoundError:
        return None, None
    return n, total

def main():
    df = pd.read_csv(os.path.join(DATA, "runs.csv")); rec = df[df.split == "recovery"].copy(); pil = df[df.split == "pilot"]
    from pgar.state import StateStore
    store = StateStore(os.path.join(REPO, "runs", "state.sqlite"))
    rows = []
    for _, r in rec.iterrows():
        f = store.query("SELECT fault_type, injected_utc, detail_json FROM faults WHERE run_id=? ORDER BY id LIMIT 1", (r.run_id,))
        ft = f[0]["fault_type"] if f else None; det = json.loads(f[0]["detail_json"]) if f else {}
        at = det.get("at_event", ""); line = None
        snap = det.get("workspace_manifest_at_kill") or {}
        raw = os.path.join(REPO, "runs", "recovery", r.run_id, "raw_stream.jsonl")
        # find the raw line index of the kill event (event_id = run:seq → seq = line number in stream)
        try: line = int(at.split(":")[1]) if at else None
        except Exception: line = None
        n_after, n_total = count_analysis_runs(raw, line)
        # lost artefacts: files present at kill but absent at the end
        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 {}
        lost = [p for p in snap if p.startswith("out/") and p[4:] not in am] if snap else []
        rep = ""
        try: rep = open(os.path.join(REPO, "runs", "recovery", r.run_id, "out", "report.md"), errors="replace").read().lower()
        except Exception: pass
        ack = any(k in rep for k in ("interrupt", "resumed", "resumption", "session was cut", "restart"))
        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"))
        match = pil[(pil.task_id == r.task_id) & (pil.treatment == r.treatment)]
        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,
                     "scientific_success": r.scientific_success, "n_violations": r.n_violations, "violations_types": r.violations_types, "feasibility_reported": None,
                     "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,
                     "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)})
        try:
            res = json.load(open(os.path.join(REPO, "runs", "recovery", r.run_id, "out", "results.json"))); rows[-1]["feasibility_reported"] = res.get("feasibility")
        except Exception: pass
    out = pd.DataFrame(rows); os.makedirs(EST, exist_ok=True); out.to_csv(os.path.join(EST, "recovery_r1_r4.csv"), index=False)
    if len(out):
        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())
        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"))
        s.to_csv(os.path.join(EST, "recovery_r1_summary.csv")); print(s)
    # ablation arm L vs A/B on confirmatory tasks
    ab = df[df.split == "ablation"]
    if len(ab):
        conf = df[df.split == "confirmatory"]; both = pd.concat([conf, ab]); both = load_runs(path=None) if False else both
        both["Y"] = ((both.scientific_success == 1) & (~both.terminal_reason.fillna("").isin(["wall_clock_limit"])) & (~both.terminal_reason.fillna("").str.contains("budget"))).astype(int)
        res = []
        for a, b in [("L", "A"), ("B", "L")]:
            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)})
        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})
        pd.DataFrame(res).to_csv(os.path.join(EST, "ablation_length_control.csv"), index=False); print(pd.DataFrame(res))

if __name__ == "__main__": main()