1"""Phase 7 — preregistered benchmark harness (protocol heldout_v1): tasks × documents × conditions × reps, blind LLM judge,2deterministic secondary outcomes. Nothing is fabricated: every run is an actual API call logged in runs.jsonl; failures are kept."""3from __future__ import annotations4import json, random, re, time, concurrent.futures as cf5from pathlib import Path6import yaml7from . import config, generation, auditing, llm, segmentation8from .provenance import stamp, now_iso, Ledger, ROOT910TASKS = { # name -> (section role used by the engine, mode, brief template)11 "draft_introduction": ("introduction", "write", "Draft the introduction of this paper from the verified material (abstract, section structure, tables, figures, references). The published introduction is withheld. Do not invent numbers beyond those in the material."),12 "rewrite_literature": ("literature", "upgrade", "The following literature-positioning passage is a weak 'laundry list' of citations. Rewrite it as a positioning paragraph set that explains how this paper relates to, differs from and builds on the cited work, citing only works that appear in the reference list:\n\n{degraded_literature}"),13 "explain_specification": ("empirical_design", "upgrade", "Improve the explanation of the empirical specification / research design described in the material: define the estimand, the identifying assumptions, the fixed effects and the inference procedure as stated; do not add specifications that are not in the material."),14 "interpret_results": ("results", "write", "Write the interpretation of the main results for the Results section, using only the tables' captions/notes and the facts stated in the material. Discuss economic and statistical significance only as far as the material supports; never invent coefficients."),15 "improve_table": ("results", "upgrade", "Propose an improved caption and notes for the first main regression table listed in the material (keep its label), stating what is shown, sample, unit of observation, period, variable definitions, standard errors and significance conventions, without inventing values."),16 "flag_causal_claims": ("results", "write", "Identify every causal claim in the material and classify it as supported by the stated design, over-stated, or associational; for each, quote the sentence (briefly), name the design element that would support it, and propose a bounded rewording. Output as a LaTeX itemize list."),17 "referee_report": ("results", "write", "Write a referee report on this manuscript material: summary, contribution assessment, major comments (identification, inference, measurement, robustness, presentation), minor comments, recommendation. Tie every comment to a section/table."),18 "improve_abstract": ("abstract", "upgrade", "Improve the abstract without altering any finding, number or claim: sharpen the question, make the contribution explicit, keep magnitudes exactly as stated, 100–150 words."),19}20CONDITIONS = ["A_baseline", "B_strong_prompt", "C_conventional_rag", "D_profiles_retrieval", "E_full_engine"]21ABLATIONS = {"E_no_retrieval": {"no_retrieval": True}, "E_no_profile": {"no_profile": True}, "E_no_referee": {"no_referee": True}, "E_no_factcheck": {"no_factcheck": True}}22REP2_TASKS = ["draft_introduction", "improve_abstract", "referee_report", "interpret_results"]23OOD_TASKS = ["draft_introduction", "improve_abstract", "referee_report"]2425def load_fixtures() -> dict:26 return json.loads((ROOT / "experiments" / "fixtures" / "heldout_v1" / "manifest.json").read_text())2728def plan(cfg: dict, tasks=None, conditions=None, reps=None) -> list[dict]:29 fx = load_fixtures(); items = []30 tasks = tasks or list(TASKS); conds = conditions or (CONDITIONS + list(ABLATIONS))31 for doc in fx["finance_docs"]:32 for t in tasks:33 for c in conds:34 n_rep = reps if reps is not None else (2 if (t in REP2_TASKS and c in CONDITIONS) else 1)35 for r in range(1, n_rep + 1): items.append({"task": t, "doc": doc["doc_id"], "domain": "finance", "condition": c, "rep": r, "paper": str(ROOT / "experiments" / "fixtures" / "heldout_v1" / doc["doc_id"] / "manuscript.md"), "degraded_literature": doc["degraded_literature"]})36 ood = fx["out_of_domain"]37 for t in [x for x in OOD_TASKS if x in tasks]:38 for c in [x for x in conds if x in CONDITIONS]:39 items.append({"task": t, "doc": ood["id"], "domain": "out_of_domain", "condition": c, "rep": 1, "paper": ood["path"], "degraded_literature": ""})40 return items4142def run_item(it: dict, effort: str, model: str) -> dict:43 sec, mode, brief = TASKS[it["task"]]; brief = brief.replace("{degraded_literature}", it.get("degraded_literature", ""))44 cond = it["condition"]; ab = ABLATIONS.get(cond); c = "E_full_engine" if ab else cond45 t0 = time.time()46 try:47 res = generation.run(Path(it["paper"]), section=sec, mode=mode, condition=c, brief=brief, model=model, effort=effort, ablation=ab, meta={"benchmark": "heldout_v1", "task": it["task"], "doc": it["doc"], "rep": it["rep"]})48 ms = auditing.load_manuscript(Path(it["paper"])); material = generation.manuscript_material(ms, sec, brief=brief)49 refs_text = "\n".join(s["full_text"] for s in ms["sections"] if "reference" in s["heading"].lower()) if it["domain"] == "finance" else ""50 out = {**it, "ok": True, "text": res["text"], "change_log": res["change_log"], "flags": res["flags"], "cost_usd": res["cost_usd"], "latency_s": res["latency_s"], "usage": res["usage"], "n_calls": res["n_calls"], "model": model, "effort": effort,51 "metrics": secondary_metrics(res["text"], material, refs_text, ms), "retrieved": res["retrieved"], "profile": res["profile"], "ts": now_iso()}52 except Exception as e:53 out = {**it, "ok": False, "error": repr(e)[:400], "latency_s": round(time.time() - t0, 1), "ts": now_iso()}54 return out5556def secondary_metrics(text: str, material: str, refs_text: str, ms: dict) -> dict:57 num = auditing.numeric_preservation(material, text); cit = auditing.citation_check(text, ms["bib_keys"]); uns = auditing.unsupported_claims(text); ov = auditing.overlap_with_corpus(text)58 # unverifiable author–year citations: surname (year) strings whose surname+year do not co-occur in the source reference list / material59 loose = re.findall(r"\b([A-Z][A-Za-z\-']+)(?:,? (?:and|&) [A-Z][A-Za-z\-']+)?(?: et al\.)? \(((?:19|20)\d{2})[a-z]?\)", text); src = (refs_text + "\n" + material)60 unver = [f"{a} ({y})" for a, y in loose if not re.search(re.escape(a) + r".{0,120}" + y, src, re.S) and not re.search(y + r".{0,120}" + re.escape(a), src, re.S)]61 return {"n_words": len(text.split()), "preserved_rate": num["preserved_rate"], "new_numbers": len(num["new_in_output"]), "new_rate": num["new_rate"], "n_author_year": len(loose), "n_unverifiable_citations": len(unver), "unverifiable_examples": unver[:5],62 "unknown_cite_keys": len(cit["unknown_keys"]), "n_unsupported_claims": len(uns), "max_overlap_share": ov["max_overlap_share"], "todo_placeholders": len(re.findall(r"\\todo\{", text))}6364def judge_set(task: str, doc: str, rep: int, items: list[dict], judge_model: str, effort: str, second_pass: bool = False) -> dict:65 """Blind judge: all condition outputs of one (task, doc, rep) in random order under random letters."""66 body, pid, pv = generation.load_prompt("evaluation/judge_rubric_v1.md")67 rng = random.Random(f"{task}|{doc}|{rep}|{'B' if second_pass else 'A'}"); order = items[:]; rng.shuffle(order); letters = "KLMNPQRSTUVWXYZ"68 mapping = {letters[i]: it["condition"] for i, it in enumerate(order)}69 ms = auditing.load_manuscript(Path(order[0]["paper"])); sec, mode, brief = TASKS[task]; brief = brief.replace("{degraded_literature}", order[0].get("degraded_literature", ""))70 material = generation.manuscript_material(ms, sec, brief=brief, max_chars=30000)71 user = f"TASK: {task} — {brief[:600]}\n\n=== SOURCE MATERIAL ===\n{material}\n\n" + "\n\n".join(f"=== OUTPUT {letters[i]} ===\n{it['text'][:9000]}" for i, it in enumerate(order))72 r = llm.call(body, user, model=judge_model, max_tokens=4000, effort=effort, purpose="judge", prompt_id=pid, prompt_version=pv, use_cache=False, parse_json=True, meta={"task": task, "doc": doc, "rep": rep, "pass": "B" if second_pass else "A"})73 j = r["json"] or {}; scores = {}74 for L, sc in (j.get("scores") or {}).items():75 if L in mapping: scores[mapping[L]] = sc76 return {"task": task, "doc": doc, "rep": rep, "pass": "B" if second_pass else "A", "judge_model": r["model"], "mapping": mapping, "scores": scores, "best": mapping.get(j.get("best")), "notes": j.get("notes"), "cost_usd": r["cost_usd"], "ok": bool(scores), "raw": None if scores else r["text"][:1000], "ts": now_iso()}7778def run_benchmark(config_path: Path, protocol: str | None = None, dry_run: bool = False, tasks: str | None = None, conditions: str | None = None, reps: int | None = None, workers: int = 4, judge_second_pass_share: float = 0.25) -> dict:79 cfg = yaml.safe_load(Path(config_path).read_text()); protocol = protocol or cfg["protocol"]; mcfg = config.load("models")80 tasks_l = tasks.split(",") if tasks else None; conds_l = conditions.split(",") if conditions else None81 items = plan(cfg, tasks_l, conds_l, reps); out_dir = ROOT / "experiments" / "outputs" / protocol; out_dir.mkdir(parents=True, exist_ok=True); (out_dir / "outputs").mkdir(exist_ok=True)82 est = sum(0.30 if it["condition"].startswith("E") else 0.14 for it in items); n_sets = len({(it["task"], it["doc"], it["rep"]) for it in items})83 summary = {"protocol": protocol, "n_runs_planned": len(items), "n_judge_sets": n_sets, "est_cost_usd": round(est + n_sets * 0.12 * (1 + judge_second_pass_share), 2), "by_condition": {c: sum(1 for i in items if i["condition"] == c) for c in CONDITIONS + list(ABLATIONS)}, "generator": mcfg["writing_model"], "judge": mcfg["judge_model"], "effort": "medium"}84 if dry_run: return {**summary, "dry_run": True}85 runs = Ledger(out_dir / "runs.jsonl"); done = {(r["task"], r["doc"], r["condition"], r["rep"]) for r in runs.read() if r.get("ok")}86 todo = [it for it in items if (it["task"], it["doc"], it["condition"], it["rep"]) not in done]87 with cf.ThreadPoolExecutor(max_workers=workers) as ex:88 futs = [ex.submit(run_item, it, "medium", mcfg["writing_model"]) for it in todo]89 for f in cf.as_completed(futs):90 r = f.result(); runs.append(r)91 if r.get("ok"): (out_dir / "outputs" / f"{r['task']}__{r['doc']}__{r['condition']}__r{r['rep']}.tex").write_text(r["text"])92 allruns = [r for r in runs.read() if r.get("ok")]93 # judging94 jl = Ledger(out_dir / "judgments.jsonl"); jdone = {(j["task"], j["doc"], j["rep"], j["pass"]) for j in jl.read() if j.get("ok")}95 sets = {}96 for r in allruns: sets.setdefault((r["task"], r["doc"], r["rep"]), []).append(r)97 rng = random.Random(45); keys = sorted(sets); second = set(rng.sample(keys, max(1, int(len(keys) * judge_second_pass_share))))98 jobs = [(k, False) for k in keys if (k[0], k[1], k[2], "A") not in jdone] + [(k, True) for k in keys if k in second and (k[0], k[1], k[2], "B") not in jdone]99 with cf.ThreadPoolExecutor(max_workers=workers) as ex:100 futs = [ex.submit(judge_set, k[0], k[1], k[2], sets[k], mcfg["judge_model"], mcfg["effort_judge"], sp) for k, sp in jobs]101 for f in cf.as_completed(futs): jl.append(f.result())102 summary.update({"n_runs_ok": len(allruns), "n_runs_failed": sum(1 for r in runs.read() if not r.get("ok")), "n_judgments": sum(1 for j in jl.read() if j.get("ok")), "spent_usd_total": llm.spent_usd(), "completed": now_iso()})103 (out_dir / "summary.json").write_text(json.dumps(summary, indent=1)); return summary104