scripts/robodojo_export.py

#!/usr/bin/env python3
"""Append the audited RoboDojo first pass without rebuilding prior benchmarks.

Reads retained results only. No simulator, policy, upload, or deployment runs.
Media encoding resumes from a private cache; catalogs change only after all 42
videos have been encoded and their original frames and frame rates verified.
"""
from __future__ import annotations

import argparse
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from copy import deepcopy
from datetime import datetime, timezone
import hashlib
import json
import math
from pathlib import Path
import re

from export_gallery import aggregate, encode, publish_metadata, read_json, sha256, title, write_json


RUN = "robodojo42_astra_firstpass_20260919"
MANIFEST_SHA256 = "fe327cacc7615e868096a025c1287a8ad070408851bac9104cf132c1349f6fd4"
GROUPS = {"generalization": 12, "memory": 6, "precision": 8, "long-horizon": 8, "open": 8}
GROUP_NAMES = {"generalization": "Generalization", "memory": "Memory", "precision": "Precision",
               "long-horizon": "Long horizon", "open": "Open"}
CAMERAS = ["cam_head", "cam_left_wrist", "cam_right_wrist"]
ENCODING = {"codec": "h264", "pixelFormat": "yuv420p", "crf": 26, "maxRate": "1500k",
            "bufferSize": "3000k", "preset": "medium", "fastStart": True,
            "resize": False, "preserveFramesAndFps": True}


def require(condition, message):
    if not condition:
        raise ValueError(message)


def digest(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def load_audited_robodojo(run_dir, output_root):
    """Return a whitelisted benchmark, encoding jobs, provenance and read hashes."""
    root, output = Path(run_dir).resolve(), Path(output_root).resolve()
    require(root != output and not root.is_relative_to(output) and not output.is_relative_to(root),
            "Source run and output gallery must be separate")
    require(root.name == RUN, "Expected the completed RoboDojo first-pass run")
    reads = {}

    def read(path):
        path = Path(path)
        value = read_json(path)
        reads[path] = sha256(path)
        return value

    manifest = read(root / "manifest.json")
    mh = digest({key: value for key, value in manifest.items() if key != "manifest_sha256"})
    require(mh == manifest.get("manifest_sha256") == MANIFEST_SHA256, "Frozen manifest differs")
    config, specs = manifest["config"], manifest["episodes"]
    require(config["model"] == "gpt-6-astra" and config["reasoning_effort"] == "high"
            and config["workers"] == 8 and config["episodes_per_task"] == 1
            and config["seed_base"] == 0 and config["max_tool_calls"] == 1500,
            "Unexpected RoboDojo policy protocol")
    names = {s["task_name"] for s in specs}
    require(len(specs) == len(names) == 42 and Counter(s["task_group"] for s in specs) == GROUPS,
            "Expected all 42 basic tasks and native groups")
    require(all(s["variant"] == "standard" and s["rollout_id"] == 0 and s["seed"] == 0
                and not s["task_name"].endswith("_random")
                and s["max_steps"] == s["native_horizon"] for s in specs), "Unexpected task variant or horizon")
    require(len({s["prompt"] for s in specs}) == 1, "Policy prompts differ by task")
    for relative, expected in manifest["provenance"]["code_files"].items():
        path = root / "source_snapshot" / relative
        require(path.resolve().is_relative_to(root / "source_snapshot") and sha256(path) == expected,
                "Frozen source snapshot differs")
        reads[path] = expected

    audit_path, boundary_path = root / "reports/final_audit.json", root / "reports/final_policy_boundary_review.json"
    audit, boundary, summary = read(audit_path), read(boundary_path), read(root / "reports/summary.json")
    require(audit.get("passed") is True and audit.get("scope") == "all_42_basic_tasks"
            and audit.get("audited_episodes") == 42 and audit.get("videos_decoded") is True
            and audit.get("maximum_simultaneous_policy_episodes") == 8 and audit.get("issues") == [],
            "A complete successful 42-task video audit is required")
    require(boundary.get("passed") is True and boundary.get("full_scope_complete") is True
            and boundary.get("scope") == "all_42_terminal_episodes"
            and boundary.get("terminal_episodes_reviewed") == 42 and boundary.get("issues") == [],
            "A complete successful policy-boundary audit is required")
    require(all(record.get("manifest_sha256") == mh for record in (audit, boundary, summary)),
            "Audit and summary manifest identities differ")
    require(summary["overall"]["valid"] == summary["overall"]["total_attempts"] == 42
            and summary["overall"]["retries"] == summary["overall"]["errors"] == 0,
            "Source collection has missing episodes, retries or errors")

    def indexed(rows):
        result = {row["task_name"]: row for row in rows}
        require(len(rows) == len(result) == 42 and set(result) == names, "Incomplete or duplicated audit task coverage")
        return result

    audited, bounded, summarized = map(indexed, (audit["episodes"], boundary["episodes"], summary["episodes"]))
    tasks, jobs, selections = [], [], []
    for spec in specs:
        name, key = spec["task_name"], spec["episode_key"]
        require(re.fullmatch(r"[A-Za-z][A-Za-z0-9_]*", name) and key == f"{name}_r00", "Invalid task identity")
        attempts = sorted((root / "episodes" / key).glob("attempt_*"))
        require(len(attempts) == 1 and attempts[0].name == "attempt_001", "Exactly one attempt is required")
        directory = attempts[0]
        result_path = directory / "result.json"
        result, env = read(result_path), read(directory / "env_result.json")
        a, b, s = audited[name], bounded[name], summarized[name]
        require(a["passed"] is True and b["passed"] is True and b["result_sha256"] == reads[result_path],
                "Result changed after the boundary audit")
        events = directory / "codex_events.jsonl"
        require(sha256(events) == b["event_log_sha256"], "Policy evidence changed after its audit")
        reads[events] = b["event_log_sha256"]
        require(result["manifest_sha256"] == env["manifest_sha256"] == mh
                and result["episode_key"] == env["episode_key"] == key,
                "Result assignment differs")
        require(result["status"] in {"success", "failure", "timeout"}
                and result["status"] == a["status"] == b["result_status"] == s["status"], "Unscored or changed outcome")
        require(result["environment"] == env and not env.get("error") and not env.get("invalid_native_layout")
                and env["policy_ready"] is True, "Invalid native environment result")
        require(type(env["success"]) is bool and env["success"] == result["success"]
                == (result["status"] == "success"), "Native success differs from the result")
        score = env["score"]
        require(type(score) in (int, float) and math.isfinite(score) and 0 <= score <= 1
                and score == a["score"] == s["native_score"], "Invalid native score")
        require(type(env["steps"]) is int and 0 < env["steps"] <= spec["max_steps"]
                and env["steps"] == a["steps"] == s["steps"] == result["steps"], "Native action count differs")
        require(env["seed"] == env["layout_id"] == 0, "Unexpected evaluation layout")
        policy_audit = result["audit"]
        require(policy_audit.get("stream_complete") and not policy_audit.get("isolation_violation")
                and not policy_audit.get("sensor_transport_error"), "Invalid policy sensor or isolation evidence")
        video = Path(env["video"]["path"]).resolve()
        require(video.is_relative_to(directory.resolve()) and video.is_file() and str(video) == a["video"],
                "Source video is outside its selected attempt")
        instruction_record, native = read(video.with_name("instruction.json")), read(video.with_name("native_state.json"))
        require(reads[video.with_name("native_state.json")] == env.get("native_state_sha256"),
                "Native checkpoint changed after its result was finalized")
        instruction = instruction_record["instruction"]
        require(isinstance(instruction, str) and instruction.strip() and len(instruction) <= 10000,
                "Native instruction is missing")
        ih = hashlib.sha256(instruction.encode()).hexdigest()
        require(instruction_record.get("augmented") is False and ih == instruction_record["sha256"]
                == env["instruction_sha256"] == native["instruction_sha256"] == b["instruction_sha256"]
                == a["instruction_sha256"] == policy_audit["instruction_sha256"], "Native instruction parity differs")
        require(all(native[field] == env[field] for field in ("steps", "score", "success")), "Native checkpoint differs")
        metadata = env["video"]
        require(metadata["closed"] is True and not metadata.get("error") and metadata["fps"] == 25
                and metadata["width"] == 1920 and metadata["height"] == 480
                and metadata["frames"] == env["steps"] + 1 == a["decoded_frames"]
                and metadata["cameras_left_to_right"] == CAMERAS, "Native video protocol differs")
        tid = f"robodojo_{name.lower()}"
        eid = f"{tid}_r00"
        suite = f"robodojo_{spec['task_group'].replace('-', '_')}"
        episode = {"id": eid, "index": 0, "status": result["status"], "seed": 0, "initStateId": None,
                   "layoutId": 0, "steps": env["steps"], "maxSteps": spec["max_steps"],
                   "toolCalls": policy_audit["mcp_calls"], "wallSeconds": result["wall_seconds"],
                   "nativeScore": score, "terminationReason": result["reason"],
                   "durationSeconds": metadata["frames"] / 25, "video": f"media/robodojo/{eid}.mp4",
                   "poster": f"media/robodojo/{eid}.jpg", "width": 1920, "height": 480, "frames": metadata["frames"]}
        task = {"id": tid, "name": title(name), "nativeTaskName": name, "taskGroup": spec["task_group"],
                "instruction": instruction, "instructionSha256": ih, "suite": suite,
                "suiteName": GROUP_NAMES[spec["task_group"]], "episodes": [episode],
                "successes": int(env["success"]), "failures": int(not env["success"]), "successRate": float(env["success"])}
        selection = {"episode": eid, "sourceEpisodeKey": key, "selectedAttempt": "attempt_001",
                     "resultSha256": reads[result_path], "sourceVideoSha256": sha256(video),
                     "instructionSha256": ih, "nativeStateSha256": reads[video.with_name("native_state.json")],
                     "nativeSuccess": env["success"], "nativeScore": score, "status": result["status"],
                     "steps": env["steps"], "seed": 0, "excludedAttempts": []}
        tasks.append(task)
        selections.append(selection)
        jobs.append({"source": video, "episode": episode, "benchmark": "robodojo", "expectedFps": "25",
                     "selection": selection, "outputRoot": output, "encoding": ENCODING})
    counts = aggregate([t["episodes"][0] for t in tasks], 42)
    require(counts["successes"] == 6 and counts["failures"] == 36 and counts["timeouts"] == 0,
            "Native first-pass aggregate differs")
    source = {"benchmark": "robodojo", "run": RUN, "manifestSha256": mh,
              "auditKind": "robodojo_firstpass", "auditSha256": reads[audit_path],
              "policyBoundaryAuditSha256": reads[boundary_path], "selectedEpisodes": 42,
              "excludedAttemptCount": 0, "attemptSelection": selections}
    benchmark = {"id": "robodojo", "name": "RoboDojo", "evaluationDate": "2026-09-19",
                 "summary": counts, "tasks": tasks, "suites": [],
                 "provenance": {k: source[k] for k in ("run", "manifestSha256", "auditKind", "auditSha256", "policyBoundaryAuditSha256")},
                 "protocol": {"label": "42 basic tasks · 1 episode per task", "episodesPerTask": 1,
                              "description": "42 basic tasks, one independent Codex episode each, with eight simultaneous policies. Native VLA instructions are delivered verbatim, with no extra task descriptions. Policies see RGB and robot proprioception; no camera calibration, depth, reward or native success labels. All failures are retained and no episode was retried.",
                              "cameras": ["Head camera", "Left wrist", "Right wrist"], "stepsLabel": "Native actions",
                              "trainingDemoNote": "RoboDojo training demonstrations have not been imported into this gallery.",
                              "videoNote": "25 fps: initial frame and every native action, with head, left-wrist and right-wrist views side by side. All recorded frames and native dimensions are preserved; thinking and transport delays are omitted."}}
    for group in GROUPS:
        selected = [t for t in tasks if t["taskGroup"] == group]
        benchmark["suites"].append({"id": selected[0]["suite"], "name": GROUP_NAMES[group],
                                    **aggregate([t["episodes"][0] for t in selected], len(selected))})
    return benchmark, jobs, source, reads


def import_results(run_dir, output, workers=4, plan_only=False):
    output = Path(output).resolve()
    gallery_path, report_path = output / "data/gallery.json", output / "data/export-report.json"
    gallery, report = read_json(gallery_path), read_json(report_path)
    unchanged = {gallery_path: sha256(gallery_path), report_path: sha256(report_path)}
    require(report.get("complete") is True, "Existing gallery export must be complete")
    historical = [b for b in gallery["benchmarks"] if b["id"] != "robodojo"]
    old_media = [r for r in report["media"] if r["benchmark"] != "robodojo"]
    require({b["id"] for b in historical} == {"libero", "robotwin", "robocasa"} and len(old_media) == 815,
            "Expected the existing complete 815-episode gallery")
    benchmark, jobs, source, frozen_reads = load_audited_robodojo(run_dir, output)
    print(json.dumps({"planned": 42, "existingEpisodesPreserved": 815, "summary": benchmark["summary"],
                      "sourceBytes": sum(j["source"].stat().st_size for j in jobs)}), flush=True)
    if plan_only:
        return
    cache_path = output / ".gallery-cache/robodojo-export.json"
    cached = read_json(cache_path) if cache_path.exists() else {}
    prior = {r["episode"]: r for r in cached.get("media", [])} if cached.get("encoding") == ENCODING else {}
    if report.get("encodingByBenchmark", {}).get("robodojo") == ENCODING:
        prior.update({r["episode"]: r for r in report["media"] if r["benchmark"] == "robodojo"})
    complete = {}
    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(encode, job, prior.get(job["episode"]["id"])): job for job in jobs}
        for future in as_completed(futures):
            record = future.result()
            complete[record["episode"]] = record
            write_json(cache_path, {"encoding": ENCODING, "manifestSha256": MANIFEST_SHA256,
                                    "media": [complete[k] for k in sorted(complete)]})
            print(json.dumps({"encoded": len(complete), "total": 42, "episode": record["episode"]}), flush=True)
    require(len(complete) == 42, "Incomplete RoboDojo encoding")
    for path, expected in {**unchanged, **frozen_reads}.items():
        require(sha256(path) == expected, "Source evidence or existing gallery changed during import")
    # Whitelist validation before these dictionaries enter the public catalog.
    from validate_gallery import public_metadata
    public_metadata(benchmark, "robodojo")
    public_metadata(source, "robodojo provenance")
    result_gallery, result_report = deepcopy(gallery), deepcopy(report)
    timestamp = datetime.now(timezone.utc).isoformat()
    result_gallery.update(generatedAt=timestamp, evaluationDate="2026-09-19",
                          evaluationDates=["2026-09-17", "2026-09-18", "2026-09-19"],
                          benchmarks=[*historical, benchmark])
    records = [*old_media, *[complete[j["episode"]["id"]] for j in jobs]]
    result_report.update(generatedAt=timestamp, complete=True,
                         runs=[*[r for r in report["runs"] if r["benchmark"] != "robodojo"], source],
                         media=records, expectedEpisodes=857, verifiedEpisodes=857)
    result_report["selectionRule"] = report["selectionRule"].split(" RoboDojo uses")[0] + " RoboDojo uses exactly one attempt for each of its 42 basic tasks, verified by the complete native/video and policy-boundary audits."
    result_report.setdefault("encodingByBenchmark", {})["robodojo"] = ENCODING
    for field in ("sourceBytes", "videoBytes", "posterBytes"):
        result_report[field] = sum(r[field] for r in records)
    result_report["publishedMediaBytes"] = result_report["videoBytes"] + result_report["posterBytes"]
    result_report["externalVideoBytes"] = sum(r["videoBytes"] for r in records if r["benchmark"] in result_report["videoHosting"])
    result_report["pagesMediaBytes"] = result_report["publishedMediaBytes"] - result_report["externalVideoBytes"]
    require(result_report["pagesMediaBytes"] < 800_000_000, "Presentation media exceeds the existing budget")
    result_report["validation"].update(expectedEpisodes=857, verifiedEpisodes=857,
                                       decodedFrameCount=sum(r["frames"] for r in records),
                                       allFramesAndFpsPreserved=all(r["verified"] for r in records),
                                       allBrowserH264Yuv420pFastStart=all(r["codec"] == "h264" and r["pixelFormat"] == "yuv420p" and r["fastStart"] for r in records),
                                       sourceFilesUnmodified=True, within800MBPagesMediaBudget=True)
    write_json(report_path, result_report)
    publish_metadata(output, result_gallery)
    print(json.dumps({"complete": True, "tasks": 497, "episodes": 857,
                      "robodojoVideoBytes": sum(r["videoBytes"] for r in complete.values()),
                      "robodojoFrames": sum(r["frames"] for r in complete.values())}), flush=True)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--run-dir", type=Path, required=True)
    parser.add_argument("--output", type=Path, default=Path(__file__).resolve().parents[1])
    parser.add_argument("--jobs", type=int, choices=range(1, 7), default=4)
    parser.add_argument("--plan-only", action="store_true")
    args = parser.parse_args()
    import_results(args.run_dir, args.output, args.jobs, args.plan_only)


if __name__ == "__main__":
    main()