| |
| """Stage G: checkpoints, validation and the per-patient manifest. |
| |
| assign one checkpoint visit per question -> checkpoints.jsonl |
| audit every (question, checkpoint) pair, with the memory depth -> checkpoint_audit.json |
| validate oracle and no-memory probes judged by an LLM, and the dialogue checks of abstention and axis-1 |
| never-again questions -> validation.jsonl (+ _validation in the banks) |
| count-errors number of validation rows whose probes or judge calls failed |
| manifest MANIFEST.json: counts, models and the file inventory |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import glob |
| import json |
| import os |
| import re |
| from collections import Counter |
| from concurrent.futures import ThreadPoolExecutor, as_completed |
| from pathlib import Path |
|
|
| from pipeline.common import BASE, Progress, has_text, lists_options, model, no_memory_exempt_axes, patient_id |
| from pipeline.llm import ChatClient |
|
|
| PATIENT = patient_id() |
| ENGINE = model("generation") |
|
|
|
|
| def assign() -> None: |
| """Assign one checkpoint visit to every question. |
| |
| The checkpoint is the visit at which a question is posed; the method has seen the dialogues up to and |
| including it. It must come after the evidence and fall inside the question's appearance window. The |
| latest visit in the window is chosen. Axis-8 questions keep their computed checkpoint; one that has to be |
| moved, since its gold was computed for it, is reported. Questions with no valid checkpoint are reported. |
| """ |
| chain = [json.loads(l) for l in open(BASE / "event_chain.jsonl") if l.strip()] |
| by_v = {r["visit_number"]: r for r in chain} |
| vs = sorted(by_v) |
| date_of = {v: by_v[v]["date"] for v in vs} |
| def last_visit_on_or_before(d): |
| cand = [v for v in vs if date_of[v] <= d] |
| return cand[-1] if cand else None |
|
|
| def first_visit_on_or_after(d): |
| cand = [v for v in vs if date_of[v] >= d] |
| return cand[0] if cand else None |
|
|
| checkpoints, problems = [], [] |
| for f in sorted(glob.glob(str(BASE / "questions_axis*.jsonl"))): |
| ax = Path(f).stem.split("axis")[1] |
| for i, l in enumerate(open(f)): |
| q = json.loads(l) |
| qid = q.get("qid", f"{PATIENT}/axis{ax}/{i:03d}") |
| gt = q.get("gt_visit") |
| win = q.get("question_appearance_window") or {} |
| wf, wt = win.get("from"), win.get("to") |
|
|
| ck = (q.get("axis_specific") or {}).get("checkpoint_visit") |
| computed = ck |
| if ck not in by_v: |
| ck = None |
|
|
| if ck is None and wt: |
| ck = last_visit_on_or_before(wt) |
| if ck is None and wf: |
| ck = first_visit_on_or_after(wf) |
| if ck is None: |
| ck = vs[-1] |
|
|
| if isinstance(gt, int) and ck is not None and ck <= gt: |
| after = [v for v in vs if v > gt and (not wt or date_of[v] <= wt)] |
| if after: |
| ck = after[-1] |
| else: |
| after_any = [v for v in vs if v > gt] |
| ck = after_any[0] if after_any else ck |
| problems.append(f"{qid}: no in-window visit after gt=v{gt}; placed ckpt=v{ck} " |
| f"(out of window)") |
|
|
| if computed is not None and ck != computed: |
| problems.append(f"{qid}: computed checkpoint v{computed} moved to v{ck}; the gold was computed " |
| f"for v{computed}") |
|
|
| ck_date = date_of.get(ck) |
| in_window = bool(wf and wt and ck_date and wf <= ck_date <= wt) |
| ckpt_gt_ok = not (isinstance(gt, int)) or (ck is not None and ck > gt) |
| if not ckpt_gt_ok: |
| problems.append(f"{qid}: ckpt=v{ck} <= gt=v{gt} (unanswerable at ask time)") |
|
|
| checkpoints.append({ |
| "qid": qid, "axis": q.get("axis", f"{ax}_"), "ckpt_visit": ck, "ckpt_date": ck_date, |
| "window_from": wf, "window_to": wt, "in_window": in_window, |
| "gt_visit": gt, "ckpt_gt_ok": ckpt_gt_ok, |
| "grounding": q.get("grounding")}) |
|
|
| out = BASE / "checkpoints.jsonl" |
| with open(out, "w", encoding="utf-8") as fh: |
| for c in checkpoints: |
| fh.write(json.dumps(c, ensure_ascii=False) + "\n") |
|
|
| n = len(checkpoints) |
| inw = sum(1 for c in checkpoints if c["in_window"]) |
| gtok = sum(1 for c in checkpoints if c["ckpt_gt_ok"]) |
| print(f"assigned {n} checkpoints -> {out}") |
| print(f" ckpt > gt (answerable) : {gtok}/{n}") |
| print(f" in appearance window : {inw}/{n}") |
| print(f" by axis: {dict(Counter(c['axis'].split('_')[0] for c in checkpoints))}") |
| if problems: |
| print(f"\n {len(problems)} placement problem(s):") |
| for p in problems[:12]: |
| print(f" {p}") |
| else: |
| print(" no placement problems; every checkpoint is after its evidence.") |
|
|
|
|
| def toks(s): |
| return {w for w in re.sub(r"[^a-z0-9 ]", " ", str(s).lower()).split() if len(w) >= 4} |
|
|
|
|
| def audit() -> None: |
| """Audit every (question, checkpoint) pair. |
| |
| Checks that the checkpoint follows the evidence, lies in the appearance window, and that the evidence is |
| present in the dialogues up to the checkpoint. Reports the memory depth: visits from the latest first |
| statement of an evidence fact to the checkpoint. |
| """ |
| chain = [json.loads(l) for l in open(BASE / "event_chain.jsonl") if l.strip()] |
| by_v = {r["visit_number"]: r for r in chain} |
| vs = sorted(by_v) |
| ck = {c["qid"]: c for c in (json.loads(l) for l in open(BASE / "checkpoints.jsonl") if l.strip())} |
| dia = {r["visit_number"]: r for r in |
| (json.loads(l) for l in open(BASE / "dialogues.jsonl") if l.strip())} |
| spoken = {v: json.dumps(dia[v].get("turns"), ensure_ascii=False).lower() for v in dia} |
|
|
| def said_by(atom, ckpt_visit): |
| """Whether `atom` is recoverable from the dialogues up to and including ckpt_visit (at least 60% of its |
| content tokens appear).""" |
| at = toks(atom) |
| if not at: |
| return True, None |
| for v in [x for x in vs if x <= ckpt_visit and x in spoken]: |
| if len(at & toks(spoken[v])) / len(at) >= 0.6: |
| return True, v |
| return False, None |
|
|
| rows, fails = [], [] |
| for f in sorted(glob.glob(str(BASE / "questions_axis*.jsonl"))): |
| for l in open(f): |
| q = json.loads(l) |
| qid = q["qid"] |
| c = ck.get(qid) |
| if not c: |
| fails.append((qid, "no checkpoint")) |
| continue |
| cv, gt = c.get("ckpt_visit"), c.get("gt_visit") |
| after_evidence = not isinstance(gt, int) or (cv is not None and cv > gt) |
| in_window = c.get("in_window", False) |
|
|
| ev = (q.get("derivation") or {}).get("evidence") or [] |
| ev_atoms = [e.get("atom") for e in ev if e.get("atom")] |
| if not ev_atoms and gt in by_v: |
| ev_atoms = [q.get("answer")] |
| answerable, last_said = True, None |
| for a in ev_atoms: |
| ok, v = said_by(a, cv if isinstance(cv, int) else vs[-1]) |
| if not ok: |
| answerable = False |
| elif v is not None: |
| last_said = v if last_said is None else max(last_said, v) |
| depth = (vs.index(cv) - vs.index(last_said)) if (isinstance(cv, int) and last_said |
| and cv in vs and last_said in vs) else None |
| rows.append({"qid": qid, "ckpt_after_evidence": after_evidence, "in_window": in_window, |
| "answerable": answerable, "depth": depth}) |
| if not (after_evidence and answerable): |
| fails.append((qid, f"ckpt_after_evidence={after_evidence} answerable={answerable}")) |
|
|
| (BASE / "checkpoint_audit.json").write_text(json.dumps(rows, ensure_ascii=False, indent=1)) |
|
|
| n = len(rows) |
| n_after = sum(1 for r in rows if r["ckpt_after_evidence"]) |
| n_window = sum(1 for r in rows if r["in_window"]) |
| n_answerable = sum(1 for r in rows if r["answerable"]) |
| depths = [r["depth"] for r in rows if r["depth"] is not None] |
| print(f"audited {n} questions") |
| print(f" checkpoint after evidence : {n_after}/{n}") |
| print(f" checkpoint in window : {n_window}/{n}") |
| print(f" answerable from dialogues : {n_answerable}/{n}") |
| if depths: |
| depths.sort() |
| print(f" depth (first statement -> checkpoint): min {depths[0]}, median {depths[len(depths)//2]}, " |
| f"max {depths[-1]}") |
| if fails: |
| print(f"\n {len(fails)} question(s) fail the audit:") |
| for qid, why in fails[:15]: |
| print(f" {qid}: {why}") |
| else: |
| print(" all questions pass the audit.") |
| print(f"\n-> {BASE/'checkpoint_audit.json'}") |
|
|
|
|
| def client(): |
| return ChatClient(model=ENGINE, reasoning_effort="medium", max_completion_tokens=1500) |
|
|
|
|
| def qtext(q): |
| """Render the question, folding MC options in so the oracle sees the choices, unless the question text |
| already lists them.""" |
| s = q.get("question") or "" |
| if q.get("options") and not lists_options(s, q["options"]): |
| s += "\n" + "\n".join(f" {o.get('label')}. {o.get('content')}" for o in q["options"]) |
| return s |
|
|
|
|
| def answer_probe(cli, q, context, as_of=None): |
| ans_type = q.get("answer_type", "open_exact") |
| hint = {"multiple_choice": "Answer with one letter (A, B, C or D) and the option text.", |
| "abstain": "If the record does not contain the answer, say exactly 'not documented'.", |
| "state": "State the current status and why."}.get(ans_type, "Answer concisely.") |
| today = f"\nToday's date, as of which you answer: {as_of}.\n" if as_of else "" |
| question = qtext(q) |
| if context: |
| sysm = ("You answer a question about one patient using only the clinical dialogue provided. Do not use\n" |
| "outside knowledge of what is typical; answer only from the record. " + hint) |
| usr = f"Clinical dialogue, the patient's record:\n{context}\n{today}\nQuestion:\n{question}\n\nAnswer:" |
| else: |
| sysm = ("You answer a question about a patient whose record you have not seen. Use general clinical\n" |
| "knowledge and give your single best guess; never say you do not know. " + hint) |
| usr = (f"A question about a patient whose record you have not been given:\n{question}\n\n" |
| "Answer from the question alone and give your best guess; do not answer 'unknown'.") |
| try: |
| r = cli.complete_json(sysm, usr + '\n\nReturn JSON in this form: {"answer": "<your answer>"}') |
| if r.get("answer") is None or not str(r["answer"]).strip(): |
| raise ValueError(f"reply has no answer: {str(r)[:60]}") |
| return str(r["answer"]).strip() |
| except Exception as e: |
| return f"__ERROR__ {str(e)[:60]}" |
|
|
|
|
| def judge(cli, q, candidate): |
| gold = q.get("answer") or (q.get("derivation") or {}).get("gold") |
| sysm = ( |
| "You judge whether a candidate answer matches the gold answer in meaning, for a specific patient.\n" |
| "Ignore wording, formatting and extra detail, and match on the core clinical fact. For a\n" |
| "multiple-choice question, the same option, by letter or by its text, is a match.\n\n" |
| "A non-answer is never a match:\n" |
| "- 'unknown', 'I don't know', 'no record was provided' and 'cannot determine without the record'\n" |
| " are non-answers. They never match, not even a gold answer that says 'not documented'. Saying\n" |
| " that you have not seen the record is different from stating, after reviewing it, that\n" |
| " something is absent.\n" |
| "- A gold answer of 'not documented', 'never recorded' or 'no' is a checked negative. Only an\n" |
| " explicit statement of absence matches it, such as 'not documented', 'never', 'no such record'\n" |
| " or 'it did not recur'. An admission of ignorance such as 'unknown' does not.\n\n" |
| "For a detailed gold answer with many values, the candidate matches if it captures the key fact\n" |
| "the question asks about, even if it leaves out secondary numbers. Return only JSON.") |
| question_text = qtext(q) |
| usr = (f"Question: {question_text}\nGold answer: {gold}\nCandidate answer: {candidate}\n\n" |
| 'Return JSON in this form: {"match": true|false, "is_non_answer": true|false}') |
| r = cli.complete_json(sysm, usr) |
| if not (isinstance(r.get("match"), bool) and isinstance(r.get("is_non_answer"), bool)): |
| raise ValueError(f"{q.get('qid')}: judge reply lacks boolean match / is_non_answer: {str(r)[:80]}") |
| return r["match"] and not r["is_non_answer"] |
|
|
|
|
| def mentioned_in_dialogues(q, corpus): |
| """Whether the entity that an abstention question calls undocumented is mentioned in the dialogues.""" |
| rx = ((q.get("derivation") or {}).get("operands") or {}).get("absent_regex") |
| return bool(rx and re.search(rx, corpus, re.I)) |
|
|
|
|
| def validate(a) -> None: |
| """Question validation: answerability and memory requirement, judged by an LLM. |
| |
| Oracle probe: the model answers from the dialogues of the evidence visits only. No-memory probe: the |
| model answers from the question alone. A separate judge call compares each answer with the gold. A |
| question is sound iff the oracle answer is correct, the no-memory answer is not (not required on the axes |
| of MEMEBENCH_NO_MEMORY_EXEMPT_AXES), for an abstention question the entity it calls undocumented is not |
| mentioned in any dialogue, and for an axis-1 never-again question the entity is not named in a dialogue after |
| its last documented date, up to the checkpoint. A question whose probes or judge calls fail gets a row with |
| `error` and is not sound. |
| """ |
|
|
| dia = {r["visit_number"]: r for r in |
| (json.loads(l) for l in open(BASE / "dialogues.jsonl") if l.strip())} |
| corpus = " ".join(t["text"] for v in sorted(dia) for t in dia[v]["turns"]) |
| ck = {c["qid"]: c for c in (json.loads(l) for l in open(BASE / "checkpoints.jsonl") if l.strip())} |
|
|
| exempt = no_memory_exempt_axes() |
| questions = [] |
| for f in sorted(glob.glob(str(BASE / "questions_axis*.jsonl"))): |
| questions += [json.loads(l) for l in open(f) if l.strip()] |
|
|
| def dialogues_text(vis, ckpt): |
| parts = [] |
| for v in vis: |
| if v in dia and (not isinstance(ckpt, int) or v <= ckpt): |
| turns = dia[v].get("turns") or [] |
| body = " ".join(f'{t.get("speaker","")}: {t.get("text","")}' for t in turns) |
| parts.append(f"[visit {v}, {dia[v].get('date')}] {body}") |
| return "\n\n".join(parts)[:16000] |
|
|
| def later_mention(q, ckpt): |
| """For an axis-1 never-again question: the first later visit (after gt_date, up to the checkpoint) whose |
| dialogue names every content word of the entity, or None.""" |
| if not str(q.get("axis", "")).startswith("1") or q.get("form") != "never_again": |
| return None |
| entity = str((q.get("axis_specific") or {}).get("entity") or "") |
| words = set(re.findall(r"[a-z][a-z0-9]{3,}", entity.lower())) |
| if not words: |
| return None |
| for v in sorted(dia): |
| if (dia[v].get("date") or "") <= str(q.get("gt_date") or "") or (isinstance(ckpt, int) and v > ckpt): |
| continue |
| text = " ".join(t.get("text", "") for t in dia[v].get("turns") or []).lower() |
| if all(w in text for w in words): |
| return v |
| return None |
|
|
| def one(q): |
| if not has_text(q): |
| return {"qid": q.get("qid"), "axis": q.get("axis"), "grounding": q.get("grounding"), |
| "oracle_ok": False, "nomem_ok": False, "absent_entity_mentioned": False, |
| "never_again_mentioned_later": None, "sound": False, |
| "oracle_ans": "", "nomem_ans": "", "problem": "no question text"} |
| cli = client() |
| cinfo = ck.get(q["qid"], {}) |
| cv = cinfo.get("ckpt_visit") |
| as_of = cinfo.get("ckpt_date") |
| ev = q.get("evidence_visits") or ([q.get("gt_visit")] if q.get("gt_visit") else []) |
| ctx = dialogues_text(ev, cv) |
| oracle_ans = answer_probe(cli, q, ctx, as_of=as_of) |
| nomem_ans = answer_probe(cli, q, "") |
| for ans in (oracle_ans, nomem_ans): |
| if ans.startswith("__ERROR__"): |
| raise RuntimeError(f"{q['qid']}: answer probe failed ({ans[10:]})") |
| later = later_mention(q, cv) |
| oracle_ok = judge(cli, q, oracle_ans) |
| nomem_ok = judge(cli, q, nomem_ans) |
| mentioned = mentioned_in_dialogues(q, corpus) |
| needs_memory = int(str(q.get("axis")).split("_")[0]) in exempt or not nomem_ok |
| return {"qid": q["qid"], "axis": q.get("axis"), "grounding": q.get("grounding"), |
| "oracle_ok": oracle_ok, "nomem_ok": nomem_ok, "absent_entity_mentioned": mentioned, |
| "never_again_mentioned_later": later, |
| "sound": bool(oracle_ok and needs_memory and not mentioned and not later), |
| "oracle_ans": oracle_ans[:200], "nomem_ans": nomem_ans[:200]} |
|
|
| def checked(q): |
| """one(q), or a row with the error when a probe or judge call fails.""" |
| try: |
| return one(q) |
| except Exception as e: |
| return {"qid": q.get("qid"), "axis": q.get("axis"), "grounding": q.get("grounding"), |
| "oracle_ok": False, "nomem_ok": False, "absent_entity_mentioned": False, |
| "never_again_mentioned_later": None, "sound": False, |
| "oracle_ans": "", "nomem_ans": "", "error": f"{type(e).__name__}: {str(e)[:200]}"} |
|
|
| print(f"validating {len(questions)} questions, engine={ENGINE}, {a.workers} workers", flush=True) |
| res = [] |
| with Progress("validate", total=len(questions), out=str(BASE / "progress_validate.json")) as pg: |
| with ThreadPoolExecutor(max_workers=a.workers) as ex: |
| futs = [ex.submit(checked, q) for q in questions] |
| for f in as_completed(futs): |
| r = f.result() |
| res.append(r) |
| pg.tick(ok=not r.get("error")) |
|
|
| (BASE / "validation.jsonl").write_text( |
| "\n".join(json.dumps(r, ensure_ascii=False) for r in res) + "\n") |
|
|
| verdict = {r["qid"]: {"sound": r["sound"], "oracle_ok": r["oracle_ok"], "nomem_ok": r["nomem_ok"], |
| **({"error": r["error"]} if r.get("error") else {})} |
| for r in res} |
| for bf in sorted(glob.glob(str(BASE / "questions_axis*.jsonl"))): |
| rows = [json.loads(l) for l in open(bf)] |
| hit = False |
| for q in rows: |
| if q.get("qid") in verdict: |
| q["_validation"] = verdict[q["qid"]] |
| hit = True |
| if hit: |
| with open(bf, "w", encoding="utf-8") as fh: |
| for q in rows: |
| fh.write(json.dumps(q, ensure_ascii=False) + "\n") |
| print(f" embedded _validation into banks ({len(verdict)} verdicts)") |
|
|
| n = len(res) |
| sound = sum(1 for r in res if r["sound"]) |
| unans = [r for r in res if not r["oracle_ok"] and not r.get("error")] |
| guess = [r for r in res if r["nomem_ok"]] |
| mentioned = [r for r in res if r["absent_entity_mentioned"]] |
| later = [r for r in res if r.get("never_again_mentioned_later")] |
| errors = [r for r in res if r.get("error")] |
| print(f"\n SOUND: {sound}/{n}") |
| print(f" errors (probe or judge failed): {len(errors)}") |
| print(f" unanswerable (oracle could not): {len(unans)}") |
| print(f" guessable (no-memory answered): {len(guess)}") |
| print(f" abstention entity mentioned in the dialogues: {len(mentioned)}") |
| print(f" never-again entity named in a later dialogue: {len(later)}") |
| print(f" sound by axis: {dict(Counter(r['axis'].split('_')[0] for r in res if r['sound']))}") |
| if unans: |
| print("\n unanswerable:") |
| for r in unans[:8]: |
| print(f" {r['qid']}: oracle said {r['oracle_ans'][:60]!r}") |
| if guess: |
| print("\n guessable:") |
| for r in guess[:8]: |
| print(f" {r['qid']}: no-mem said {r['nomem_ans'][:60]!r}") |
| if errors: |
| print("\n errors:") |
| for r in errors[:8]: |
| print(f" {r['qid']}: {r['error'][:100]}") |
| print(f"\n-> {BASE/'validation.jsonl'}") |
|
|
|
|
| def count_errors() -> None: |
| """Print the number of validation rows whose probes or judge calls failed (rows with `error`).""" |
| print(sum(1 for l in open(BASE / "validation.jsonl") if l.strip() and json.loads(l).get("error"))) |
|
|
|
|
| def engines() -> dict: |
| """The model of every LLM stage, as configured in build_patient.sh.""" |
| x, g, d = model("extraction"), model("generation"), model("dialogue") |
| return {"extract": x, "evidence": g, "tags": x, "axes1-2_generate": g, |
| "axes3-8_discover": g, "synthetic_questions": g, "persona_gate": g, "dialogues": d, "validation": g} |
|
|
|
|
| def nlines(p): |
| return sum(1 for _ in open(p)) if os.path.exists(p) else 0 |
|
|
|
|
| def manifest() -> None: |
| """Write MANIFEST.json for a finished patient: counts, models and the file inventory.""" |
| D = {} |
| for name in ["event_chain.jsonl", "evidence_chain.jsonl", "tags.jsonl", "vitals.json", |
| "dialogues.jsonl", "checkpoints.jsonl", "checkpoint_audit.json", |
| "validation.jsonl", "injection_registry.json", "injection_plan.json"]: |
| p = BASE / name |
| if p.exists(): |
| D[name] = {"rows": nlines(p) if name.endswith(".jsonl") else None, |
| "bytes": p.stat().st_size} |
|
|
| q_files = sorted(glob.glob(str(BASE / "questions_axis*.jsonl"))) |
| per_axis, sound_axis, synth = {}, {}, 0 |
| total = sound = 0 |
| for f in q_files: |
| ax = Path(f).stem.split("axis")[1] |
| rows = [json.loads(l) for l in open(f)] |
| per_axis[ax] = len(rows) |
| sa = sum(1 for q in rows if (q.get("_validation") or {}).get("sound")) |
| sound_axis[ax] = sa |
| synth += sum(1 for q in rows if "synthetic" in str(q.get("grounding", ""))) |
| total += len(rows) |
| sound += sa |
| D[os.path.basename(f)] = {"rows": len(rows), "sound": sa} |
|
|
| depths = [] |
| ap = BASE / "checkpoint_audit.json" |
| if ap.exists(): |
| depths = sorted(r["depth"] for r in json.loads(ap.read_text()) if r.get("depth") is not None) |
|
|
| manifest = { |
| "patient_id": PATIENT, |
| "engines": engines(), |
| "questions": {"total": total, "sound": sound, "synthetic": synth, |
| "record": total - synth, "per_axis": per_axis, "sound_per_axis": sound_axis}, |
| "corpus": {"dialogues": nlines(BASE / "dialogues.jsonl"), |
| "event_chain_visits": nlines(BASE / "event_chain.jsonl"), |
| "evidence_timelines": nlines(BASE / "evidence_chain.jsonl")}, |
| "checkpoints": {"n": nlines(BASE / "checkpoints.jsonl"), |
| "depth_min": depths[0] if depths else None, |
| "depth_median": depths[len(depths) // 2] if depths else None, |
| "depth_max": depths[-1] if depths else None}, |
| "no_memory_baseline": { |
| "note": "questions a no-record model answered correctly = do not require memory", |
| "guessable": sum(1 for l in open(BASE / "validation.jsonl") |
| if json.loads(l).get("nomem_ok")) if (BASE / "validation.jsonl").exists() else None}, |
| "deliverables": D, |
| } |
| (BASE / "MANIFEST.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=1)) |
|
|
| print(f"MANIFEST, patient {PATIENT}") |
| print(f" questions: {total} ({total-synth} record + {synth} synthetic), SOUND {sound}") |
| print(f" corpus: {manifest['corpus']['dialogues']} dialogues, {manifest['corpus']['event_chain_visits']} visits") |
| print(f" checkpoints: {manifest['checkpoints']['n']}, " |
| f"depth median {manifest['checkpoints']['depth_median']} visits") |
| print(f" no-memory baseline: {manifest['no_memory_baseline']['guessable']} guessable (rest require memory)") |
| print(f" deliverables: {len(D)} files") |
| print(f"-> {BASE/'MANIFEST.json'}") |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser(description="Stage G: checkpoints, validation, manifest.") |
| sp = ap.add_subparsers(dest="cmd", required=True) |
| for name in ("assign", "audit", "count-errors", "manifest"): |
| sp.add_parser(name) |
| sp.add_parser("validate").add_argument("--workers", type=int, default=8) |
| a = ap.parse_args() |
| {"assign": assign, "audit": audit, "validate": lambda: validate(a), "count-errors": count_errors, |
| "manifest": manifest}[a.cmd]() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
