Update benchmark runtime image

entrypoint.py

@@ -110,7 +110,7 @@ def _start_health_server() -> HTTPServer:
     return server
 
 
-def upload_artifact(api: HfApi, path: Path, dest: str) -> None:
+def upload_artifact(api: HfApi, path: Path, dest: str) -> None:
     if not path.exists():
         print(f'[upload] skip missing {path}', flush=True)
         return
@@ -120,7 +120,20 @@ def upload_artifact(api: HfApi, path: Path, dest: str) -> None:
         repo_id=OUTPUT_REPO,
         repo_type='model',
     )
-    print(f'[upload] uploaded {path} -> {OUTPUT_REPO}/{dest}', flush=True)
+    print(f'[upload] uploaded {path} -> {OUTPUT_REPO}/{dest}', flush=True)
+
+
+def build_benchmark_mode_command() -> list[str]:
+    return [
+        'python',
+        str(REPO_ROOT / 'scripts' / 'benchmark_runner.py'),
+        '--benchmark', os.environ.get('HYDRA_BENCHMARK_NAME', 'GSM8K'),
+        '--generator-mode', 'hydra',
+        '--variant', os.environ.get('HYDRA_BENCHMARK_VARIANT', 'hydra_full'),
+        '--seed', os.environ.get('HYDRA_SEED', '42'),
+        '--out', str(REPO_ROOT / 'benchmark_result.json'),
+        '--ledger', str(REPO_ROOT / 'benchmark_ledger.json'),
+    ]
 
 
 def _wait_for_cuda_ready(timeout_s: int = 120) -> None:
@@ -158,7 +171,7 @@ def _wait_for_cuda_ready(timeout_s: int = 120) -> None:
     print(f'[job] CUDA still not ready after {timeout_s}s — continuing anyway (training will likely fail)', flush=True)
 
 
-def run_job_mode() -> int:
+def run_job_mode() -> int:
     os.chdir(REPO_ROOT)
     os.environ.setdefault('HYDRA_TIME_BUDGET', '43200')
     os.environ.setdefault('HYDRA_TARGET_SHARDS', '2048')
@@ -203,7 +216,31 @@ def run_job_mode() -> int:
     else:
         print('[upload] HF_TOKEN not set; skipping artifact upload', flush=True)
 
-    return proc.returncode
+    return proc.returncode
+
+
+def run_benchmark_mode() -> int:
+    os.chdir(REPO_ROOT)
+    cmd = build_benchmark_mode_command()
+    print(f'[benchmark] command={cmd}', flush=True)
+    proc = subprocess.run(cmd, check=False)
+
+    if TOKEN:
+        api = HfApi(token=TOKEN)
+        try:
+            api.create_repo(repo_id=OUTPUT_REPO, repo_type='model', private=True, exist_ok=True)
+        except Exception as e:
+            print(f'[upload] create_repo warning: {type(e).__name__}: {e}', flush=True)
+        prefix = f'jobs/{JOB_ID}'
+        try:
+            upload_artifact(api, REPO_ROOT / 'benchmark_result.json', f'{prefix}/benchmark_result.json')
+            upload_artifact(api, REPO_ROOT / 'benchmark_ledger.json', f'{prefix}/benchmark_ledger.json')
+        except Exception as e:
+            print(f'[upload] upload warning: {type(e).__name__}: {e}', flush=True)
+    else:
+        print('[upload] HF_TOKEN not set; skipping benchmark artifact upload', flush=True)
+
+    return proc.returncode
 
 
 def run_space_mode() -> int:
@@ -217,10 +254,12 @@ def run_space_mode() -> int:
         server.server_close()
 
 
-def main() -> int:
-    if RUNTIME_MODE == 'job':
-        return run_job_mode()
-    return run_space_mode()
+def main() -> int:
+    if RUNTIME_MODE == 'job':
+        return run_job_mode()
+    if RUNTIME_MODE == 'benchmark':
+        return run_benchmark_mode()
+    return run_space_mode()
 
 
 if __name__ == '__main__':

overlay/hydra/model.py

@@ -32,11 +32,28 @@ from __future__ import annotations
 
 import os
 
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-
-from mamba_ssm import Mamba3
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+
+try:
+    from mamba_ssm import Mamba3
+except Exception:  # pragma: no cover - depends on optional runtime install
+    Mamba3 = None  # type: ignore[assignment]
+
+
+def _get_mamba3_cls():
+    global Mamba3
+    if Mamba3 is None:
+        try:
+            from mamba_ssm import Mamba3 as _Mamba3  # type: ignore
+            Mamba3 = _Mamba3  # type: ignore[assignment]
+        except Exception as exc:  # pragma: no cover - environment dependent
+            raise ImportError(
+                "mamba_ssm is required for Mamba-based HYDRA blocks. "
+                "Install mamba-ssm or use HYDRA_BASELINE_ARCH=transformer."
+            ) from exc
+    return Mamba3
 
 
 def _ensure_triton_cuda_backend_registered() -> None:
@@ -81,9 +98,30 @@ from hydra.hyena_block import HyenaBlock
 from hydra.optimizer import MuonAdamW
 
 
-def norm(x: torch.Tensor) -> torch.Tensor:
-    """RMSNorm over the last dim — stateless, autocast-friendly."""
-    return F.rms_norm(x, (x.size(-1),))
+def norm(x: torch.Tensor) -> torch.Tensor:
+    """RMSNorm over the last dim — stateless, autocast-friendly."""
+    return F.rms_norm(x, (x.size(-1),))
+
+
+class TransformerBaselineBlock(nn.Module):
+    """Transformer-style delta block for matched baseline experiments.
+
+    This block returns a transformed delta tensor rather than owning the outer
+    residual connection, because ManifoldHyperConnection already handles stream
+    mixing and residual injection around the block function.
+    """
+
+    def __init__(self, d_model: int, n_heads: int, expand: int, dropout: float) -> None:
+        super().__init__()
+        self.self_attn = nn.MultiheadAttention(d_model, n_heads, dropout=dropout, batch_first=True)
+        self.ff_in = nn.Linear(d_model, expand * d_model, bias=False)
+        self.ff_out = nn.Linear(expand * d_model, d_model, bias=False)
+        self.dropout = nn.Dropout(dropout)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        attn_out, _ = self.self_attn(x, x, x, need_weights=False)
+        ff = self.ff_out(F.gelu(self.ff_in(attn_out)))
+        return self.dropout(attn_out + ff)
 
 
 class PostSemClawModel(nn.Module):
@@ -103,6 +141,7 @@ class PostSemClawModel(nn.Module):
         _ensure_triton_cuda_backend_registered()
         self.config = config
         self._throughput_mode = os.environ.get("HYDRA_THROUGHPUT_MODE", "0") == "1"
+        self._baseline_arch = os.environ.get("HYDRA_BASELINE_ARCH", "mamba3").strip().lower()
 
         # Token embedding
         self.wte = nn.Embedding(config.vocab_size, config.d_model)
@@ -124,23 +163,31 @@ class PostSemClawModel(nn.Module):
             print(f"[WARN] layers in both hyena_layers and gdn_layers; using Hyena: {sorted(_both)}", flush=True)
             _gdn_layer_set -= _hyena_layer_set
 
-        if _gdn_layer_set:
-            from hydra.gdn_block import GDNBlock  # requires `fla` package
-
-        def _build_block(i: int) -> nn.Module:
-            if i in _hyena_layer_set:
-                return HyenaBlock(
+        if _gdn_layer_set:
+            from hydra.gdn_block import GDNBlock  # requires `fla` package
+
+        def _build_block(i: int) -> nn.Module:
+            if self._baseline_arch == "transformer":
+                return TransformerBaselineBlock(
+                    d_model=config.d_model,
+                    n_heads=config.n_heads,
+                    expand=config.expand,
+                    dropout=float(os.environ.get("HYDRA_DROPOUT", "0.2")),
+                )
+            if i in _hyena_layer_set:
+                return HyenaBlock(
                     d_model=config.d_model,
                     seq_len=config.sequence_len,
                     order=int(os.environ.get("HYDRA_HYENA_ORDER", "2")),
                     filter_order=int(os.environ.get("HYDRA_HYENA_FILTER_DIM", "64")),
                 )
-            if i in _gdn_layer_set:
-                return GDNBlock(
-                    d_model=config.d_model,
-                    n_heads=config.n_heads,
-                )
-            return Mamba3(
+            if i in _gdn_layer_set:
+                return GDNBlock(
+                    d_model=config.d_model,
+                    n_heads=config.n_heads,
+                )
+            mamba3_cls = _get_mamba3_cls()
+            return mamba3_cls(
                 d_model=config.d_model,
                 d_state=config.d_state,
                 expand=config.expand,

overlay/hydra/training.py

@@ -4,17 +4,20 @@ Extracted from the monolithic train.py (W1 modularization). Semantics
 preserved. Public entrypoint: `main()`.
 """
 
-from __future__ import annotations
-
-import gc
-import json
-import math
-import os
-import sys
-import threading
-import time
-from dataclasses import asdict
-from pathlib import Path
+from __future__ import annotations
+
+import gc
+import hashlib
+import json
+import math
+import os
+import sys
+import threading
+import time
+from collections.abc import Mapping
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any
 
 import torch
 
@@ -130,7 +133,7 @@ def _ckpt_snapshot_state_dicts(
     return msd, osd
 
 
-def save_ckpt(
+def save_ckpt(
     model: PostSemClawModel,
     optimizer: torch.optim.Optimizer,
     config: PostSemClawConfig,
@@ -211,11 +214,233 @@ def save_ckpt(
             target=_write, daemon=True, name=f"ckpt-save-{step}"
         )
         _CKPT_WORKER_THREAD.start()
-    except Exception as e:
-        print(f"[ckpt] SNAPSHOT FAILED {path}: {type(e).__name__}: {e}", flush=True)
-
-
-def config_from_dict(cfg_dict: dict) -> PostSemClawConfig:
+    except Exception as e:
+        print(f"[ckpt] SNAPSHOT FAILED {path}: {type(e).__name__}: {e}", flush=True)
+
+
+def _env_flag_enabled(env: Mapping[str, str], key: str) -> bool:
+    value = str(env.get(key, "0") or "0").strip().lower()
+    return value not in {"", "0", "false", "no", "off"}
+
+
+def _env_int(env: Mapping[str, str], key: str, default: int) -> int:
+    try:
+        return int(str(env.get(key, str(default)) or str(default)))
+    except ValueError:
+        return default
+
+
+def architecture_compliance_payload(env: Mapping[str, str]) -> dict[str, bool | int | str]:
+    throughput_mode = _env_flag_enabled(env, "HYDRA_THROUGHPUT_MODE")
+    fastpath = _env_flag_enabled(env, "HYDRA_FASTPATH")
+    force_htm_cpu = _env_flag_enabled(env, "HYDRA_FORCE_HTM_CPU")
+    inert_mamba = _env_flag_enabled(env, "HYDRA_INERT_MAMBA")
+    synthetic_retina = _env_flag_enabled(env, "HYDRA_ALLOW_SYNTHETIC_RETINA")
+    hyena_layers = str(env.get("HYDRA_HYENA_LAYERS", "") or "")
+    engram_subsample = _env_int(env, "HYDRA_ENGRAM_SUBSAMPLE", 1)
+    htm_subsample = _env_int(env, "HYDRA_HTM_SUBSAMPLE", 1)
+    full_arch_compliant = not any((
+        throughput_mode,
+        fastpath,
+        force_htm_cpu,
+        inert_mamba,
+        synthetic_retina,
+        bool(hyena_layers.strip()),
+    ))
+    return {
+        'full_arch_compliant': full_arch_compliant,
+        'throughput_mode': throughput_mode,
+        'fastpath': fastpath,
+        'force_htm_cpu': force_htm_cpu,
+        'inert_mamba': inert_mamba,
+        'synthetic_retina': synthetic_retina,
+        'hyena_layers': hyena_layers,
+        'engram_subsample': engram_subsample,
+        'htm_subsample': htm_subsample,
+    }
+
+
+def eval_attempt_batches(*, requested_batch: int, min_batch: int) -> list[int]:
+    requested = max(1, int(requested_batch))
+    minimum = max(1, int(min_batch))
+    batches: list[int] = []
+    current = requested
+    while current >= minimum:
+        if current not in batches:
+            batches.append(current)
+        if current == minimum:
+            break
+        next_batch = max(minimum, current // 2)
+        if next_batch == current:
+            break
+        current = next_batch
+    if minimum not in batches:
+        batches.append(minimum)
+    return batches
+
+
+def build_eval_plan(*, eval_tokens: int, requested_batch: int, max_seq_len: int, chunk_tokens: int, min_batch: int) -> dict[str, Any]:
+    effective_chunk_tokens = max(int(chunk_tokens), int(requested_batch) * int(max_seq_len))
+    chunk_count = max(1, math.ceil(int(eval_tokens) / effective_chunk_tokens))
+    return {
+        'eval_tokens': int(eval_tokens),
+        'eval_requested_batch': int(requested_batch),
+        'eval_chunk_tokens': int(effective_chunk_tokens),
+        'eval_chunk_count': int(chunk_count),
+        'eval_attempt_batches': eval_attempt_batches(requested_batch=requested_batch, min_batch=min_batch),
+        'eval_min_batch': int(max(1, min_batch)),
+    }
+
+
+def _fingerprint_descriptor(descriptor: Mapping[str, Any]) -> str:
+    payload = json.dumps(dict(descriptor), sort_keys=True, separators=(",", ":"))
+    return hashlib.sha1(payload.encode("utf-8")).hexdigest()[:12]
+
+
+def dataset_domain_payload(*, env: Mapping[str, str], prepare_module: Any, nemotron_module: Any | None) -> dict[str, Any]:
+    use_nemotron = _env_flag_enabled(env, "HYDRA_USE_NEMOTRON")
+    vocab_size = int(getattr(prepare_module, "VOCAB_SIZE", 0))
+
+    if use_nemotron and nemotron_module is not None:
+        use_full_blend = _env_flag_enabled(env, "HYDRA_USE_FULL_BLEND")
+        phase = str(env.get("HYDRA_NEMOTRON_PHASE", "phase1") or "phase1").strip().lower()
+        if use_full_blend:
+            train_weights = dict(getattr(nemotron_module, "FULL_BLEND_WEIGHTS", {}))
+            val_weights = dict(train_weights)
+        else:
+            train_weights = dict(
+                getattr(nemotron_module, "PHASE2_WEIGHTS", {}) if phase == "phase2" else getattr(nemotron_module, "PHASE1_WEIGHTS", {})
+            )
+            val_weights = {"Nemotron-Pretraining-Multiple-Choice": 1.0}
+        train_descriptor = {
+            "backend": "nemotron_stream",
+            "phase": "full_blend" if use_full_blend else phase,
+            "weights": train_weights,
+            "factual_inject_rate": _env_int(env, "HYDRA_FACTUAL_INJECT_RATE", 50),
+            "vocab_size": vocab_size,
+        }
+        val_descriptor = {
+            "backend": "nemotron_stream",
+            "phase": "full_blend" if use_full_blend else "val_multiple_choice",
+            "weights": val_weights,
+            "vocab_size": vocab_size,
+        }
+        data_backend = "nemotron_stream"
+    else:
+        all_files = list(getattr(prepare_module, "list_parquet_files", lambda: [])())
+        val_filename = str(getattr(prepare_module, "VAL_FILENAME", ""))
+        train_files = [str(path) for path in all_files if not str(path).endswith(val_filename)]
+        val_files = [str(path) for path in all_files if str(path).endswith(val_filename)]
+        train_descriptor = {
+            "backend": "climbmix_parquet",
+            "train_shard_count": len(train_files),
+            "train_shard_examples": sorted(Path(path).name for path in train_files[:3]),
+            "vocab_size": vocab_size,
+        }
+        val_descriptor = {
+            "backend": "climbmix_parquet",
+            "val_filename": val_filename,
+            "val_shard_count": len(val_files),
+            "vocab_size": vocab_size,
+        }
+        data_backend = "climbmix_parquet"
+
+    train_fingerprint = _fingerprint_descriptor(train_descriptor)
+    val_fingerprint = _fingerprint_descriptor(val_descriptor)
+    return {
+        "data_backend": data_backend,
+        "train_domain_descriptor": train_descriptor,
+        "val_domain_descriptor": val_descriptor,
+        "train_domain_fingerprint": train_fingerprint,
+        "val_domain_fingerprint": val_fingerprint,
+        "train_val_domain_match": train_fingerprint == val_fingerprint,
+    }
+
+
+def build_lineage_payload(
+    *,
+    env: Mapping[str, str],
+    seed: int,
+    resume_requested: bool,
+    resume_requested_path: str | None,
+    resume_loaded_path: str | None,
+    resume_step: int,
+    resume_epoch: int,
+) -> dict[str, Any]:
+    warmstart = _env_flag_enabled(env, "HYDRA_WARMSTART")
+    resume_applied = resume_loaded_path is not None and int(resume_step) > 0
+    if resume_applied and warmstart:
+        lineage_mode = "warmstart_resume"
+    elif resume_applied:
+        lineage_mode = "resume"
+    else:
+        lineage_mode = "fresh"
+    return {
+        "seed": int(seed),
+        "warmstart": warmstart,
+        "resume_requested": bool(resume_requested),
+        "resume_applied": resume_applied,
+        "resume_requested_path": resume_requested_path,
+        "resume_loaded_path": resume_loaded_path,
+        "resume_step": int(resume_step),
+        "resume_epoch": int(resume_epoch),
+        "lineage_mode": lineage_mode,
+    }
+
+
+def build_final_metrics_payload(
+    *,
+    secondary_metrics: dict[str, Any],
+    val_bpb: float | None,
+    val_ppl: float | None,
+    eval_status: str,
+    eval_error: str | None,
+    n_layer: int,
+    d_model: int,
+    num_params: int,
+    step: int,
+    total_tokens: int,
+    peak_vram_mb: float,
+    total_training_time: float,
+    sdr_target_active: int,
+    architecture_env: Mapping[str, str] | None = None,
+    eval_diagnostics: Mapping[str, Any] | None = None,
+    domain_fingerprints: Mapping[str, Any] | None = None,
+    lineage_payload: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build final run metrics without conflating skipped eval and validation.
+
+    This helper deliberately preserves ``val_bpb=None`` when final eval did not
+    complete. HPO can then prune or explicitly label a fallback instead of
+    accidentally treating live training BPB as validation BPB.
+    """
+    payload = dict(secondary_metrics)
+    payload.update({
+        'eval_status': eval_status,
+        'eval_error': eval_error,
+        'objective_source': 'final_val' if val_bpb is not None else 'missing_final_val',
+        'val_bpb': float(val_bpb) if val_bpb is not None else None,
+        'val_ppl': float(val_ppl) if val_ppl is not None else None,
+        'n_layer': int(n_layer),
+        'd_model': int(d_model),
+        'num_params_M': float(num_params / 1e6),
+        'num_steps': int(step),
+        'total_tokens_M': float(total_tokens / 1e6),
+        'peak_vram_mb': float(peak_vram_mb),
+        'training_seconds': float(total_training_time),
+        'sdr_target_active': int(sdr_target_active),
+    })
+    payload.update(architecture_compliance_payload(architecture_env or dict(os.environ)))
+    if eval_diagnostics:
+        payload.update(dict(eval_diagnostics))
+    if domain_fingerprints:
+        payload.update(dict(domain_fingerprints))
+    if lineage_payload:
+        payload.update(dict(lineage_payload))
+    return payload
+
+
+def config_from_dict(cfg_dict: dict) -> PostSemClawConfig:
     """Reconstruct a PostSemClawConfig from a checkpoint's asdict() payload.
 
     Newly-added fields (e.g. `hyena_layers`) are defaulted when absent in
@@ -275,14 +500,14 @@ def _try_load_ckpt(path: Path, model, optimizer, device):
     return step, total_training_time, smooth_train_loss, bpt_ema, epoch
 
 
-def maybe_resume_ckpt(
-    model: PostSemClawModel,
-    optimizer: torch.optim.Optimizer,
-    device: torch.device,
-) -> tuple[int, float, float, float, int]:
-    if not RESUME_CKPT or RESUME_CKPT.lower() == "none":
-        print("[ckpt] resume disabled; starting fresh", flush=True)
-        return 0, 0.0, 0.0, 0.0, 0
+def maybe_resume_ckpt(
+    model: PostSemClawModel,
+    optimizer: torch.optim.Optimizer,
+    device: torch.device,
+) -> tuple[int, float, float, float, int, str | None]:
+    if not RESUME_CKPT or RESUME_CKPT.lower() == "none":
+        print("[ckpt] resume disabled; starting fresh", flush=True)
+        return 0, 0.0, 0.0, 0.0, 0, None
 
     resume_path = Path(os.path.expanduser(RESUME_CKPT))
     # Try the primary path, then rotated backups. This is crucial because a
@@ -296,17 +521,18 @@ def maybe_resume_ckpt(
         if not cand.exists():
             continue
         try:
-            result = _try_load_ckpt(cand, model, optimizer, device)
-            if result is not None:
-                if cand != resume_path:
-                    print(f"[ckpt] fell back to rotation {cand.name}", flush=True)
-                return result
+            result = _try_load_ckpt(cand, model, optimizer, device)
+            if result is not None:
+                if cand != resume_path:
+                    print(f"[ckpt] fell back to rotation {cand.name}", flush=True)
+                step, total_training_time, smooth_train_loss, bpt_ema, epoch = result
+                return step, total_training_time, smooth_train_loss, bpt_ema, epoch, str(cand)
         except Exception as e:
             print(f"[ckpt] {cand.name} load failed: {type(e).__name__}: {e}", flush=True)
             continue
 
-    print(f"[ckpt] no usable checkpoint in {resume_path} + rotations; starting fresh", flush=True)
-    return 0, 0.0, 0.0, 0.0, 0
+    print(f"[ckpt] no usable checkpoint in {resume_path} + rotations; starting fresh", flush=True)
+    return 0, 0.0, 0.0, 0.0, 0, None
 
 
 # ---------------------------------------------------------------------------
@@ -388,9 +614,18 @@ def main() -> None:
         weight_decay=WEIGHT_DECAY,
     )
 
-    step, total_training_time, smooth_train_loss, bpt_ema, resume_epoch = maybe_resume_ckpt(
-        model, optimizer, device,
-    )
+    step, total_training_time, smooth_train_loss, bpt_ema, resume_epoch, resume_loaded_path = maybe_resume_ckpt(
+        model, optimizer, device,
+    )
+    lineage_payload = build_lineage_payload(
+        env=dict(os.environ),
+        seed=SEED,
+        resume_requested=bool(RESUME_CKPT and RESUME_CKPT.lower() != "none"),
+        resume_requested_path=RESUME_CKPT if RESUME_CKPT and RESUME_CKPT.lower() != "none" else None,
+        resume_loaded_path=resume_loaded_path,
+        resume_step=step,
+        resume_epoch=resume_epoch,
+    )
 
     # Learnability #4: inform the model of the BOS token id so it can mask
     # doc-separator positions in packed sequences. Always set (the mask only
@@ -785,10 +1020,22 @@ def main() -> None:
     # does not benefit from overlap with backward). HYDRA_EVAL_TOKENS controls
     # how many val tokens to sweep (default 2 M, short enough for autoresearch
     # 5-min budgets).
-    val_bpb: float | None = None
-    _eval_B = int(os.environ.get("HYDRA_EVAL_BATCH", str(max(1, DEVICE_BATCH_SIZE // 2))))
-    _eval_tokens = int(os.environ.get("HYDRA_EVAL_TOKENS", str(2 * 524288)))
-    try:
+    val_bpb: float | None = None
+    val_ppl: float | None = None
+    eval_status = "not_started"
+    eval_error: str | None = None
+    _eval_B = int(os.environ.get("HYDRA_EVAL_BATCH", str(max(1, DEVICE_BATCH_SIZE // 2))))
+    _eval_tokens = int(os.environ.get("HYDRA_EVAL_TOKENS", str(2 * 524288)))
+    _eval_chunk_tokens = int(os.environ.get("HYDRA_EVAL_CHUNK_TOKENS", str(_eval_tokens)))
+    _eval_min_batch = int(os.environ.get("HYDRA_EVAL_MIN_BATCH", "1"))
+    eval_diagnostics = build_eval_plan(
+        eval_tokens=_eval_tokens,
+        requested_batch=_eval_B,
+        max_seq_len=MAX_SEQ_LEN,
+        chunk_tokens=_eval_chunk_tokens,
+        min_batch=_eval_min_batch,
+    )
+    try:
         # Aggressive VRAM reclaim for 6GB cards. Peak training VRAM = 5.1GB
         # which leaves < 1GB for the eval forward — the driver can't satisfy
         # the allocation. Free EVERY tensor we don't strictly need:
@@ -810,34 +1057,70 @@ def main() -> None:
             model._last_sdr = None
         import gc as _gc
         _gc.collect()
-        torch.cuda.empty_cache()
-        torch.cuda.synchronize()
-        try:
-            _free_mb = torch.cuda.mem_get_info()[0] / 1024 / 1024
-            print(f"[VAL] free_vram_mb={_free_mb:.0f} (cleared optimizer state)", flush=True)
-        except Exception:
-            pass
-        print(f"[VAL] running eval on {_eval_tokens} tokens at B={_eval_B}...", flush=True)
-        model.eval()
-        _orig = _prepare_mod.EVAL_TOKENS
-        _prepare_mod.EVAL_TOKENS = _eval_tokens
-        with autocast_ctx:
-            val_bpb = evaluate_bpb(model, tokenizer, _eval_B)
-        _prepare_mod.EVAL_TOKENS = _orig
-        val_ppl = 2 ** val_bpb
-        print(f"[VAL] step={step} val_bpb={val_bpb:.4f} val_ppl={val_ppl:.3f}", flush=True)
-    except torch.cuda.OutOfMemoryError as e:
-        print(f"[VAL] SKIPPED (OOM): {e}", flush=True)
-        torch.cuda.empty_cache()
-    except Exception as e:
-        import traceback as _tb
-        print(f"[VAL] SKIPPED ({type(e).__name__}): {e}", flush=True)
-        _tb.print_exc()
-        try:
-            _free = torch.cuda.mem_get_info()[0] / 1024 / 1024
-            print(f"[VAL] post-crash free_vram_mb={_free:.0f}", flush=True)
-        except Exception:
-            pass
+        torch.cuda.empty_cache()
+        torch.cuda.synchronize()
+        try:
+            _free_mb = torch.cuda.mem_get_info()[0] / 1024 / 1024
+            eval_diagnostics["eval_free_vram_before_mb"] = float(_free_mb)
+            print(f"[VAL] free_vram_mb={_free_mb:.0f} (cleared optimizer state)", flush=True)
+        except Exception:
+            pass
+        print(
+            f"[VAL] running eval on {_eval_tokens} tokens at B={_eval_B} "
+            f"chunk_tokens={eval_diagnostics['eval_chunk_tokens']} attempts={eval_diagnostics['eval_attempt_batches']}...",
+            flush=True,
+        )
+        model.eval()
+        _orig = _prepare_mod.EVAL_TOKENS
+        _orig_chunk = getattr(_prepare_mod, "EVAL_CHUNK_TOKENS", _eval_tokens)
+        _prepare_mod.EVAL_TOKENS = _eval_tokens
+        _prepare_mod.EVAL_CHUNK_TOKENS = int(eval_diagnostics["eval_chunk_tokens"])
+        _successful_batch: int | None = None
+        _attempts: list[int] = []
+        try:
+            for _attempt_batch in eval_diagnostics["eval_attempt_batches"]:
+                _attempts.append(int(_attempt_batch))
+                eval_diagnostics["eval_attempted_batch"] = int(_attempt_batch)
+                try:
+                    with autocast_ctx:
+                        val_bpb = evaluate_bpb(model, tokenizer, int(_attempt_batch))
+                    _successful_batch = int(_attempt_batch)
+                    break
+                except torch.cuda.OutOfMemoryError as _attempt_oom:
+                    eval_error = str(_attempt_oom)
+                    eval_status = "oom"
+                    torch.cuda.empty_cache()
+                    if int(_attempt_batch) == eval_diagnostics["eval_attempt_batches"][-1]:
+                        raise
+        finally:
+            _prepare_mod.EVAL_TOKENS = _orig
+            _prepare_mod.EVAL_CHUNK_TOKENS = _orig_chunk
+        eval_diagnostics["eval_attempt_batches"] = _attempts
+        eval_diagnostics["eval_effective_batch"] = _successful_batch
+        val_ppl = 2 ** val_bpb
+        eval_status = "completed"
+        print(f"[VAL] step={step} val_bpb={val_bpb:.4f} val_ppl={val_ppl:.3f}", flush=True)
+    except torch.cuda.OutOfMemoryError as e:
+        eval_status = "oom"
+        eval_error = str(e)
+        print(f"[VAL] SKIPPED (OOM): {e}", flush=True)
+        torch.cuda.empty_cache()
+        try:
+            eval_diagnostics["eval_free_vram_after_mb"] = float(torch.cuda.mem_get_info()[0] / 1024 / 1024)
+        except Exception:
+            pass
+    except Exception as e:
+        import traceback as _tb
+        eval_status = type(e).__name__
+        eval_error = str(e)
+        print(f"[VAL] SKIPPED ({type(e).__name__}): {e}", flush=True)
+        _tb.print_exc()
+        try:
+            _free = torch.cuda.mem_get_info()[0] / 1024 / 1024
+            eval_diagnostics["eval_free_vram_after_mb"] = float(_free)
+            print(f"[VAL] post-crash free_vram_mb={_free:.0f}", flush=True)
+        except Exception:
+            pass
 
     # Final ckpts with val_bpb filled in (if eval succeeded).
     save_ckpt(
@@ -881,8 +1164,13 @@ def main() -> None:
         / total_training_time / GPU_BF16_PEAK_FLOPS
         if total_training_time > 0 else 0
     )
-    peak_vram_mb = torch.cuda.max_memory_allocated() / 1024 / 1024
-    metrics = model.get_secondary_metrics()
+    peak_vram_mb = torch.cuda.max_memory_allocated() / 1024 / 1024
+    metrics = model.get_secondary_metrics()
+    domain_fingerprints = dataset_domain_payload(
+        env=dict(os.environ),
+        prepare_module=_prepare_mod,
+        nemotron_module=globals().get("_p_nemo"),
+    )
 
     print("---")
     print(f"val_bpb:          {val_bpb:.6f}" if val_bpb is not None else "val_bpb:          SKIPPED")
@@ -918,22 +1206,28 @@ def main() -> None:
     # Emit full metrics dictionary as JSON for sweep aggregation. Path from
     # HYDRA_METRICS_OUT env var; default=/tmp/hydra_run_metrics.json. Always
     # written (even without diagnostics) so the aggregator can compare runs.
-    _metrics_out = os.environ.get("HYDRA_METRICS_OUT", "/tmp/hydra_run_metrics.json")
-    try:
-        _dump = dict(metrics)
-        _dump.update({
-            'val_bpb': float(val_bpb),
-            'val_ppl': float(val_ppl),
-            'n_layer': int(N_LAYER),
-            'd_model': int(D_MODEL),
-            'num_params_M': float(num_params / 1e6),
-            'num_steps': int(step),
-            'total_tokens_M': float(total_tokens / 1e6),
-            'peak_vram_mb': float(peak_vram_mb),
-            'training_seconds': float(total_training_time),
-            'sdr_target_active': int(os.environ.get("HYDRA_SDR_TARGET_ACTIVE", "327")),
-        })
-        Path(_metrics_out).parent.mkdir(parents=True, exist_ok=True)
+    _metrics_out = os.environ.get("HYDRA_METRICS_OUT", "/tmp/hydra_run_metrics.json")
+    try:
+        _dump = build_final_metrics_payload(
+            secondary_metrics=metrics,
+            val_bpb=val_bpb,
+            val_ppl=val_ppl,
+            eval_status=eval_status,
+            eval_error=eval_error,
+            n_layer=N_LAYER,
+            d_model=D_MODEL,
+            num_params=num_params,
+            step=step,
+            total_tokens=total_tokens,
+            peak_vram_mb=peak_vram_mb,
+            total_training_time=total_training_time,
+            sdr_target_active=int(os.environ.get("HYDRA_SDR_TARGET_ACTIVE", "327")),
+            architecture_env=dict(os.environ),
+            eval_diagnostics=eval_diagnostics,
+            domain_fingerprints=domain_fingerprints,
+            lineage_payload=lineage_payload,
+        )
+        Path(_metrics_out).parent.mkdir(parents=True, exist_ok=True)
         with open(_metrics_out, 'w') as _f:
             json.dump(_dump, _f, indent=2, sort_keys=True)
         print(f"[METRICS] wrote {_metrics_out}", flush=True)

overlay/prepare.py

@@ -13,9 +13,10 @@ import os
 import sys
 import time
 import math
-import argparse
-import pickle
-from multiprocessing import Pool
+import argparse
+import pickle
+from multiprocessing import Pool
+from typing import Any
 
 import requests
 import pyarrow.parquet as pq
@@ -29,7 +30,8 @@ import torch
 
 MAX_SEQ_LEN = int(os.environ.get("HYDRA_SEQ_LEN", "512"))  # context length
 TIME_BUDGET = 300        # training time budget in seconds (5 minutes)
-EVAL_TOKENS = 40 * 524288  # number of tokens for val eval
+EVAL_TOKENS = 40 * 524288  # number of tokens for val eval
+EVAL_CHUNK_TOKENS = int(os.environ.get("HYDRA_EVAL_CHUNK_TOKENS", str(EVAL_TOKENS)))
 
 # ---------------------------------------------------------------------------
 # Configuration
@@ -158,7 +160,8 @@ def train_tokenizer():
     print("Tokenizer: training BPE tokenizer...")
     t0 = time.time()
 
-    tokenizer = rustbpe.Tokenizer()
+    tokenizer_cls = getattr(rustbpe, "Tokenizer")
+    tokenizer: Any = tokenizer_cls()
     vocab_size_no_special = VOCAB_SIZE - len(SPECIAL_TOKENS)
     tokenizer.train_from_iterator(text_iterator(), vocab_size_no_special, pattern=SPLIT_PATTERN)
 
@@ -225,9 +228,10 @@ class Tokenizer:
     def get_bos_token_id(self):
         return self.bos_token_id
 
-    def encode(self, text, prepend=None, num_threads=8):
-        if prepend is not None:
-            prepend_id = prepend if isinstance(prepend, int) else self.enc.encode_single_token(prepend)
+    def encode(self, text, prepend=None, num_threads=8):
+        prepend_id = None
+        if prepend is not None:
+            prepend_id = prepend if isinstance(prepend, int) else self.enc.encode_single_token(prepend)
         if isinstance(text, str):
             ids = self.enc.encode_ordinary(text)
             if prepend is not None:
@@ -245,7 +249,7 @@ class Tokenizer:
         return self.enc.decode(ids)
 
 
-_TOKEN_BYTES_CACHE: dict = {}
+_TOKEN_BYTES_CACHE: dict[str, torch.Tensor] = {}
 
 def get_token_bytes(device="cpu"):
     key = str(device)
@@ -341,12 +345,30 @@ def make_dataloader(tokenizer, B, T, split, buffer_size=1000):
         gpu_buffer.copy_(cpu_buffer, non_blocking=True)
         yield inputs, targets, epoch
 
-# ---------------------------------------------------------------------------
-# Evaluation (DO NOT CHANGE — this is the fixed metric)
-# ---------------------------------------------------------------------------
-
-@torch.no_grad()
-def evaluate_bpb(model, tokenizer, batch_size):
+# ---------------------------------------------------------------------------
+# Evaluation (DO NOT CHANGE — this is the fixed metric)
+# ---------------------------------------------------------------------------
+
+def compute_bpb_from_totals(total_nats: torch.Tensor, total_bytes: torch.Tensor) -> torch.Tensor:
+    if int(total_bytes.item()) <= 0:
+        raise ValueError("BPB normalization requires at least one non-special token")
+    return total_nats.to(dtype=torch.float64) / (math.log(2) * total_bytes.to(dtype=torch.float64))
+
+
+def compute_bpb_from_losses(loss_flat: torch.Tensor, nbytes: torch.Tensor) -> torch.Tensor:
+    """Convert per-token losses and token byte lengths into bits-per-byte.
+
+    Tokens with zero byte length (special tokens) are excluded from both the
+    numerator and denominator so BPB remains comparable across tokenizer
+    special-token conventions.
+    """
+    mask = nbytes > 0
+    total_nats = (loss_flat * mask).sum(dtype=torch.float64)
+    total_bytes = nbytes[mask].sum(dtype=torch.int64)
+    return compute_bpb_from_totals(total_nats, total_bytes)
+
+@torch.no_grad()
+def evaluate_bpb(model, tokenizer, batch_size):
     """
     Bits per byte (BPB): vocab size-independent evaluation metric.
     Sums per-token cross-entropy (in nats), sums target byte lengths,
@@ -357,31 +379,35 @@ def evaluate_bpb(model, tokenizer, batch_size):
     Perf: accumulates on GPU (single sync at end), prefetches next batch
     while current forward runs.
     """
-    token_bytes = get_token_bytes(device="cuda")
-    val_loader = make_dataloader(tokenizer, batch_size, MAX_SEQ_LEN, "val")
-    steps = EVAL_TOKENS // (batch_size * MAX_SEQ_LEN)
+    token_bytes = get_token_bytes(device="cuda")
+    val_loader = make_dataloader(tokenizer, batch_size, MAX_SEQ_LEN, "val")
+    steps = EVAL_TOKENS // (batch_size * MAX_SEQ_LEN)
+    chunk_steps = max(1, EVAL_CHUNK_TOKENS // (batch_size * MAX_SEQ_LEN))
 
     # GPU-resident accumulators — avoid per-batch .item() sync
     total_nats_t = torch.zeros(1, device="cuda", dtype=torch.float64)
     total_bytes_t = torch.zeros(1, device="cuda", dtype=torch.int64)
 
     # Prefetch first batch
-    next_batch = next(val_loader)
-    for _ in range(steps):
-        x, y, _epoch = next_batch
-        # Prefetch NEXT batch while GPU computes current forward
-        next_batch = next(val_loader)
-        loss_flat = model(x, y, reduction='none').view(-1)
-        y_flat = y.view(-1)
-        nbytes = token_bytes[y_flat]
-        mask = nbytes > 0
-        total_nats_t += (loss_flat * mask).sum()
-        total_bytes_t += nbytes.sum()
-
-    # Single GPU→CPU sync at end
-    total_nats = total_nats_t.item()
-    total_bytes = total_bytes_t.item()
-    return total_nats / (math.log(2) * total_bytes)
+    next_batch = next(val_loader)
+    steps_done = 0
+    while steps_done < steps:
+        this_chunk = min(chunk_steps, steps - steps_done)
+        for _ in range(this_chunk):
+            x, y, _epoch = next_batch
+            # Prefetch NEXT batch while GPU computes current forward
+            next_batch = next(val_loader)
+            loss_flat = model(x, y, reduction='none').view(-1)
+            y_flat = y.view(-1)
+            nbytes = token_bytes[y_flat]
+            total_nats_t += (loss_flat * (nbytes > 0)).sum(dtype=torch.float64)
+            total_bytes_t += nbytes[nbytes > 0].sum(dtype=torch.int64)
+        steps_done += this_chunk
+        if steps_done < steps:
+            torch.cuda.empty_cache()
+
+    # Single GPU→CPU sync at end
+    return float(compute_bpb_from_totals(total_nats_t, total_bytes_t).item())
 
 # ---------------------------------------------------------------------------
 # Main

overlay/scripts/audit_overlay_sync.py

@@ -0,0 +1,100 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+
+DEFAULT_INCLUDE_PATHS = [
+    "hydra",
+    "subsystems",
+    "scripts",
+    "htm_rust",
+    "harness",
+    "configs",
+    "prepare.py",
+    "prepare_nemotron.py",
+    "train.py",
+    "pyproject.toml",
+    "uv.lock",
+]
+
+
+def _iter_files(path: Path) -> list[Path]:
+    if not path.exists():
+        return []
+    if path.is_file():
+        return [path]
+    return sorted(p for p in path.rglob("*") if p.is_file())
+
+
+def classify_overlay_pairs(*, repo_root: Path, include_paths: list[str]) -> dict[str, list[str]]:
+    overlay_root = repo_root / "hf_jobs" / "feather_h200_image" / "overlay"
+    identical: list[str] = []
+    root_ahead: list[str] = []
+    overlay_only: list[str] = []
+    missing_overlay: list[str] = []
+
+    for rel in include_paths:
+        root_path = repo_root / rel
+        overlay_path = overlay_root / rel
+
+        root_files = {p.relative_to(root_path).as_posix(): p for p in _iter_files(root_path)} if root_path.exists() and root_path.is_dir() else {}
+        overlay_files = {p.relative_to(overlay_path).as_posix(): p for p in _iter_files(overlay_path)} if overlay_path.exists() and overlay_path.is_dir() else {}
+
+        if root_path.is_file() or overlay_path.is_file():
+            rel_name = rel.replace("\\", "/")
+            if root_path.exists() and overlay_path.exists():
+                if root_path.read_bytes() == overlay_path.read_bytes():
+                    identical.append(rel_name)
+                else:
+                    root_ahead.append(rel_name)
+            elif root_path.exists():
+                missing_overlay.append(rel_name)
+            elif overlay_path.exists():
+                overlay_only.append(rel_name)
+            continue
+
+        for subrel, root_file in root_files.items():
+            rel_name = f"{rel}/{subrel}".replace("\\", "/")
+            overlay_file = overlay_files.get(subrel)
+            if overlay_file is None:
+                missing_overlay.append(rel_name)
+            elif root_file.read_bytes() == overlay_file.read_bytes():
+                identical.append(rel_name)
+            else:
+                root_ahead.append(rel_name)
+
+        for subrel in overlay_files:
+            if subrel not in root_files:
+                overlay_only.append(f"{rel}/{subrel}".replace("\\", "/"))
+
+    for bucket in (identical, root_ahead, overlay_only, missing_overlay):
+        bucket.sort()
+
+    return {
+        "identical": identical,
+        "root_ahead": root_ahead,
+        "overlay_only": overlay_only,
+        "missing_overlay": missing_overlay,
+    }
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Audit mirrored H200 overlay files against root source-of-truth paths")
+    parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[1])
+    parser.add_argument("--include-path", action="append", default=[])
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    include_paths = args.include_path or DEFAULT_INCLUDE_PATHS
+    payload = classify_overlay_pairs(repo_root=args.repo_root, include_paths=include_paths)
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/benchmark_assets.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import os
+import shutil
+from pathlib import Path
+
+from scripts.benchmark_checkpoint import checkpoint_candidates
+
+try:
+    from huggingface_hub import HfApi
+except Exception:  # pragma: no cover - optional import for offline test envs
+    HfApi = None
+
+
+def _download_file(*, repo_id: str, filename: str, local_dir: str, token: str | None, subfolder: str | None = None) -> Path:
+    from huggingface_hub import hf_hub_download
+
+    path = hf_hub_download(
+        repo_id=repo_id,
+        repo_type="model",
+        filename=filename,
+        subfolder=subfolder,
+        token=token,
+        local_dir=local_dir,
+        local_dir_use_symlinks=False,
+    )
+    return Path(path)
+
+
+def resolve_tokenizer_cache_repo(*, output_repo: str, retina_cache_repo: str) -> str:
+    return (
+        os.environ.get("HYDRA_TOKENIZER_CACHE_REPO")
+        or os.environ.get("FEATHER_HF_OUTPUT_REPO")
+        or os.environ.get("HF_REPO_ID")
+        or os.environ.get("HYDRA_RETINA_CACHE_REPO")
+        or os.environ.get("FEATHER_HF_RETINA_CACHE_REPO")
+        or output_repo
+        or retina_cache_repo
+    )
+
+
+def tokenizer_cache_prefix() -> str:
+    vocab_size = int(os.environ.get("HYDRA_VOCAB_SIZE", "65536"))
+    return f"tokenizer/vocab{vocab_size}"
+
+
+def choose_remote_checkpoint_path(files: list[str]) -> str | None:
+    preferred = [
+        path for path in files
+        if path.endswith("/pretrain_final.pt") or path.endswith("/best_bpb.pt") or path.endswith("/latest.pt")
+    ]
+    if not preferred:
+        return None
+    pretrain = sorted([p for p in preferred if p.endswith("/pretrain_final.pt")])
+    if pretrain:
+        return pretrain[-1]
+    best = sorted([p for p in preferred if p.endswith("/best_bpb.pt")])
+    if best:
+        return best[-1]
+    latest = sorted([p for p in preferred if p.endswith("/latest.pt")])
+    if latest:
+        return latest[-1]
+    return None
+
+
+def hydrate_benchmark_assets(*, cache_dir: Path, output_repo: str, tokenizer_repo: str, token: str | None) -> dict[str, str]:
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    tok_dir = cache_dir / "tokenizer"
+    tok_dir.mkdir(parents=True, exist_ok=True)
+    tok_repo = resolve_tokenizer_cache_repo(output_repo=tokenizer_repo, retina_cache_repo=tokenizer_repo)
+    tok_prefix = tokenizer_cache_prefix()
+
+    ckpt_path = None
+    for candidate in checkpoint_candidates(cache_dir):
+        if candidate.exists():
+            ckpt_path = candidate
+            break
+        try:
+            ckpt_path = _download_file(repo_id=output_repo, filename=candidate.name, local_dir=str(cache_dir), token=token)
+            break
+        except Exception:
+            continue
+    if ckpt_path is None:
+        try:
+            if HfApi is None:
+                raise RuntimeError("huggingface_hub unavailable")
+            files = HfApi(token=token).list_repo_files(repo_id=output_repo, repo_type="model", token=token)
+            remote_path = choose_remote_checkpoint_path(files)
+            if remote_path is not None:
+                parent, filename = remote_path.rsplit("/", 1)
+                downloaded_path = _download_file(
+                    repo_id=output_repo,
+                    filename=filename,
+                    local_dir=str(cache_dir),
+                    token=token,
+                    subfolder=parent,
+                )
+                canonical_path = cache_dir / filename
+                if downloaded_path != canonical_path:
+                    canonical_path.parent.mkdir(parents=True, exist_ok=True)
+                    shutil.copy2(downloaded_path, canonical_path)
+                ckpt_path = canonical_path
+        except Exception:
+            pass
+    if ckpt_path is None:
+        raise FileNotFoundError(f"No benchmark checkpoint found in cache or repo {output_repo}")
+
+    tok_path = tok_dir / "tokenizer.pkl"
+    if not tok_path.exists():
+        downloaded_tok = _download_file(repo_id=tok_repo, filename="tokenizer.pkl", local_dir=str(tok_dir), token=token, subfolder=tok_prefix)
+        if downloaded_tok != tok_path:
+            shutil.copy2(downloaded_tok, tok_path)
+
+    token_bytes_path = tok_dir / "token_bytes.pt"
+    if not token_bytes_path.exists():
+        downloaded_token_bytes = _download_file(repo_id=tok_repo, filename="token_bytes.pt", local_dir=str(tok_dir), token=token, subfolder=tok_prefix)
+        if downloaded_token_bytes != token_bytes_path:
+            shutil.copy2(downloaded_token_bytes, token_bytes_path)
+
+    return {
+        "checkpoint_path": str(ckpt_path),
+        "tokenizer_dir": str(tok_dir),
+    }

overlay/scripts/benchmark_checkpoint.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+from pathlib import Path
+
+from scripts.hf_routing import resolve_routing
+
+
+def checkpoint_candidates(cache_dir: Path) -> list[Path]:
+    return [
+        cache_dir / "best_bpb.pt",
+        cache_dir / "pretrain_final.pt",
+        cache_dir / "latest.pt",
+    ]
+
+
+def choose_checkpoint_candidate(cache_dir: Path) -> Path | None:
+    for path in checkpoint_candidates(cache_dir):
+        if path.exists():
+            return path
+    return None
+
+
+def resolve_checkpoint_source(*, cache_dir: Path, output_repo: str | None) -> dict[str, str]:
+    local = choose_checkpoint_candidate(cache_dir)
+    if local is not None:
+        return {"mode": "local", "path": str(local)}
+    if output_repo:
+        return {"mode": "remote", "repo_id": output_repo}
+    routing = resolve_routing(token=None)
+    return {"mode": "remote", "repo_id": routing.output_repo}
+
+
+def _download_checkpoint_file(*, repo_id: str, filename: str, local_dir: str, token: str | None) -> str:
+    from huggingface_hub import hf_hub_download
+
+    return hf_hub_download(
+        repo_id=repo_id,
+        repo_type="model",
+        filename=filename,
+        token=token,
+        local_dir=local_dir,
+        local_dir_use_symlinks=False,
+    )
+
+
+def hydrate_checkpoint(*, cache_dir: Path, output_repo: str | None, token: str | None) -> Path | None:
+    local = choose_checkpoint_candidate(cache_dir)
+    if local is not None:
+        return local
+    source = resolve_checkpoint_source(cache_dir=cache_dir, output_repo=output_repo)
+    if source["mode"] != "remote":
+        return None
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    for filename in ("best_bpb.pt", "pretrain_final.pt", "latest.pt"):
+        try:
+            path = Path(
+                _download_checkpoint_file(
+                    repo_id=source["repo_id"],
+                    filename=filename,
+                    local_dir=str(cache_dir),
+                    token=token,
+                )
+            )
+            if path.exists():
+                return path
+        except Exception:
+            continue
+    return None

overlay/scripts/benchmark_checkpoint_report.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+
+
+def build_checkpoint_report(files: list[str]) -> dict[str, object]:
+    by_job: dict[str, dict[str, object]] = {}
+    for path in files:
+        parts = path.split("/")
+        if len(parts) < 3 or parts[0] != "jobs":
+            continue
+        job_id = parts[1]
+        filename = parts[-1]
+        if filename not in {"best_bpb.pt", "pretrain_final.pt", "latest.pt"}:
+            continue
+        row = by_job.setdefault(job_id, {"job_id": job_id, "paths": []})
+        row["paths"].append(path)
+
+    candidates = []
+    for job_id, row in by_job.items():
+        paths = list(row["paths"])
+        preferred = None
+        for suffix in ("pretrain_final.pt", "best_bpb.pt", "latest.pt"):
+            for path in paths:
+                if path.endswith(suffix):
+                    preferred = path
+                    break
+            if preferred is not None:
+                break
+        candidates.append({
+            "job_id": job_id,
+            "preferred_path": preferred,
+            "available_paths": sorted(paths),
+        })
+
+    candidates.sort(key=lambda row: row["job_id"], reverse=True)
+    return {
+        "n_candidates": len(candidates),
+        "candidates": candidates,
+    }
+
+
+def main() -> int:
+    print(json.dumps(build_checkpoint_report([]), indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/benchmark_contract.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+
+def _require_path(payload: dict[str, Any], path: str) -> None:
+    current: Any = payload
+    for part in path.split('.'):
+        if not isinstance(current, dict) or part not in current:
+            raise ValueError(f"missing required field: {path}")
+        current = current[part]
+
+
+def validate_benchmark_contract(payload: dict[str, Any]) -> None:
+    for field in [
+        "cycle_id",
+        "hardware_class",
+        "seeds",
+        "budget_modes",
+        "coding_benchmarks.fast_iteration",
+        "coding_benchmarks.milestone",
+        "reasoning_benchmarks.fast_iteration",
+        "reasoning_benchmarks.milestone",
+        "variants.hydra_full",
+        "variants.baseline_mamba_matched",
+    ]:
+        _require_path(payload, field)
+
+    for section in [
+        payload["coding_benchmarks"]["fast_iteration"],
+        payload["coding_benchmarks"]["milestone"],
+        payload["reasoning_benchmarks"]["fast_iteration"],
+        payload["reasoning_benchmarks"]["milestone"],
+    ]:
+        if "name" not in section or "primary_metric" not in section or "decode" not in section:
+            raise ValueError("benchmark sections require name, primary_metric, and decode")
+
+    if not isinstance(payload["seeds"], list) or len(payload["seeds"]) < 3:
+        raise ValueError("seeds must contain at least three values")
+
+    if payload["variants"]["hydra_full"].get("status") != "runnable_now":
+        raise ValueError("hydra_full must be runnable_now")
+
+    if payload["variants"]["baseline_mamba_matched"].get("status") != "runnable_now":
+        raise ValueError("baseline_mamba_matched must be runnable_now")
+
+
+def load_benchmark_contract(path: Path) -> dict[str, Any]:
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict):
+        raise ValueError("benchmark contract must be a JSON object")
+    validate_benchmark_contract(payload)
+    return payload
+
+
+def main() -> int:
+    path = Path("artifacts/cycle_1_execution_freeze.json")
+    payload = load_benchmark_contract(path)
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/benchmark_datasets.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+from pathlib import Path
+
+
+CANONICAL_SUBSETS = {
+    "MBPP": Path("data/benchmarks/mbpp.cycle1.jsonl"),
+    "GSM8K": Path("data/benchmarks/gsm8k.cycle1.jsonl"),
+}
+
+
+def resolve_benchmark_dataset(benchmark_name: str, explicit_path: Path | None) -> Path:
+    if explicit_path is not None:
+        return explicit_path
+    if benchmark_name not in CANONICAL_SUBSETS:
+        raise ValueError(f"Unsupported benchmark dataset: {benchmark_name}")
+    return Path.cwd() / CANONICAL_SUBSETS[benchmark_name]

overlay/scripts/benchmark_preflight.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+from pathlib import Path
+
+from scripts.bootstrap_benchmark_env import build_bootstrap_report
+from scripts.benchmark_checkpoint import choose_checkpoint_candidate
+
+
+def build_readiness_report(*, cache_dir: Path, hf_token_present: bool, dependencies_present: bool = True, missing_dependencies: list[str] | None = None, output_repo: str | None = None, tokenizer_repo: str | None = None) -> dict[str, object]:
+    checkpoint = choose_checkpoint_candidate(cache_dir)
+    tokenizer_dir = cache_dir / "tokenizer"
+    tokenizer_ready = (tokenizer_dir / "tokenizer.pkl").exists() and (tokenizer_dir / "token_bytes.pt").exists()
+    checkpoint_present = checkpoint is not None
+    runtime = build_bootstrap_report(missing_dependencies=list(missing_dependencies or []))
+    return {
+        "cache_dir": str(cache_dir),
+        "checkpoint_present": checkpoint_present,
+        "checkpoint_path": str(checkpoint) if checkpoint is not None else None,
+        "tokenizer_ready": tokenizer_ready,
+        "hf_token_present": hf_token_present,
+        "dependencies_present": dependencies_present,
+        "missing_dependencies": list(missing_dependencies or []),
+        "install_hint": runtime["install_hint"],
+        "install_command": runtime["install_command"],
+        "install_blockers": runtime["install_blockers"],
+        "output_repo": output_repo,
+        "tokenizer_repo": tokenizer_repo,
+        "hydration_possible": bool(hf_token_present and output_repo and tokenizer_repo),
+        "ready_for_hydra_benchmarks": checkpoint_present and tokenizer_ready and dependencies_present,
+    }

overlay/scripts/benchmark_runner.py

@@ -0,0 +1,248 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+from typing import Any, Callable
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+LEDGER_TEMPLATE_PATH = REPO_ROOT / "artifacts" / "benchmark_ledger.template.json"
+
+from scripts.hydra_generation import build_hydra_generator
+from scripts.benchmark_datasets import resolve_benchmark_dataset as resolve_canonical_dataset
+from scripts.benchmark_suite import build_prompt, validate_sample
+
+
+def load_jsonl_samples(path: Path) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if line.strip():
+            rows.append(json.loads(line))
+    return rows
+
+
+def _score_mbpp(samples: list[dict[str, Any]], generate_fn: Callable[[str], str]) -> float:
+    passed = 0
+    for sample in samples:
+        validate_sample("MBPP", sample)
+        code = generate_fn(build_prompt("MBPP", sample))
+        namespace: dict[str, Any] = {}
+        exec(code, namespace, namespace)
+        for test in sample["tests"]:
+            exec(test, namespace, namespace)
+        passed += 1
+    return passed / len(samples) if samples else 0.0
+
+
+def _extract_last_number(text: str) -> str | None:
+    matches = re.findall(r"-?\d+(?:\.\d+)?", text)
+    return matches[-1] if matches else None
+
+
+def _score_gsm8k(samples: list[dict[str, Any]], generate_fn: Callable[[str], str]) -> float:
+    passed = 0
+    for sample in samples:
+        validate_sample("GSM8K", sample)
+        output = generate_fn(build_prompt("GSM8K", sample))
+        pred = _extract_last_number(output)
+        if pred is not None and pred == str(sample["answer"]):
+            passed += 1
+    return passed / len(samples) if samples else 0.0
+
+
+def _score_humaneval(samples: list[dict[str, Any]], generate_fn: Callable[[str], str]) -> float:
+    passed = 0
+    for sample in samples:
+        validate_sample("HumanEval", sample)
+        code = generate_fn(build_prompt("HumanEval", sample))
+        namespace: dict[str, Any] = {}
+        exec(code, namespace, namespace)
+        exec(sample["test"], namespace, namespace)
+        passed += 1
+    return passed / len(samples) if samples else 0.0
+
+
+def _score_arc(samples: list[dict[str, Any]], generate_fn: Callable[[str], str]) -> float:
+    passed = 0
+    for sample in samples:
+        validate_sample("ARC-Challenge", sample)
+        output = generate_fn(build_prompt("ARC-Challenge", sample)).strip()
+        if output == str(sample["answer"]):
+            passed += 1
+    return passed / len(samples) if samples else 0.0
+
+
+def run_benchmark(benchmark_name: str, path: Path, generate_fn: Callable[[str], str]) -> dict[str, Any]:
+    samples = load_jsonl_samples(path)
+    if benchmark_name == "MBPP":
+        return {
+            "benchmark": "MBPP",
+            "primary_metric": "pass_at_1",
+            "score": _score_mbpp(samples, generate_fn),
+            "n_samples": len(samples),
+        }
+    if benchmark_name == "GSM8K":
+        return {
+            "benchmark": "GSM8K",
+            "primary_metric": "exact_match",
+            "score": _score_gsm8k(samples, generate_fn),
+            "n_samples": len(samples),
+        }
+    if benchmark_name == "HumanEval":
+        return {
+            "benchmark": "HumanEval",
+            "primary_metric": "pass_at_1",
+            "score": _score_humaneval(samples, generate_fn),
+            "n_samples": len(samples),
+        }
+    if benchmark_name == "ARC-Challenge":
+        return {
+            "benchmark": "ARC-Challenge",
+            "primary_metric": "accuracy",
+            "score": _score_arc(samples, generate_fn),
+            "n_samples": len(samples),
+        }
+    raise ValueError(f"Unsupported runnable benchmark: {benchmark_name}")
+
+
+def write_benchmark_result(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+
+
+def append_benchmark_run_record(
+    ledger_path: Path,
+    result: dict[str, Any],
+    *,
+    benchmark_name: str,
+    variant: str,
+    seed: int,
+    samples_path: Path,
+) -> None:
+    if not ledger_path.exists():
+        ledger_path.parent.mkdir(parents=True, exist_ok=True)
+        ledger_path.write_text(LEDGER_TEMPLATE_PATH.read_text(encoding="utf-8"), encoding="utf-8")
+    payload = json.loads(ledger_path.read_text(encoding="utf-8"))
+    run_records = payload.setdefault("run_records", [])
+    if len(run_records) == 1 and run_records[0].get("run_id") == "example-run-0001":
+        run_records.clear()
+    run_records.append(
+        {
+            "run_id": result.get("run_id", f"{benchmark_name.lower()}-{seed}"),
+            "commit": "HEAD",
+            "model_family": "hydra",
+            "variant": variant,
+            "seed": seed,
+            "hardware": {
+                "hardware_class": payload.get("benchmark_cycle", {}).get("hardware_class", "unknown"),
+            },
+            "budget": {
+                "budget_mode": payload.get("benchmark_cycle", {}).get("budget_modes", [None])[0],
+            },
+            "capability": {
+                "coding_score": result["score"] if benchmark_name in {"MBPP", "HumanEval"} else None,
+                "reasoning_score": result["score"] if benchmark_name in {"GSM8K", "ARC-Challenge"} else None,
+            },
+            "artifacts": {
+                "samples_path": str(samples_path),
+            },
+        }
+    )
+    ledger_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+
+
+def resolve_samples_path(benchmark_name: str, samples: Path | None, suite_path: Path) -> Path:
+    if samples is not None:
+        return samples
+    payload = json.loads(suite_path.read_text(encoding="utf-8"))
+    for section in ("coding_benchmarks", "reasoning_benchmarks"):
+        if section not in payload:
+            continue
+        for slot in ("fast_iteration", "milestone"):
+            entry = payload[section].get(slot)
+            if isinstance(entry, dict) and entry.get("name") == benchmark_name and "sample_path" in entry:
+                return Path(entry["sample_path"])
+    try:
+        return resolve_canonical_dataset(benchmark_name, None)
+    except ValueError:
+        raise ValueError(f"No sample path found for benchmark: {benchmark_name}")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run a local benchmark against JSONL samples")
+    parser.add_argument("--benchmark", required=True, choices=["MBPP", "GSM8K", "HumanEval", "ARC-Challenge"])
+    parser.add_argument("--samples", type=Path)
+    parser.add_argument("--suite", type=Path, default=REPO_ROOT / "artifacts" / "benchmark_suite.cycle1.json")
+    parser.add_argument("--out", type=Path)
+    parser.add_argument("--ledger", type=Path)
+    parser.add_argument("--variant", default="hydra_full")
+    parser.add_argument("--seed", type=int, default=42)
+    parser.add_argument("--generator-mode", choices=["stub", "hydra"], default="stub")
+    parser.add_argument("--checkpoint", type=Path)
+    parser.add_argument("--device")
+    parser.add_argument("--max-new-tokens", type=int, default=256)
+    parser.add_argument("--temperature", type=float, default=0.2)
+    parser.add_argument("--top-p", type=float, default=0.95)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    sample_path = resolve_samples_path(args.benchmark, args.samples, args.suite)
+    try:
+        if args.generator_mode == "hydra":
+            generator = build_hydra_generator(
+                checkpoint_path=args.checkpoint,
+                device=args.device,
+                max_new_tokens=args.max_new_tokens,
+                temperature=args.temperature,
+                top_p=args.top_p,
+            )
+        else:
+            def generator(prompt: str) -> str:
+                return prompt
+
+        result = run_benchmark(args.benchmark, sample_path, generator)
+        exit_code = 0
+    except FileNotFoundError as exc:
+        result = {
+            "benchmark": args.benchmark,
+            "status": "failed",
+            "failure_type": "missing_checkpoint",
+            "error": str(exc),
+            "n_samples": 0,
+        }
+        exit_code = 1
+    except Exception as exc:  # noqa: BLE001
+        result = {
+            "benchmark": args.benchmark,
+            "status": "failed",
+            "failure_type": type(exc).__name__,
+            "error": str(exc),
+            "n_samples": 0,
+        }
+        exit_code = 1
+
+    if args.out is not None:
+        write_benchmark_result(args.out, result)
+    if args.ledger is not None and exit_code == 0:
+        append_benchmark_run_record(
+            args.ledger,
+            result,
+            benchmark_name=args.benchmark,
+            variant=args.variant,
+            seed=args.seed,
+            samples_path=sample_path,
+        )
+    print(json.dumps(result, indent=2, sort_keys=True))
+    return exit_code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/benchmark_suite.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+@dataclass(frozen=True)
+class BenchmarkSpec:
+    name: str
+    family: str
+    required_fields: tuple[str, ...]
+
+
+REGISTRY: dict[str, BenchmarkSpec] = {
+    "MBPP": BenchmarkSpec("MBPP", "coding", ("task_id", "prompt", "tests")),
+    "HumanEval": BenchmarkSpec("HumanEval", "coding", ("task_id", "prompt", "test")),
+    "GSM8K": BenchmarkSpec("GSM8K", "reasoning", ("question", "answer")),
+    "ARC-Challenge": BenchmarkSpec("ARC-Challenge", "reasoning", ("question", "choices", "answer")),
+}
+
+
+def validate_sample(benchmark_name: str, sample: dict[str, Any]) -> None:
+    spec = REGISTRY[benchmark_name]
+    for field in spec.required_fields:
+        if field not in sample:
+            raise ValueError(f"{benchmark_name} sample missing required field: {field}")
+
+
+def build_prompt(benchmark_name: str, sample: dict[str, Any]) -> str:
+    validate_sample(benchmark_name, sample)
+    if benchmark_name == "MBPP":
+        tests = sample["tests"]
+        rendered_tests = "\n".join(str(t) for t in tests)
+        return (
+            "Write a Python function that solves the task below.\n\n"
+            f"Task:\n{sample['prompt']}\n\n"
+            f"Tests:\n{rendered_tests}\n"
+        )
+    if benchmark_name == "HumanEval":
+        return (
+            "Complete the following Python function exactly as specified.\n\n"
+            f"Prompt:\n{sample['prompt']}\n\n"
+            f"Reference test:\n{sample['test']}\n"
+        )
+    if benchmark_name == "GSM8K":
+        return f"Solve the following math word problem. Return only the final answer.\n\nQuestion: {sample['question']}\n"
+    if benchmark_name == "ARC-Challenge":
+        choices = sample["choices"]
+        rendered_choices = "\n".join(f"- {choice}" for choice in choices)
+        return (
+            "Answer the following multiple-choice science question. Return only the correct option text or label.\n\n"
+            f"Question: {sample['question']}\nChoices:\n{rendered_choices}\n"
+        )
+    raise ValueError(f"Unknown benchmark: {benchmark_name}")
+
+
+def load_cycle_benchmark_suite(path: Path) -> dict[str, dict[str, BenchmarkSpec]]:
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    out: dict[str, dict[str, BenchmarkSpec]] = {"coding_benchmarks": {}, "reasoning_benchmarks": {}}
+    for section in ("coding_benchmarks", "reasoning_benchmarks"):
+        if section not in payload:
+            raise ValueError(f"missing benchmark section: {section}")
+        for slot in ("fast_iteration", "milestone"):
+            if slot not in payload[section]:
+                raise ValueError(f"missing benchmark slot: {section}.{slot}")
+            name = payload[section][slot]["name"]
+            if name not in REGISTRY:
+                raise ValueError(f"unsupported benchmark: {name}")
+            out[section][slot] = REGISTRY[name]
+    return out
+
+
+def main() -> int:
+    path = Path("artifacts/benchmark_suite.cycle1.json")
+    suite = load_cycle_benchmark_suite(path)
+    print(json.dumps({k: {slot: spec.name for slot, spec in section.items()} for k, section in suite.items()}, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/bootstrap_benchmark_env.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+import shutil
+
+import torch
+
+
+PACKAGE_MAP = {
+    "mamba_ssm": "mamba-ssm",
+    "transformers": "transformers",
+}
+
+
+def build_install_command(*, missing_dependencies: list[str]) -> list[str]:
+    packages = [PACKAGE_MAP.get(name, name) for name in missing_dependencies]
+    return [] if not packages else ["python", "-m", "pip", "install", *packages]
+
+
+def diagnose_install_blockers(
+    *,
+    missing_dependencies: list[str],
+    torch_version: str,
+    cuda_available: bool,
+    nvcc_present: bool,
+) -> list[str]:
+    blockers: list[str] = []
+    if "mamba_ssm" in missing_dependencies:
+        if "+cpu" in torch_version or not cuda_available:
+            blockers.append("mamba_ssm install likely blocked by CPU-only torch runtime")
+        if not nvcc_present:
+            blockers.append("mamba_ssm install likely blocked because nvcc is unavailable")
+    return blockers
+
+
+def build_bootstrap_report(*, missing_dependencies: list[str]) -> dict[str, object]:
+    ready = len(missing_dependencies) == 0
+    packages = [PACKAGE_MAP.get(name, name) for name in missing_dependencies]
+    install_hint = "" if ready else f"Install missing benchmark dependencies: {', '.join(packages)}"
+    blockers = diagnose_install_blockers(
+        missing_dependencies=missing_dependencies,
+        torch_version=getattr(torch, "__version__", "unknown"),
+        cuda_available=torch.cuda.is_available(),
+        nvcc_present=shutil.which("nvcc") is not None,
+    )
+    return {
+        "ready": ready,
+        "missing_dependencies": list(missing_dependencies),
+        "install_hint": install_hint,
+        "install_command": build_install_command(missing_dependencies=missing_dependencies),
+        "install_blockers": blockers,
+    }
+
+
+def main() -> int:
+    report = build_bootstrap_report(missing_dependencies=["mamba_ssm"])
+    print(json.dumps(report, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/bootstrap_benchmark_runtime.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import importlib.util
+import json
+
+
+PACKAGE_MAP = {
+    "mamba_ssm": "mamba-ssm",
+    "transformers": "transformers",
+}
+
+
+def detect_missing_modules(required: list[str] | None = None) -> list[str]:
+    names = required or list(PACKAGE_MAP)
+    return [name for name in names if importlib.util.find_spec(name) is None]
+
+
+def build_install_command(*, missing_modules: list[str]) -> list[str]:
+    packages = [PACKAGE_MAP[name] for name in missing_modules if name in PACKAGE_MAP]
+    if not packages:
+        return []
+    return ["python", "-m", "pip", "install", *packages]
+
+
+def build_runtime_report(*, missing_modules: list[str]) -> dict[str, object]:
+    return {
+        "ready": len(missing_modules) == 0,
+        "missing_modules": list(missing_modules),
+        "packages": {name: PACKAGE_MAP[name] for name in missing_modules if name in PACKAGE_MAP},
+        "install_command": build_install_command(missing_modules=missing_modules),
+    }
+
+
+def main() -> int:
+    missing = detect_missing_modules()
+    print(json.dumps(build_runtime_report(missing_modules=missing), indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/cycle_executor.py

@@ -0,0 +1,312 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import importlib
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+from scripts.benchmark_preflight import build_readiness_report
+from scripts.hf_routing import resolve_routing
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+FREEZE_PATH = REPO_ROOT / "artifacts" / "cycle_1_execution_freeze.json"
+RUNNER_PATH = REPO_ROOT / "scripts" / "benchmark_runner.py"
+
+
+def active_hf_token() -> str | None:
+    token = os.environ.get("HF_TOKEN")
+    if token:
+        return token
+    try:
+        from huggingface_hub.utils import get_token
+        return get_token()
+    except Exception:
+        return None
+
+
+def missing_benchmark_dependencies() -> list[str]:
+    required = ["mamba_ssm", "transformers"]
+    missing: list[str] = []
+    for name in required:
+        try:
+            spec = importlib.util.find_spec(name)
+        except (ImportError, ValueError):
+            spec = None
+        if spec is None:
+            try:
+                importlib.import_module(name)
+            except Exception:
+                missing.append(name)
+    return missing
+
+
+def load_cycle_freeze(path: Path) -> dict[str, Any]:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def load_cycle_benchmarks(path: Path) -> list[str]:
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    out: list[str] = []
+    for section in ("coding_benchmarks", "reasoning_benchmarks"):
+        for slot in ("fast_iteration", "milestone"):
+            entry = payload.get(section, {}).get(slot)
+            if isinstance(entry, dict) and entry.get("name"):
+                out.append(str(entry["name"]))
+    return out
+
+
+def build_preflight_report(
+    *,
+    cache_dir: Path,
+    output_repo: str | None = None,
+    tokenizer_repo: str | None = None,
+) -> dict[str, object]:
+    return build_readiness_report(
+        cache_dir=cache_dir,
+        hf_token_present=bool(active_hf_token()),
+        dependencies_present=not bool(missing_benchmark_dependencies()),
+        missing_dependencies=missing_benchmark_dependencies(),
+        output_repo=output_repo,
+        tokenizer_repo=tokenizer_repo,
+    )
+
+
+def write_preflight_report(path: Path, payload: dict[str, object]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+
+
+def write_cycle_summary(path: Path, payload: list[dict[str, Any]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+
+
+def build_remote_checkpoint_report(output_repo: str, token: str | None) -> dict[str, Any]:
+    from huggingface_hub import HfApi
+
+    from scripts.benchmark_checkpoint_report import build_checkpoint_report
+
+    files = HfApi(token=token).list_repo_files(repo_id=output_repo, repo_type="model", token=token)
+    return build_checkpoint_report(files)
+
+
+def ensure_benchmark_assets(
+    *,
+    cache_dir: Path,
+    output_repo: str,
+    tokenizer_repo: str,
+    token: str | None,
+    hydrate: bool,
+) -> dict[str, str] | None:
+    if not hydrate:
+        return None
+    from scripts.benchmark_assets import hydrate_benchmark_assets
+
+    return hydrate_benchmark_assets(
+        cache_dir=cache_dir,
+        output_repo=output_repo,
+        tokenizer_repo=tokenizer_repo,
+        token=token,
+    )
+
+
+def build_benchmark_command(
+    freeze: dict[str, Any],
+    *,
+    benchmark: str,
+    variant: str,
+    seed: int,
+    out_dir: Path,
+) -> tuple[list[str], dict[str, str]]:
+    variant_cfg = freeze["variants"][variant]
+    env = os.environ.copy()
+    env.update({str(k): str(v) for k, v in variant_cfg.get("env", {}).items()})
+    env["HYDRA_SEED"] = str(seed)
+
+    out_dir.mkdir(parents=True, exist_ok=True)
+    result_path = out_dir / f"{benchmark.lower()}_{variant}_seed{seed}.json"
+    ledger_path = out_dir / "benchmark_ledger.json"
+    cmd = [
+        sys.executable,
+        str(RUNNER_PATH),
+        "--benchmark",
+        benchmark,
+        "--generator-mode",
+        "hydra",
+        "--out",
+        str(result_path),
+        "--ledger",
+        str(ledger_path),
+        "--variant",
+        variant,
+        "--seed",
+        str(seed),
+    ]
+    return cmd, env
+
+
+def build_cycle_plan(freeze: dict[str, Any], *, benchmark: str, out_dir: Path) -> list[dict[str, Any]]:
+    runnable_variants = [
+        name for name, cfg in freeze.get("variants", {}).items()
+        if isinstance(cfg, dict) and cfg.get("status") == "runnable_now"
+    ]
+    seeds = [int(seed) for seed in freeze.get("seeds", [])]
+    plan: list[dict[str, Any]] = []
+    for variant in runnable_variants:
+        for seed in seeds:
+            cmd, env = build_benchmark_command(
+                freeze,
+                benchmark=benchmark,
+                variant=variant,
+                seed=seed,
+                out_dir=out_dir,
+            )
+            plan.append({
+                "benchmark": benchmark,
+                "variant": variant,
+                "seed": seed,
+                "command": cmd,
+                "env": env,
+            })
+    return plan
+
+
+def execute_cycle_plan(plan: list[dict[str, Any]], *, repo_root: Path) -> list[dict[str, Any]]:
+    results: list[dict[str, Any]] = []
+    for item in plan:
+        proc = subprocess.run(item["command"], cwd=str(repo_root), env=item["env"])
+        results.append(
+            {
+                "benchmark": item["benchmark"],
+                "variant": item["variant"],
+                "seed": item["seed"],
+                "returncode": proc.returncode,
+            }
+        )
+    return results
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Execute a frozen Cycle 1 benchmark run")
+    parser.add_argument("--freeze", type=Path, default=FREEZE_PATH)
+    parser.add_argument("--suite", type=Path, default=REPO_ROOT / "artifacts" / "benchmark_suite.cycle1.json")
+    parser.add_argument("--benchmark", required=True)
+    parser.add_argument("--variant", required=True)
+    parser.add_argument("--seed", type=int, required=True)
+    parser.add_argument("--out-dir", type=Path, default=REPO_ROOT / "artifacts" / "runs")
+    parser.add_argument("--preflight-out", type=Path)
+    parser.add_argument("--summary-out", type=Path)
+    parser.add_argument("--hydrate-assets", action="store_true")
+    parser.add_argument("--all-runnable", action="store_true")
+    parser.add_argument("--all-benchmarks", action="store_true")
+    parser.add_argument("--require-ready", action="store_true")
+    parser.add_argument("--output-repo")
+    parser.add_argument("--tokenizer-repo")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    cache_dir = Path(os.path.expanduser("~/.cache/autoresearch"))
+    report = None
+    token = active_hf_token()
+    routing = resolve_routing(token=token)
+    output_repo = args.output_repo or routing.output_repo
+    tokenizer_repo = args.tokenizer_repo or routing.output_repo
+    if args.hydrate_assets:
+        try:
+            ensure_benchmark_assets(
+                cache_dir=cache_dir,
+                output_repo=output_repo,
+                tokenizer_repo=tokenizer_repo,
+                token=token,
+                hydrate=True,
+            )
+        except FileNotFoundError as exc:
+            checkpoint_report = None
+            try:
+                checkpoint_report = build_remote_checkpoint_report(output_repo, token)
+            except Exception:
+                checkpoint_report = None
+            if args.summary_out is not None:
+                write_cycle_summary(
+                    args.summary_out,
+                    [{
+                        "status": "blocked",
+                        "reason": "asset_hydration_failed",
+                        "error": str(exc),
+                        "checkpoint_candidates": checkpoint_report,
+                    }],
+                )
+            return 3
+    if args.preflight_out is not None:
+        report = build_preflight_report(
+            cache_dir=cache_dir,
+            output_repo=output_repo,
+            tokenizer_repo=tokenizer_repo,
+        )
+        write_preflight_report(args.preflight_out, report)
+    if args.require_ready:
+        if report is None:
+            report = build_preflight_report(
+                cache_dir=cache_dir,
+                output_repo=output_repo,
+                tokenizer_repo=tokenizer_repo,
+            )
+        if not bool(report.get("ready_for_hydra_benchmarks")):
+            checkpoint_report = None
+            try:
+                checkpoint_report = build_remote_checkpoint_report(output_repo, token)
+            except Exception:
+                checkpoint_report = None
+            if args.summary_out is not None:
+                write_cycle_summary(
+                    args.summary_out,
+                    [{
+                        "status": "blocked",
+                        "reason": "preflight_not_ready",
+                        "preflight": report,
+                        "checkpoint_candidates": checkpoint_report,
+                    }],
+                )
+            return 2
+    freeze = load_cycle_freeze(args.freeze)
+    if args.all_runnable:
+        benchmarks = load_cycle_benchmarks(args.suite) if args.all_benchmarks else [args.benchmark]
+        plan = []
+        for benchmark in benchmarks:
+            plan.extend(build_cycle_plan(freeze, benchmark=benchmark, out_dir=args.out_dir))
+        results = execute_cycle_plan(plan, repo_root=REPO_ROOT)
+        if args.summary_out is not None:
+            write_cycle_summary(args.summary_out, results)
+        return 0 if all(item["returncode"] == 0 for item in results) else 1
+    cmd, env = build_benchmark_command(
+        freeze,
+        benchmark=args.benchmark,
+        variant=args.variant,
+        seed=args.seed,
+        out_dir=args.out_dir,
+    )
+    proc = subprocess.run(cmd, cwd=str(REPO_ROOT), env=env)
+    if args.summary_out is not None:
+        write_cycle_summary(
+            args.summary_out,
+            [{
+                "benchmark": args.benchmark,
+                "variant": args.variant,
+                "seed": args.seed,
+                "returncode": proc.returncode,
+            }],
+        )
+    return proc.returncode
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/export_hpo_priors.py

@@ -9,6 +9,8 @@ from typing import Any
 
 import optuna
 
+from scripts.hpo_leaderboard import build_leaderboard
+
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Export top Optuna trials as transfer-learning priors")
@@ -35,38 +37,56 @@ def _serialize_trial(trial: optuna.trial.FrozenTrial) -> dict[str, Any]:
     }
 
 
-def main() -> int:
-    args = parse_args()
-    study_names = args.study_name or ["hydra_hpo"]
-    merged_trials: list[dict[str, Any]] = []
-    total_trials = 0
-    total_completed = 0
-
-    for study_name in study_names:
-        study = optuna.load_study(study_name=study_name, storage=args.storage)
-        ranked = _completed_trials(study)
-        selected = ranked[: max(0, args.top_k)]
-        total_trials += len(study.trials)
-        total_completed += len(ranked)
-        for t in selected:
-            row = _serialize_trial(t)
-            row["study_name"] = study_name
-            merged_trials.append(row)
-
-    payload = {
-        "schema_version": 1,
+def collect_prior_trials(*, storage: str, study_names: list[str], top_k: int, metric: str) -> dict[str, Any]:
+    leaderboard = build_leaderboard(storage=storage, study_names=study_names, metric=metric)
+    selected = leaderboard["clean_trials"][: max(0, top_k)]
+    trials = [
+        {
+            "study_name": row["study_name"],
+            "trial_number": row["trial_number"],
+            "value": row["value"],
+            "params": row["params"],
+            "user_attrs": row["user_attrs"],
+        }
+        for row in selected
+    ]
+    quarantined = [
+        {
+            "study_name": row["study_name"],
+            "trial_number": row["trial_number"],
+            "value": row["value"],
+            "params": row["params"],
+            "user_attrs": row["user_attrs"],
+            "contamination_reason": row["contamination_reason"],
+        }
+        for row in leaderboard["contaminated_trials"]
+    ]
+    return {
+        "schema_version": 2,
         "generated_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
         "study_names": study_names,
-        "metric": args.metric,
-        "n_total_trials": total_trials,
-        "n_completed_trials": total_completed,
-        "top_k_per_study": args.top_k,
-        "trials": merged_trials,
+        "metric": metric,
+        "n_total_trials": sum(int(s["n_trials"]) for s in leaderboard["studies"]),
+        "n_completed_trials": sum(int(s["n_completed"]) for s in leaderboard["studies"]),
+        "n_exported_trials": len(trials),
+        "n_quarantined_trials": len(quarantined),
+        "top_k": top_k,
+        "trials": trials,
+        "quarantined_trials": quarantined,
     }
 
+
+def main() -> int:
+    args = parse_args()
+    study_names = args.study_name or ["hydra_hpo"]
+    payload = collect_prior_trials(storage=args.storage, study_names=study_names, top_k=args.top_k, metric=args.metric)
+
     args.out.parent.mkdir(parents=True, exist_ok=True)
     args.out.write_text(json.dumps(payload, indent=2), encoding="utf-8")
-    print(f"[hpo-priors] wrote {args.out} with {len(merged_trials)} merged trials")
+    print(
+        f"[hpo-priors] wrote {args.out} with {payload['n_exported_trials']} clean trials "
+        f"({payload['n_quarantined_trials']} quarantined)"
+    )
     return 0
 
 

overlay/scripts/hpo_component_report.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+import math
+from collections import defaultdict
+from pathlib import Path
+from typing import Any
+
+from scripts.hpo_leaderboard import build_leaderboard
+
+
+_COMPONENT_KEYS = [
+    "engram_subsample",
+    "htm_subsample",
+    "htm_learn_every",
+    "engram_n_columns",
+    "engram_layer_idx",
+    "sdr_target_active",
+    "mamba3_chunk",
+    "dropout",
+    "hyena_layers",
+]
+
+
+def _recover_params(row: dict[str, Any]) -> dict[str, Any]:
+    params = dict(row.get("params") or {})
+    attrs = row.get("user_attrs") or {}
+    for key, value in attrs.items():
+        if key.startswith("param_"):
+            params.setdefault(key.removeprefix("param_"), value)
+    return params
+
+
+def _pearson(xs: list[float], ys: list[float]) -> float | None:
+    if len(xs) < 2 or len(xs) != len(ys):
+        return None
+    mean_x = sum(xs) / len(xs)
+    mean_y = sum(ys) / len(ys)
+    cov = sum((x - mean_x) * (y - mean_y) for x, y in zip(xs, ys))
+    var_x = sum((x - mean_x) ** 2 for x in xs)
+    var_y = sum((y - mean_y) ** 2 for y in ys)
+    if var_x <= 0 or var_y <= 0:
+        return None
+    return cov / math.sqrt(var_x * var_y)
+
+
+def build_component_report(*, storage: str, study_names: list[str], metric: str = "val_bpb") -> dict[str, Any]:
+    leaderboard = build_leaderboard(storage=storage, study_names=study_names, metric=metric)
+    clean_trials = leaderboard["clean_trials"]
+
+    ablations: dict[str, list[dict[str, Any]]] = {}
+    numeric_correlations: list[dict[str, Any]] = []
+
+    for key in _COMPONENT_KEYS:
+        grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+        numeric_x: list[float] = []
+        metric_y: list[float] = []
+        tps_y: list[float] = []
+        for row in clean_trials:
+            params = _recover_params(row)
+            if key not in params:
+                continue
+            value = params[key]
+            grouped[str(value)].append({"value": value, "metric": float(row["value"]), "tps": row.get("tps")})
+            if isinstance(value, (int, float)) and isinstance(row.get("tps"), (int, float)):
+                numeric_x.append(float(value))
+                metric_y.append(float(row["value"]))
+                tps_y.append(float(row["tps"]))
+
+        rows: list[dict[str, Any]] = []
+        for grouped_rows in grouped.values():
+            value = grouped_rows[0]["value"]
+            metric_vals = [r["metric"] for r in grouped_rows]
+            tps_vals = [float(r["tps"]) for r in grouped_rows if isinstance(r["tps"], (int, float))]
+            rows.append({
+                "value": value,
+                "n_trials": len(grouped_rows),
+                "mean_metric": sum(metric_vals) / len(metric_vals),
+                "mean_tps": (sum(tps_vals) / len(tps_vals)) if tps_vals else None,
+            })
+        if rows:
+            rows.sort(key=lambda row: str(row["value"]))
+            ablations[key] = rows
+
+        pearson_metric = _pearson(numeric_x, metric_y)
+        pearson_tps = _pearson(numeric_x, tps_y)
+        if pearson_metric is not None or pearson_tps is not None:
+            numeric_correlations.append({
+                "param": key,
+                "pearson_with_metric": pearson_metric,
+                "pearson_with_tps": pearson_tps,
+                "n_points": len(numeric_x),
+            })
+
+    numeric_correlations.sort(key=lambda row: row["param"])
+    return {
+        "schema_version": 1,
+        "generated_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
+        "metric": metric,
+        "study_names": study_names,
+        "n_clean_trials": len(clean_trials),
+        "component_ablations": ablations,
+        "numeric_correlations": numeric_correlations,
+    }
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Build component ablation and correlation report from clean HPO trials")
+    parser.add_argument("--storage", default="sqlite:///optuna_hpo.db")
+    parser.add_argument("--study-name", action="append", default=[])
+    parser.add_argument("--metric", default="val_bpb")
+    parser.add_argument("--out", type=Path, default=Path(".tmp") / "optuna" / "component_report.json")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    study_names = args.study_name or ["hydra_hpo"]
+    payload = build_component_report(storage=args.storage, study_names=study_names, metric=args.metric)
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/hpo_leaderboard.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+from pathlib import Path
+from typing import Any
+
+import optuna
+
+
+def _trial_direction(study: optuna.Study) -> str:
+    return "maximize" if study.direction == optuna.study.StudyDirection.MAXIMIZE else "minimize"
+
+
+def _contamination_reason(trial: optuna.trial.FrozenTrial, metric: str) -> str | None:
+    if trial.value is None:
+        return "missing_value"
+    attrs = trial.user_attrs
+    source = attrs.get("objective_source")
+    eval_status = attrs.get("eval_status")
+    objective_metric = attrs.get("objective_metric")
+
+    if source in {"train_log_fallback", "missing_metric", "missing_metrics", "missing_final_val"}:
+        return f"objective_source={source}"
+    if eval_status not in {None, "completed"}:
+        return f"eval_status={eval_status}"
+    if objective_metric not in {None, metric}:
+        return f"objective_metric={objective_metric}"
+    return None
+
+
+def _serialize_trial(study_name: str, trial: optuna.trial.FrozenTrial, metric: str) -> dict[str, Any]:
+    attrs = dict(trial.user_attrs)
+    source = attrs.get("objective_source") or "legacy_completed_value"
+    row = {
+        "study_name": study_name,
+        "trial_number": trial.number,
+        "value": float(trial.value) if trial.value is not None else None,
+        "metric": metric,
+        "objective_source": source,
+        "objective_metric": attrs.get("objective_metric", metric),
+        "eval_status": attrs.get("eval_status"),
+        "hf_job_id": attrs.get("hf_job_id"),
+        "tps": attrs.get("tps"),
+        "params": dict(trial.params),
+        "user_attrs": attrs,
+    }
+    reason = _contamination_reason(trial, metric)
+    if reason is not None:
+        row["contamination_reason"] = reason
+    return row
+
+
+def _is_pareto_dominated(candidate: dict[str, Any], peers: list[dict[str, Any]]) -> bool:
+    candidate_value = float(candidate["value"])
+    candidate_tps = float(candidate["tps"])
+    for peer in peers:
+        if peer is candidate or peer.get("tps") is None:
+            continue
+        peer_value = float(peer["value"])
+        peer_tps = float(peer["tps"])
+        no_worse = peer_value <= candidate_value and peer_tps >= candidate_tps
+        strictly_better = peer_value < candidate_value or peer_tps > candidate_tps
+        if no_worse and strictly_better:
+            return True
+    return False
+
+
+def _annotate_pareto(clean_trials: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    pareto_trials: list[dict[str, Any]] = []
+    comparable = [row for row in clean_trials if row.get("tps") is not None]
+    for row in clean_trials:
+        if row.get("tps") is None:
+            row["pareto_frontier"] = False
+            row["pareto_dominated"] = None
+            row["pareto_reason"] = "missing_tps"
+            continue
+        dominated = _is_pareto_dominated(row, comparable)
+        row["pareto_frontier"] = not dominated
+        row["pareto_dominated"] = dominated
+        row["pareto_reason"] = "frontier" if not dominated else "dominated"
+        if not dominated:
+            pareto_trials.append(row)
+    pareto_trials.sort(key=lambda row: (float(row["value"]), -float(row["tps"])))
+    return pareto_trials
+
+
+def build_leaderboard(*, storage: str, study_names: list[str], metric: str = "val_bpb") -> dict[str, Any]:
+    clean_trials: list[dict[str, Any]] = []
+    contaminated_trials: list[dict[str, Any]] = []
+    study_summaries: list[dict[str, Any]] = []
+    direction = "minimize"
+
+    for study_name in study_names:
+        study = optuna.load_study(study_name=study_name, storage=storage)
+        direction = _trial_direction(study)
+        completed = [t for t in study.trials if t.value is not None]
+        study_summaries.append({
+            "study_name": study_name,
+            "direction": direction,
+            "n_trials": len(study.trials),
+            "n_completed": len(completed),
+        })
+        for trial in completed:
+            row = _serialize_trial(study_name, trial, metric)
+            if "contamination_reason" in row:
+                contaminated_trials.append(row)
+            else:
+                clean_trials.append(row)
+
+    reverse = direction == "maximize"
+    clean_trials.sort(key=lambda row: float(row["value"]), reverse=reverse)
+    contaminated_trials.sort(key=lambda row: float(row["value"]), reverse=reverse)
+    pareto_trials = _annotate_pareto(clean_trials)
+
+    return {
+        "schema_version": 1,
+        "generated_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
+        "metric": metric,
+        "direction": direction,
+        "study_names": study_names,
+        "studies": study_summaries,
+        "n_clean_trials": len(clean_trials),
+        "n_contaminated_trials": len(contaminated_trials),
+        "pareto_metric_x": metric,
+        "pareto_metric_y": "tps",
+        "n_pareto_trials": len(pareto_trials),
+        "clean_trials": clean_trials,
+        "contaminated_trials": contaminated_trials,
+        "pareto_trials": pareto_trials,
+    }
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Build a clean Optuna HPO leaderboard")
+    parser.add_argument("--storage", default="sqlite:///optuna_hpo.db")
+    parser.add_argument("--study-name", action="append", default=[], help="Repeat to merge multiple studies")
+    parser.add_argument("--metric", default="val_bpb")
+    parser.add_argument("--out", type=Path, default=Path(".tmp") / "optuna" / "leaderboard.json")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    study_names = args.study_name or ["hydra_hpo"]
+    payload = build_leaderboard(storage=args.storage, study_names=study_names, metric=args.metric)
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/hpo_retest.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+from pathlib import Path
+from typing import Any
+
+import optuna
+
+from scripts.hpo_leaderboard import build_leaderboard
+
+
+_PARAM_TO_ENV = {
+    "d_model": "HYDRA_D_MODEL",
+    "n_layer": "HYDRA_N_LAYER",
+    "d_state": "HYDRA_D_STATE",
+    "headdim": "HYDRA_HEADDIM",
+    "expand": "HYDRA_EXPAND",
+    "seq_len": "HYDRA_SEQ_LEN",
+    "batch_size": "HYDRA_BATCH_SIZE",
+    "matrix_lr": "HYDRA_MATRIX_LR",
+    "embed_lr": "HYDRA_EMBED_LR",
+    "unembed_lr": "HYDRA_UNEMBED_LR",
+    "engram_n_columns": "HYDRA_ENGRAM_N_COLUMNS",
+    "engram_layer_idx": "HYDRA_ENGRAM_LAYER_IDX",
+    "sdr_target_active": "HYDRA_SDR_TARGET_ACTIVE",
+    "htm_learn_every": "HYDRA_HTM_LEARN_EVERY",
+    "htm_subsample": "HYDRA_HTM_SUBSAMPLE",
+    "engram_subsample": "HYDRA_ENGRAM_SUBSAMPLE",
+    "mamba3_chunk": "HYDRA_MAMBA3_CHUNK",
+    "dropout": "HYDRA_DROPOUT",
+}
+
+_DEFAULT_ENV = {
+    "HYDRA_USE_NEMOTRON": "1",
+    "HYDRA_LOCAL_SHARDS_ONLY": "0",
+    "HYDRA_THROUGHPUT_MODE": "0",
+    "HYDRA_FASTPATH": "0",
+    "HYDRA_FORCE_HTM_CPU": "0",
+    "HYDRA_INERT_MAMBA": "0",
+    "HYDRA_ALLOW_SYNTHETIC_RETINA": "0",
+    "HYDRA_HTM_FUSED": "1",
+    "HYDRA_HYENA_LAYERS": "",
+    "HYDRA_CKPT_INTERVAL": "0",
+    "HYDRA_ENGRAM_SUBSAMPLE": "1",
+    "HYDRA_HTM_SUBSAMPLE": "2",
+    "HYDRA_HTM_LEARN_EVERY": "8",
+}
+
+
+def _recover_params(row: dict[str, Any]) -> dict[str, Any]:
+    params = dict(row.get("params") or {})
+    attrs = row.get("user_attrs") or {}
+    for key, value in attrs.items():
+        if key.startswith("param_"):
+            params.setdefault(key.removeprefix("param_"), value)
+    return params
+
+
+def _candidate_env(params: dict[str, Any], *, eval_tokens: int, eval_batch: int, time_budget: int) -> dict[str, str]:
+    env = dict(_DEFAULT_ENV)
+    env["HYDRA_EVAL_TOKENS"] = str(eval_tokens)
+    env["HYDRA_EVAL_BATCH"] = str(eval_batch)
+    env["HYDRA_TIME_BUDGET"] = str(time_budget)
+    for key, value in params.items():
+        env_key = _PARAM_TO_ENV.get(key)
+        if env_key is not None:
+            env[env_key] = str(value)
+    if "HYDRA_BATCH_SIZE" in env and "HYDRA_SEQ_LEN" in env:
+        grad_accum = int(params.get("grad_accum", 16))
+        env["HYDRA_TOTAL_BATCH"] = str(int(env["HYDRA_BATCH_SIZE"]) * int(env["HYDRA_SEQ_LEN"]) * grad_accum)
+    return env
+
+
+def build_retest_plan(
+    *,
+    storage: str,
+    study_names: list[str],
+    top_k: int,
+    metric: str = "val_bpb",
+    eval_tokens: int = 16384,
+    eval_batch: int = 2,
+    time_budget: int = 420,
+) -> dict[str, Any]:
+    leaderboard = build_leaderboard(storage=storage, study_names=study_names, metric=metric)
+    rows = [*leaderboard["contaminated_trials"], *leaderboard["clean_trials"]]
+    reverse = leaderboard["direction"] == "maximize"
+    rows.sort(key=lambda row: float(row["value"]), reverse=reverse)
+    candidates = []
+    for row in rows[: max(0, top_k)]:
+        params = _recover_params(row)
+        env = _candidate_env(params, eval_tokens=eval_tokens, eval_batch=eval_batch, time_budget=time_budget)
+        reason = row.get("contamination_reason") or "canonical_truth_eval_retest"
+        candidates.append({
+            "study_name": row["study_name"],
+            "trial_number": row["trial_number"],
+            "source_value": row["value"],
+            "source_objective": row["objective_source"],
+            "source_job_id": row.get("hf_job_id"),
+            "needs_retest_reason": reason,
+            "params": params,
+            "env": env,
+        })
+    return {
+        "schema_version": 1,
+        "generated_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
+        "metric": metric,
+        "study_names": study_names,
+        "eval_tokens": eval_tokens,
+        "eval_batch": eval_batch,
+        "time_budget": time_budget,
+        "n_candidates": len(candidates),
+        "candidates": candidates,
+    }
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Plan canonical-eval retests for historical HPO configs")
+    parser.add_argument("--storage", default="sqlite:///optuna_hpo.db")
+    parser.add_argument("--study-name", action="append", default=[])
+    parser.add_argument("--metric", default="val_bpb")
+    parser.add_argument("--top-k", type=int, default=10)
+    parser.add_argument("--eval-tokens", type=int, default=16384)
+    parser.add_argument("--eval-batch", type=int, default=2)
+    parser.add_argument("--time-budget", type=int, default=420)
+    parser.add_argument("--out", type=Path, default=Path(".tmp") / "optuna" / "retest_plan.json")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    study_names = args.study_name or ["hydra_hpo"]
+    payload = build_retest_plan(
+        storage=args.storage,
+        study_names=study_names,
+        top_k=args.top_k,
+        metric=args.metric,
+        eval_tokens=args.eval_tokens,
+        eval_batch=args.eval_batch,
+        time_budget=args.time_budget,
+    )
+    args.out.parent.mkdir(parents=True, exist_ok=True)
+    args.out.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/hydra_generation.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Callable
+
+import torch
+
+from scripts.benchmark_checkpoint import hydrate_checkpoint
+from scripts.hf_routing import resolve_routing
+
+
+def default_checkpoint_path() -> Path:
+    return Path(os.path.expanduser("~/.cache/autoresearch/latest.pt"))
+
+
+def checkpoint_candidates(*, cache_dir: Path | None = None) -> list[Path]:
+    base = cache_dir or Path(os.path.expanduser("~/.cache/autoresearch"))
+    return [
+        base / "best_bpb.pt",
+        base / "pretrain_final.pt",
+        base / "latest.pt",
+    ]
+
+
+def resolve_checkpoint_path(explicit_path: Path | None, *, cache_dir: Path | None = None) -> Path:
+    if explicit_path is not None:
+        return explicit_path
+    for candidate in checkpoint_candidates(cache_dir=cache_dir):
+        if candidate.exists():
+            return candidate
+    return default_checkpoint_path()
+
+
+def validate_checkpoint_compatibility(
+    *,
+    baseline_arch: str,
+    missing_keys: list[str],
+    unexpected_keys: list[str],
+    total_model_keys: int,
+) -> None:
+    if baseline_arch == "transformer" and (missing_keys or unexpected_keys):
+        raise RuntimeError(
+            "checkpoint incompatible with transformer baseline architecture; "
+            "use a transformer-trained checkpoint or keep HYDRA_BASELINE_ARCH=mamba3"
+        )
+    mismatch_count = len(missing_keys) + len(unexpected_keys)
+    if total_model_keys > 0 and mismatch_count > max(8, total_model_keys // 2):
+        raise RuntimeError("checkpoint incompatible with requested model architecture")
+
+
+def generate_from_callable(
+    generator: Callable[[str], str] | Callable[..., str],
+    prompt: str,
+    *,
+    max_new_tokens: int,
+    temperature: float,
+    top_p: float,
+) -> str:
+    text = generator(
+        prompt,
+        max_new_tokens=max_new_tokens,
+        temperature=temperature,
+        top_p=top_p,
+    )
+    return str(text).strip()
+
+
+def load_hydra_causal_lm(checkpoint_path: Path | None = None, device: str | None = None):
+    ckpt_path = resolve_checkpoint_path(checkpoint_path)
+    if not ckpt_path.exists():
+        hydrated = hydrate_checkpoint(
+            cache_dir=ckpt_path.parent,
+            output_repo=resolve_routing(token=os.environ.get("HF_TOKEN")).output_repo,
+            token=os.environ.get("HF_TOKEN"),
+        )
+        if hydrated is not None:
+            ckpt_path = hydrated
+        if not ckpt_path.exists():
+            raise FileNotFoundError(f"Checkpoint not found: {ckpt_path}")
+
+    from transformers import GenerationConfig, GenerationMixin, PretrainedConfig, PreTrainedModel
+    from transformers.modeling_outputs import CausalLMOutputWithPast
+
+    from hydra.config import PostSemClawConfig
+    from hydra.model import PostSemClawModel
+    from prepare import Tokenizer
+
+    resolved_device = device or ("cuda" if torch.cuda.is_available() else "cpu")
+
+    class _HydraGenConfig(PretrainedConfig):
+        model_type = "hydra"
+
+        def __init__(self, vocab_size: int = 65536, **kw):
+            super().__init__(**kw)
+            self.vocab_size = vocab_size
+
+    class HydraForCausalLM(PreTrainedModel, GenerationMixin):
+        config_class = _HydraGenConfig
+
+        def __init__(self, gen_config, inner_model):
+            super().__init__(gen_config)
+            self.inner = inner_model
+            self.config.vocab_size = gen_config.vocab_size
+
+        def forward(self, input_ids, attention_mask=None, **kw):
+            logits = self.inner(input_ids)
+            return CausalLMOutputWithPast(loss=None, logits=logits, past_key_values=None)
+
+        def prepare_inputs_for_generation(self, input_ids, **kw):
+            return {"input_ids": input_ids}
+
+        def get_input_embeddings(self):
+            return self.inner.wte
+
+        def can_generate(self) -> bool:
+            return True
+
+        @property
+        def _supports_cache_class(self):
+            return False
+
+    tokenizer = Tokenizer.from_directory()
+    vocab_size = tokenizer.get_vocab_size()
+    bos = tokenizer.get_bos_token_id()
+    ckpt = torch.load(str(ckpt_path), map_location="cpu", weights_only=False)
+    cfg = PostSemClawConfig(**ckpt["config"])
+    with torch.device("meta"):
+        inner = PostSemClawModel(cfg)
+    inner.to_empty(device=resolved_device)
+    missing, unexpected = inner.load_state_dict(ckpt["model_state_dict"], strict=False)
+    validate_checkpoint_compatibility(
+        baseline_arch=os.environ.get("HYDRA_BASELINE_ARCH", "mamba3").strip().lower(),
+        missing_keys=list(missing),
+        unexpected_keys=list(unexpected),
+        total_model_keys=len(inner.state_dict()),
+    )
+    inner.eval()
+
+    gen_cfg = _HydraGenConfig(vocab_size=vocab_size)
+    gen_cfg.bos_token_id = bos
+    gen_cfg.eos_token_id = bos
+    gen_cfg.pad_token_id = bos
+    model = HydraForCausalLM(gen_cfg, inner).to(resolved_device)
+    model.eval()
+    return tokenizer, model, bos, resolved_device, GenerationConfig
+
+
+def build_hydra_generator(
+    *,
+    checkpoint_path: Path | None = None,
+    device: str | None = None,
+    max_new_tokens: int,
+    temperature: float,
+    top_p: float,
+):
+    tokenizer, model, bos, resolved_device, GenerationConfig = load_hydra_causal_lm(checkpoint_path=checkpoint_path, device=device)
+
+    def _generate(prompt: str) -> str:
+        ids = torch.tensor([tokenizer.encode(prompt)], dtype=torch.long, device=resolved_device)
+        gen_config = GenerationConfig(
+            max_new_tokens=max_new_tokens,
+            use_cache=False,
+            do_sample=temperature > 0.0,
+            temperature=temperature,
+            top_p=top_p,
+            bos_token_id=bos,
+            eos_token_id=bos,
+            pad_token_id=bos,
+        )
+        if str(resolved_device).startswith("cuda"):
+            with torch.no_grad(), torch.amp.autocast(device_type="cuda", dtype=torch.bfloat16):
+                out = model.generate(ids, generation_config=gen_config)
+        else:
+            with torch.no_grad():
+                out = model.generate(ids, generation_config=gen_config)
+        return tokenizer.decode(out[0].tolist())
+
+    return _generate

overlay/scripts/launch_benchmark_hf_job.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from huggingface_hub import HfApi
+from huggingface_hub.utils import get_token
+
+from scripts.hf_routing import resolve_routing
+from scripts.launch_feather_hf_job import IMAGE_DIR, sync_overlay_from_repo, wait_for_space
+
+
+def build_benchmark_job_env(
+    *,
+    benchmark: str,
+    variant: str,
+    seed: int,
+    output_repo: str,
+    tokenizer_repo: str,
+) -> dict[str, str]:
+    env = {
+        "FEATHER_HF_OUTPUT_REPO": output_repo,
+        "FEATHER_RUNTIME_MODE": "benchmark",
+        "HYDRA_TOKENIZER_CACHE_REPO": tokenizer_repo,
+        "HYDRA_BENCHMARK_NAME": benchmark,
+        "HYDRA_BENCHMARK_VARIANT": variant,
+        "HYDRA_SEED": str(seed),
+        "PYTHONUNBUFFERED": "1",
+    }
+    for key, value in os.environ.items():
+        if key.startswith("HYDRA_") and key not in env:
+            env[key] = value
+    return env
+
+
+def build_benchmark_job_command(*, benchmark: str, variant: str, seed: int) -> list[str]:
+    return [
+        "python",
+        "/app/entrypoint.py",
+    ]
+
+
+def submit_benchmark_job(
+    *,
+    api,
+    image: str,
+    command: list[str],
+    env: dict[str, str],
+    token: str,
+    namespace: str,
+    flavor: str,
+    timeout: str,
+) -> dict[str, str]:
+    job = api.run_job(
+        image=image,
+        command=command,
+        env=env,
+        secrets={"HF_TOKEN": token},
+        flavor=flavor,
+        timeout=timeout,
+        namespace=namespace,
+        token=token,
+    )
+    return {
+        "job_id": job.id,
+        "job_url": job.url,
+        "job_stage": str(job.status.stage),
+    }
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    routing = resolve_routing(token=os.environ.get("HF_TOKEN"))
+    parser = argparse.ArgumentParser(description="Prepare or submit a remote HF benchmark job")
+    parser.add_argument("--benchmark", required=True)
+    parser.add_argument("--variant", required=True)
+    parser.add_argument("--seed", type=int, required=True)
+    parser.add_argument("--output-repo", default=routing.output_repo)
+    parser.add_argument("--tokenizer-repo", default=routing.output_repo)
+    parser.add_argument("--image", default=f"hf.co/spaces/{routing.space_repo}")
+    parser.add_argument("--namespace", default=routing.job_namespace)
+    parser.add_argument("--flavor", default="a10g-small")
+    parser.add_argument("--timeout", default="30m")
+    parser.add_argument("--summary-out", type=Path)
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--refresh-image", action="store_true")
+    parser.add_argument("--sync-overlay", action="store_true")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    env = build_benchmark_job_env(
+        benchmark=args.benchmark,
+        variant=args.variant,
+        seed=args.seed,
+        output_repo=args.output_repo,
+        tokenizer_repo=args.tokenizer_repo,
+    )
+    command = build_benchmark_job_command(benchmark=args.benchmark, variant=args.variant, seed=args.seed)
+    payload = {
+        "benchmark": args.benchmark,
+        "variant": args.variant,
+        "seed": args.seed,
+        "output_repo": args.output_repo,
+        "tokenizer_repo": args.tokenizer_repo,
+        "image": args.image,
+        "namespace": args.namespace,
+        "command": command,
+        "env": env,
+        "dry_run": args.dry_run,
+    }
+    if not args.dry_run:
+        token = os.environ.get("HF_TOKEN") or get_token()
+        if not token:
+            raise SystemExit("HF_TOKEN must be set or cached via huggingface-cli login")
+        api = HfApi(token=token)
+        if args.refresh_image:
+            space_repo = args.image.removeprefix("hf.co/spaces/")
+            if args.sync_overlay:
+                sync_overlay_from_repo()
+            api.upload_folder(
+                repo_id=space_repo,
+                repo_type="space",
+                folder_path=str(IMAGE_DIR),
+                commit_message="Update benchmark runtime image",
+                token=token,
+            )
+            wait_for_space(api, space_repo, token=token)
+        payload.update(
+            submit_benchmark_job(
+                api=api,
+                image=args.image,
+                command=command,
+                env=env,
+                token=token,
+                namespace=args.namespace,
+                flavor=args.flavor,
+                timeout=args.timeout,
+            )
+        )
+    if args.summary_out is not None:
+        args.summary_out.parent.mkdir(parents=True, exist_ok=True)
+        args.summary_out.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+    print(json.dumps(payload, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/optuna_hpo.py

@@ -108,6 +108,28 @@ def _enqueue_transfer_priors(study: optuna.Study, priors_file: Path, apply_prior
         if after > before:
             enqueued += 1
     return enqueued
+
+
+def _enqueue_quality_anchors(study: optuna.Study, priors_file: Path, quality_mode_local: bool, top_k: int) -> int:
+    if not quality_mode_local or top_k <= 0:
+        return 0
+
+    priors = _load_prior_param_sets(priors_file)[:top_k]
+    enqueued = 0
+    for params in priors:
+        before = len(study.get_trials(deepcopy=False))
+        try:
+            study.enqueue_trial(
+                params,
+                user_attrs={"seed_source": "quality_anchor"},
+                skip_if_exists=True,
+            )
+        except TypeError:
+            study.enqueue_trial(params, user_attrs={"seed_source": "quality_anchor"})
+        after = len(study.get_trials(deepcopy=False))
+        if after > before:
+            enqueued += 1
+    return enqueued
 
 
 def _parse_metrics_from_stdout(stdout: str) -> dict[str, Any] | None:
@@ -143,15 +165,99 @@ def _parse_metrics_from_log_lines(lines: list[str]) -> dict[str, Any] | None:
 
 
 def _parse_last_train_bpb_from_logs(lines: list[str]) -> float | None:
-    """Best-effort fallback when final eval crashes before metrics JSON write."""
-    last: float | None = None
-    for line in lines:
-        m = re.search(r"\bbpb=([0-9]+(?:\.[0-9]+)?)", line)
+    """Best-effort fallback when final eval crashes before metrics JSON write."""
+    last: float | None = None
+    for line in lines:
+        m = re.search(r"\bbpb=([0-9]+(?:\.[0-9]+)?)", line)
         if m:
-            last = float(m.group(1))
+            last = float(m.group(1))
     return last
 
 
+def _persist_trial_artifacts(
+    *,
+    trial_dir: Path,
+    metrics: dict[str, Any] | None,
+    log_lines: list[str] | None,
+    log_name: str,
+    metadata: dict[str, Any],
+) -> dict[str, str | None]:
+    trial_dir.mkdir(parents=True, exist_ok=True)
+    metrics_path = trial_dir / "metrics.json"
+    log_path = trial_dir / log_name
+    manifest_path = trial_dir / "trial_artifacts.json"
+
+    if metrics is not None:
+        metrics_path.write_text(json.dumps(metrics, indent=2, sort_keys=True), encoding="utf-8")
+    if log_lines is not None:
+        log_path.write_text("\n".join(log_lines), encoding="utf-8")
+
+    manifest = {
+        **metadata,
+        "metrics_path": str(metrics_path) if metrics is not None else None,
+        "log_path": str(log_path) if log_lines is not None else None,
+    }
+    manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True), encoding="utf-8")
+    return {
+        "metrics_path": str(metrics_path) if metrics is not None else None,
+        "log_path": str(log_path) if log_lines is not None else None,
+        "manifest_path": str(manifest_path),
+    }
+
+
+def _resolve_objective_metric(
+    trial: optuna.Trial,
+    *,
+    metric_key: str,
+    metrics: dict[str, Any] | None,
+    allow_log_metric_fallback: bool,
+    fallback_bpb: float | None,
+    tps_seen: float | None,
+) -> float:
+    """Resolve the objective value while labeling where it came from.
+
+    Validation metrics and live training-log fallbacks are intentionally
+    different sources. Keeping that distinction in trial attrs prevents a
+    skipped/OOM eval from being mistaken for a real validation result.
+    """
+    if metrics is None:
+        if allow_log_metric_fallback and metric_key == "val_bpb" and fallback_bpb is not None:
+            trial.set_user_attr("objective_source", "train_log_fallback")
+            trial.set_user_attr("objective_metric", "train_bpb")
+            trial.set_user_attr("eval_status", "missing_metrics")
+            trial.set_user_attr("train_bpb_fallback", float(fallback_bpb))
+            if tps_seen is not None:
+                trial.set_user_attr("tps", float(tps_seen))
+            return float(fallback_bpb)
+        trial.set_user_attr("objective_source", "missing_metrics")
+        raise optuna.TrialPruned("No metrics payload found")
+
+    eval_status = str(
+        metrics.get(
+            "eval_status",
+            "completed" if metrics.get("val_bpb") is not None else "unknown",
+        )
+    )
+    trial.set_user_attr("eval_status", eval_status)
+
+    if fallback_bpb is not None:
+        trial.set_user_attr("train_bpb_fallback", float(fallback_bpb))
+
+    if metric_key not in metrics or metrics[metric_key] is None:
+        trial.set_user_attr("objective_source", "missing_metric")
+        trial.set_user_attr("objective_metric", metric_key)
+        raise optuna.TrialPruned(f"Metric '{metric_key}' missing in metrics payload")
+
+    value = float(metrics[metric_key])
+    trial.set_user_attr("objective_metric", metric_key)
+    if metric_key == "val_bpb":
+        trial.set_user_attr("objective_source", "final_val")
+        trial.set_user_attr("final_val_bpb", value)
+    else:
+        trial.set_user_attr("objective_source", "metrics_json")
+    return value
+
+
 def _fetch_job_logs_safe(
     api,
     *,
@@ -180,12 +286,20 @@ def _fetch_job_logs_safe(
     if last_exc is not None:
         raise last_exc
     return []
+
+
+def _effective_min_tps(args: argparse.Namespace) -> float | None:
+    min_tps = args.min_tps
+    if getattr(args, "quality_mode_local", False) and min_tps == 50000.0:
+        return 0.0
+    return min_tps
 
 
 def _trial_env(trial: optuna.Trial, args: argparse.Namespace, metrics_path: Path) -> dict[str, str]:
     env = os.environ.copy()
     full_arch_hpo = env.get("HYDRA_HPO_FULL_ARCH", "0") == "1"
     speed_arch_hpo = full_arch_hpo and env.get("HYDRA_HPO_SPEED_ARCH", "0") == "1"
+    quality_mode_local = bool(getattr(args, "quality_mode_local", False))
 
     # Runtime and reporting
     env["HYDRA_METRICS_OUT"] = str(metrics_path)
@@ -203,6 +317,12 @@ def _trial_env(trial: optuna.Trial, args: argparse.Namespace, metrics_path: Path
         env["HYDRA_D_STATE"] = str(trial.suggest_categorical("d_state", [16, 32]))
         env["HYDRA_HEADDIM"] = str(trial.suggest_categorical("headdim", [16, 32]))
         env["HYDRA_EXPAND"] = str(trial.suggest_categorical("expand", [1, 2]))
+    elif quality_mode_local and full_arch_hpo:
+        env["HYDRA_D_MODEL"] = str(trial.suggest_categorical("d_model", [64, 96, 128]))
+        env["HYDRA_N_LAYER"] = str(trial.suggest_int("n_layer", 2, 3))
+        env["HYDRA_D_STATE"] = str(trial.suggest_categorical("d_state", [16, 32]))
+        env["HYDRA_HEADDIM"] = str(trial.suggest_categorical("headdim", [16, 32]))
+        env["HYDRA_EXPAND"] = str(trial.suggest_categorical("expand", [1, 2]))
     else:
         env["HYDRA_D_MODEL"] = str(trial.suggest_categorical("d_model", [64, 96, 128, 160, 192]))
         env["HYDRA_N_LAYER"] = str(trial.suggest_int("n_layer", 1, 4))
@@ -214,6 +334,10 @@ def _trial_env(trial: optuna.Trial, args: argparse.Namespace, metrics_path: Path
         seq_len = trial.suggest_categorical("seq_len", [64, 128])
         batch_size = trial.suggest_categorical("batch_size", [8, 16, 32])
         grad_accum = trial.suggest_categorical("grad_accum", [4, 8, 16])
+    elif quality_mode_local and full_arch_hpo:
+        seq_len = trial.suggest_categorical("seq_len", [64])
+        batch_size = trial.suggest_categorical("batch_size", [4, 8])
+        grad_accum = trial.suggest_categorical("grad_accum", [4, 8, 16])
     else:
         seq_len = trial.suggest_categorical("seq_len", [32, 64])
         batch_size = trial.suggest_categorical("batch_size", [4, 8] if full_arch_hpo else [4, 8, 16])
@@ -224,22 +348,41 @@ def _trial_env(trial: optuna.Trial, args: argparse.Namespace, metrics_path: Path
     env["HYDRA_BATCH_SIZE"] = str(batch_size)
     env["HYDRA_TOTAL_BATCH"] = str(total_batch)
 
-    env["HYDRA_MATRIX_LR"] = str(trial.suggest_float("matrix_lr", 0.005, 0.2, log=True))
-    env["HYDRA_EMBED_LR"] = str(trial.suggest_float("embed_lr", 0.05, 1.0, log=True))
-    env["HYDRA_UNEMBED_LR"] = str(trial.suggest_float("unembed_lr", 0.0005, 0.02, log=True))
+    if quality_mode_local and full_arch_hpo:
+        env["HYDRA_MATRIX_LR"] = str(trial.suggest_float("matrix_lr", 0.008, 0.03, log=True))
+        env["HYDRA_EMBED_LR"] = str(trial.suggest_float("embed_lr", 0.15, 0.6, log=True))
+        env["HYDRA_UNEMBED_LR"] = str(trial.suggest_float("unembed_lr", 0.001, 0.01, log=True))
+    else:
+        env["HYDRA_MATRIX_LR"] = str(trial.suggest_float("matrix_lr", 0.005, 0.2, log=True))
+        env["HYDRA_EMBED_LR"] = str(trial.suggest_float("embed_lr", 0.05, 1.0, log=True))
+        env["HYDRA_UNEMBED_LR"] = str(trial.suggest_float("unembed_lr", 0.0005, 0.02, log=True))
 
     if full_arch_hpo:
         env["HYDRA_HYENA_LAYERS"] = ""
         env["HYDRA_ENGRAM_N_COLUMNS"] = str(
-            trial.suggest_categorical("engram_n_columns", [512, 1024] if speed_arch_hpo else [512, 1024, 2048])
+            trial.suggest_categorical(
+                "engram_n_columns",
+                [512, 1024] if (speed_arch_hpo or quality_mode_local) else [512, 1024, 2048],
+            )
         )
         env["HYDRA_ENGRAM_LAYER_IDX"] = str(trial.suggest_int("engram_layer_idx", 0, max(0, int(env["HYDRA_N_LAYER"]) - 1)))
-        env["HYDRA_SDR_TARGET_ACTIVE"] = str(trial.suggest_categorical("sdr_target_active", [164, 327] if speed_arch_hpo else [164, 327, 512]))
-        env["HYDRA_HTM_LEARN_EVERY"] = str(trial.suggest_categorical("htm_learn_every", [8, 16] if speed_arch_hpo else [4, 8, 16]))
-        env["HYDRA_HTM_SUBSAMPLE"] = str(trial.suggest_categorical("htm_subsample", [4, 8, 16] if speed_arch_hpo else [1, 2, 4, 8]))
-        env["HYDRA_ENGRAM_SUBSAMPLE"] = str(trial.suggest_categorical("engram_subsample", [1, 2, 4] if speed_arch_hpo else [1]))
+        env["HYDRA_SDR_TARGET_ACTIVE"] = str(
+            trial.suggest_categorical(
+                "sdr_target_active",
+                [327] if quality_mode_local else ([164, 327] if speed_arch_hpo else [164, 327, 512]),
+            )
+        )
+        env["HYDRA_HTM_LEARN_EVERY"] = str(
+            trial.suggest_categorical("htm_learn_every", [8, 16] if (speed_arch_hpo or quality_mode_local) else [4, 8, 16])
+        )
+        env["HYDRA_HTM_SUBSAMPLE"] = str(
+            trial.suggest_categorical("htm_subsample", [1, 2] if quality_mode_local else ([4, 8, 16] if speed_arch_hpo else [1, 2, 4, 8]))
+        )
+        env["HYDRA_ENGRAM_SUBSAMPLE"] = str(
+            trial.suggest_categorical("engram_subsample", [1, 2] if quality_mode_local else ([1, 2, 4] if speed_arch_hpo else [1]))
+        )
         env["HYDRA_MAMBA3_CHUNK"] = str(trial.suggest_categorical("mamba3_chunk", [32, 64]))
-        env["HYDRA_DROPOUT"] = str(trial.suggest_categorical("dropout", [0.0, 0.1] if speed_arch_hpo else [0.0, 0.1, 0.2]))
+        env["HYDRA_DROPOUT"] = str(trial.suggest_categorical("dropout", [0.0, 0.1] if (speed_arch_hpo or quality_mode_local) else [0.0, 0.1, 0.2]))
     else:
         env["HYDRA_HYENA_LAYERS"] = trial.suggest_categorical("hyena_layers", ["", "0", "1", "0,1"])
 
@@ -299,8 +442,10 @@ def _space_repo_from_hf_image(image: str, namespace: str) -> str:
     return os.environ.get("FEATHER_HF_SPACE_REPO", f"{namespace}/feather-a10-runtime")
 
 
-def _objective_local(args: argparse.Namespace):
-    def objective(trial: optuna.Trial) -> float:
+def _objective_local(args: argparse.Namespace):
+    effective_min_tps = _effective_min_tps(args)
+
+    def objective(trial: optuna.Trial) -> float:
         trial_dir = Path(tempfile.mkdtemp(prefix=f"optuna_trial_{trial.number}_", dir=str(args.work_dir)))
         metrics_path = trial_dir / "metrics.json"
 
@@ -315,44 +460,67 @@ def _objective_local(args: argparse.Namespace):
             timeout=args.trial_timeout,
         )
 
-        metrics: dict[str, Any] | None = None
+        metrics: dict[str, Any] | None = None
         if metrics_path.exists():
             try:
                 metrics = json.loads(metrics_path.read_text(encoding="utf-8"))
             except json.JSONDecodeError:
                 metrics = None
-        if metrics is None:
-            metrics = _parse_metrics_from_stdout(proc.stdout)
-
-        if metrics is None:
-            raise optuna.TrialPruned("No metrics found (HYDRA_METRICS_OUT/[METRICS_JSON])")
+        if metrics is None:
+            metrics = _parse_metrics_from_stdout(proc.stdout)
+
+        artifact_paths = _persist_trial_artifacts(
+            trial_dir=trial_dir,
+            metrics=metrics,
+            log_lines=(proc.stdout or "").splitlines(),
+            log_name="train_stdout.log",
+            metadata={"runner": "local", "returncode": proc.returncode},
+        )
+        (trial_dir / "train_stderr.log").write_text(proc.stderr or "", encoding="utf-8")
+
+        fallback_bpb = _parse_last_train_bpb_from_logs(proc.stdout.splitlines())
+        if metrics is None:
+            _resolve_objective_metric(
+                trial,
+                metric_key=args.metric,
+                metrics=None,
+                allow_log_metric_fallback=args.allow_log_metric_fallback,
+                fallback_bpb=fallback_bpb,
+                tps_seen=None,
+            )
+            raise optuna.TrialPruned("No metrics found (HYDRA_METRICS_OUT/[METRICS_JSON])")
 
         if proc.returncode != 0:
             raise optuna.TrialPruned(f"Training failed rc={proc.returncode}")
 
-        metric_key = args.metric
-        if metric_key not in metrics or metrics[metric_key] is None:
-            raise optuna.TrialPruned(f"Metric '{metric_key}' missing in metrics payload")
+        metric_key = args.metric
 
         tps_val = metrics.get("tps")
         if tps_val is not None:
             tps_f = float(tps_val)
             trial.set_user_attr("tps", tps_f)
-            if args.min_tps is not None and tps_f < args.min_tps:
-                raise optuna.TrialPruned(f"TPS below floor: {tps_f} < {args.min_tps}")
-
-        value = float(metrics[metric_key])
-
-        # Keep useful context on trial
-        trial.set_user_attr("summary_path", metrics.get("summary_path"))
-        trial.set_user_attr("run_log_path", metrics.get("run_log_path"))
+            if effective_min_tps is not None and tps_f < effective_min_tps:
+                raise optuna.TrialPruned(f"TPS below floor: {tps_f} < {effective_min_tps}")
+
+        value = _resolve_objective_metric(
+            trial,
+            metric_key=metric_key,
+            metrics=metrics,
+            allow_log_metric_fallback=args.allow_log_metric_fallback,
+            fallback_bpb=fallback_bpb,
+            tps_seen=None,
+        )
 
-        return value
+        # Keep useful context on trial
+        trial.set_user_attr("summary_path", metrics.get("summary_path") or artifact_paths["manifest_path"])
+        trial.set_user_attr("run_log_path", metrics.get("run_log_path") or artifact_paths["log_path"])
+
+        return value
 
     return objective
 
 
-def _objective_hf_job(args: argparse.Namespace):
+def _objective_hf_job(args: argparse.Namespace):
     from huggingface_hub import HfApi
     from huggingface_hub.utils import get_token
 
@@ -362,8 +530,9 @@ def _objective_hf_job(args: argparse.Namespace):
             f"No Hugging Face token found. Set {args.hf_token_env} or run huggingface-cli login."
         )
 
-    api = HfApi(token=token)
-    terminal_states = {"ERROR", "COMPLETED", "CANCELLED", "TIMEOUT", "FAILED", "CANCELED"}
+    api = HfApi(token=token)
+    terminal_states = {"ERROR", "COMPLETED", "CANCELLED", "TIMEOUT", "FAILED", "CANCELED"}
+    effective_min_tps = _effective_min_tps(args)
 
     def objective(trial: optuna.Trial) -> float:
         trial_dir = Path(tempfile.mkdtemp(prefix=f"optuna_trial_{trial.number}_", dir=str(args.work_dir)))
@@ -474,50 +643,66 @@ def _objective_hf_job(args: argparse.Namespace):
         except Exception:
             pass
 
-        # Save logs for debugging
-        (trial_dir / "hf_job.log").write_text("\n".join(log_lines), encoding="utf-8")
-        trial.set_user_attr("hf_stage", stage)
-        trial.set_user_attr("hf_log_lines", len(log_lines))
+        artifact_paths = _persist_trial_artifacts(
+            trial_dir=trial_dir,
+            metrics=metrics,
+            log_lines=log_lines,
+            log_name="hf_job.log",
+            metadata={"runner": "hf-job", "hf_job_id": job_id, "hf_stage": stage},
+        )
+        trial.set_user_attr("hf_stage", stage)
+        trial.set_user_attr("hf_log_lines", len(log_lines))
         if terminal_detail:
             trial.set_user_attr("hf_status_message", terminal_detail)
 
-        if metrics is None:
-            if args.allow_log_metric_fallback and args.metric == "val_bpb":
-                fallback_bpb = _parse_last_train_bpb_from_logs(log_lines)
-                if fallback_bpb is not None:
-                    trial.set_user_attr("metric_source", "log_bpb_fallback")
-                    if tps_seen is not None:
-                        trial.set_user_attr("tps", tps_seen)
-                        if args.min_tps is not None and tps_seen < args.min_tps:
-                            raise optuna.TrialPruned(f"TPS below floor: {tps_seen} < {args.min_tps}")
-                    return float(fallback_bpb)
-            if tps_seen is not None:
-                trial.set_user_attr("tps", tps_seen)
-            detail = f"stage={stage}, logs={len(log_lines)}"
-            if terminal_detail:
-                detail = f"{detail}, message={terminal_detail}"
+        fallback_bpb = _parse_last_train_bpb_from_logs(log_lines)
+        if metrics is None:
+            try:
+                value = _resolve_objective_metric(
+                    trial,
+                    metric_key=args.metric,
+                    metrics=None,
+                    allow_log_metric_fallback=args.allow_log_metric_fallback,
+                    fallback_bpb=fallback_bpb,
+                    tps_seen=tps_seen,
+                )
+                if tps_seen is not None and effective_min_tps is not None and tps_seen < effective_min_tps:
+                    raise optuna.TrialPruned(f"TPS below floor: {tps_seen} < {effective_min_tps}")
+                return value
+            except optuna.TrialPruned:
+                pass
+            if tps_seen is not None:
+                trial.set_user_attr("tps", tps_seen)
+            detail = f"stage={stage}, logs={len(log_lines)}"
+            if terminal_detail:
+                detail = f"{detail}, message={terminal_detail}"
             raise optuna.TrialPruned(f"No metrics found from HF job ({detail})")
 
-        metric_key = args.metric
-        if metric_key not in metrics or metrics[metric_key] is None:
-            raise optuna.TrialPruned(f"Metric '{metric_key}' missing in metrics payload")
+        metric_key = args.metric
 
         tps_val = metrics.get("tps")
         if tps_val is not None:
             tps_f = float(tps_val)
             trial.set_user_attr("tps", tps_f)
-            if args.min_tps is not None and tps_f < args.min_tps:
-                raise optuna.TrialPruned(f"TPS below floor: {tps_f} < {args.min_tps}")
-
-        value = float(metrics[metric_key])
-        trial.set_user_attr("summary_path", metrics.get("summary_path"))
-        trial.set_user_attr("run_log_path", metrics.get("run_log_path"))
-        return value
+            if effective_min_tps is not None and tps_f < effective_min_tps:
+                raise optuna.TrialPruned(f"TPS below floor: {tps_f} < {effective_min_tps}")
+
+        value = _resolve_objective_metric(
+            trial,
+            metric_key=metric_key,
+            metrics=metrics,
+            allow_log_metric_fallback=args.allow_log_metric_fallback,
+            fallback_bpb=fallback_bpb,
+            tps_seen=tps_seen,
+        )
+        trial.set_user_attr("summary_path", metrics.get("summary_path") or artifact_paths["manifest_path"])
+        trial.set_user_attr("run_log_path", metrics.get("run_log_path") or artifact_paths["log_path"])
+        return value
 
     return objective
 
 
-def _objective_hf_launcher(args: argparse.Namespace):
+def _objective_hf_launcher(args: argparse.Namespace):
     from huggingface_hub import HfApi
     from huggingface_hub.utils import get_token
 
@@ -527,8 +712,9 @@ def _objective_hf_launcher(args: argparse.Namespace):
             f"No Hugging Face token found. Set {args.hf_token_env} or run huggingface-cli login."
         )
 
-    api = HfApi(token=token)
-    terminal_states = {"ERROR", "COMPLETED", "CANCELLED", "TIMEOUT", "FAILED", "CANCELED"}
+    api = HfApi(token=token)
+    terminal_states = {"ERROR", "COMPLETED", "CANCELLED", "TIMEOUT", "FAILED", "CANCELED"}
+    effective_min_tps = _effective_min_tps(args)
 
     def objective(trial: optuna.Trial) -> float:
         trial_dir = Path(tempfile.mkdtemp(prefix=f"optuna_trial_{trial.number}_", dir=str(args.work_dir)))
@@ -610,44 +796,61 @@ def _objective_hf_launcher(args: argparse.Namespace):
         except Exception:
             pass
 
-        (trial_dir / "hf_job.log").write_text("\n".join(log_lines), encoding="utf-8")
-        trial.set_user_attr("hf_stage", stage)
-        trial.set_user_attr("hf_log_lines", len(log_lines))
+        artifact_paths = _persist_trial_artifacts(
+            trial_dir=trial_dir,
+            metrics=metrics,
+            log_lines=log_lines,
+            log_name="hf_job.log",
+            metadata={"runner": "hf-launcher", "hf_job_id": job_id, "hf_stage": stage},
+        )
+        trial.set_user_attr("hf_stage", stage)
+        trial.set_user_attr("hf_log_lines", len(log_lines))
         if terminal_detail:
             trial.set_user_attr("hf_status_message", terminal_detail)
 
-        if metrics is None:
-            if args.allow_log_metric_fallback and args.metric == "val_bpb":
-                fallback_bpb = _parse_last_train_bpb_from_logs(log_lines)
-                if fallback_bpb is not None:
-                    trial.set_user_attr("metric_source", "log_bpb_fallback")
-                    if tps_seen is not None:
-                        trial.set_user_attr("tps", tps_seen)
-                        if args.min_tps is not None and tps_seen < args.min_tps:
-                            raise optuna.TrialPruned(f"TPS below floor: {tps_seen} < {args.min_tps}")
-                    return float(fallback_bpb)
-            if tps_seen is not None:
-                trial.set_user_attr("tps", tps_seen)
-            detail = f"stage={stage}, logs={len(log_lines)}"
-            if terminal_detail:
-                detail = f"{detail}, message={terminal_detail}"
+        fallback_bpb = _parse_last_train_bpb_from_logs(log_lines)
+        if metrics is None:
+            try:
+                value = _resolve_objective_metric(
+                    trial,
+                    metric_key=args.metric,
+                    metrics=None,
+                    allow_log_metric_fallback=args.allow_log_metric_fallback,
+                    fallback_bpb=fallback_bpb,
+                    tps_seen=tps_seen,
+                )
+                if tps_seen is not None and effective_min_tps is not None and tps_seen < effective_min_tps:
+                    raise optuna.TrialPruned(f"TPS below floor: {tps_seen} < {effective_min_tps}")
+                return value
+            except optuna.TrialPruned:
+                pass
+            if tps_seen is not None:
+                trial.set_user_attr("tps", tps_seen)
+            detail = f"stage={stage}, logs={len(log_lines)}"
+            if terminal_detail:
+                detail = f"{detail}, message={terminal_detail}"
             raise optuna.TrialPruned(f"No metrics found from HF launcher job ({detail})")
 
-        metric_key = args.metric
-        if metric_key not in metrics or metrics[metric_key] is None:
-            raise optuna.TrialPruned(f"Metric '{metric_key}' missing in metrics payload")
+        metric_key = args.metric
 
         tps_val = metrics.get("tps")
         if tps_val is not None:
             tps_f = float(tps_val)
             trial.set_user_attr("tps", tps_f)
-            if args.min_tps is not None and tps_f < args.min_tps:
-                raise optuna.TrialPruned(f"TPS below floor: {tps_f} < {args.min_tps}")
-
-        value = float(metrics[metric_key])
-        trial.set_user_attr("summary_path", metrics.get("summary_path"))
-        trial.set_user_attr("run_log_path", metrics.get("run_log_path"))
-        return value
+            if effective_min_tps is not None and tps_f < effective_min_tps:
+                raise optuna.TrialPruned(f"TPS below floor: {tps_f} < {effective_min_tps}")
+
+        value = _resolve_objective_metric(
+            trial,
+            metric_key=metric_key,
+            metrics=metrics,
+            allow_log_metric_fallback=args.allow_log_metric_fallback,
+            fallback_bpb=fallback_bpb,
+            tps_seen=tps_seen,
+        )
+        trial.set_user_attr("summary_path", metrics.get("summary_path") or artifact_paths["manifest_path"])
+        trial.set_user_attr("run_log_path", metrics.get("run_log_path") or artifact_paths["log_path"])
+        return value
 
     return objective
 
@@ -690,6 +893,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--priors-file", type=Path, default=REPO_ROOT / "docs" / "hpo_transfer_priors.json", help="Path to transfer-learning prior trials JSON")
     parser.add_argument("--apply-priors", action="store_true", default=True, help="Enqueue transfer-learning prior trials before optimize")
     parser.add_argument("--no-apply-priors", action="store_false", dest="apply_priors")
+    parser.add_argument("--quality-mode-local", action="store_true", default=False, help="Narrow local full-architecture search around the proven quality-winning region")
+    parser.add_argument("--quality-anchor-top-k", type=int, default=3, help="Number of top clean priors to enqueue as deterministic local quality anchors")
     parser.add_argument("--seed", type=int, default=42, help="Seed for sampler")
     parser.add_argument("--n-startup-trials", type=int, default=5, help="Pruner startup trials before pruning")
     parser.add_argument("--n-warmup-steps", type=int, default=0, help="Pruner warmup steps")
@@ -720,6 +925,10 @@ def main() -> int:
         pruner=pruner,
     )
 
+    enqueued_quality_anchors = _enqueue_quality_anchors(study, args.priors_file, args.quality_mode_local, args.quality_anchor_top_k)
+    if enqueued_quality_anchors:
+        print(f"[hpo] enqueued {enqueued_quality_anchors} local quality anchors from {args.priors_file}")
+
     enqueued_priors = _enqueue_transfer_priors(study, args.priors_file, args.apply_priors)
     if enqueued_priors:
         print(f"[hpo] enqueued {enqueued_priors} transfer priors from {args.priors_file}")
@@ -784,6 +993,8 @@ def main() -> int:
             "n_completed": len(completed),
             "patience_trials": args.patience_trials,
             "min_improvement": args.min_improvement,
+            "quality_mode_local": args.quality_mode_local,
+            "enqueued_quality_anchors": enqueued_quality_anchors,
             "enqueued_priors": enqueued_priors,
         }
     else:
@@ -793,10 +1004,12 @@ def main() -> int:
             "metric": args.metric,
             "best_value": None,
             "best_params": {},
-            "best_trial_number": None,
+            "best_trial_number": None,
             "best_trial_user_attrs": {},
             "n_trials": len(study.trials),
             "n_completed": 0,
+            "quality_mode_local": args.quality_mode_local,
+            "enqueued_quality_anchors": enqueued_quality_anchors,
             "enqueued_priors": enqueued_priors,
             "note": "No completed trials with metrics found.",
         }

overlay/scripts/run_cycle1a.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from scripts import cycle_executor
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run the full local Cycle 1a benchmark suite")
+    parser.add_argument("--out-dir", type=Path, default=REPO_ROOT / "artifacts" / "cycle1a_runs")
+    parser.add_argument("--preflight-out", type=Path, default=REPO_ROOT / "artifacts" / "cycle1a_preflight.json")
+    parser.add_argument("--summary-out", type=Path, default=REPO_ROOT / "artifacts" / "cycle1a_summary.json")
+    parser.add_argument("--hydrate-assets", action="store_true")
+    parser.add_argument("--require-ready", action="store_true")
+    parser.add_argument("--output-repo")
+    parser.add_argument("--tokenizer-repo")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    return cycle_executor.main([
+        "--benchmark", "GSM8K",
+        "--variant", "hydra_full",
+        "--seed", "42",
+        "--out-dir", str(args.out_dir),
+        "--preflight-out", str(args.preflight_out),
+        "--summary-out", str(args.summary_out),
+        "--all-runnable",
+        "--all-benchmarks",
+        *( ["--hydrate-assets"] if args.hydrate_assets else [] ),
+        *( ["--require-ready"] if args.require_ready else [] ),
+        *( ["--output-repo", args.output_repo] if args.output_repo else [] ),
+        *( ["--tokenizer-repo", args.tokenizer_repo] if args.tokenizer_repo else [] ),
+    ])
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

overlay/scripts/sweep_depth_aggregate.py

@@ -26,6 +26,8 @@ type MetricsDict = dict[str, MetricValue]
 MANIFEST = Path(sys.argv[1] if len(sys.argv) > 1 else '/tmp/sweep_depth_manifest.txt')
 STEP_TPS_PATTERN = re.compile(r"step=(\d+).*?\btps=(\d+)\b")
 MIN_TPS = float(os.environ.get('SWEEP_MIN_TPS', '0'))
+TARGET_TOKENS_M = float(os.environ.get('SWEEP_TARGET_TOKENS_M', '0'))
+TARGET_SECONDS = float(os.environ.get('SWEEP_TARGET_SECONDS', '0'))
 
 
 def _zero_shot_score(result: MetricsDict) -> float:
@@ -47,6 +49,25 @@ def _metric_int(result: MetricsDict, key: str, default: int = 0) -> int:
     return int(value) if isinstance(value, int) else default
 
 
+def _fixed_budget_ranking(results: dict[int, MetricsDict], *, metric_key: str, target: float) -> list[tuple[int, MetricsDict, float]]:
+    ranked: list[tuple[int, MetricsDict, float]] = []
+    for n_layer, row in results.items():
+        budget_val = row.get(metric_key)
+        if not isinstance(budget_val, (int, float)):
+            continue
+        gap = abs(float(budget_val) - target)
+        ranked.append((n_layer, row, gap))
+    ranked.sort(
+        key=lambda item: (
+            item[2],
+            _metric_float(item[1], 'val_bpb', float('inf')),
+            -_zero_shot_score(item[1]),
+            -_metric_float(item[1], 'tps_median', 0.0),
+        )
+    )
+    return ranked
+
+
 def _percentile_linear(sorted_values: list[float], pct: float) -> float:
     if not sorted_values:
         return 0.0
@@ -210,6 +231,28 @@ def compare(results: dict[int, MetricsDict]) -> None:
         )
     if MIN_TPS > 0:
         print(f"[agg] throughput gate: tps_median >= {MIN_TPS:.0f}; feasible={feasible_count}/{len(ranked)}")
+
+    if TARGET_TOKENS_M > 0:
+        print('\n=== Fixed-token champion comparison ===')
+        print(f'  target_tokens_M={TARGET_TOKENS_M:.4f}')
+        for n, r, gap in _fixed_budget_ranking(results, metric_key='total_tokens_M', target=TARGET_TOKENS_M):
+            print(
+                f"  n_layer={n:2d} val_bpb={_metric_float(r, 'val_bpb', float('nan')):.4f} "
+                f"total_tokens_M={_metric_float(r, 'total_tokens_M', float('nan')):.4f} "
+                f"token_gap_M={gap:.4f} tps_median={_metric_float(r, 'tps_median', 0.0):.0f}",
+                flush=True,
+            )
+
+    if TARGET_SECONDS > 0:
+        print('\n=== Fixed-time champion comparison ===')
+        print(f'  target_seconds={TARGET_SECONDS:.1f}')
+        for n, r, gap in _fixed_budget_ranking(results, metric_key='training_seconds', target=TARGET_SECONDS):
+            print(
+                f"  n_layer={n:2d} val_bpb={_metric_float(r, 'val_bpb', float('nan')):.4f} "
+                f"training_seconds={_metric_float(r, 'training_seconds', float('nan')):.1f} "
+                f"time_gap_s={gap:.1f} tps_median={_metric_float(r, 'tps_median', 0.0):.0f}",
+                flush=True,
+            )
 
 
 def main() -> int:

overlay/scripts/watch_benchmark_hf_job.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+
+def parse_benchmark_result_from_logs(lines: list[str]):
+    for line in reversed(lines):
+        text = line.strip()
+        if not text.startswith("{"):
+            continue
+        try:
+            payload = json.loads(text)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(payload, dict) and "benchmark" in payload:
+            return payload
+    return None
+
+
+def write_watch_summary(path: Path, payload: dict[str, object]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Watch or snapshot a remote benchmark job")
+    parser.add_argument("--job-id", required=True)
+    parser.add_argument("--namespace", default="jackoatmon")
+    parser.add_argument("--summary-out", type=Path)
+    return parser.parse_args(argv)