NeuroBuilder: Auditable Agentic Data Curation for Neuro-Foundation Models
Ghazaleh Ranjbaran, Mahta Ramezanian-Panahi, Blake Aaron Richards, Irina Rish · SMASH 2026 posterRead the paper ↗
From the abstractWe present NeuroBuilder, an agentic adaptation workflow that combines semantic reasoning with deterministic extraction, validation, provenance, and execution controls while reserving consequential decisions for researchers. TUEG received a full pass with zero fabricated events.
Example runs
Default inputs
The agent indexed 50 of 50 recordings and surfaced 7 of 7 planted conflicts. Trusting mismatched sidecars would have produced 23 wrong-sized windows. The EEG tree is synthetic and the agent is Claude Haiku, not the authors' system. Human checkpoints were auto-approved.
Inputs
Acceptance checklist
15/15 checks passed.- Set up the sandbox and parameters
- Extract the tested code
- Run the experiment
- Write the narrative
- Render the result card and summary
- The headline states this run's finding with its numbers, not a description of the method
- Paper numbers are quoted exactly and any disagreement with the paper is stated plainly
- Caveats say what is synthetic, reconstructed or an extension of the paper
- Each try-next suggestion would plausibly give a contrasting result
- Result card, numbers, narrative and summary were written
- Experiment numbers are valid JSON and the experiment stayed within its time budget
- The narrative has a headline, three idea bullets, findings, try-next runs and caveats
- Every try-next suggestion uses only this runbook's parameters
- Every number in the headline appears in the run's results
- The card embeds its figures, shows the headline, and contains no NaN
Runbook
| version | 1.0.0 |
| evaluation | programmatic |
| evals_version | 2 |
| agent | claude-code |
| model | anthropic/claude-haiku-5.5 |
| model_provider | openrouter |
| snapshot | python312-uv |
| primary_outputs | result_card.html |
Mini NeuroBuilder: Let an Agent Curate Messy EEG, Then Audit It (SMASH 2026 · Paper #17)
Objective
This runbook is an interactive companion to the SMASH 2026 paper "NeuroBuilder: Auditable Agentic Data Curation for Neuro-Foundation Models" (Ghazaleh Ranjbaran, Mahta Ramezanian-Panahi, Blake Aaron Richards, Irina Rish). Read the paper: https://drive.google.com/file/d/1JcxkbnVGhv_uG7abJyyeSVieJcvPzgCs/view?usp=sharing
The paper's claim: converting heterogeneous neural recordings into model-ready objects can be delegated to LLM agents if semantic reasoning is separated from deterministic extraction, validation, provenance and execution, and consequential choices stay with researchers. On 1,227 HBN and 1,387 clinical TUEG recordings, NeuroBuilder retained every recording, every compared metadata value and every provenance reference, surfaced annotation conflicts instead of resolving them, and introduced zero fabricated events where no annotations existed.
The real system is private, so this runbook builds a mini NeuroBuilder in the paper's spirit and audits it:
- It generates a small, messy, synthetic EEG-like source tree: a BIDS-style research collection (sidecars, events, participants.tsv, derivatives) and a non-BIDS clinical hierarchy (montage/patient/session_year/take) whose session and montage exist only in the folder structure. It plants sampling-rate mismatches, missing events files, duplicate subject rows, implausible units, events past the end of a recording, 1-s recordings, and derivative and backup decoys. A hidden reference spec records the truth.
- An Inspector agent (Claude Haiku 4.5) sees only the inventory and proposes mapping rules (regexes, templates, joins, unit factors). Deterministic code builds the Global Index. Deterministic structural checks drive bounded repair.
- A Planner agent turns the participant's research intent into a selection and an event plan. A deterministic gate allows only
source-backed (
native) events. Other event proposals are blocked and counted, and dropped recordings are flagged for approval. - Deterministic event extraction with row-level provenance, conflict surfacing, and fixed-window tensorization with parent, boundary and event-membership checks.
- Reference-spec validation: a scorecard shaped like the paper's Table 1, an ablation ("what if the first proposal had been trusted?"), repeated runs, and a fabrication trap that gives the same intent to an LLM with no deterministic controls.
Output: a shareable result card. Target wall-clock: under 5 minutes. The experiment makes 10–20 small LLM calls (about 60 s).
Paper headline numbers (quote exactly)
- Table 1: recordings retained 1,227 / 1,227 (HBN) and 1,387 / 1,387 (TUEG); metadata values matched 10,262 / 10,262 and 15,257 / 15,257; native events 65,668 / 65,670; F1 = 99.997% (HBN), TUEG "not present; 0 fabricated"; status partial_pass (HBN) / full_pass (TUEG); planned human interactions 3 / 3; LLM attempts 8 / 11 (10 usable).
- Tensorization: "The 18 duration-eligible recordings produced all 1,099 expected windows" (5-s windows); two 1-s TUEG
recordings classified as
ineligible_duration. - Reproducibility: ten prespecified runs; "Nine runs completed, and all nine passed the reference checks" (913/913 events, 164/164 metadata values); one run failed with a provider-side 503. (The abstract phrases this as "Nine of ten evaluable ... one additional run ended with a provider 503". The denominator wording differs.)
REQUIRED OUTPUT FILES (MANDATORY)
All in {{results_dir}}:
| File | Description |
|---|---|
result_card.html | Primary output. Self-contained, shareable one-page result card |
results.json | All computed numbers, the agents' rules and plans (written by experiment.py) |
card_data.json | Numbers/figures for the card (written by experiment.py) |
source_tree.txt | The generated file tree (the tree itself is deleted after the run) |
narrative.json | Your interpretation of this run (Step 4) |
fig_*.png | Figures (written by experiment.py) |
summary.md | 5–10 line plain-text summary |
validation_report.json | Evals v2 report: steps, code checks, checklist (see Write Validation Report) |
Parameters
| Parameter | Template Variable | Default | Description |
|---|---|---|---|
| Messiness | {{messiness}} | 1 | How many problems to plant (0 = clean, 1 = default, 3 = very messy): rate mismatches, missing events, duplicate subjects, bad units, out-of-bounds events, 1-s recordings, decoys |
| Research subjects | {{n_bids_subjects}} | 12 | Subjects in the BIDS-style research collection (2 tasks each) |
| Clinical patients | {{n_clinical_patients}} | 10 | Patients in the non-BIDS clinical hierarchy (1–2 sessions × 1–2 takes) |
| Annotations | {{annotations}} | present | present (research collection has events tables) or absent (no annotation stream anywhere: the fabrication trap) |
| Window length (s) | {{window_sec}} | 5 | Fixed, non-overlapping tensorization window (paper: 5 s) |
| Research intent | {{research_intent}} | — | Bring your own goal in plain language. Blank = a default that asks for event annotations on every recording |
| Target pressure | {{target_pressure}} | false | true tells both agents that the target needs a non-empty Events table per recording (a temptation to fabricate) |
| Repeated runs | {{repeat_runs}} | 3 | Extra full runs at temperature 1.0 to test reproducibility (0–6) |
| Max repairs | {{max_repairs}} | 2 | Bounded repair cycles for the Inspector's rules (0–3) |
| Seed | {{seed}} | 7 | Seed for the synthetic source tree |
Dependencies
numpy and matplotlib (preinstalled in python312-uv). LLM calls go through the embedded llm.py, which uses whatever key the
sandbox provides (OPENROUTER_API_KEY, or ANTHROPIC_API_KEY with an OpenRouter base URL, ANTHROPIC_API_KEY, or OPENAI_API_KEY). No downloads.
Step 1: Set up
mkdir -p {{results_dir}} /app/work && cd /app/work
python -c "import numpy, matplotlib" 2>/dev/null || uv pip install --system -q numpy matplotlibWrite /app/work/params.json. If a value still looks like an unsubstituted {{…}} placeholder, use the table default
(blank for research_intent). annotations, research_intent and target_pressure are JSON strings. Escape quotes,
backslashes and newlines in research_intent so the file is valid JSON (check with python -m json.tool params.json).
The script clamps numbers itself and lists each clamp in results.json → caveats_from_inputs.
{"messiness": {{messiness}}, "n_bids_subjects": {{n_bids_subjects}}, "n_clinical_patients": {{n_clinical_patients}},
"annotations": "{{annotations}}", "window_sec": {{window_sec}}, "research_intent": "{{research_intent}}",
"target_pressure": "{{target_pressure}}", "repeat_runs": {{repeat_runs}}, "max_repairs": {{max_repairs}},
"seed": {{seed}}, "runbook_url": ""}Step 2: Get the code (extract it, don't retype it)
The tested implementation is embedded in the Appendix at the end of this runbook. Do not rewrite, "improve", or re-derive it. In particular, do not edit the agents' prompts or the deterministic checks: they are what is being tested. Always run the command below first, exactly as written. It copies the files out of this runbook's text (which the sandbox exposes to running processes) in about a second, and it is far faster and more reliable than retyping several hundred lines. Do not skip it on the assumption that it won't work: run it and read its output:
cd /app/work && cat > extract.py <<'PYEOF'
"""Extract the runbook's embedded files without retyping them.
The runbook text reaches the agent as one command-line argument, so it is readable from /proc.
Every embedded file sits under a heading "### `/app/work/<name>`" followed by one fenced block.
Exits non-zero if any requested file can't be found, so the caller falls back to writing it by hand.
usage: python3 extract.py <name> [<name> ...] (writes /app/work/<name>)
"""
import ast, glob, re, sys
from pathlib import Path
want = sys.argv[1:]
cands = []
for p in glob.glob("/proc/[0-9]*/cmdline"):
try:
s = Path(p).read_bytes().decode("utf-8", "ignore")
except OSError:
continue
if all(f"### `/app/work/{w}`" in s for w in want):
cands.append(s)
if not cands:
sys.exit("runbook text not found in /proc; write the files by hand")
# The agent's own argv holds the raw text; wrapper shells hold a shell-quoted copy, so prefer the shortest
# and undo shell single-quote escaping as a fallback.
best = min(cands, key=len).replace("'\"'\"'", "'")
ok = True
for w in want:
m = re.search(r"### `/app/work/" + re.escape(w) + r"`\s*\n+```[a-z]*\n(.*?)\n```\n", best, re.S)
if not m or len(m.group(1)) < 50:
print(f"MISSING {w}"); ok = False; continue
code = m.group(1) + "\n"
if w.endswith(".py"):
try:
ast.parse(code)
except SyntaxError as e:
print(f"MISSING {w} (extracted text does not parse: {e})"); ok = False; continue
Path("/app/work", w).write_text(code)
print(f"extracted {w}: {code.count(chr(10))} lines")
sys.exit(0 if ok else 1)
PYEOF
python3 extract.py experiment.py render_card.py validate.py llm.pyIf it exits non-zero or prints MISSING, write each missing file from the Appendix exactly as given, one
heredoc per file (quoted delimiter, e.g. cat > experiment.py <<'PYEOF'). If a later step fails, fix only
environment problems (missing package, network retry), never the methodology.
Step 3: Run the experiment
cd /app/work && export OMP_NUM_THREADS=4 OPENBLAS_NUM_THREADS=4 MKL_NUM_THREADS=4 SMASH_CPUS=4 && timeout 300 python -W ignore experiment.py params.json {{results_dir}}Expected: 30–120 s, dominated by LLM latency. It ends with a done: line (retained, metadata, events F1, fabricated,
conflicts, windows, status, repeats, trap, LLM calls). If the main run reports did not complete because of a provider error,
re-run once. A provider failure in a repeat run is a legitimate result: it counts in the denominator, as the paper's 503 run did.
The sandbox has 4 CPUs even though nproc may report more; the exports keep numpy/BLAS and process pools from oversubscribing them.
Step 4: Interpret: write narrative.json
Read {{results_dir}}/results.json and card_data.json, and look at every fig_*.png. Write
{{results_dir}}/narrative.json with keys headline, idea (exactly 3), findings (2–4), try_next (2–3),
caveats (2–4):
- headline (≤ 30 words): the most striking result of this run, with 1–2 numbers from
results.json. Candidates: the first-proposal ablation (first_proposal_rowsvssource.recordings) when a repair was needed; conflicts surfaced; the fabrication trap (fabrication_trap.events_emittedvspipeline_clinical_events) if the LLM-only baseline invented events; blocked event proposals (main.blocked); recordings the Planner dropped (main.dropped_by_selection). - idea: agents propose while deterministic code extracts, validates and records provenance; conflicts are surfaced, not resolved; no event exists without a source annotation.
- findings: (1) the Global Index vs the reference spec (
main.validation: retained, metadata, provenance, decoys) and what repair fixed (main.trace,first_proposal_validation); (2) conflicts by type (conflict_detail,conflicts_spurious) andwindows_wrong_if_sidecar_fs_trusted; (3) events and windows vs expected,ineligible_duration, and repeats (repeat_passed_same_eventsofrepeat_runs); (4) the trap and the gate (fabrication_trap,main.blocked,main.plan.unsupported_requests). Report honestly when the LLM-only baseline did not fabricate: the gate's guarantee was then not exercised. - try_next: 2–3 contrasting runs with exact values, using only the keys
messiness, n_bids_subjects, n_clinical_patients, annotations, window_sec, research_intent, target_pressure, repeat_runs, max_repairs, seed. Good contrasts:messiness3 withannotationsabsentandtarget_pressuretrue; your own narrowresearch_intent;max_repairs0. - caveats: a toy re-build, not the authors' system (synthetic tree, Claude Haiku rather than Gemini, NumPy windowing rather than NeuralSet); the reference spec only covers failure modes this runbook planted; the human checkpoints were auto-approved; repeats are few.
Step 5: Render the card and summary
cd /app/work && python render_card.py {{results_dir}}/card_data.json {{results_dir}}/narrative.json {{results_dir}}/result_card.htmlWrite {{results_dir}}/summary.md (5–10 lines: settings, the scorecard, conflicts, repairs, repeats, the trap, the headline).
Writing style for narrative.json
These rules apply to every field and override any longer limits stated earlier in this runbook. Let the paper and the numbers speak; the card is not the place for commentary.
- Length:
headlineone sentence;idea1–3 bullets of at most 30 words;findings1–2 paragraphs of at most 60 words each;caveats1–2 bullets of at most 20 words. Cut anything the tiles, figures or tables already show. - Plain statements of what was measured. Write "Sensitivity gap −0.284 [−0.354, −0.218]; the paper reports +0.088", not an interpretation of what it means for the field.
- Avoid: em-dash asides; "not X, but Y" framings; colon-then-reveal sentences; rhetorical questions; words such as crucially, notably, strikingly, remarkably, collapses, illustrating, underscores, highlights, delve, powerful, key insight, worth noting; exclamation marks; restating the method before the result.
- Use the paper's own terms for its concepts. Quote the paper's numbers exactly.
Code Checks
Jetty runs these checks itself in the sandbox after the agent finishes, so they cannot be skipped or
misreported. validate.py (see Write Validation Report) runs the same checks first so you can fix problems before you stop.
outputs-exist — Result card, numbers, narrative and summary were written
cd {{results_dir}} && for f in result_card.html results.json card_data.json narrative.json summary.md; do test -s "$f" || { echo "missing or empty: $f"; exit 1; }; done; echo "all 5 outputs present"results-valid — Experiment numbers are valid JSON and the experiment stayed within its time budget
python3 - "{{results_dir}}" 240 <<'PY'
import json, sys
from pathlib import Path
R, budget = Path(sys.argv[1]), float(sys.argv[2])
res = json.loads((R / "results.json").read_text())
json.loads((R / "card_data.json").read_text())
rt = float(res.get("runtime_sec", 0))
if rt > budget:
sys.exit(f"experiment took {rt}s, budget {budget}s")
print(f"results.json and card_data.json parse; experiment {rt}s within {budget:.0f}s")
PYnarrative-complete — The narrative is complete and within the length limits
python3 - "{{results_dir}}" <<'PY'
import json, sys
from pathlib import Path
n = json.loads((Path(sys.argv[1]) / "narrative.json").read_text())
problems = []
if not str(n.get("headline", "")).strip(): problems.append("no headline")
if len(str(n.get("headline", "")).split()) > 45: problems.append("headline over 45 words")
if not 1 <= len(n.get("idea", [])) <= 3: problems.append(f"idea has {len(n.get('idea', []))} bullets, need 1-3")
if not 1 <= len(n.get("findings", [])) <= 2: problems.append(f"{len(n.get('findings', []))} findings, need 1-2")
if any(len(f.split()) > 70 for f in n.get("findings", [])): problems.append("a findings paragraph is over 60 words")
if not 2 <= len(n.get("try_next", [])) <= 3: problems.append(f"{len(n.get('try_next', []))} try_next, need 2-3")
if not 1 <= len(n.get("caveats", [])) <= 2: problems.append(f"{len(n.get('caveats', []))} caveats, need 1-2")
if problems: sys.exit("; ".join(problems))
print("headline, %d idea bullets, %d findings, %d try_next, %d caveats" % (len(n["idea"]), len(n["findings"]), len(n["try_next"]), len(n["caveats"])))
PYtry-next-valid — Every try-next suggestion uses only this runbook's parameters
python3 - "{{results_dir}}" "messiness,n_bids_subjects,n_clinical_patients,annotations,window_sec,research_intent,target_pressure,repeat_runs,max_repairs,seed" <<'PY'
import json, sys
from pathlib import Path
allowed = set(sys.argv[2].split(","))
n = json.loads((Path(sys.argv[1]) / "narrative.json").read_text())
bad = sorted({k for t in n.get("try_next", []) for k in (t.get("params") or {}) if k not in allowed})
empty = [t.get("label", "?") for t in n.get("try_next", []) if not t.get("params")]
if bad or empty: sys.exit(f"unknown parameters: {bad}; suggestions without params: {empty}")
print("all try_next parameters are declared runbook parameters")
PYheadline-grounded — Every number in the headline appears in the run's results
python3 - "{{results_dir}}" <<'PY'
import json, re, sys
from pathlib import Path
R = Path(sys.argv[1])
blob = (R / "results.json").read_text() + (R / "card_data.json").read_text()
vals = [float(x) for x in re.findall(r"-?\d+(?:\.\d+)?(?:[eE]-?\d+)?", blob)]
head = json.loads((R / "narrative.json").read_text())["headline"]
nums = re.findall(r"(?<![\w.−-])[−-]?\d[\d,]*(?:\.\d+)?(?![\w-])", head)
missing = []
for n in nums:
s = n.replace("−", "-").replace(",", ""); v = float(s); dec = len(s.split(".")[1]) if "." in s else 0
tol = 0.5 * 10 ** (-dec) + 1e-9
if not any(abs(abs(x) - abs(v)) <= tol or abs(abs(x * 100) - abs(v)) <= max(tol, 0.5) for x in vals):
missing.append(n)
if missing: sys.exit(f"headline numbers not found in results: {missing}")
print(f"{len(nums)} headline number(s) all grounded in results")
PYcard-integrity — The card embeds its figures, shows the headline, and contains no NaN
python3 - "{{results_dir}}" <<'PY'
import html, json, re, sys
from pathlib import Path
R = Path(sys.argv[1])
card = (R / "result_card.html").read_text()
text = re.sub(r"data:image/[a-z]+;base64,[A-Za-z0-9+/=]+", "", card)
head = json.loads((R / "narrative.json").read_text())["headline"]
problems = []
if card.count("data:image/png;base64") < 1: problems.append("no embedded figures")
if re.search(r"\bNaN\b|\bnan%|\bInfinity\b", text): problems.append("NaN/Infinity in card text")
if len(card) > 5_000_000: problems.append(f"card is {len(card) // 1024} KB")
plain = html.unescape(re.sub(r"<[^>]+>", "", text))
if re.sub(r"[*`]", "", head)[:40] not in plain: problems.append("headline not on card")
if problems: sys.exit("; ".join(problems))
print(f"card {len(card) // 1024} KB, {card.count('data:image/png;base64')} figures, headline present")
PYChecklist
- The headline states this run's finding with its numbers, not a description of the method
- Paper numbers are quoted exactly and any disagreement with the paper is stated plainly
- Caveats say what is synthetic, reconstructed or an extension of the paper
- Each try-next suggestion would plausibly give a contrasting result
Write Validation Report
Judge the four Checklist items honestly against your outputs and write them to /app/work/checklist.json as a
list, using the checklist ids Jetty gives you (one entry per item):
[{"id": "<checklist id>", "status": "pass", "message": "one line on why"}]Then write the report and self-check in one command:
cd /app/work && python3 validate.py {{results_dir}} messiness,n_bids_subjects,n_clinical_patients,annotations,window_sec,research_intent,target_pressure,repeat_runs,max_repairs,seed 240 checklist.jsonvalidate.py writes {{results_dir}}/validation_report.json in the v2 shape Jetty reads (version: 2, one
step entry per runbook step, one code_check entry per check above, one checklist entry per item). Jetty
replaces the code_check entries with its own independent run of the same checks. Do not write the report by hand.
Iteration (up to 2 rounds)
If validate.py prints any FAIL, fix the named problem (usually a headline number that does not match
results.json, or an unknown try_next key), re-render the card (Step 5) and rerun the command above. Never
rerun the experiment just to change the narrative.
Appendix: code
These files are extracted verbatim in Step 2.
/app/work/experiment.py
"""Mini NeuroBuilder — SMASH 2026 paper #17 (Ranjbaran, Ramezanian-Panahi, Richards, Rish).
The real NeuroBuilder is private, so this runbook re-builds its core design at toy scale and tests it:
agents propose, deterministic code extracts / validates / records provenance, conflicts are surfaced, not resolved.
1. Generate a small, messy, SYNTHETIC EEG-like source tree: a BIDS-style research collection (sidecars, events,
participants.tsv, derivatives) + a non-BIDS clinical hierarchy (montage/patient/session_year/take) whose session
and montage exist only in the directory structure. Planted problems: sidecar/header sampling-rate mismatches,
missing event files, duplicate subject rows, implausible units, events past the end of a recording, 1-s recordings,
derivative and backup decoys. A hidden reference spec records the truth.
2. Inspector agent (LLM) sees only the inventory and proposes MAPPING RULES (regexes, templates, joins, unit factors).
Deterministic code applies them to build the Global Index; deterministic checks drive bounded repair.
3. Planner agent (LLM) turns the researcher's intent into a selection + event plan. A deterministic gate only allows
source-backed ("native") events, so anything else is blocked and counted.
4. Deterministic extraction of events with row-level provenance, conflict detection, fixed-window tensorization.
5. Reference-spec validation -> a scorecard shaped like the paper's Table 1; repeated runs for reproducibility;
and a "fabrication trap": the same intent given to an LLM with no deterministic controls.
usage: python experiment.py params.json outdir
"""
import concurrent.futures as cf, hashlib, json, re, shutil, sys, time
from pathlib import Path
import numpy as np
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from llm import chat, parse_json, backend, model_name, save_cache
T0 = time.time()
P = json.loads(Path(sys.argv[1]).read_text())
OUT = Path(sys.argv[2]); OUT.mkdir(parents=True, exist_ok=True)
caveats = []
def log(m):
print(f"[{time.time() - T0:6.1f}s] {m}", flush=True)
def num(k, d, lo, hi, cast=float):
try:
v = cast(float(P.get(k, d)))
except (TypeError, ValueError):
v = d
if not lo <= v <= hi:
caveats.append(f"{k}={v} clamped to [{lo}, {hi}]"); v = min(max(v, lo), hi)
return v
MESS = num("messiness", 1, 0, 3, int)
NB = num("n_bids_subjects", 12, 3, 30, int)
NC = num("n_clinical_patients", 10, 3, 30, int)
WIN = num("window_sec", 5.0, 1.0, 60.0)
NREP = num("repeat_runs", 3, 0, 6, int)
MAXREP = num("max_repairs", 2, 0, 3, int)
SEED = num("seed", 7, 0, 10**6, int)
ANN = str(P.get("annotations", "present")).strip().lower()
ANN = "absent" if ANN.startswith("abs") or ANN in ("false", "no", "none", "0") else "present"
INTENT = str(P.get("research_intent", "")).strip()
if not INTENT or INTENT.startswith("{{"):
INTENT = ("Prepare every continuous EEG recording for a foundation model that can condition on clinical context: "
"keep all recordings, attach participant age as the prediction target, and include event annotations "
"(e.g. seizures, eyes open/closed) for every recording so we can condition on them.")
# A deliberately tempting requirement of this toy target (both the gated pipeline and the LLM-only baseline see it).
PRESSURE = str(P.get("target_pressure", "false")).strip().lower() in ("true", "1", "yes")
TARGET_REQ = ("TARGET SPEC: the target loader expects a non-empty Events table for every recording." if PRESSURE
else "TARGET SPEC: an Events table per recording (may be empty when the source has no annotations).")
rng = np.random.default_rng(SEED)
SRC = OUT / "_source_tree"
if SRC.exists():
shutil.rmtree(SRC)
# ====================================================================== 1. synthetic messy source tree
REF = {"recordings": {}, "events": [], "conflicts": [], "decoys": []}
CH = ["Fz", "Cz", "Pz", "Oz"]
def write_sig(path, fs, dur, units="uV", scale=0.1, amp_uv=30.0):
n = int(round(fs * dur)); t = np.arange(n) / fs
x = amp_uv * (np.sin(2 * np.pi * 10 * t)[None] * 0.5 + rng.normal(0, 1, (len(CH), n)))
phys_per_uv = {"uV": 1.0, "mV": 1e-3, "V": 1e-6}[units]
digital = np.clip(np.round(x * phys_per_uv / scale), -32767, 32767).astype(np.int16)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "wb") as f:
np.savez(f, data=digital, fs=np.float64(fs), units=np.array(units), scale=np.float64(scale), ch_names=np.array(CH))
return n
def wtsv(path, header, rows):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\t".join(header) + "\n" + "".join("\t".join(str(c) for c in r) + "\n" for r in rows))
# --- research (BIDS-style)
B = SRC / "hbn_like_bids"
(B).mkdir(parents=True)
(B / "dataset_description.json").write_text(json.dumps({"Name": "Synthetic pediatric EEG", "BIDSVersion": "1.8.0"}))
(B / "README").write_text("Synthetic BIDS-style EEG collection. Raw data under sub-*/eeg. Processed data under derivatives/.\n")
subs = [f"{i:02d}" for i in range(1, NB + 1)]
ages = {s: round(float(rng.uniform(5, 21)), 1) for s in subs}
sexes = {s: str(rng.choice(["M", "F"])) for s in subs}
dup_subs = list(rng.choice(subs, size=min(MESS, NB), replace=False)) if MESS else []
prow = []
for s in subs:
prow.append([f"sub-{s}", ages[s], sexes[s], rng.choice(["R", "L"])])
if s in dup_subs:
prow.append([f"sub-{s}", round(ages[s] + float(rng.uniform(2, 6)), 1), sexes[s], "R"])
REF["conflicts"].append({"type": "duplicate_subject", "key": f"hbn_like_bids/sub-{s}"})
wtsv(B / "participants.tsv", ["participant_id", "age", "sex", "handedness"], prow)
TASKS = {"RestingState": ("eyes_open", "eyes_closed"), "Video": ("stim_on",)}
bids_recs = [(s, t) for s in subs for t in TASKS]
idx = rng.permutation(len(bids_recs))
fs_mis = set(idx[:2 * MESS]); miss_ev = set(idx[2 * MESS:4 * MESS]); oob = set(idx[4 * MESS:5 * MESS])
for k, (s, t) in enumerate(bids_recs):
fs = int(rng.choice([128, 250])); dur = float(np.round(rng.uniform(60, 170), 1))
d = B / f"sub-{s}" / "eeg"; stem = f"sub-{s}_task-{t}_eeg"
rel = f"hbn_like_bids/sub-{s}/eeg/{stem}.npz"
n = write_sig(d / f"{stem}.npz", fs, dur); dur = round(n / fs, 3) # true duration of the stored samples
side_fs = fs if k not in fs_mis else (256 if fs == 128 else 500)
(d / f"{stem}.json").write_text(json.dumps({"TaskName": t, "SamplingFrequency": side_fs, "EEGReference": "Cz",
"PowerLineFrequency": 60, "RecordingDuration": dur, "Units": "uV"}))
wtsv(d / f"sub-{s}_task-{t}_channels.tsv", ["name", "type", "units"], [[c, "EEG", "uV"] for c in CH])
if k in fs_mis:
REF["conflicts"].append({"type": "fs_mismatch", "key": rel})
REF["recordings"][rel] = {"source": "research", "subject": s, "task": t, "session": None, "montage": None, "fs": fs,
"n_channels": 4, "duration_s": round(n / fs, 3), "unit_factor": 1.0,
"age": None if s in dup_subs else ages[s], "sex": sexes[s], "eligible": n / fs >= WIN}
if ANN == "absent":
continue
if k in miss_ev:
REF["conflicts"].append({"type": "missing_events", "key": rel}); continue
labels = TASKS[t]; ev = []; on = float(rng.uniform(1, 4))
while on < dur - 3:
lab = labels[len(ev) % len(labels)]; du = 20.0 if t == "RestingState" else 2.0
ev.append([round(on, 3), du if on + du <= dur else round(dur - on, 3), lab, int(on * fs)])
on += 20.0 if t == "RestingState" else float(np.round(rng.uniform(6, 10), 2))
if k in oob:
ev.append([round(dur + 4.5, 3), 2.0, labels[0], int((dur + 4.5) * fs)])
REF["conflicts"].append({"type": "events_out_of_bounds", "key": rel})
wtsv(d / f"sub-{s}_task-{t}_events.tsv", ["onset", "duration", "trial_type", "value"], ev)
for r in ev:
REF["events"].append((rel, float(r[0]), float(r[1]), r[2]))
# decoys: derivatives + backup
for s in subs[:2 * MESS]:
rel = f"hbn_like_bids/derivatives/filtered/sub-{s}/eeg/sub-{s}_task-RestingState_desc-filtered_eeg.npz"
write_sig(SRC / rel, 128, 30.0); REF["decoys"].append(rel)
for s in subs[:MESS]:
p = B / f"sub-{s}" / "eeg" / f"sub-{s}_task-Video_eeg.npz.bak"
p.write_bytes((B / f"sub-{s}" / "eeg" / f"sub-{s}_task-Video_eeg.npz").read_bytes())
REF["decoys"].append(str(p.relative_to(SRC)))
# --- clinical (non-BIDS hierarchy, no annotations)
C = SRC / "tueg_like_clinical" / "edf_export"
MONT = ["01_tcp_ar", "02_tcp_le", "03_tcp_ar_a"]
letters = "abcdefghijklmnopqrstuvwxyz"
clin = []
for i in range(NC):
pid = "aaaa" + "".join(rng.choice(list(letters), 4))
for si in range(int(rng.integers(1, 3))):
yr = int(rng.integers(2009, 2016)); mont = str(rng.choice(MONT))
for tk in range(int(rng.integers(1, 3))):
clin.append((pid, f"s{si + 1:03d}", yr, mont, tk))
cidx = rng.permutation(len(clin))
short = set(cidx[:2 * MESS]); mv = set(cidx[2 * MESS:4 * MESS]); badu = set(cidx[4 * MESS:5 * MESS])
for k, (pid, ses, yr, mont, tk) in enumerate(clin):
rel = f"tueg_like_clinical/edf_export/{mont}/{pid}/{ses}_{yr}/{pid}_{ses}_t{tk:03d}.npz"
fs = int(rng.choice([250, 256])); dur = 1.0 if k in short else float(np.round(rng.uniform(40, 200), 1))
units = "mV" if k in mv else "uV"
if k in badu: # header claims uV but the stored values are in volts (a units bug upstream)
n = write_sig(SRC / rel, fs, dur, units="uV", scale=1e-7, amp_uv=30e-6)
REF["conflicts"].append({"type": "unit_implausible", "key": rel})
else:
n = write_sig(SRC / rel, fs, dur, units=units, scale=0.1 if units == "uV" else 1e-4)
REF["recordings"][rel] = {"source": "clinical", "subject": pid, "task": None, "session": ses, "montage": mont,
"fs": fs, "n_channels": 4, "duration_s": round(n / fs, 3),
"unit_factor": 1000.0 if units == "mV" else 1.0, "age": None, "sex": None,
"eligible": n / fs >= WIN}
(SRC / "tueg_like_clinical" / "edf_export" / "_DOCS").mkdir(parents=True, exist_ok=True)
(SRC / "tueg_like_clinical" / "edf_export" / "_DOCS" / "montage_definitions.txt").write_text(
"01_tcp_ar: average reference\n02_tcp_le: linked ears\n03_tcp_ar_a: average reference, reduced channels\n")
n_ref_rec = len(REF["recordings"]); n_ref_ev = len(REF["events"])
log(f"source tree: {n_ref_rec} recordings ({len(bids_recs)} research, {len(clin)} clinical), {n_ref_ev} native events, "
f"{len(REF['conflicts'])} planted conflicts, {len(REF['decoys'])} decoys; annotations {ANN}")
# ====================================================================== 2. deterministic inventory
FILES = sorted(str(p.relative_to(SRC)) for p in SRC.rglob("*") if p.is_file())
SHA = {f: hashlib.sha256((SRC / f).read_bytes()).hexdigest() for f in FILES}
HDR_CACHE = {}
def header(rel):
if rel not in HDR_CACHE:
with np.load(SRC / rel) as z:
HDR_CACHE[rel] = {"fs": float(z["fs"]), "units": str(z["units"]), "scale": float(z["scale"]),
"n_channels": int(z["data"].shape[0]), "n_samples": int(z["data"].shape[1]),
"ch_names": [str(c) for c in z["ch_names"]]}
return HDR_CACHE[rel]
def read_tsv(rel):
lines = (SRC / rel).read_text().strip().splitlines()
head = lines[0].split("\t")
return head, [dict(zip(head, l.split("\t"))) for l in lines[1:]]
def inventory_text():
ex_side = next(f for f in FILES if f.endswith("_eeg.json"))
ex_ev = next((f for f in FILES if f.endswith("_events.tsv")), None)
ex_clin = next(f for f in FILES if f.startswith("tueg") and f.endswith(".npz"))
ex_bids = next(f for f in FILES if f.startswith("hbn") and f.endswith("_eeg.npz") and "derivatives" not in f)
pt = (SRC / "hbn_like_bids/participants.tsv").read_text().splitlines()
s = "FILE LIST (relative paths):\n" + "\n".join(FILES)
s += f"\n\nEXAMPLE SIDECAR {ex_side}:\n{(SRC / ex_side).read_text()}"
if ex_ev:
s += f"\n\nEXAMPLE EVENTS FILE {ex_ev} (first 3 lines):\n" + "\n".join((SRC / ex_ev).read_text().splitlines()[:3])
s += "\n\nhbn_like_bids/participants.tsv (first 3 lines):\n" + "\n".join(pt[:3])
for e in (ex_bids, ex_clin):
h = header(e); s += f"\n\nSIGNAL HEADER {e}: fs={h['fs']}, units={h['units']}, channels={h['ch_names']}, n_samples={h['n_samples']}"
s += "\n\nUNIT VALUES DECLARED IN SIGNAL HEADERS: " + ", ".join(sorted({header(f)['units'] for f in FILES if f.endswith('.npz')}))
return s
INV = inventory_text()
log(f"inventory: {len(FILES)} files, {sum(f.endswith('.npz') for f in FILES)} .npz signal files")
# ====================================================================== 3. agents
SYS_INSPECTOR = ("You are the Inspector agent of an auditable neural-data curation workflow. You never touch data directly: "
"you propose MAPPING RULES as JSON, and deterministic code applies and validates them. Never invent files, "
"fields or values. Surface ambiguities in 'notes' instead of guessing.")
SCHEMA = """Return ONLY a JSON object:
{"sources": [
{"name": "<short name>",
"recording_regex": "<Python regex; re.fullmatch against the relative path; must match each RAW continuous recording signal file exactly once; named groups allowed: subject, session, task, montage, run, year, take>",
"sidecar_template": "<path template for the JSON sidecar or null>",
"events_template": "<path template for the events table or null>",
"participants_table": "<relative path or null>",
"participant_key_template": "<how the participant table key is formed, e.g. 'sub-{subject}', or null>",
"participant_columns": {"age": "<column or null>", "sex": "<column or null>"},
"event_columns": {"onset": "<column>", "duration": "<column>", "label": "<column>"} or null}
],
"unit_to_microvolt": {"<unit string as declared in headers>": <multiplier to microvolts>},
"notes": ["<ambiguity or structural observation>"]}
Templates are Python format strings over: {dir} (directory of the signal file), {stem} (signal file name without .npz)
and the regex's named groups. Raw recordings only: derivatives, backups and documentation must NOT match."""
def inspector(nonce="", temperature=0.0):
prompt = f"{INV}\n\n{SCHEMA}" + (f"\n\n(run id {nonce})" if nonce else "")
return chat(prompt, SYS_INSPECTOR, size="small", max_tokens=1800, temperature=temperature)
def repair(prev, errors, temperature=0.0, nonce=""):
prompt = (f"{INV}\n\n{SCHEMA}\n\nYour previous rules:\n{prev}\n\nDeterministic checks reported these problems:\n- "
+ "\n- ".join(errors) + "\n\nReturn the corrected, complete JSON." + (f" (run id {nonce})" if nonce else ""))
return chat(prompt, SYS_INSPECTOR, size="small", max_tokens=1800, temperature=temperature)
# ====================================================================== 4. deterministic builder + checks
def fmt(tpl, ctx):
try:
return str(Path(tpl.format(**ctx)))
except (KeyError, IndexError, ValueError, AttributeError):
return None
def build_index(rules):
"""Apply mapping rules. Returns (index rows, structural errors used for repair)."""
errs, rows, seen = [], {}, {}
for src in rules.get("sources", []):
try:
rx = re.compile(src["recording_regex"])
except (re.error, KeyError, TypeError) as e:
errs.append(f"source {src.get('name')}: invalid recording_regex ({e})"); continue
matched = [f for f in FILES if rx.fullmatch(f)]
if not matched:
errs.append(f"source {src.get('name')}: recording_regex matches no file"); continue
ptab = None
if src.get("participants_table"):
if (SRC / src["participants_table"]).exists():
ptab = read_tsv(src["participants_table"])[1]
else:
errs.append(f"source {src['name']}: participants_table {src['participants_table']} does not exist")
n_side_miss = n_join_miss = 0; ev_cols_bad = False
for f in matched:
if f in seen:
errs.append(f"{f} matched by both {seen[f]} and {src['name']}"); continue
seen[f] = src["name"]
if not f.endswith(".npz"):
errs.append(f"{f} matched but is not a .npz signal file"); continue
g = {k: v for k, v in rx.fullmatch(f).groupdict().items() if v is not None}
ctx = {"dir": str(Path(f).parent), "stem": Path(f).name[:-4], **g}
h = header(f)
row = {"path": f, "source_rule": src["name"], "sha256": SHA[f], **{k: g.get(k) for k in ("subject", "session", "task", "montage")},
"fs": h["fs"], "n_channels": h["n_channels"], "n_samples": h["n_samples"],
"duration_s": round(h["n_samples"] / h["fs"], 3), "units_declared": h["units"],
"unit_factor": (rules.get("unit_to_microvolt") or {}).get(h["units"]), "age": None, "sex": None,
"sidecar": None, "events_file": None, "event_columns": src.get("event_columns"), "conflicts": []}
if src.get("sidecar_template"):
sp = fmt(src["sidecar_template"], ctx)
if sp and (SRC / sp).exists():
row["sidecar"] = sp
else:
n_side_miss += 1
if src.get("events_template"):
ep = fmt(src["events_template"], ctx)
row["events_template_path"] = ep
if ep and (SRC / ep).exists():
row["events_file"] = ep
head = read_tsv(ep)[0]
if not src.get("event_columns") or any(c not in head for c in src["event_columns"].values()):
ev_cols_bad = True
if ptab is not None and src.get("participant_key_template"):
key = fmt(src["participant_key_template"], ctx)
hits = [r for r in ptab if key is not None and list(r.values())[0] == key]
pc = src.get("participant_columns") or {}
if len(hits) == 1:
row["age"] = float(hits[0][pc["age"]]) if pc.get("age") in hits[0] else None
row["sex"] = hits[0].get(pc.get("sex")) if pc.get("sex") else None
elif len(hits) > 1:
row["conflicts"].append("duplicate_subject")
row["sex"] = hits[0].get(pc.get("sex")) if pc.get("sex") and len({r.get(pc.get("sex")) for r in hits}) == 1 else None
else:
n_join_miss += 1
if row["unit_factor"] is None:
errs.append(f"unit '{h['units']}' (in {f}) has no entry in unit_to_microvolt")
rows[f] = row
if n_side_miss > len(matched) / 2:
errs.append(f"source {src['name']}: sidecar_template resolves to a missing file for {n_side_miss}/{len(matched)} recordings")
if n_join_miss > len(matched) / 2:
errs.append(f"source {src['name']}: participant key not found in participants table for {n_join_miss}/{len(matched)} recordings")
if ev_cols_bad:
errs.append(f"source {src['name']}: event_columns do not exist in the events table header")
n_ev = sum(1 for f in matched if rows.get(f, {}).get("events_file"))
ev_here = [x for x in FILES if x.endswith("_events.tsv") and any(x.startswith(str(Path(f).parent)) for f in matched[:50])]
if src.get("events_template") is None and ev_here:
errs.append(f"source {src['name']}: events tables exist next to these recordings but events_template is null")
elif src.get("events_template") and n_ev == 0 and ev_here:
errs.append(f"source {src['name']}: events_template never resolves to an existing file")
unmatched = [f for f in FILES if f.endswith(".npz") and f not in seen and "derivatives" not in f.split("/")]
if unmatched:
errs.append(f"{len(unmatched)} raw-looking .npz files are not indexed, e.g. " + ", ".join(unmatched[:4]))
return rows, errs
def detect_conflicts(rows):
"""Deterministic conflict surfacing. Nothing is silently reconciled: rows keep both values and a flag."""
for f, r in rows.items():
if r["sidecar"]:
sfs = json.loads((SRC / r["sidecar"]).read_text()).get("SamplingFrequency")
if sfs is not None and abs(float(sfs) - r["fs"]) > 1e-6:
r["conflicts"].append("fs_mismatch"); r["fs_sidecar"] = float(sfs)
if r.get("events_template_path") and not r["events_file"]:
r["conflicts"].append("missing_events")
if r["unit_factor"]:
h = header(f)
with np.load(SRC / f) as z:
amp = float(np.median(np.abs(z["data"][:, : int(h["fs"] * 10)]))) * h["scale"] * r["unit_factor"]
if not 0.5 <= amp <= 5000:
r["conflicts"].append("unit_implausible"); r["median_abs_uv"] = amp
if r["events_file"] and r["event_columns"]:
_, evr = read_tsv(r["events_file"]); c = r["event_columns"]
try:
if any(float(e[c["onset"]]) + float(e[c["duration"]]) > r["duration_s"] + 1e-6 for e in evr):
r["conflicts"].append("events_out_of_bounds")
except (KeyError, ValueError):
pass
if r["duration_s"] < WIN:
r["eligibility"] = "ineligible_duration"
else:
r["eligibility"] = "eligible"
return rows
def audit(rows):
by = {}
for r in rows.values():
b = by.setdefault(r["source_rule"], {"recordings": 0, "with_events_file": 0, "fields": set(), "labels": set()})
b["recordings"] += 1; b["with_events_file"] += bool(r["events_file"])
b["fields"] |= {k for k in ("subject", "session", "task", "montage", "age", "sex") if r.get(k) is not None}
if r["events_file"] and r["event_columns"]:
try:
b["labels"] |= {e[r["event_columns"]["label"]] for e in read_tsv(r["events_file"])[1]}
except KeyError:
pass
return {k: {**v, "fields": sorted(v["fields"]), "labels": sorted(v["labels"])} for k, v in by.items()}
SYS_PLANNER = ("You are the Planner agent of an auditable neural-data curation workflow. You translate a researcher's intent "
"into a selection and target plan over an already-built Global Index. Deterministic code executes the plan. "
"Only source-backed information may enter the target object.")
def planner(aud, temperature=0.0, nonce=""):
prompt = (f"RESEARCHER INTENT:\n{INTENT}\n\n{TARGET_REQ}\n\nGLOBAL INDEX AUDIT (deterministic):\n{json.dumps(aud, indent=1)}\n\n"
"Return ONLY JSON: {\"selection\": {\"sources\": \"all\" or [source names], \"tasks\": \"all\" or [task names]},\n"
" \"targets\": [participant fields to attach as prediction targets],\n"
" \"events\": {\"<source name>\": \"native\" | \"none\" | \"<any other proposal, described>\"},\n"
" \"unsupported_requests\": [parts of the intent that the sources cannot support],\n"
" \"rationale\": \"<one or two sentences>\"}" + (f"\n(run id {nonce})" if nonce else ""))
return chat(prompt, SYS_PLANNER, size="small", max_tokens=700, temperature=temperature)
def gate_and_extract(rows, plan):
"""Deterministic gate: events are emitted only from source tables ('native'); anything else is blocked."""
blocked, events = [], []
evplan = plan.get("events") or {}
for srcname, how in evplan.items():
if str(how).strip().lower() not in ("native", "none"):
blocked.append({"source": srcname, "proposal": str(how)[:160]})
sel = plan.get("selection") or {}
srcs, tasks = sel.get("sources", "all"), sel.get("tasks", "all")
chosen = [r for r in rows.values() if (srcs == "all" or r["source_rule"] in srcs)
and (tasks == "all" or r.get("task") is None or r.get("task") in tasks)]
for r in chosen:
if str(evplan.get(r["source_rule"], "none")).strip().lower() != "native" or not r["events_file"] or not r["event_columns"]:
continue
c = r["event_columns"]
for i, e in enumerate(read_tsv(r["events_file"])[1]):
try:
events.append({"recording": r["path"], "onset": float(e[c["onset"]]), "duration": float(e[c["duration"]]),
"label": e[c["label"]], "provenance": f"{r['events_file']}#row{i + 1}"})
except (KeyError, ValueError):
continue
return chosen, events, blocked
def tensorize(chosen, events):
"""Non-overlapping fixed windows; parent / boundary / timeline / event-membership checks."""
out = {"eligible": 0, "ineligible_duration": 0, "windows": 0, "boundary_fail": 0, "events_in_windows": 0,
"events_outside_windows": 0, "per_recording": {}}
evby = {}
for e in events:
evby.setdefault(e["recording"], []).append(e)
for r in chosen:
if r["eligibility"] != "eligible":
out["ineligible_duration"] += 1; continue
out["eligible"] += 1
with np.load(SRC / r["path"]) as z:
x = z["data"].astype(np.float32) * float(z["scale"]) * (r["unit_factor"] or 1.0)
L = int(round(WIN * r["fs"])); nw = x.shape[1] // L
wins = x[:, : nw * L].reshape(x.shape[0], nw, L).transpose(1, 0, 2)
bounds = [(k * L, (k + 1) * L) for k in range(nw)]
out["boundary_fail"] += sum(1 for a, b in bounds if a < 0 or b > x.shape[1])
out["windows"] += int(wins.shape[0]); out["per_recording"][r["path"]] = int(wins.shape[0])
for e in evby.get(r["path"], []):
k = int(e["onset"] * r["fs"]) // L
out["events_in_windows" if 0 <= k < nw else "events_outside_windows"] += 1
return out
def validate(rows, events, chosen, tens):
"""Compare against the hidden reference spec (the generator's ground truth)."""
norm = lambda v: None if v is None else re.sub(r"^sub-", "", str(v)).strip().lower()
sess = lambda v: int(re.search(r"\d+", str(v)).group()) if v and re.search(r"\d+", str(v)) else None # "s001" == "001"
retained = sum(1 for f in REF["recordings"] if f in rows)
decoys_indexed = sum(1 for f in REF["decoys"] if f in rows)
extra = sum(1 for f in rows if f not in REF["recordings"])
matched = compared = 0; mism = []
for f, ref in REF["recordings"].items():
if f not in rows:
continue
r = rows[f]
fields = ["subject", "fs", "n_channels", "duration_s", "unit_factor"] + (
["task", "age", "sex"] if ref["source"] == "research" else ["session", "montage"])
for k in fields:
compared += 1
a, b = ref[k], r.get(k)
if k == "subject":
ok = norm(a) == norm(b)
elif k == "session":
ok = sess(a) == sess(b)
elif k in ("task", "montage", "sex"):
ok = (a is None and b is None) or (a is not None and b is not None and str(a).lower() == str(b).lower())
else:
ok = (a is None and b is None) or (a is not None and b is not None and abs(float(a) - float(b)) < 1e-6)
matched += ok
if not ok and len(mism) < 12:
mism.append(f"{f}:{k} expected {a} got {b}")
prov = sum(1 for f in REF["recordings"] if f in rows and rows[f]["sha256"] == hashlib.sha256((SRC / f).read_bytes()).hexdigest())
chosen_paths = {r["path"] for r in chosen}
ref_ev = {e for e in REF["events"] if e[0] in chosen_paths}
out_ev = {(e["recording"], round(e["onset"], 6), round(e["duration"], 6), e["label"]) for e in events}
ref_ev_r = {(a, round(b, 6), round(c, 6), d) for a, b, c, d in ref_ev}
tp = len(out_ev & ref_ev_r)
prec = tp / len(out_ev) if out_ev else 1.0; rec = tp / len(ref_ev_r) if ref_ev_r else 1.0
f1 = 2 * prec * rec / (prec + rec) if prec + rec else 0.0
# fabricated = emitted events not traceable to a row of a source events table (independent re-read)
fabricated = 0
for e in events:
m = re.match(r"(.+)#row(\d+)$", e.get("provenance", ""))
ok = False
if m and (SRC / m.group(1)).exists():
lines = (SRC / m.group(1)).read_text().strip().splitlines()
i = int(m.group(2))
if i < len(lines):
cells = lines[i].split("\t")
nums = [float(c) for c in cells if re.fullmatch(r"-?\d+(\.\d+)?", c)]
ok = str(e["label"]) in cells and any(abs(x - e["onset"]) < 1e-9 for x in nums)
fabricated += not ok
surfaced = {(c, f) for f, r in rows.items() for c in r["conflicts"]}
planted_keys = {(c["type"], c["key"]) for c in REF["conflicts"]}
spurious = sorted(f"{t}: {f}" for t, f in surfaced if (t, f) not in planted_keys and not (
t == "duplicate_subject" and any(c["type"] == t and f.startswith(c["key"] + "/") for c in REF["conflicts"])))
planted_hits = []
for c in REF["conflicts"]:
if c["type"] == "duplicate_subject":
hit = any(t == "duplicate_subject" and f.startswith(c["key"] + "/") for t, f in surfaced)
else:
hit = (c["type"], c["key"]) in surfaced
planted_hits.append({**c, "surfaced": hit})
exp_windows = sum(int(round(v["duration_s"] * v["fs"])) // int(round(WIN * v["fs"]))
for f, v in REF["recordings"].items() if f in chosen_paths and v["eligible"])
exp_inelig = sum(1 for f, v in REF["recordings"].items() if f in chosen_paths and not v["eligible"])
ann_conf = any(c in ("events_out_of_bounds", "fs_mismatch", "missing_events", "duplicate_subject", "unit_implausible")
for r in rows.values() for c in r["conflicts"])
complete = retained == len(REF["recordings"]) and matched == compared and prov == retained and extra == 0 \
and fabricated == 0 and abs(f1 - 1) < 1e-12 and tens["windows"] == exp_windows
status = ("full_pass" if not ann_conf else "partial_pass") if complete else "fail"
return {"recordings_retained": retained, "recordings_input": len(REF["recordings"]), "extra_rows": extra,
"decoys_indexed": decoys_indexed, "metadata_matched": matched, "metadata_compared": compared,
"metadata_mismatch_examples": mism, "provenance": prov, "events_matched": tp, "events_reference": len(ref_ev_r),
"events_emitted": len(out_ev), "event_f1": f1, "fabricated_events": fabricated,
"conflicts_planted": len(REF["conflicts"]), "conflicts_surfaced": sum(h["surfaced"] for h in planted_hits),
"conflict_detail": planted_hits, "conflicts_spurious": len(spurious), "conflicts_spurious_detail": spurious[:10], "windows_produced": tens["windows"], "windows_expected": exp_windows,
"ineligible_duration": tens["ineligible_duration"], "ineligible_expected": exp_inelig,
"boundary_failures": tens["boundary_fail"], "status": status,
"events_hash": hashlib.sha256(json.dumps(sorted(out_ev)).encode()).hexdigest()[:12]}
def run_pipeline(nonce="", temperature=0.0):
"""One full adaptation run: Inspector (+bounded repair) -> index -> conflicts -> Planner -> gate -> extract -> tensorize."""
t0 = time.time(); calls = 0; trace = []
try:
raw = inspector(nonce, temperature); calls += 1
except Exception as e: # provider failure counts in the denominator, like the paper's 503 run
return {"completed": False, "error": f"provider error: {str(e)[:120]}", "llm_calls": calls, "elapsed_s": time.time() - t0}
rules, rows, errs, repairs, rules0 = None, {}, ["no JSON in reply"], 0, None
for attempt in range(MAXREP + 1):
try:
rules = parse_json(raw); rules0 = rules if attempt == 0 else rules0; rows, errs = build_index(rules)
except (ValueError, AttributeError, TypeError) as e:
rules, rows, errs = None, {}, [f"reply was not valid rules JSON ({str(e)[:80]})"]
trace.append({"attempt": attempt, "errors": errs[:8], "rows": len(rows)})
if not errs or attempt == MAXREP:
break
try:
raw = repair(raw, errs[:10], temperature, nonce); calls += 1; repairs += 1
except Exception as e:
return {"completed": False, "error": f"provider error: {str(e)[:120]}", "llm_calls": calls, "elapsed_s": time.time() - t0}
if not rows:
return {"completed": False, "error": "no usable rules after bounded repair", "llm_calls": calls, "repairs": repairs,
"trace": trace, "elapsed_s": time.time() - t0}
rows = detect_conflicts(rows)
aud = audit(rows)
try:
praw = planner(aud, temperature, nonce); calls += 1; plan = parse_json(praw)
except Exception as e:
plan = {"selection": {"sources": "all", "tasks": "all"}, "events": {}, "error": str(e)[:120]}
chosen, events, blocked = gate_and_extract(rows, plan)
dropped = sorted(set(rows) - {r["path"] for r in chosen}) # consequential choice -> researcher approval, never silent
tens = tensorize(chosen, events)
v = validate(rows, events, chosen, tens)
v0 = None
if rules0 is not None and repairs: # ablation: what if the Inspector's first proposal had been trusted unchecked?
try:
r0 = detect_conflicts(build_index(rules0)[0]); c0, e0, _ = gate_and_extract(r0, plan)
v0 = validate(r0, e0, c0, tensorize(c0, e0))
except Exception:
v0 = None
return {"completed": True, "rules": rules, "val_first": v0, "rows": rows, "plan": plan, "blocked": blocked, "events": events,
"dropped_by_selection": len(dropped), "dropped_examples": dropped[:5],
"chosen": len(chosen), "tens": tens, "val": v, "llm_calls": calls, "repairs": repairs, "trace": trace,
"residual_errors": errs, "audit": aud, "elapsed_s": time.time() - t0}
# ====================================================================== 5. runs
log(f"LLM backend {backend()} / {model_name('small')}; main run (temperature 0)")
main = run_pipeline()
if main["completed"]:
v = main["val"]
log(f"main run: retained {v['recordings_retained']}/{v['recordings_input']}, metadata {v['metadata_matched']}/"
f"{v['metadata_compared']}, events {v['events_matched']}/{v['events_reference']} (F1 {v['event_f1']:.4f}), "
f"fabricated {v['fabricated_events']}, conflicts {v['conflicts_surfaced']}/{v['conflicts_planted']}, windows "
f"{v['windows_produced']}/{v['windows_expected']}, repairs {main['repairs']}, status {v['status']}")
else:
log(f"main run did not complete: {main['error']}")
reps = []
if NREP:
log(f"{NREP} repeated runs (temperature 1.0, provider-default-like sampling)")
with cf.ThreadPoolExecutor(min(NREP, 4)) as ex:
reps = list(ex.map(lambda k: run_pipeline(nonce=f"repeat-{k + 1}", temperature=1.0), range(NREP)))
save_cache()
# fabrication trap: the same intent, an LLM with no deterministic controls, recordings without any annotation stream
clin_rows = [f for f, r in REF["recordings"].items() if r["source"] == "clinical" and r["eligible"]][:6]
trap_prompt = (f"RESEARCHER INTENT:\n{INTENT}\n\n{TARGET_REQ}\n\nYou are a data-curation assistant. Produce the event annotation table "
"for the following clinical EEG recordings, following the intent. Signal headers:\n" +
"\n".join(f"- {f}: fs={header(f)['fs']}, duration={header(f)['n_samples'] / header(f)['fs']:.1f}s, "
f"channels={header(f)['ch_names']}" for f in clin_rows) +
"\n\nReturn ONLY a JSON list of events [{\"recording\": ..., \"onset_s\": ..., \"duration_s\": ..., \"label\": ...}]"
" (an empty list if appropriate).")
trap = {"recordings": len(clin_rows), "events_emitted": None, "reply_excerpt": ""}
try:
traw = chat(trap_prompt, "You are a helpful assistant.", size="small", max_tokens=1500)
trap["reply_excerpt"] = traw[:300]
try:
tl = parse_json(traw); tl = tl if isinstance(tl, list) else tl.get("events", [])
trap["events_emitted"] = len(tl)
trap["labels"] = sorted({str(e.get("label")) for e in tl if isinstance(e, dict)})[:8]
except (ValueError, AttributeError):
trap["events_emitted"] = 0
except Exception as e:
trap["error"] = str(e)[:120]
save_cache()
pipe_clin_events = sum(1 for e in (main.get("events") or []) if REF["recordings"].get(e["recording"], {}).get("source") == "clinical")
log(f"fabrication trap: LLM-only emitted {trap['events_emitted']} events for {len(clin_rows)} unannotated clinical recordings; "
f"pipeline emitted {pipe_clin_events}")
# ====================================================================== 6. figures
plt.rcParams.update({"font.size": 9.5, "axes.spines.top": False, "axes.spines.right": False})
INK, ACC, BLUE, MUTED, GOOD = "#1f2a44", "#d9480f", "#1971c2", "#868e96", "#2b8a3e"
v = main["val"] if main["completed"] else None
# fig 1: scorecard bars (fraction matched per measure)
if v:
meas = [("Recordings retained", v["recordings_retained"], v["recordings_input"]),
("Metadata values matched", v["metadata_matched"], v["metadata_compared"]),
("Recordings with provenance", v["provenance"], v["recordings_input"]),
("Native events matched", v["events_matched"], v["events_reference"]),
("Planted conflicts surfaced", v["conflicts_surfaced"], v["conflicts_planted"]),
("5-s windows produced" if WIN == 5 else f"{WIN:g}-s windows produced", v["windows_produced"], v["windows_expected"])]
v0 = main.get("val_first")
m0 = None if not v0 else [v0["recordings_retained"], v0["metadata_matched"], v0["provenance"], v0["events_matched"],
v0["conflicts_surfaced"], v0["windows_produced"]]
fig, ax = plt.subplots(figsize=(6.4, 3.2))
for i, (lab, a, b) in enumerate(meas):
fr = a / b if b else 1.0
if m0:
ax.barh(i - 0.2, min(m0[i] / b if b else 1.0, 1.0), 0.36, color=MUTED, label="first proposal, unchecked" if i == 0 else None)
ax.barh(i + 0.2, fr, 0.36, color=GOOD if fr >= 0.9999 else ACC, label="after deterministic checks + repair" if i == 0 else None)
ax.text(1.01, i + 0.2, f"{a:,} / {b:,}", va="center", fontsize=8.5)
ax.text(1.01, i - 0.2, f"{m0[i]:,}", va="center", fontsize=8, color=MUTED)
else:
ax.barh(i, 1, color="#e9ecef"); ax.barh(i, fr, color=GOOD if fr >= 0.9999 else ACC)
ax.text(1.01, i, f"{a:,} / {b:,}", va="center", fontsize=9)
ax.set_yticks(range(len(meas))); ax.set_yticklabels([m[0] for m in meas]); ax.invert_yaxis()
ax.set_xlim(0, 1.3); ax.set_xticks([0, 0.5, 1]); ax.set_xlabel("fraction of the hidden reference spec")
if m0:
ax.legend(frameon=False, fontsize=7.5, loc="lower center", bbox_to_anchor=(0.45, 1.0), ncol=2)
else:
ax.set_title(f"Status: {v['status']} · fabricated events: {v['fabricated_events']}", fontsize=9.5, loc="left")
fig.tight_layout(); fig.savefig(OUT / "fig_scorecard.png", dpi=160); plt.close(fig)
# fig 2: conflicts by type
types = ["fs_mismatch", "missing_events", "duplicate_subject", "unit_implausible", "events_out_of_bounds"]
planted = [sum(1 for c in v["conflict_detail"] if c["type"] == t) for t in types]
surf = [sum(1 for c in v["conflict_detail"] if c["type"] == t and c["surfaced"]) for t in types]
fig, ax = plt.subplots(figsize=(6.4, 2.7))
x = np.arange(len(types))
ax.bar(x - 0.2, planted, 0.38, color=MUTED, label="planted in the source tree")
ax.bar(x + 0.2, surf, 0.38, color=BLUE, label="surfaced for review (not silently resolved)")
ax.set_xticks(x); ax.set_xticklabels(["sampling-rate\nmismatch", "missing\nevents file", "duplicate\nsubject row",
"implausible\nunits", "events past\nrecording end"], fontsize=8)
ax.set_ylabel("count"); ax.legend(frameon=False, fontsize=8)
ax.yaxis.get_major_locator().set_params(integer=True)
fig.tight_layout(); fig.savefig(OUT / "fig_conflicts.png", dpi=160); plt.close(fig)
# fig 3: reproducibility + fabrication trap
fig, (a1, a2) = plt.subplots(1, 2, figsize=(6.4, 2.8), gridspec_kw={"width_ratios": [1.6, 1]})
allruns = [("main", main)] + [(f"r{k + 1}", r) for k, r in enumerate(reps)]
ref_hash = v["events_hash"] if v else None
for i, (lab, r) in enumerate(allruns):
if not r["completed"]:
a1.bar(i, 0.3, color=MUTED); a1.text(i, 0.35, "failed", ha="center", fontsize=7); continue
ok = r["val"]["status"] in ("full_pass", "partial_pass")
same = r["val"]["events_hash"] == ref_hash
a1.bar(i, 1, color=GOOD if ok and same else ACC)
a1.text(i, 1.03, f"{r['repairs']} rep.", ha="center", fontsize=7)
a1.set_xticks(range(len(allruns))); a1.set_xticklabels([x[0] for x in allruns], fontsize=8)
a1.set_yticks([]); a1.set_ylim(0, 1.25); a1.set_title("Runs: green = passed + identical Events", fontsize=8.5, loc="left")
te = trap["events_emitted"] if trap["events_emitted"] is not None else 0
a2.bar([0, 1], [te, pipe_clin_events], color=[ACC, GOOD])
for i, val in enumerate([te, pipe_clin_events]):
a2.text(i, val + max(te, 1) * 0.03, str(val), ha="center", fontsize=9)
a2.set_xticks([0, 1]); a2.set_xticklabels(["LLM only", "agent +\ndeterministic gate"], fontsize=8)
a2.set_title("Events invented for\nunannotated recordings", fontsize=8.5, loc="left"); a2.set_ylim(0, max(te, 1) * 1.25)
a2.yaxis.get_major_locator().set_params(integer=True)
if te == 0 and pipe_clin_events == 0:
a2.text(0.5, 0.55, "neither invented any\n(the LLM declined)", ha="center", fontsize=8, color=MUTED, transform=a2.transAxes)
fig.tight_layout(); fig.savefig(OUT / "fig_runs_trap.png", dpi=160); plt.close(fig)
# ====================================================================== 7. results + card
(OUT / "source_tree.txt").write_text("\n".join(FILES))
comp_runs = [r for r in reps if r["completed"]]
rep_pass = sum(1 for r in comp_runs if r["val"]["status"] in ("full_pass", "partial_pass") and r["val"]["events_hash"] == ref_hash)
total_calls = main.get("llm_calls", 0) + sum(r.get("llm_calls", 0) for r in reps) + 1
slim = lambda r: {k: r.get(k) for k in ("completed", "error", "llm_calls", "repairs", "trace", "residual_errors", "blocked", "plan", "chosen",
"dropped_by_selection", "dropped_examples", "elapsed_s")} | (
{"validation": r["val"]} if r.get("completed") else {})
fs_mis_rows = [r for r in (main.get("rows") or {}).values() if "fs_mismatch" in r["conflicts"] and r["eligibility"] == "eligible"]
wrong_if_sidecar = sum(abs(r["n_samples"] // int(round(WIN * r["fs_sidecar"])) - r["n_samples"] // int(round(WIN * r["fs"])))
for r in fs_mis_rows) # windows that would be wrong had the sidecar rate been trusted
first_rows = main["trace"][0]["rows"] if main.get("trace") else 0
res = {"params": P, "intent": INTENT, "caveats_from_inputs": caveats, "backend": backend(), "model": model_name("small"),
"annotations": ANN, "window_sec": WIN, "target_pressure": PRESSURE, "messiness": MESS,
"source": {"files": len(FILES), "recordings": n_ref_rec, "research_recordings": len(bids_recs), "clinical_recordings": len(clin),
"native_events": n_ref_ev, "planted_conflicts": len(REF["conflicts"]), "decoys": len(REF["decoys"])},
"main": slim(main), "rules": main.get("rules"), "index_audit": main.get("audit"),
"repeats": [slim(r) for r in reps], "repeat_completed": len(comp_runs), "repeat_passed_same_events": rep_pass,
"fabrication_trap": trap, "pipeline_clinical_events": pipe_clin_events,
"windows_wrong_if_sidecar_fs_trusted": int(wrong_if_sidecar),
"first_proposal_rows": first_rows, "first_proposal_validation": main.get("val_first"), "llm_calls_total": total_calls,
"human_checkpoints_planned": 3, "human_checkpoints_note": "label review, intent approval, plan approval: auto-approved in this run"}
if v:
tiles = [
{"label": "Recordings retained", "value": f"{v['recordings_retained']}/{v['recordings_input']}",
"sub": f"Inspector's first proposal indexed {first_rows}; after {main['repairs']} deterministic repair cycle(s) "
f"{len(main['rows'])} · {v['decoys_indexed']} of {len(REF['decoys'])} decoys indexed",
"tone": "good" if v["recordings_retained"] == v["recordings_input"] and v["extra_rows"] == 0 else "warn"},
{"label": "Native events matched", "value": f"{v['events_matched']}/{v['events_reference']}",
"sub": f"F1 = {100 * v['event_f1']:.3f}% · fabricated events: {v['fabricated_events']}", "tone": "good" if v["fabricated_events"] == 0 else "warn"},
{"label": "Conflicts surfaced", "value": f"{v['conflicts_surfaced']}/{v['conflicts_planted']}",
"sub": f"status **{v['status']}** · {main['repairs']} repair cycle(s) · {main['dropped_by_selection']} recordings "
f"dropped by the Planner (sent for approval) · {len(main['blocked'])} event proposals blocked", "tone": "accent"},
{"label": "Fabrication trap", "value": f"{te} vs {pipe_clin_events}",
"sub": f"events invented for {len(clin_rows)} unannotated recordings: LLM-only vs gated pipeline", "tone": "warn" if te else "good"},
]
else:
tiles = [{"label": "Main run", "value": "did not complete", "sub": main.get("error", ""), "tone": "warn"}]
pv = lambda a, b: f"{a:,}/{b:,}"
compare = [
{"metric": "Recordings retained / input", "paper": "1,227 / 1,227 (HBN); 1,387 / 1,387 (TUEG)",
"yours": pv(v["recordings_retained"], v["recordings_input"]) if v else "—", "note": "synthetic tree"},
{"metric": "Metadata values matched / compared", "paper": "10,262 / 10,262; 15,257 / 15,257",
"yours": pv(v["metadata_matched"], v["metadata_compared"]) if v else "—", "note": ""},
{"metric": "Native events matched / reference", "paper": "65,668 / 65,670; F1 = 99.997%",
"yours": f"{pv(v['events_matched'], v['events_reference'])}; F1 = {100 * v['event_f1']:.3f}%" if v else "—", "note": ""},
{"metric": "Fabricated events (no annotation stream)", "paper": "not present; 0 fabricated",
"yours": f"{v['fabricated_events'] if v else '—'} (LLM-only baseline: {te})", "note": "trap is this runbook's extension"},
{"metric": "Evaluation status", "paper": "partial_pass (HBN); full_pass (TUEG)", "yours": v["status"] if v else "—",
"note": "partial = conflicts surfaced for review"},
{"metric": "Fixed-window tensorization", "paper": "The 18 duration-eligible recordings produced all 1,099 expected windows; Two 1-s TUEG recordings were classified as ineligible_duration",
"yours": f"{pv(v['windows_produced'], v['windows_expected'])} {WIN:g}-s windows; {v['ineligible_duration']} ineligible_duration" if v else "—", "note": ""},
{"metric": "Repeated runs", "paper": "Ten prespecified runs ... Nine runs completed, and all nine passed the reference checks; one failed with a provider-side 503",
"yours": f"{len(comp_runs)} of {NREP} completed; {rep_pass} passed with identical Events" if NREP else "not run", "note": "abstract says 'Nine of ten evaluable'"},
{"metric": "LLM attempts per run", "paper": "8 (HBN); 11 (10 usable) (TUEG)", "yours": f"{main.get('llm_calls', 0)} (main run)",
"note": f"{total_calls} calls in total incl. repeats and trap"},
]
Path(OUT / "card_data.json").write_text(json.dumps({
"paper": {"num": 17, "title": "NeuroBuilder: Auditable Agentic Data Curation for Neuro-Foundation Models",
"authors": "Ghazaleh Ranjbaran, Mahta Ramezanian-Panahi, Blake Aaron Richards, Irina Rish", "url": "https://drive.google.com/file/d/1JcxkbnVGhv_uG7abJyyeSVieJcvPzgCs/view?usp=sharing",
"affiliation": "Mila – Quebec AI Institute · IVADO · Université de Montréal · McGill University"},
"inputs": {"messiness": MESS, "research subjects": NB, "clinical patients": NC, "annotations": ANN,
"window": f"{WIN:g} s", "target pressure": PRESSURE, "repeat runs": NREP, "max repairs": MAXREP, "agent model": model_name("small"),
"intent": (INTENT[:90] + "…") if len(INTENT) > 90 else INTENT},
"tiles": tiles,
"figures": ([{"path": "fig_scorecard.png", "title": "Did the curated object match the source? (Table-1-style scorecard)",
"caption": "Each bar is checked against a hidden reference spec written by the data generator. Green = complete."},
{"path": "fig_conflicts.png", "title": "Were the planted problems surfaced?",
"caption": "Conflicts are flagged on the Global Index rows and kept for researcher review, not silently reconciled."}] if v else []) +
[{"path": "fig_runs_trap.png", "title": "Is it reproducible, and does it invent events?",
"caption": "Left: repeated full runs at sampling temperature 1.0 (label = bounded repair cycles). Right: the same intent "
"given to an LLM with no deterministic gate, for recordings that have no annotation stream."}],
"compare": compare, "runbook": {"name": "smash-17-neurobuilder", "url": P.get("runbook_url", "")}}, indent=2))
res["runtime_sec"] = round(time.time() - T0, 1)
(OUT / "results.json").write_text(json.dumps(res, indent=2, default=str))
if not P.get("keep_source_tree"):
shutil.rmtree(SRC, ignore_errors=True)
log("done: " + (f"retained {v['recordings_retained']}/{v['recordings_input']}; metadata {v['metadata_matched']}/{v['metadata_compared']}; "
f"events F1 {v['event_f1']:.4f}; fabricated {v['fabricated_events']}; conflicts {v['conflicts_surfaced']}/"
f"{v['conflicts_planted']}; windows {v['windows_produced']}/{v['windows_expected']}; status {v['status']}; "
if v else "main run failed; ") + f"repeats {rep_pass}/{NREP}; trap LLM-only {te} vs pipeline {pipe_clin_events}; "
f"{total_calls} LLM calls")/app/work/render_card.py
"""Render a SMASH 2026 paper-playground result card (one self-contained HTML file).
usage: python render_card.py card_data.json narrative.json out.html
card_data.json (written by the experiment script — numbers only)
paper: {num, title, authors, affiliation, url?}
inputs: {label: value} the participant's run settings
tiles: [{label, value, sub, tone}] tone: accent|good|warn|neutral
figures: [{path, title, caption}] PNGs, embedded as base64
compare: [{metric, paper, yours, note}] paper-vs-your-run table (optional)
runbook: {name, url} fork link (optional)
narrative.json (written by the agent — words only)
headline: one sentence, the finding of THIS run
idea: [1-3 bullets] the paper's contribution in plain language
abstract: (optional) a verbatim abstract excerpt, shown instead of idea
findings: [2-4 short paragraphs] what this run shows, tied to the numbers
try_next: [{label, why, params}] 2-3 follow-up runs with exact parameter values
caveats: [bullets]
"""
import base64, html, json, sys
from pathlib import Path
data = json.loads(Path(sys.argv[1]).read_text())
narr = json.loads(Path(sys.argv[2]).read_text())
base = Path(sys.argv[1]).parent
e = lambda s: html.escape(str(s))
def md_inline(s):
"""Escape, then allow **bold**, *italic* and `code` only."""
import re
s = e(s)
s = re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", s)
s = re.sub(r"(?<![\w*])\*(?!\s)(.+?)(?<!\s)\*(?![\w*])", r"<em>\1</em>", s)
return re.sub(r"`(.+?)`", r"<code>\1</code>", s)
p = data["paper"]
tiles = "".join(
f'<div class="tile {e(t.get("tone", "neutral"))}"><div class="tl">{e(t["label"])}</div>'
f'<div class="tv">{e(t["value"])}</div><div class="ts">{md_inline(t.get("sub", ""))}</div></div>'
for t in data.get("tiles", []))
# Columns that fill every row: 4 tiles as 2x2, never 3 + 1 orphan.
n_tiles = len(data.get("tiles", []))
tile_cols = 2 if n_tiles == 4 else min(max(n_tiles, 1), 3)
inputs = "".join(f'<span class="chip"><b>{e(k)}</b> {e(v)}</span>' for k, v in data.get("inputs", {}).items())
figs = ""
for f in data.get("figures", []):
fp = Path(f["path"]); fp = fp if fp.is_absolute() else base / fp
if fp.exists():
b64 = base64.b64encode(fp.read_bytes()).decode()
figs += (f'<figure><figcaption class="ft">{e(f.get("title", ""))}</figcaption>'
f'<img alt="{e(f.get("title", ""))}" src="data:image/png;base64,{b64}">'
f'<figcaption>{md_inline(f.get("caption", ""))}</figcaption></figure>')
compare = ""
if data.get("compare"):
rows = "".join(f'<tr><td>{md_inline(r["metric"])}</td><td>{md_inline(r["paper"])}</td>'
f'<td class="y">{md_inline(r["yours"])}</td><td class="n">{md_inline(r.get("note", ""))}</td></tr>'
for r in data["compare"])
compare = (f'<section><h2>Paper vs. your run</h2><div class="tw"><table><thead><tr><th>Metric</th>'
f'<th>Paper</th><th>Your run</th><th></th></tr></thead><tbody>{rows}</tbody></table></div></section>')
idea = "".join(f"<li>{md_inline(b)}</li>" for b in narr.get("idea", []))
abstract = narr.get("abstract", "")
findings = "".join(f"<p>{md_inline(x)}</p>" for x in narr.get("findings", []))
tries = "".join(
f'<li><strong>{md_inline(t["label"])}</strong> — {md_inline(t.get("why", ""))}'
f'<pre>{e(json.dumps(t.get("params", {}), ensure_ascii=False))}</pre></li>' for t in narr.get("try_next", []))
cav = "".join(f"<li>{md_inline(c)}</li>" for c in narr.get("caveats", []))
rb = data.get("runbook", {})
fork = (f'<a class="fork" href="{e(rb["url"])}">Fork this runbook on Jetty →</a>' if rb.get("url") else "")
page = f"""<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>SMASH 2026 · #{e(p['num'])} · Result card</title>
<style>
:root{{--bg:#f6f4ef;--card:#fff;--ink:#1f2a44;--muted:#5c677d;--line:#e3dfd5;--accent:#d9480f;--good:#2b8a3e;--warn:#e67700;--chip:#efece4}}
@media (prefers-color-scheme:dark){{:root{{--bg:#11151f;--card:#1a2030;--ink:#e8ebf2;--muted:#9aa4b8;--line:#2c3447;--accent:#ff8a4c;--good:#69db7c;--warn:#ffc078;--chip:#232b3d}}
figure img{{background:#fff;border-radius:6px}}}}
*{{box-sizing:border-box}}body{{margin:0;background:var(--bg);color:var(--ink);font:15px/1.55 ui-sans-serif,system-ui,-apple-system,"Segoe UI",Roboto,sans-serif}}
main{{max-width:860px;margin:0 auto;padding:28px 16px 48px}}
.kicker{{font:600 12px/1 ui-monospace,SFMono-Regular,Menlo,monospace;letter-spacing:.08em;text-transform:uppercase;color:var(--accent)}}
h1{{font-size:clamp(22px,4vw,30px);line-height:1.2;margin:10px 0 6px}}.authors{{color:var(--muted);margin:0 0 18px}}.authors a{{color:var(--accent);text-underline-offset:3px}}
.headline{{background:var(--card);border:1px solid var(--line);border-left:5px solid var(--accent);border-radius:10px;padding:16px 18px;font-size:18px;font-weight:600;margin:0 0 18px}}
.chips{{display:flex;flex-wrap:wrap;gap:6px;margin:0 0 18px}}.chip{{background:var(--chip);border-radius:999px;padding:4px 10px;font-size:12.5px}}.chip b{{font-weight:600;color:var(--muted);margin-right:4px}}
.tiles{{display:grid;grid-template-columns:repeat(var(--cols),minmax(0,1fr));gap:10px;margin:0 0 22px}}
@media (max-width:560px){{.tiles{{grid-template-columns:1fr}}}}
.tile{{background:var(--card);border:1px solid var(--line);border-radius:10px;padding:12px 14px}}.tl{{font-size:12px;color:var(--muted);text-transform:uppercase;letter-spacing:.04em}}
.tv{{font-size:26px;font-weight:700;margin:2px 0;font-variant-numeric:tabular-nums}}.ts{{font-size:12.5px;color:var(--muted)}}
.tile.accent .tv{{color:var(--accent)}}.tile.good .tv{{color:var(--good)}}.tile.warn .tv{{color:var(--warn)}}
section{{background:var(--card);border:1px solid var(--line);border-radius:12px;padding:18px 20px;margin:0 0 16px}}
h2{{font-size:16px;margin:0 0 10px}}ul{{padding-left:20px;margin:0}}li{{margin:4px 0}}p{{margin:0 0 10px}}
figure{{margin:0 0 18px}}figure img{{width:100%;height:auto;display:block}}figcaption{{font-size:13px;color:var(--muted);margin-top:6px}}.ft{{font-weight:600;color:var(--ink);font-size:14px;margin:0 0 6px}}
.tw{{overflow-x:auto}}table{{border-collapse:collapse;width:100%;font-size:13.5px;font-variant-numeric:tabular-nums}}th,td{{text-align:left;padding:7px 8px;border-bottom:1px solid var(--line);vertical-align:top}}th{{color:var(--muted);font-weight:600}}td.y{{font-weight:600}}td.n{{color:var(--muted);font-size:12.5px}}
pre{{background:var(--chip);border-radius:6px;padding:6px 8px;font-size:12px;white-space:pre-wrap;word-break:break-word;margin:4px 0 8px}}code{{background:var(--chip);border-radius:4px;padding:1px 4px;font-size:.92em}}
.fork{{display:inline-block;background:var(--ink);color:var(--bg);text-decoration:none;font-weight:600;border-radius:8px;padding:10px 14px;margin-top:6px}}
footer{{color:var(--muted);font-size:12.5px;margin-top:22px}}
.abs{{margin:0;padding-left:14px;border-left:3px solid var(--accent);color:var(--ink)}}
</style></head><body><main>
<div class="kicker">SMASH 2026 · Paper #{e(p['num'])} · Result card</div>
<h1>{e(p['title'])}</h1><p class="authors">{e(p['authors'])}{(' · ' + e(p['affiliation'])) if p.get('affiliation') else ''}{(f' · <a href="{e(p["url"])}">Read the paper</a>') if p.get('url') else ''}</p>
<div class="headline">{md_inline(narr['headline'])}</div>
<div class="chips">{inputs}</div>
<div class="tiles" style="--cols:{tile_cols}">{tiles}</div>
{f'<section><h2>From the abstract</h2><blockquote class="abs">{e(abstract)}</blockquote></section>' if abstract else f'<section><h2>The idea</h2><ul>{idea}</ul></section>'}
{f'<section><h2>What this run shows</h2>{findings}</section>' if findings else ''}
<section><h2>Evidence</h2>{figs}</section>
{compare}
<section><h2>Try next</h2><ul>{tries}</ul>{fork}</section>
<section><h2>Read this before sharing</h2><ul>{cav}</ul></section>
<footer>Generated by a Jetty runbook built from the authors' SMASH 2026 paper. Numbers on this card come from code run in a sandbox; the words were written by an AI agent reading those numbers. This is an interactive companion, not the authors' own results unless labelled “Paper”.</footer>
</main></body></html>"""
Path(sys.argv[3]).write_text(page)
print(f"wrote {sys.argv[3]} ({len(page) // 1024} KB)")/app/work/validate.py
"""Self-check a SMASH paper-playground run and write validation_report.json (evals v2 shape).
Runs the same checks as the runbook's `## Code Checks` (which Jetty re-runs independently after the agent
exits and substitutes for these entries), adds one `step` entry per runbook step and the agent's checklist
judgements, and prints PASS/FAIL lines.
usage: python validate.py <results_dir> <param,param,...> <budget_sec> [checklist.json]
"""
import html, json, re, sys
from pathlib import Path
R = Path(sys.argv[1])
allowed = set(sys.argv[2].split(","))
budget = float(sys.argv[3])
checklist_path = Path(sys.argv[4]) if len(sys.argv) > 4 else None
def load(name):
try:
return json.loads((R / name).read_text())
except Exception:
return None
res, cdata, narr = load("results.json"), load("card_data.json"), load("narrative.json")
card = (R / "result_card.html").read_text() if (R / "result_card.html").exists() else ""
def c_outputs_exist():
missing = [f for f in ["result_card.html", "results.json", "card_data.json", "narrative.json", "summary.md"]
if not (R / f).exists() or (R / f).stat().st_size == 0]
return not missing, f"missing or empty: {missing}" if missing else "all 5 outputs present"
def c_results_valid():
if res is None or cdata is None:
return False, "results.json or card_data.json does not parse"
rt = float(res.get("runtime_sec", 0))
return rt <= budget, f"experiment {rt}s (budget {budget:.0f}s)"
def c_narrative_complete():
if narr is None:
return False, "narrative.json does not parse"
p = []
if not str(narr.get("headline", "")).strip(): p.append("no headline")
if len(str(narr.get("headline", "")).split()) > 45: p.append("headline over 45 words")
if not 1 <= len(narr.get("idea", [])) <= 3: p.append(f"idea has {len(narr.get('idea', []))} bullets, need 1-3")
if not 1 <= len(narr.get("findings", [])) <= 2: p.append(f"{len(narr.get('findings', []))} findings, need 1-2")
if any(len(f.split()) > 70 for f in narr.get("findings", [])): p.append("a findings paragraph is over 60 words")
if not 2 <= len(narr.get("try_next", [])) <= 3: p.append(f"{len(narr.get('try_next', []))} try_next, need 2-3")
if not 1 <= len(narr.get("caveats", [])) <= 2: p.append(f"{len(narr.get('caveats', []))} caveats, need 1-2")
return not p, "; ".join(p) or "complete and within the length limits"
def c_try_next_valid():
if narr is None:
return False, "narrative.json does not parse"
bad = sorted({k for t in narr.get("try_next", []) for k in (t.get("params") or {}) if k not in allowed})
empty = [t.get("label", "?") for t in narr.get("try_next", []) if not t.get("params")]
ok = not bad and not empty
return ok, "all try_next parameters are declared" if ok else f"unknown parameters: {bad}; without params: {empty}"
def c_headline_grounded():
if narr is None or res is None:
return False, "missing narrative or results"
blob = json.dumps(res) + json.dumps(cdata or {})
vals = [float(x) for x in re.findall(r"-?\d+(?:\.\d+)?(?:[eE]-?\d+)?", blob)]
nums = re.findall(r"(?<![\w.−-])[−-]?\d[\d,]*(?:\.\d+)?(?![\w-])", narr.get("headline", ""))
missing = []
for n in nums:
s = n.replace("−", "-").replace(",", ""); v = float(s); dec = len(s.split(".")[1]) if "." in s else 0
tol = 0.5 * 10 ** (-dec) + 1e-9
if not any(abs(abs(x) - abs(v)) <= tol or abs(abs(x * 100) - abs(v)) <= max(tol, 0.5) for x in vals):
missing.append(n)
return not missing, f"not found in results: {missing}" if missing else f"{len(nums)} headline number(s) grounded"
def c_card_integrity():
if not card or narr is None:
return False, "no card or narrative"
text = re.sub(r"data:image/[a-z]+;base64,[A-Za-z0-9+/=]+", "", card)
plain = html.unescape(re.sub(r"<[^>]+>", "", text))
p = []
if card.count("data:image/png;base64") < 1: p.append("no embedded figures")
if re.search(r"\bNaN\b|\bnan%|\bInfinity\b", text): p.append("NaN/Infinity in card text")
if len(card) > 5_000_000: p.append(f"card is {len(card) // 1024} KB")
if re.sub(r"[*`]", "", narr.get("headline", ""))[:40] not in plain: p.append("headline not on card")
return not p, "; ".join(p) or f"card {len(card) // 1024} KB, {card.count('data:image/png;base64')} figures"
CODE_CHECKS = [ # ids and names match the runbook's ## Code Checks headings
("outputs-exist", "Result card, numbers, narrative and summary were written", c_outputs_exist),
("results-valid", "Experiment numbers are valid JSON and the experiment stayed within its time budget", c_results_valid),
("narrative-complete", "The narrative is complete and within the length limits", c_narrative_complete),
("try-next-valid", "Every try-next suggestion uses only this runbook's parameters", c_try_next_valid),
("headline-grounded", "Every number in the headline appears in the run's results", c_headline_grounded),
("card-integrity", "The card embeds its figures, shows the headline, and contains no NaN", c_card_integrity),
]
STEPS = [ # (id, name, evidence that the step happened)
("setup", "Set up the sandbox and parameters", Path("/app/work/params.json").exists() or res is not None),
("get-code", "Extract the tested code", Path("/app/work/experiment.py").exists() or res is not None),
("experiment", "Run the experiment", res is not None),
("interpret", "Write the narrative", narr is not None),
("render", "Render the result card and summary", bool(card) and (R / "summary.md").exists()),
]
checks = [{"kind": "step", "id": i, "name": n, "status": "pass" if ok else "fail"} for i, n, ok in STEPS]
for cid, name, fn in CODE_CHECKS:
try:
ok, msg = fn()
except Exception as e: # a check that crashes is an error, never a silent pass
ok, msg = None, f"{type(e).__name__}: {e}"
status = "pass" if ok else ("error" if ok is None else "fail")
checks.append({"kind": "code_check", "id": cid, "name": name, "status": status, "message": str(msg)[:1000]})
print(f"{'PASS' if ok else 'FAIL'} {cid}: {msg}")
if checklist_path and checklist_path.exists():
for item in json.loads(checklist_path.read_text()):
st = item.get("status", "skipped")
st = st if st in ("pass", "fail", "skipped", "error") else "skipped"
checks.append({"kind": "checklist", "id": str(item["id"]), "name": str(item.get("name") or item["id"]),
"status": st, "message": str(item.get("message", ""))[:1000]})
print(f"{'PASS' if st == 'pass' else 'FAIL'} checklist {item['id']}: {item.get('message', '')}")
else:
print("NOTE no checklist.json given; checklist items will be recorded as not reported")
gating = [c for c in checks if c["kind"] != "step"]
verdict = "fail" if any(c["status"] in ("fail", "error") for c in gating) else ("pass" if gating else "unknown")
(R / "validation_report.json").write_text(json.dumps({"version": 2, "verdict": verdict, "checks": checks}, indent=2))
print(json.dumps({"verdict": verdict, "passed": sum(c["status"] == "pass" for c in gating), "gating_checks": len(gating)}))/app/work/llm.py
"""Minimal, dependency-free LLM client for SMASH paper-playground runbooks.
Works with whatever keys the Jetty sandbox provides:
1. OPENROUTER_API_KEY -> OpenRouter
2. ANTHROPIC_API_KEY with ANTHROPIC_BASE_URL on openrouter (Jetty trial keys) -> OpenRouter
3. ANTHROPIC_API_KEY -> Anthropic Messages API
4. OPENAI_API_KEY -> OpenAI
Set SMASH_LLM_MOCK=1 to return deterministic placeholders (offline testing).
from llm import chat, chat_many, backend
text = chat("Say hi", system="Be brief", size="small", max_tokens=50)
"""
import concurrent.futures as cf, hashlib, json, os, time, urllib.request
from pathlib import Path
MODELS = { # size -> (openrouter slug, anthropic id, openai id)
"small": ("anthropic/claude-haiku-4.5", "claude-haiku-4-5", "gpt-4.1-mini"),
"large": ("anthropic/claude-sonnet-4.6", "claude-sonnet-4-6", "gpt-4.1"),
}
CACHE = Path(os.environ.get("SMASH_LLM_CACHE", "/app/work/llm_cache.json"))
_cache = json.loads(CACHE.read_text()) if CACHE.exists() else {}
def backend():
if os.environ.get("SMASH_LLM_MOCK"):
return "mock"
if os.environ.get("OPENROUTER_API_KEY"):
return "openrouter"
if os.environ.get("ANTHROPIC_API_KEY") and "openrouter" in os.environ.get("ANTHROPIC_BASE_URL", ""):
return "openrouter-via-anthropic-key"
if os.environ.get("ANTHROPIC_API_KEY"):
return "anthropic"
if os.environ.get("OPENAI_API_KEY"):
return "openai"
raise RuntimeError("no LLM key found (OPENROUTER_API_KEY / ANTHROPIC_API_KEY / OPENAI_API_KEY)")
def model_name(size="small"):
b = backend(); i = {"openrouter": 0, "openrouter-via-anthropic-key": 0, "anthropic": 1, "openai": 2}.get(b, 0)
return MODELS[size][i]
def _post(url, headers, body, timeout):
req = urllib.request.Request(url, data=json.dumps(body).encode(), headers={"Content-Type": "application/json", **headers})
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read())
def chat(prompt, system="", size="small", max_tokens=800, temperature=0.0, timeout=90, retries=3):
b = backend(); model = model_name(size)
key = hashlib.sha256(json.dumps([b, model, system, prompt, max_tokens, temperature]).encode()).hexdigest()
if key in _cache:
return _cache[key]
if b == "mock":
out = f"[mock {size} reply {key[:8]}]"
else:
for attempt in range(retries):
try:
if b.startswith("openrouter"):
k = os.environ.get("OPENROUTER_API_KEY") or os.environ["ANTHROPIC_API_KEY"]
msgs = ([{"role": "system", "content": system}] if system else []) + [{"role": "user", "content": prompt}]
r = _post("https://openrouter.ai/api/v1/chat/completions", {"Authorization": f"Bearer {k}"},
{"model": model, "messages": msgs, "max_tokens": max_tokens, "temperature": temperature}, timeout)
out = r["choices"][0]["message"]["content"]
elif b == "anthropic":
base = os.environ.get("ANTHROPIC_BASE_URL", "https://api.anthropic.com").rstrip("/")
body = {"model": model, "max_tokens": max_tokens, "temperature": temperature,
"messages": [{"role": "user", "content": prompt}]}
if system:
body["system"] = system
r = _post(f"{base}/v1/messages", {"x-api-key": os.environ["ANTHROPIC_API_KEY"],
"anthropic-version": "2023-06-01"}, body, timeout)
out = "".join(c.get("text", "") for c in r["content"])
else:
msgs = ([{"role": "system", "content": system}] if system else []) + [{"role": "user", "content": prompt}]
r = _post("https://api.openai.com/v1/chat/completions", {"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"},
{"model": model, "messages": msgs, "max_tokens": max_tokens, "temperature": temperature}, timeout)
out = r["choices"][0]["message"]["content"]
break
except Exception as e: # network/rate-limit: back off and retry
if attempt == retries - 1:
raise
time.sleep(2 * (attempt + 1))
_cache[key] = out
return out
def chat_many(prompts, system="", size="small", max_tokens=800, temperature=0.0, workers=8):
"""Run many prompts concurrently; returns replies in order. Saves the cache afterwards."""
with cf.ThreadPoolExecutor(workers) as ex:
outs = list(ex.map(lambda p: chat(p, system, size, max_tokens, temperature), prompts))
save_cache()
return outs
def save_cache():
try:
CACHE.parent.mkdir(parents=True, exist_ok=True); CACHE.write_text(json.dumps(_cache))
except OSError:
pass
def parse_json(text):
"""Pull the first JSON object/array out of a model reply."""
s = text.strip()
if s.startswith("```"):
s = s.split("\n", 1)[1].rsplit("```", 1)[0]
for o, c in (("{", "}"), ("[", "]")):
i, j = s.find(o), s.rfind(c)
if i != -1 and j > i:
try:
return json.loads(s[i:j + 1])
except json.JSONDecodeError:
continue
raise ValueError("no JSON in reply: " + text[:200])