{error && (
data load failed: {error} — expected public/data/cryobrain.json
)}
);
}
```
### tasks.py
```python
"""CryoBrain task bindings for `hud eval tasks.py`."""
from env import cryo_task, env, verilog_task # noqa: F401
from task_catalog import CRYO_CURRICULUM, TASK_SPECS
tasks = []
for _spec in TASK_SPECS:
if _spec.template == "cryo_task":
_curriculum = CRYO_CURRICULUM[_spec.slug]
_task = cryo_task(
slug=_spec.slug,
task_id=_spec.task_id,
distance=_curriculum["distance"],
noise_rate=_curriculum["noise_rate"],
max_latency_cycles=_curriculum["max_latency_cycles"],
max_area_mm2=_curriculum["max_area_mm2"],
max_power_mw=_curriculum["max_power_mw"],
)
else:
_task = verilog_task(slug=_spec.slug, task_id=_spec.task_id)
_task.slug = _spec.slug
_task.columns = {
"task_id": _spec.task_id,
"track": _spec.track,
"variant": _spec.variant,
"language": _spec.language,
"toolchain": _spec.toolchain,
"module": _spec.module,
}
tasks.append(_task)
```
### grader.py
```python
"""Hand-built EvaluationResult that preserves the hidden grader's hard cap."""
import importlib.util
from pathlib import Path
from typing import Any
from hud.graders import EvaluationResult, SubScore # v6 home (was hud.tools.types)
from scenario_helpers import WORKSPACE_ROOT, hidden_dir as _hidden_dir
from task_catalog import TASK_SPECS_BY_TASK_ID
def _load_grade_module(task_id: str, hidden_dir: Path):
grade_path = hidden_dir / "grade.py"
spec = importlib.util.spec_from_file_location(f"{task_id}_hidden_grade", grade_path)
if spec is None or spec.loader is None:
raise RuntimeError(f"cannot import hidden grader at {grade_path}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _subscores_from_result(result: dict[str, Any]) -> list[SubScore]:
subscores = []
for name, data in result.get("subscores", {}).items():
raw_value = data.get("raw_score")
if raw_value is None:
weight = float(data.get("weight", 1.0))
raw_value = float(data.get("weighted_score", 0.0)) / weight if weight else 0.0
subscores.append(
SubScore(
name=name,
weight=float(data.get("weight", 0.0)),
value=max(0.0, min(1.0, float(raw_value or 0.0))),
metadata=data.get("result"),
)
)
return subscores
def evaluate_task(task_id: str) -> EvaluationResult:
task_spec = TASK_SPECS_BY_TASK_ID[task_id]
workdir = WORKSPACE_ROOT
hidden_dir = _hidden_dir(task_id)
grade_mod = _load_grade_module(task_id, hidden_dir)
try:
result: dict[str, Any] = grade_mod.grade(
workdir,
rtl_override=None,
hidden_root=hidden_dir,
)
except Exception as exc: # noqa: BLE001 - fail closed; never error the episode
# A missing/renamed editable file or any grader crash must score 0, not raise out
# and error the rollout: an agent WILL delete the target file, and that should be a
# 0, not a failed run. (Reproduced: missing props -> FileNotFoundError -> episode error.)
return EvaluationResult(
reward=0.0,
done=True,
content=f"{task_id}: submission could not be graded ({type(exc).__name__}: {exc})",
info={
"task_id": task_id,
"module": task_spec.module,
"track": task_spec.track,
"hard_caps": ["grader_error"],
"error": f"{type(exc).__name__}: {exc}",
"workdir": str(workdir),
"hidden_dir": str(hidden_dir),
},
subscores=[],
)
# Build EvaluationResult by hand: do NOT route through hud.graders.combine(),
# which renormalizes positive weights and would erase the grader's hard cap
# (functional==0 -> reward forced to 0). The negative-weight hard_cap_penalty subscore
# reconciles the displayed subscores to the capped reward.
subscores = _subscores_from_result(result)
weighted_sum = sum(subscore.weight * subscore.value for subscore in subscores)
reward = float(result.get("reward", 0.0))
if reward + 1e-9 < weighted_sum:
subscores.append(
SubScore(
name="hard_cap_penalty",
weight=reward - weighted_sum,
value=1.0,
metadata={"hard_caps": result.get("hard_caps", [])},
)
)
return EvaluationResult(
reward=reward,
done=True,
content=f"{task_id} graded",
info={
"task_id": task_id,
"module": task_spec.module,
"track": task_spec.track,
"hard_caps": result.get("hard_caps", []),
"workdir": str(workdir),
"hidden_dir": str(hidden_dir),
},
subscores=subscores,
)
```
### task_catalog.py
```python
"""CryoBrain + fallback task catalog."""
from dataclasses import dataclass
from pathlib import Path
ROOT_DIR = Path(__file__).resolve().parent
TASK_ROOT = ROOT_DIR / "tasks"
@dataclass(frozen=True)
class TaskSpec:
task_id: str
slug: str
prompt: str
track: str
variant: str
language: str
module: str
toolchain: str
template: str = "cryo_task"
def _prompt(task_id: str) -> str:
return (TASK_ROOT / task_id / "prompt.md").read_text(encoding="utf-8")
CRYO_BRAIN_DECODER_D3 = TaskSpec(
task_id="cryo_brain_decoder",
slug="cryo-brain-decoder-d3",
prompt=_prompt("cryo_brain_decoder"),
track="qec-codesign",
variant="distance_3",
language="systemverilog+python",
module="cryo_brain_decoder",
toolchain="verilator+yosys+stim",
template="cryo_task",
)
CRYO_BRAIN_DECODER_D5 = TaskSpec(
task_id="cryo_brain_decoder",
slug="cryo-brain-decoder-d5",
prompt=_prompt("cryo_brain_decoder"),
track="qec-codesign",
variant="distance_5",
language="systemverilog+python",
module="cryo_brain_decoder",
toolchain="verilator+yosys+stim",
template="cryo_task",
)
CRYO_BRAIN_DECODER_D7 = TaskSpec(
task_id="cryo_brain_decoder",
slug="cryo-brain-decoder-d7",
prompt=_prompt("cryo_brain_decoder"),
track="qec-codesign",
variant="distance_7",
language="systemverilog+python",
module="cryo_brain_decoder",
toolchain="verilator+yosys+stim",
template="cryo_task",
)
STREAM_ARB_FIFO_REPAIR = TaskSpec(
task_id="stream_arb_fifo_repair",
slug="stream-arb-fifo-repair",
prompt=_prompt("stream_arb_fifo_repair"),
track="fallback-design",
variant="repair_debug",
language="systemverilog",
module="stream_arb_fifo",
toolchain="verilator+yosys",
template="verilog_task",
)
STREAM_ARB_FIFO_COCOTB_DV = TaskSpec(
task_id="stream_arb_fifo_cocotb_dv",
slug="stream-arb-fifo-cocotb-dv",
prompt=_prompt("stream_arb_fifo_cocotb_dv"),
track="fallback-verification",
variant="repair_debug",
language="python+cocotb",
module="stream_arb_fifo",
toolchain="cocotb+verilator",
template="verilog_task",
)
STREAM_ARB_FIFO_FORMAL = TaskSpec(
task_id="stream_arb_fifo_formal",
slug="stream-arb-fifo-formal",
prompt=_prompt("stream_arb_fifo_formal"),
track="fallback-formal",
variant="repair_debug",
language="systemverilog-formal",
module="stream_arb_fifo",
toolchain="symbiyosys+yosys",
template="verilog_task",
)
TASK_SPECS = [
CRYO_BRAIN_DECODER_D3,
CRYO_BRAIN_DECODER_D5,
CRYO_BRAIN_DECODER_D7,
STREAM_ARB_FIFO_REPAIR,
STREAM_ARB_FIFO_COCOTB_DV,
STREAM_ARB_FIFO_FORMAL,
]
# HUD eval binds by slug (6 curriculum/fallback variants). On-disk graders share task_id.
TASK_SPECS_BY_SLUG = {spec.slug: spec for spec in TASK_SPECS}
# One metadata row per filesystem task directory (4 on-disk tasks).
TASK_SPECS_BY_TASK_ID: dict[str, TaskSpec] = {}
for _spec in TASK_SPECS:
TASK_SPECS_BY_TASK_ID.setdefault(_spec.task_id, _spec)
# Back-compat alias used by grader/setup (filesystem task_id, not eval slug).
TASK_SPECS_BY_ID = TASK_SPECS_BY_TASK_ID
# RSI distance curriculum bindings (SPEC F7 / CP6)
CRYO_CURRICULUM = {
"cryo-brain-decoder-d3": {
"distance": 3,
"noise_rate": 0.001,
"max_latency_cycles": 64,
"max_area_mm2": 0.06,
"max_power_mw": 8.0,
},
"cryo-brain-decoder-d5": {
"distance": 5,
"noise_rate": 0.002,
"max_latency_cycles": 96,
"max_area_mm2": 0.08,
"max_power_mw": 10.0,
},
"cryo-brain-decoder-d7": {
"distance": 7,
"noise_rate": 0.003,
"max_latency_cycles": 128,
"max_area_mm2": 0.10,
"max_power_mw": 12.0,
},
}
CRYO_SLUGS = tuple(CRYO_CURRICULUM)
FALLBACK_SLUGS = tuple(
spec.slug for spec in TASK_SPECS if spec.track.startswith("fallback")
)
def curriculum_for_slug(slug: str) -> dict[str, float | int]:
"""Return RSI distance/noise/budget knobs for a cryo eval slug."""
return dict(CRYO_CURRICULUM[slug])
```
### scenario_helpers.py
```python
"""Per-task workspace setup for CryoBrain + fallback tracks."""
import json
import os
import shutil
import tempfile
from pathlib import Path
ROOT_DIR = Path(__file__).resolve().parent
LOCAL_TASK_ROOT = ROOT_DIR / "tasks"
HIDDEN_ROOT = Path(os.environ.get("HIDDEN_ROOT", "/donotaccess"))
AGENT_USER = "agent"
AGENT_GROUP = "agent"
AGENT_FILES = (
"Makefile",
"filelist.f",
"prompt.md",
"rtl",
"dv",
"synth",
"scripts",
"formal",
"vendor",
"design_config.json",
"scenario.json",
)
_KEEP = {".hud"}
GOLDEN_OVERLAY = {
"stream_arb_fifo_repair": ("stream_arb_fifo_golden.sv", "rtl/stream_arb_fifo.sv"),
"stream_arb_fifo_cocotb_dv": ("solution/test_stream_arb_fifo.py", "dv/cocotb/test_stream_arb_fifo.py"),
"stream_arb_fifo_formal": ("solution/stream_arb_fifo_props.sv", "formal/stream_arb_fifo_props.sv"),
"cryo_brain_decoder": ("cryo_brain_decoder_golden.sv", "rtl/cryo_brain_decoder.sv"),
}
def _resolve_workspace_root() -> Path:
explicit = os.environ.get("WORKSPACE_ROOT")
if explicit:
return Path(explicit)
if Path("/workdir").is_dir():
return Path("/workdir")
root = Path(tempfile.gettempdir()) / "hud-cryobrain" / f"workdir-{os.getpid()}"
root.mkdir(parents=True, exist_ok=True)
return root
WORKSPACE_ROOT = _resolve_workspace_root()
def hidden_dir(task_id: str) -> Path:
candidate = HIDDEN_ROOT / task_id
if candidate.is_dir():
return candidate
return LOCAL_TASK_ROOT / task_id / "donotaccess"
def write_scenario_files(workdir: Path, scenario: dict[str, object]) -> None:
"""Bind curriculum distance/noise/budget into agent-visible scenario.json."""
scenario_path = workdir / "scenario.json"
base = {}
if scenario_path.is_file():
base = json.loads(scenario_path.read_text(encoding="utf-8"))
base.update(scenario)
scenario_path.write_text(json.dumps(base, indent=2), encoding="utf-8")
def _is_root() -> bool:
return hasattr(os, "geteuid") and os.geteuid() == 0
def _make_agent_writable(path: Path) -> None:
if not _is_root():
return
for item in [path, *path.rglob("*")]:
if item.name in _KEEP:
continue
try:
shutil.chown(item, user=AGENT_USER, group=AGENT_GROUP)
item.chmod(0o755 if item.is_dir() else 0o644)
except (LookupError, PermissionError, FileNotFoundError):
pass
def _clear_workspace(workdir: Path) -> None:
for child in workdir.iterdir():
if child.name in _KEEP:
continue
if child.is_dir() and not child.is_symlink():
shutil.rmtree(child, ignore_errors=True)
else:
child.unlink(missing_ok=True)
def _copy_into(src: Path, dst: Path) -> None:
for item in src.iterdir():
target = dst / item.name
if item.is_dir():
shutil.copytree(item, target, symlinks=True, dirs_exist_ok=True)
else:
shutil.copy2(item, target)
def _populate_from_repo(task_id: str, workdir: Path) -> None:
src = LOCAL_TASK_ROOT / task_id
for name in AGENT_FILES:
item = src / name
if not item.exists():
continue
target = workdir / name
if item.is_dir():
shutil.copytree(item, target, ignore=shutil.ignore_patterns("__pycache__"), dirs_exist_ok=True)
else:
shutil.copy2(item, target)
calib = workdir / "scripts" / "check_calibration.py"
if calib.is_file():
calib.unlink(missing_ok=True)
def setup_task(
task_id: str,
validate_mode: str | None = None,
scenario: dict[str, object] | None = None,
) -> dict[str, object]:
workdir = WORKSPACE_ROOT
hidden = hidden_dir(task_id)
baked_baseline = HIDDEN_ROOT / task_id / "baseline"
workdir.mkdir(parents=True, exist_ok=True)
_clear_workspace(workdir)
if baked_baseline.is_dir():
_copy_into(baked_baseline, workdir)
else:
_populate_from_repo(task_id, workdir)
if scenario:
write_scenario_files(workdir, scenario)
if validate_mode == "golden_pass" and task_id in GOLDEN_OVERLAY:
src_rel, dst_rel = GOLDEN_OVERLAY[task_id]
src = hidden / src_rel
dst = workdir / dst_rel
if src.is_file():
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
_make_agent_writable(workdir)
return {
"success": workdir.is_dir(),
"task_id": task_id,
"workdir": str(workdir),
"hidden_dir": str(hidden),
"validate_mode": validate_mode,
"scenario": scenario,
"baseline_reset": True,
}
```
[510 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]