1"""Phase 08 drivers: R1 interruption/resumption, R2 stale-input invalidation, R3 duplicate submission, R4 missing fixture.2Usage: PYTHONPATH=src .venv/bin/python -m pgar.recovery_experiment r1|r2|r3|r4 [--seed N]"""3import json, os, sys, posixpath4from .cli import REPO, load_cfg, get_store5from .scheduler import Controller6from . import recovery, launcher7from .exports import export_runs89def r1(seed=801):10 cfg = load_cfg("development"); store = get_store(); ctl = Controller(REPO, store, cfg, cfg["nodes"])11 existing = store.query("SELECT COUNT(*) AS n FROM runs WHERE split='recovery'")[0]["n"]12 runs = ctl.make_assignment("recovery", ["F1-dev-01", "F3-dev-01", "F6-dev-01"], 2, seed, arms=("A", "D"), start_seq=existing + 1)13 for m in runs: m.notes += "; R1 interrupt after 6th tool use then resume"14 ctl.register(runs)15 ids = [m.run_id for m in runs]16 res = ctl.run_many(ids, faults={rid: {"type": "interrupt", "trigger": {"after_tool_uses": 6}} for rid in ids})17 export_runs(store, REPO); print(json.dumps([{k: v for k, v in x.items() if k != "evaluation"} for x in res], indent=1, default=str))1819def r2():20 store = get_store(); out = []21 muts = {"F1-dev-01": None, "F6-dev-01": None}22 for m in store.list_runs(split="pilot"):23 if m["task_id"] not in muts: continue24 bundle = os.path.join(REPO, "tasks", "public_bundles", m["task_id"], "data")25 files = sorted((f for f in os.listdir(bundle) if f.endswith(".csv")), key=lambda f: -os.path.getsize(os.path.join(bundle, f)))26 if not files: continue27 data_file = posixpath.join("data", files[0]) # the largest CSV = the main input28 # mutate: multiply the LAST numeric field of the 2nd data line by 1.37 and add 1 (a non-identifier value; documented, inside the copy only)29 mutate = ("python3 -c \"import sys,re;p=sys.argv[1];L=open(p).read().split('\\n');i=2\n"30 "while i<len(L) and not re.search(r'\\d',L[i]): i+=1\n"31 "nums=list(re.finditer(r'-?\\d+(?:\\.\\d+)?',L[i]));m=nums[-1];L[i]=L[i][:m.start()]+str(round(float(m.group(0))*1.37+1,6))+L[i][m.end():];open(p,'w').write('\\n'.join(L))\" $F")32 ws = f"/Users/simon-pierreboucher/wp29/runs/{m['run_id']}"33 rec = recovery.stale_input_check(m["node"], ws, data_file, mutate)34 repro = [c for c in rec.get("checks", []) if c["id"] == "reproducibility"]35 row = {"run_id": m["run_id"], "task_id": m["task_id"], "treatment": m["treatment"], "data_file": data_file, "gate_passed_after_mutation": rec.get("passed"), "reproducibility_check_after_mutation": repro[0]["passed"] if repro else None, "detail": (repro[0]["detail"][:200] if repro else rec.get("error"))}36 fid = store.add_fault(m["run_id"], "stale_input", data_file, {"mutate": mutate}); store.set_fault_recovery(fid, repro and not repro[0]["passed"], row)37 out.append(row); print(row)38 os.makedirs(os.path.join(REPO, "results", "estimates"), exist_ok=True)39 json.dump(out, open(os.path.join(REPO, "results", "estimates", "recovery_r2_stale_input.json"), "w"), indent=1)4041def r3(seed=803):42 cfg = load_cfg("development"); store = get_store(); ctl = Controller(REPO, store, cfg, cfg["nodes"])43 existing = store.query("SELECT COUNT(*) AS n FROM runs WHERE split='recovery'")[0]["n"]44 runs = ctl.make_assignment("recovery", ["F2-dev-01"], 1, seed, arms=("B",), start_seq=existing + 1)45 runs[0].notes += "; R3 duplicate submission test"; ctl.register(runs); rid = runs[0].run_id46 res = ctl.run_many([rid, rid, rid])47 n_sessions = store.query("SELECT COUNT(*) AS n FROM sessions WHERE run_id=?", (rid,))[0]["n"]48 summary = {"run_id": rid, "submissions": 3, "results": [{k: v for k, v in x.items() if k != "evaluation"} for x in res], "n_sessions_launched": n_sessions}49 json.dump(summary, open(os.path.join(REPO, "results", "estimates", "recovery_r3_duplicate.json"), "w"), indent=1, default=str); export_runs(store, REPO); print(json.dumps(summary, indent=1, default=str))5051def r4(seed=804):52 cfg = load_cfg("development"); store = get_store(); ctl = Controller(REPO, store, cfg, cfg["nodes"])53 existing = store.query("SELECT COUNT(*) AS n FROM runs WHERE split='recovery'")[0]["n"]54 bundle = os.path.join(REPO, "tasks", "public_bundles", "F1-dev-01", "data"); main_file = sorted(f for f in os.listdir(bundle) if f.endswith(".csv"))[0]55 runs = ctl.make_assignment("recovery", ["F1-dev-01"], 1, seed, arms=("A", "D"), start_seq=existing + 1)56 for m in runs: m.notes += f"; R4 missing fixture data/{main_file}"57 ctl.register(runs); ids = [m.run_id for m in runs]58 res = ctl.run_many(ids, faults={rid: {"type": "missing_fixture", "path": f"data/{main_file}"} for rid in ids})59 export_runs(store, REPO); print(json.dumps([{k: v for k, v in x.items() if k != "evaluation"} for x in res], indent=1, default=str))6061if __name__ == "__main__":62 {"r1": r1, "r2": r2, "r3": r3, "r4": r4}[sys.argv[1]]()63