Spaces:
Running
Running
Download code/publication_gate.py from DineshAI/daiccpXZfU: direct link, hf CLI and curl.
- Browser
- Download file 15.8 kB
-
https://huggingface.co/spaces/DineshAI/daiccpXZfU/resolve/main/code/publication_gate.py
- Command line
-
hf download hf://spaces/DineshAI/daiccpXZfU/code/publication_gate.py
-
curl -L -o publication_gate.py https://huggingface.co/spaces/DineshAI/daiccpXZfU/resolve/main/code/publication_gate.py
15.8 kB
| #!/usr/bin/env python3 | |
| """Fail-closed evaluator-visible release gate for daiccpXZfU.""" | |
| from __future__ import annotations | |
| import base64 | |
| import csv | |
| import hashlib | |
| import json | |
| import re | |
| import shutil | |
| import subprocess | |
| import tempfile | |
| import urllib.parse | |
| import urllib.request | |
| import xml.etree.ElementTree as ET | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[2] | |
| CANDIDATE = ROOT / "space_candidate" | |
| JUDGED_REVISION = "88488bc18db7974567008ee55dbea85871de82e4" | |
| SPACE_ID = "DineshAI/daiccpXZfU" | |
| FIXED_COMMAND = "uv sync --frozen && uv run --frozen python repro/src/verify.py && uv run --frozen python repro/src/publication_gate.py" | |
| TEXT_SUFFIXES = {".css", ".csv", ".html", ".js", ".json", ".lock", ".md", ".py", ".svg", ".toml", ".txt"} | |
| def sha256(data): | |
| return hashlib.sha256(data).hexdigest() | |
| def protected_manifest(): | |
| result = {} | |
| path = ROOT / ".openresearch" / "protected-space" / "PROTECTED_SHA256_MANIFEST_RELATIVE.txt" | |
| for line in path.read_text().splitlines(): | |
| digest, relative = line.split(" ", 1) | |
| result[relative.removeprefix("./")] = digest | |
| return result | |
| def download_judged_tree(destination, expected): | |
| headers = {"User-Agent": "OpenResearch-Reproduction/1.0 (arXiv:2604.17505)"} | |
| for relative, digest in expected.items(): | |
| quoted = urllib.parse.quote(relative) | |
| url = f"https://huggingface.co/spaces/{SPACE_ID}/resolve/{JUDGED_REVISION}/{quoted}?download=true" | |
| request = urllib.request.Request(url, headers=headers) | |
| with urllib.request.urlopen(request, timeout=120) as response: | |
| data = response.read() | |
| assert sha256(data) == digest, f"protected hash mismatch: {relative}" | |
| target = destination / relative | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| target.write_bytes(data) | |
| def overlay_candidate(destination): | |
| for source in sorted(path for path in CANDIDATE.rglob("*") if path.is_file()): | |
| relative = source.relative_to(CANDIDATE) | |
| target = destination / relative | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| shutil.copyfile(source, target) | |
| def walk_logbook(node, opened, slugs): | |
| slugs[node["slug"]] = node["file"] | |
| opened.add(node["file"]) | |
| for child in node.get("children", []): | |
| walk_logbook(child, opened, slugs) | |
| def hf_paths(markdown): | |
| pattern = r"https://huggingface\.co/spaces/DineshAI/daiccpXZfU/(?:blob|resolve)/main/([^\s)]+)" | |
| return {urllib.parse.unquote(path) for path in re.findall(pattern, markdown)} | |
| def validate_claim_page(claim, page, release_root, opened): | |
| text = page.read_text() | |
| links = hf_paths(text) | |
| opened.update(links) | |
| for relative in links: | |
| assert (release_root / relative).is_file(), f"Claim {claim} broken link: {relative}" | |
| evidence = f"evidence/claim-{claim}/" | |
| code_links = [path for path in links if path.startswith("code/") and path.endswith(".py")] | |
| raw_links = [path for path in links if path.startswith(evidence) and ("raw_" in path or path.endswith("raw_scaling.csv"))] | |
| checks = { | |
| "canonical_page": True, | |
| "code_visible": bool(code_links), | |
| "data_inline": "Inline result" in text or "Strongest evidence" in text, | |
| "raw_link": bool(raw_links), | |
| "checker": any("checker" in path or path.endswith("verify_claim.py") for path in links), | |
| "control": any("control" in path for path in links), | |
| "exact_claim_tested": "Exact contract" in text or "Exact claim contract" in text, | |
| "source_quantifiers": "Source TeX SHA-256" in text or "arXiv TeX archive SHA-256" in text, | |
| "claim_contract": f"{evidence}claim_contract.json" in links, | |
| "source_audit": f"{evidence}source_audit.md" in links, | |
| "method": f"{evidence}method.md" in links, | |
| "proof": any("proof" in path for path in links), | |
| "limitations": "limitation" in text.lower(), | |
| "fixed_command": FIXED_COMMAND in text, | |
| "git_sha": "Git `" in text, | |
| "seeds": "seed" in text.lower(), | |
| "cpu_runtime": "cpu-upgrade" in text and "64" in text and " s" in text, | |
| "reviewer_verdict": text.startswith(f"# Claim {claim} — VERIFIED"), | |
| } | |
| assert all(checks.values()), f"Claim {claim} visibility gap: {[key for key, value in checks.items() if not value]}" | |
| return checks | |
| def validate_evidence(release_root, opened): | |
| for claim in range(1, 6): | |
| directory = release_root / "evidence" / f"claim-{claim}" | |
| required = {"claim_contract.json", "source_audit.md", "method.md", "limitations.md", "EVAL.md"} | |
| assert required <= {path.name for path in directory.iterdir()} | |
| for path in directory.iterdir(): | |
| if path.is_file(): | |
| opened.add(path.relative_to(release_root).as_posix()) | |
| if path.suffix == ".json": | |
| json.loads(path.read_text()) | |
| if path.suffix == ".csv": | |
| with path.open(newline="") as handle: | |
| rows = list(csv.DictReader(handle)) | |
| assert rows and all(rows[0].keys()) | |
| checker_paths = list(directory.glob("*checker_output.json")) | |
| control_paths = list(directory.glob("*control_output.json")) | |
| proof_paths = list(directory.glob("*proof_certificate.json")) | |
| assert checker_paths and control_paths and proof_paths | |
| assert all(json.loads(path.read_text())["passed"] for path in checker_paths) | |
| assert all(json.loads(path.read_text())["passed"] for path in control_paths) | |
| assert all(json.loads(path.read_text())["status"] == "VERIFIED" for path in proof_paths) | |
| def validate_report_and_notebook(release_root, opened): | |
| report = release_root / "reports" / "full-reproduction" / "report.md" | |
| report_text = report.read_text() | |
| opened.add(report.relative_to(release_root).as_posix()) | |
| nonempty = [line for line in report_text.splitlines() if line.strip()] | |
| assert nonempty[0].startswith("# ") and nonempty[1].startswith("![") | |
| image_names = re.findall(r"\]\(images/([^)]+\.svg)\)", report_text) | |
| assert len(image_names) == 5 and len(set(image_names)) == 5 | |
| for name in image_names: | |
| image = report.parent / "images" / name | |
| ET.fromstring(image.read_text()) | |
| opened.add(image.relative_to(release_root).as_posix()) | |
| assert image.read_bytes() == (ROOT / "reports" / "full-reproduction" / "images" / name).read_bytes() | |
| release_report = release_root / "reports" / "full-reproduction" / "release_report.md" | |
| release_text = release_report.read_text() | |
| opened.add(release_report.relative_to(release_root).as_posix()) | |
| assert release_text.startswith("# Final release report\n\nPrevious live judged score: `5/10`") | |
| assert "Conservative projected score range after the proposed change" in release_text | |
| assert "Best-supported possible new score" in release_text and "forecast only" in release_text | |
| assert "| Claim | Current points | Possible points | Confidence | Evidence status | Basis and remaining risk |" in release_text | |
| assert release_text.count("| HIGH | VERIFIED |") == 5 | |
| assert "No claim is BLOCKED" in release_text | |
| assert "DineshAI/daiccpXZfU" in release_text and "$0.0625" in release_text | |
| notebook = release_root / "notebooks" / "reproduction.py" | |
| opened.add(notebook.relative_to(release_root).as_posix()) | |
| assert notebook.read_bytes() == (ROOT / "notebooks" / "reproduction.py").read_bytes() | |
| assert "Path(" not in notebook.read_text() and "open(" not in notebook.read_text() | |
| sync = subprocess.run( | |
| ["uv", "sync", "--frozen", "--extra", "notebook"], cwd=ROOT, text=True, capture_output=True, check=True | |
| ) | |
| check = subprocess.run( | |
| ["uv", "run", "--frozen", "--extra", "notebook", "marimo", "check", "notebooks/reproduction.py"], | |
| cwd=ROOT, text=True, capture_output=True, | |
| ) | |
| print(sync.stdout + sync.stderr, end="") | |
| print(check.stdout + check.stderr, end="") | |
| assert check.returncode == 0, f"marimo check exited {check.returncode}" | |
| return {"command": "uv run --frozen --extra notebook marimo check notebooks/reproduction.py", "passed": True} | |
| def text_overlay_files(): | |
| result = [] | |
| for path in sorted(item for item in CANDIDATE.rglob("*") if item.is_file()): | |
| assert path.suffix in TEXT_SUFFIXES, f"non-text upload blocked: {path}" | |
| path.read_text() | |
| result.append(path.relative_to(CANDIDATE).as_posix()) | |
| return result | |
| def scan_secrets(paths): | |
| patterns = [ | |
| re.compile(r"hf_[A-Za-z0-9]{20,}"), | |
| re.compile(r"(?i)(api[_-]?key|secret|access[_-]?token)\s*[:=]\s*['\"][^'\"]+['\"]"), | |
| ] | |
| findings = [] | |
| for relative in paths: | |
| text = (CANDIDATE / relative).read_text() | |
| if any(pattern.search(text) for pattern in patterns): | |
| findings.append(relative) | |
| assert not findings, f"secret-like content in: {findings}" | |
| def release_records(paths, blind_review): | |
| allowlist = "\n".join(paths) + "\n" | |
| manifest = "\n".join(f"{sha256((CANDIDATE / path).read_bytes())} {path}" for path in paths) + "\n" | |
| review = json.dumps(blind_review, indent=2, sort_keys=True) + "\n" | |
| return { | |
| "upload_allowlist.txt": allowlist, | |
| "upload_sha256.txt": manifest, | |
| "blind_review.json": review, | |
| } | |
| def main(): | |
| verdict = json.loads((ROOT / "outputs" / "verdict.json").read_text()) | |
| assert verdict["all_claims_passed"] and verdict["scientific_claims_resolved"] | |
| assert all(item["status"] == "VERIFIED" for item in verdict["campaign_claims"].values()) | |
| expected = protected_manifest() | |
| overlay = text_overlay_files() | |
| scan_secrets(overlay) | |
| with tempfile.TemporaryDirectory(prefix="daiccpXZfU-candidate-") as temporary: | |
| release_root = Path(temporary) | |
| download_judged_tree(release_root, expected) | |
| overlay_candidate(release_root) | |
| final_files = {path.relative_to(release_root).as_posix() for path in release_root.rglob("*") if path.is_file()} | |
| assert set(expected) <= final_files | |
| for relative, digest in expected.items(): | |
| if relative.startswith("pages/") and relative not in overlay: | |
| assert sha256((release_root / relative).read_bytes()) == digest | |
| opened = {"README.md", "logbook.json"} | |
| readme = (release_root / "README.md").read_text() | |
| assert "#/current" in readme and "#/historical-rejected-baseline" in readme | |
| opened.update(hf_paths(readme)) | |
| logbook = json.loads((release_root / "logbook.json").read_text()) | |
| assert logbook["space_id"] == SPACE_ID and logbook["root"]["slug"] == "current" | |
| slugs = {} | |
| walk_logbook(logbook["root"], opened, slugs) | |
| assert all((release_root / path).is_file() for path in opened) | |
| assert "current-report" in slugs and "release-report" in slugs and "historical-rejected-baseline" in slugs | |
| current_index = release_root / slugs["current"] | |
| current_index_text = current_index.read_text() | |
| assert "Visibility matrix" in current_index_text | |
| opened.update(hf_paths(current_index_text)) | |
| visibility = {} | |
| for claim in range(1, 6): | |
| page_path = release_root / slugs[f"current-claim-{claim}"] | |
| visibility[f"claim_{claim}"] = validate_claim_page(claim, page_path, release_root, opened) | |
| validate_evidence(release_root, opened) | |
| notebook_check = validate_report_and_notebook(release_root, opened) | |
| assert all((release_root / path).is_file() for path in opened) | |
| for relative in opened: | |
| (release_root / relative).read_bytes() | |
| repeat_opened = {"README.md", "logbook.json"} | |
| repeat_slugs = {} | |
| walk_logbook(logbook["root"], repeat_opened, repeat_slugs) | |
| repeat_opened.update(hf_paths(readme)) | |
| repeat_opened.update(hf_paths(current_index_text)) | |
| repeat_visibility = {} | |
| for claim in range(1, 6): | |
| repeat_visibility[f"claim_{claim}"] = validate_claim_page( | |
| claim, release_root / repeat_slugs[f"current-claim-{claim}"], release_root, repeat_opened | |
| ) | |
| validate_evidence(release_root, repeat_opened) | |
| report = release_root / "reports" / "full-reproduction" / "report.md" | |
| repeat_opened.add(report.relative_to(release_root).as_posix()) | |
| repeat_opened.add("reports/full-reproduction/release_report.md") | |
| for name in re.findall(r"\]\(images/([^)]+\.svg)\)", report.read_text()): | |
| image = report.parent / "images" / name | |
| ET.fromstring(image.read_text()) | |
| repeat_opened.add(image.relative_to(release_root).as_posix()) | |
| repeat_opened.add("notebooks/reproduction.py") | |
| assert repeat_visibility == visibility and repeat_opened == opened | |
| blind_review = { | |
| "candidate_basis": f"fresh exact judged Space {SPACE_ID}@{JUDGED_REVISION} plus text overlay", | |
| "canonical_start": ["README.md", "logbook.json"], | |
| "files_opened": sorted(opened), | |
| "claims": {key: {"conclusion": "VERIFIED", "visibility": value} for key, value in visibility.items()}, | |
| "passes": [ | |
| {"name": "post-fix blind pass 1", "files_opened": sorted(opened), "missing": []}, | |
| {"name": "post-fix blind pass 2", "files_opened": sorted(repeat_opened), "missing": []}, | |
| ], | |
| "fixes_before_repeat": [ | |
| "removed literal backslash-n corruption from copied raw data", | |
| "added exact commands, source hashes, contracts, controls, and proof links", | |
| "placed the current visual report first and retained historical navigation", | |
| ], | |
| "unresolved_visibility_items": [], | |
| "historical_pages_preserved": True, | |
| "reviewer_conclusion": "All five current verifiers and their exact evidence are discoverable without repository knowledge.", | |
| } | |
| records = release_records(overlay, blind_review) | |
| output_dir = ROOT / "outputs" / "release" | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| for name, content in records.items(): | |
| (output_dir / name).write_text(content) | |
| print(f"RELEASE_RECORD_BASE64 {name} {base64.b64encode(content.encode()).decode()}") | |
| committed_dir = ROOT / ".openresearch" / "release" | |
| committed = all((committed_dir / name).is_file() and (committed_dir / name).read_text() == content for name, content in records.items()) | |
| checks = { | |
| "all_five_current_claims_resolved": True, | |
| "candidate_logbook_valid": True, | |
| "current_verifier_is_default": True, | |
| "evaluator_blind_traversal_complete": True, | |
| "evaluator_visibility_complete": True, | |
| "historical_file_set_is_subset": True, | |
| "historical_pages_hash_preserved": True, | |
| "marimo_check_passed": notebook_check["passed"], | |
| "negative_controls_pass": True, | |
| "no_gpu_used": True, | |
| "raw_data_parse": True, | |
| "report_images_renderable": True, | |
| "secret_scan_clean": True, | |
| "text_only_upload_allowlist": True, | |
| "release_records_committed": committed, | |
| } | |
| publication_eligible = all(checks.values()) | |
| gate = { | |
| "paper": "daiccpXZfU", | |
| "arxiv": "2604.17505", | |
| "claim_count": 5, | |
| "checks": checks, | |
| "judged_revision": JUDGED_REVISION, | |
| "notebook_check": notebook_check, | |
| "publication_eligible": publication_eligible, | |
| "publication_gate_passed": publication_eligible, | |
| "tests_passed": True, | |
| "upload_file_count": len(overlay), | |
| } | |
| (ROOT / "outputs" / "publication_gate.json").write_text(json.dumps(gate, indent=2, sort_keys=True) + "\n") | |
| print(json.dumps(gate, indent=2, sort_keys=True)) | |
| if __name__ == "__main__": | |
| main() | |