| |
| """Compute the representation-validation readiness gate from real artifacts. |
| |
| This does NOT invent labels or treat fallback fingerprints as POM. It inspects |
| the on-disk benchmark artifacts and reports whether each readiness check is |
| genuinely satisfied. Safe to rerun: output is deterministic apart from |
| ``generated_at``. Never overwrites benchmark manifests/results. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| def write_json(path: Path, value: Any) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n") |
|
|
|
|
| def _read(path: Path) -> Any: |
| try: |
| return json.loads(path.read_text()) |
| except Exception: |
| return None |
|
|
|
|
| def real_pom_available() -> bool: |
| man = _read(ROOT / "artifacts/pom/pom_manifest.json") |
| emb = ROOT / "artifacts/pom/pom_embeddings.npy" |
| if not (man and emb.exists()): |
| return False |
| |
| src = json.dumps(man).lower() |
| return ("openpom" in src or "v1.0.0" in src) and "fallback" not in src |
|
|
|
|
| def ablation_complete() -> bool: |
| res = _read(ROOT / "artifacts/representation_ablation/ablation_results.json") |
| if not res: |
| return False |
| txt = json.dumps(res) |
| return "morgan" in txt and "real_pom" in txt |
|
|
|
|
| def triplet_count_positive() -> bool: |
| f = ROOT / "data/benchmarks/substitution_triplets/triplets.jsonl" |
| if not f.exists(): |
| return False |
| return sum(1 for line in f.read_text().splitlines() if line.strip()) > 0 |
|
|
|
|
| def prospective_count_ok() -> bool: |
| f = ROOT / "data/benchmarks/prospective_formulas/formulas.jsonl" |
| if not f.exists(): |
| return False |
| n = sum(1 for line in f.read_text().splitlines() if line.strip()) |
| return 30 <= n <= 50 |
|
|
|
|
| def prospective_labels_sequestered() -> bool: |
| f = ROOT / "data/benchmarks/prospective_formulas/labels.sequestered.json" |
| if not f.exists(): |
| return False |
| return "evaluation_only" in f.read_text() |
|
|
|
|
| def evaluate(now: str) -> dict[str, Any]: |
| checks = { |
| "real_pom_available": real_pom_available(), |
| "ablation_complete": ablation_complete(), |
| "triplet_count_positive": triplet_count_positive(), |
| "prospective_formula_count_30_to_50": prospective_count_ok(), |
| "prospective_labels_sequestered": prospective_labels_sequestered(), |
| } |
| ready = all(checks.values()) |
| gate = { |
| "schema_version": 1, |
| "generated_at": now, |
| "ready_for_training": ready, |
| "checks": checks, |
| "decision": "READY_FOR_TRAINING" if ready else "DO_NOT_TRAIN", |
| "blockers": [name for name, passed in checks.items() if not passed], |
| } |
| write_json(ROOT / "artifacts/training_readiness.json", gate) |
| return gate |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--timestamp", help="ISO-8601 timestamp for reproducible freezing") |
| args = parser.parse_args() |
| now = args.timestamp or datetime.now(timezone.utc).isoformat() |
| gate = evaluate(now) |
| print(json.dumps(gate, indent=2, sort_keys=True)) |
| return 0 if gate["ready_for_training"] else 2 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
