diff --git a/README.md b/README.md index 216239fa..fa674508 100644 --- a/README.md +++ b/README.md @@ -211,17 +211,64 @@ It is important to note that the inference server and client must be deployed on ### Evaluation on LIBERO -Follow the official instructions to install LIBERO, then launch the server and client: +Original LIBERO and LIBERO-Plus use different task sets and trial protocols. +Select the protocol explicitly; their success rates are not directly comparable. + +For the original 10-task LIBERO suites, follow the +[official LIBERO](https://github.com/Lifelong-Robot-Learning/LIBERO) +installation instructions and launch the server and client below. This keeps +the existing evaluation behavior: 50 trials for each of the 10 tasks +(denominator: 500). ```bash # server bash evaluation/libero/launch_server.sh -# client +# original LIBERO client bash evaluation/libero/launch_client.sh ``` +For [LIBERO-Plus](https://github.com/sylvestf/LIBERO-plus), install its expanded +benchmark fork, keep the same inference server running, and use: + +```bash +bash evaluation/libero/launch_client_plus.sh \ + --checkpoint-id robbyant/lingbot-va-posttrain-libero@REVISION +``` + +Plus mode evaluates every perturbation variant in the selected suite exactly +once by default. It rejects any `--test-num` other than `1`, verifies that the +installed benchmark and `task_classification.json` have the same task count and +order, and reads the instruction from the BDDL environment rather than from a +metadata-bearing filename. The launcher forwards additional arguments to the +client. For an intentional shard, use a half-open slice such as: + +```bash +bash evaluation/libero/launch_client_plus.sh \ + --task-range 0 250 \ + --checkpoint-id robbyant/lingbot-va-posttrain-libero@REVISION +``` + +The observed denominator is then the size of that slice, but a shard is marked +non-reportable as a full-suite LIBERO-Plus score. Likewise, an interrupted full +run records its prefix success rate only as a diagnostic; the full-suite score +remains `null` until every planned variant has completed. The client pins the +official suite sizes and the semantic hash of `task_classification.json`, so a +partial or modified task set fails closed instead of being labeled LIBERO-Plus. + +Each run writes an immutable, fingerprinted manifest and an incremental summary +containing the selected protocol, task range, planned and completed denominators, +per-category numerators/denominators, score eligibility, code and benchmark +provenance, and classification file hashes. Re-running an identical command +resumes matching completed task results; a changed checkpoint, protocol, code, +benchmark, classification, or task range receives a different run fingerprint +and cannot be mixed into the old summary. Since the inference server is a +separate process, checkpoint identity cannot be detected automatically, so +`--checkpoint-id NAME@REVISION` is required in Plus mode. Original mode keeps +the legacy optional argument, but omitting it creates a fresh non-resumable run +identity to prevent two unknown checkpoints from sharing results. + ### Run Image to Video-Action Generation We also provide a script for image to video-action generation: diff --git a/evaluation/libero/client.py b/evaluation/libero/client.py index 338ef5d6..0ef73279 100644 --- a/evaluation/libero/client.py +++ b/evaluation/libero/client.py @@ -1,35 +1,71 @@ -import numpy as np -from wan_va.utils.Simple_Remote_Infer.deploy.websocket_client_policy import WebsocketClientPolicy import argparse -from libero.libero import benchmark +import importlib.metadata +import secrets +import subprocess import time -from libero.libero.envs import OffScreenRenderEnv +from collections import Counter +from datetime import datetime, timezone from pathlib import Path -from tqdm import tqdm -from lerobot.datasets.utils import write_json -import os -import imageio -import cv2 - -def save_video(real_obs_list, save_path, fps=15, video_names=["observation.images.agentview_rgb", "observation.images.eye_in_hand_rgb"]): +import cv2 +import imageio +import numpy as np +from libero.libero import benchmark +from libero.libero.envs import OffScreenRenderEnv +from tqdm import tqdm +from wan_va.utils.Simple_Remote_Infer.deploy.websocket_client_policy import ( + WebsocketClientPolicy, +) + +from eval_protocol import ( + ClassificationBundle, + EvaluationProtocolError, + LIBERO_PLUS_CLASSIFICATION_SEMANTIC_SHA256, + TaskMetadata, + add_protocol_arguments, + aggregate_task_results, + canonical_json_sha256, + describe_evaluation_status, + load_plus_task_metadata, + load_or_create_manifest, + load_resumable_task_result, + normalize_checkpoint_id, + resolve_evaluation_plan, + resolve_prompt, + sha256_file, + validate_benchmark_shape, + write_json_atomic, +) + + +def save_video( + real_obs_list, + save_path, + fps=15, + video_names=( + "observation.images.agentview_rgb", + "observation.images.eye_in_hand_rgb", + ), +): if not real_obs_list: - print("❌ No real observation frames") + print("No real observation frames; skipping video") return first_obs = real_obs_list[0] base_h, width_base = first_obs[video_names[0]].shape[:2] target_size = (width_base, base_h) - + print(f"Saving video: {len(real_obs_list)} frames...") final_frames = [ - np.hstack([cv2.resize(obs[name], target_size) for name in video_names]).astype(np.uint8) + np.hstack([cv2.resize(obs[name], target_size) for name in video_names]).astype( + np.uint8 + ) for obs in real_obs_list ] imageio.mimsave(save_path, final_frames, fps=fps) - print(f"✅ Video saved to: {save_path}") + print(f"Video saved to: {save_path}") def construct_single_env(env_args): @@ -40,8 +76,8 @@ def construct_single_env(env_args): try: env = OffScreenRenderEnv(**env_args) env_creation = True - except Exception as e: - print(f"Error!!! construct env failed: {e}") + except Exception as exc: + print(f"Error: constructing environment failed: {exc}") time.sleep(5) count += 1 if count >= 5: @@ -50,22 +86,21 @@ def construct_single_env(env_args): def _extract_obs(obs): - """ - Extract agentview and eye_in_hand images from raw env obs dict. + """Extract and vertically flip the two uint8 camera observations.""" - Avoids torch round-trip: the env already returns uint8 numpy arrays [H, W, C]. - We just flip the vertical axis ([::-1]) and make a contiguous copy once. - """ agentview = np.ascontiguousarray(obs["agentview_image"][::-1]) eye_in_hand = np.ascontiguousarray(obs["robot0_eye_in_hand_image"][::-1]) - return {"observation.images.agentview_rgb": agentview, "observation.images.eye_in_hand_rgb": eye_in_hand} + return { + "observation.images.agentview_rgb": agentview, + "observation.images.eye_in_hand_rgb": eye_in_hand, + } def init_single_env(env_in, init_state): env_in.reset() env_in.set_init_state(init_state) for _ in range(5): - obs, _, _, _ = env_in.step([0.] * 7) + obs, _, _, _ = env_in.step([0.0] * 7) return _extract_obs(obs) @@ -74,150 +109,522 @@ def env_one_step(env_in, action): return _extract_obs(obs), done -def run_one(model, libero_benchmark, task_idx, out_dir, episode_idx): - benchmark_dict = benchmark.get_benchmark_dict() - benchmark_instance = benchmark_dict[libero_benchmark]() +def run_one( + model, + benchmark_instance, + protocol, + task_idx, + video_root, + episode_idx, +): num_tasks = benchmark_instance.get_num_tasks() - assert task_idx < num_tasks, f"Error: error id must smaller than {num_tasks}" - prompt = benchmark_instance.get_task(task_idx).language + if task_idx >= num_tasks: + raise EvaluationProtocolError( + f"Task index {task_idx} must be smaller than {num_tasks}." + ) + + task = benchmark_instance.get_task(task_idx) env_args = { - "bddl_file_name": benchmark_instance.get_task_bddl_file_path(task_idx), - "camera_heights": 128, - "camera_widths": 128, - } + "bddl_file_name": benchmark_instance.get_task_bddl_file_path(task_idx), + "camera_heights": 128, + "camera_widths": 128, + } init_states = benchmark_instance.get_task_init_states(task_idx) cur_env = construct_single_env(env_args) - first_obs = init_single_env(cur_env, init_states[episode_idx % init_states.shape[0]]) - - ret = model.infer(dict(reset=True, prompt=prompt)) - - full_obs_list = [] - done = False - first = True - while cur_env.env.timestep < 800: - ret = model.infer(dict(obs=first_obs, prompt=prompt)) - action = ret['action'] - - key_frame_list = [] - assert action.shape[2] % 4 == 0 - action_per_frame = action.shape[2] // 4 - start_idx = 1 if first else 0 - for i in range(start_idx, action.shape[1]): - for j in range(action.shape[2]): - ee_action = action[:, i, j] - observes, done = env_one_step(cur_env, ee_action) + if cur_env is None: + raise RuntimeError( + f"Could not construct environment for task index {task_idx} after 5 tries." + ) + + try: + prompt = resolve_prompt( + protocol, + benchmark_prompt=task.language, + environment_prompt=getattr(cur_env, "language_instruction", None), + ) + first_obs = init_single_env( + cur_env, init_states[episode_idx % init_states.shape[0]] + ) + + model.infer(dict(reset=True, prompt=prompt)) + + full_obs_list = [] + done = False + first = True + while cur_env.env.timestep < 800: + ret = model.infer(dict(obs=first_obs, prompt=prompt)) + action = ret["action"] + + key_frame_list = [] + assert action.shape[2] % 4 == 0 + action_per_frame = action.shape[2] // 4 + start_idx = 1 if first else 0 + for i in range(start_idx, action.shape[1]): + for j in range(action.shape[2]): + ee_action = action[:, i, j] + observes, done = env_one_step(cur_env, ee_action) + if done: + break + if (j + 1) % action_per_frame == 0: + full_obs_list.append(observes) + key_frame_list.append(observes) + if done: break - if (j+1) % action_per_frame == 0: - full_obs_list.append(observes) - key_frame_list.append(observes) + + first = False if done: break - - first = False - - if done: - break - else: - model.infer(dict(obs=key_frame_list, compute_kv_cache=True, imagine=False, state=action)) - - out_file = Path(out_dir) / libero_benchmark / f"{task_idx}_{prompt.replace(' ', '_')}" / f"{episode_idx}_{done}.mp4" + model.infer( + dict( + obs=key_frame_list, + compute_kv_cache=True, + imagine=False, + state=action, + ) + ) + finally: + cur_env.close() + + artifact_name = task.name if protocol == "plus" else prompt.replace(" ", "_") + out_file = ( + Path(video_root) + / f"{task_idx}_{artifact_name}" + / f"{episode_idx}_{bool(done)}.mp4" + ) out_file.parent.mkdir(exist_ok=True, parents=True) save_video( real_obs_list=full_obs_list, save_path=out_file, fps=60, - video_names=["observation.images.agentview_rgb", "observation.images.eye_in_hand_rgb"] + video_names=( + "observation.images.agentview_rgb", + "observation.images.eye_in_hand_rgb", + ), ) - cur_env.close() - return done + return bool(done), prompt -def run(libero_benchmark, port, out_dir, test_num, task_range=None): - ''' - task_range: [start, end) for splitting tasks - ''' - if task_range is None: - benchmark_dict = benchmark.get_benchmark_dict() - benchmark_instance = benchmark_dict[libero_benchmark]() - num_tasks = benchmark_instance.get_num_tasks() - progress_bar = tqdm(range(num_tasks), total=num_tasks) +def _git_revision(path): + try: + result = subprocess.run( + ["git", "-C", str(path), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + timeout=5, + ) + except (OSError, subprocess.SubprocessError): + return None + return result.stdout.strip() or None + + +def _installed_versions(): + versions = {} + for distribution in ("libero", "robosuite", "mujoco", "numpy", "torch"): + try: + versions[distribution] = importlib.metadata.version(distribution) + except importlib.metadata.PackageNotFoundError: + versions[distribution] = None + return versions + + +def _build_provenance(classification, checkpoint_id): + client_path = Path(__file__).resolve() + repo_root = client_path.parents[2] + benchmark_path = Path(benchmark.__file__).resolve() + protocol_path = client_path.with_name("eval_protocol.py") + + classification_provenance = None + if classification is not None: + classification_provenance = { + "path": classification.source, + "sha256": classification.sha256, + "semantic_sha256": classification.semantic_sha256, + } + + return { + "lingbot_va_git_commit": _git_revision(repo_root), + "client_path": str(client_path), + "client_sha256": sha256_file(client_path), + "protocol_sha256": sha256_file(protocol_path), + "libero_benchmark_module": str(benchmark_path), + "libero_benchmark_module_sha256": sha256_file(benchmark_path), + "libero_benchmark_git_commit": _git_revision(benchmark_path.parent), + "task_classification": classification_provenance, + "checkpoint_id_declared_by_user": checkpoint_id, + "package_versions": _installed_versions(), + } + + +def _planned_category_counts(plan, task_metadata): + counts = Counter(task_metadata[task_idx].category for task_idx in plan.task_indices) + return { + category: { + f"{plan.selection_unit}s": count, + "denominator": count * plan.trials_per_task, + } + for category, count in sorted(counts.items()) + } + + +def _result_path(out_dir, protocol, suite, task_idx, run_root=None): + out_dir = Path(out_dir) + if protocol == "original": + # Preserve the existing result path for downstream original-LIBERO users. + return out_dir / f"{suite}_{task_idx}.json" + if run_root is None: + raise ValueError("run_root is required for collision-free LIBERO-Plus output.") + return Path(run_root) / "tasks" / f"{suite}_{task_idx}.json" + + +def run( + libero_benchmark, + port, + out_dir, + test_num=None, + task_range=None, + *, + protocol="original", + task_classification=None, + checkpoint_id=None, +): + """Run one explicitly selected LIBERO evaluation protocol.""" + + checkpoint_id = normalize_checkpoint_id(protocol, checkpoint_id) + + # Preserve the original CLI's optional checkpoint metadata while preventing + # two unidentified original-LIBERO invocations from sharing a resume key. + unidentified_session_nonce = ( + secrets.token_hex(16) if checkpoint_id is None else None + ) + + benchmark_dict = benchmark.get_benchmark_dict() + benchmark_instance = benchmark_dict[libero_benchmark]() + benchmark_total_tasks = benchmark_instance.get_num_tasks() + + validate_benchmark_shape( + protocol, + benchmark_total_tasks, + suite=libero_benchmark, + ) + plan = resolve_evaluation_plan( + protocol=protocol, + benchmark_total_tasks=benchmark_total_tasks, + task_range=task_range, + test_num=test_num, + ) + + task_names = [ + benchmark_instance.get_task(task_idx).name + for task_idx in range(benchmark_total_tasks) + ] + classification: ClassificationBundle | None = None + if protocol == "plus": + classification_path = ( + Path(task_classification) + if task_classification is not None + else Path(benchmark.__file__) + .resolve() + .with_name("task_classification.json") + ) + classification = load_plus_task_metadata( + classification_path, + suite=libero_benchmark, + task_names=task_names, + expected_semantic_sha256=(LIBERO_PLUS_CLASSIFICATION_SEMANTIC_SHA256), + ) + task_metadata = classification.tasks else: - assert len(task_range) == 2, f'task_range: [start, end) for splitting tasks, however, task_range: {task_range}' - num_tasks = task_range[1] - task_range[0] - progress_bar = tqdm(range(task_range[0], task_range[1]), total=num_tasks) + task_metadata = tuple( + TaskMetadata( + task_id=task_idx + 1, + name=task_name, + category="Original LIBERO", + difficulty_level=None, + ) + for task_idx, task_name in enumerate(task_names) + ) - print(f"#################### Use benchmark: {libero_benchmark}, num_tasks: {num_tasks} #############") - model = WebsocketClientPolicy(port=port) + output_root = Path(out_dir) + output_root.mkdir(exist_ok=True, parents=True) - video_save_root_dict = None + provenance = _build_provenance(classification, checkpoint_id) + classification_semantic_sha256 = ( + classification.semantic_sha256 if classification is not None else None + ) + run_identity = { + "schema_version": 1, + "benchmark_suite": libero_benchmark, + "evaluation_plan": plan.to_dict(), + "checkpoint_id_declared_by_user": checkpoint_id, + "unidentified_session_nonce": unidentified_session_nonce, + "code": { + "lingbot_va_git_commit": provenance["lingbot_va_git_commit"], + "client_sha256": provenance["client_sha256"], + "protocol_sha256": provenance["protocol_sha256"], + }, + "benchmark": { + "git_commit": provenance["libero_benchmark_git_commit"], + "module_sha256": provenance["libero_benchmark_module_sha256"], + "task_order_sha256": canonical_json_sha256(task_names), + "task_classification_semantic_sha256": (classification_semantic_sha256), + "package_versions": provenance["package_versions"], + }, + } + run_fingerprint = canonical_json_sha256(run_identity) + run_stem = ( + f"{libero_benchmark}_{protocol}_{plan.task_start}-{plan.task_end}_" + f"{run_fingerprint[:12]}" + ) + if protocol == "plus": + run_root = output_root / "plus" / "runs" / run_stem + video_root = run_root / "videos" / libero_benchmark + else: + run_root = output_root / "runs" / run_stem + # Preserve the existing original-LIBERO video location. + video_root = output_root / libero_benchmark + + manifest_path = run_root / "manifest.json" + summary_path = run_root / "summary.json" + manifest_reference = manifest_path.relative_to(output_root).as_posix() + manifest = { + "schema_version": 1, + "created_at_utc": datetime.now(timezone.utc).isoformat(), + "benchmark_suite": libero_benchmark, + "evaluation_plan": plan.to_dict(), + "planned_by_category": _planned_category_counts(plan, task_metadata), + "run_fingerprint": run_fingerprint, + "run_identity": run_identity, + "prompt_source": ( + "benchmark task language (original behavior)" + if protocol == "original" + else "BDDL :language via OffScreenRenderEnv.language_instruction" + ), + "provenance": provenance, + } + manifest = load_or_create_manifest(manifest, manifest_path) + manifest_sha256 = sha256_file(manifest_path) + + print( + "Evaluation plan: " + f"protocol={protocol}, suite={libero_benchmark}, " + f"selected_{plan.selection_unit}s={plan.selected_task_count}, " + f"trials_per_{plan.selection_unit}={plan.trials_per_task}, " + f"denominator={plan.denominator}" + ) + print(f"Run manifest: {manifest_path}") + + task_results = [] + model = None + + def write_summary(): + completed = aggregate_task_results( + task_results, + selection_unit=plan.selection_unit, + ) + evaluation_status = describe_evaluation_status(plan, completed) + observed_success_rate = completed["observed_success_rate"] + full_suite_success_rate = ( + observed_success_rate + if evaluation_status["reportable_as_full_suite_score"] + else None + ) + summary = { + "schema_version": 1, + "benchmark_suite": libero_benchmark, + "protocol": protocol, + "run_fingerprint": run_fingerprint, + "planned": plan.to_dict(), + "completed": completed, + "evaluation_status": evaluation_status, + "score": { + "metric": "micro_success_rate", + "value": full_suite_success_rate, + "reportable": evaluation_status["reportable_as_full_suite_score"], + "reason": evaluation_status["score_reason"], + }, + "planned_by_category": manifest["planned_by_category"], + "provenance": { + "manifest": manifest_reference, + "manifest_sha256": manifest_sha256, + "task_classification_sha256": ( + classification.sha256 if classification is not None else None + ), + "task_classification_semantic_sha256": (classification_semantic_sha256), + }, + } + write_json_atomic(summary, summary_path) + return summary + + summary = write_summary() + progress_bar = tqdm( + plan.task_indices, + total=plan.selected_task_count, + ) - episode_list = range(test_num) for task_idx in progress_bar: - if video_save_root_dict is not None and task_idx in video_save_root_dict: - video_save_list = os.listdir(os.path.join(out_dir, libero_benchmark, video_save_root_dict[task_idx])) - video_states = [1 for file in video_save_list if file.split('_')[1].split('.')[0] == 'True'] - succ_num = float(len(video_states)) - episode_list = range(len(video_save_list), test_num) + metadata = task_metadata[task_idx] + result_path = _result_path( + out_dir=output_root, + protocol=protocol, + suite=libero_benchmark, + task_idx=task_idx, + run_root=run_root, + ) + expected_task_fields = { + "protocol": protocol, + "suite": libero_benchmark, + "task_index": task_idx, + "task_id": metadata.task_id, + "task_name": metadata.name, + "category": metadata.category, + "difficulty_level": metadata.difficulty_level, + } + task_result = load_resumable_task_result( + result_path, + run_fingerprint=run_fingerprint, + manifest_reference=manifest_reference, + manifest_sha256=manifest_sha256, + expected_task_fields=expected_task_fields, + trials_per_task=plan.trials_per_task, + ) + if task_result is None: + successes = 0 + completed_trials = 0 else: - succ_num = 0. - - for episode_idx in tqdm(episode_list, total=len(episode_list)): - res_i = run_one(model, libero_benchmark, task_idx, out_dir, episode_idx) - succ_num += res_i - succ_rate = succ_num / (episode_idx + 1) - print(f"Success rate: {succ_rate}, success num: {succ_num}, total num: {episode_idx + 1}") - out_file = Path(out_dir) / f"{libero_benchmark}_{task_idx}.json" - out_file.parent.mkdir(exist_ok=True, parents=True) - write_json({ - "succ_num": succ_num, - "total_num": episode_idx + 1., - "succ_rate": succ_rate, - }, out_file + successes = task_result["successes"] + completed_trials = task_result["denominator"] + print( + f"Resuming task {task_idx} from " + f"{completed_trials}/{plan.trials_per_task} completed trials." + ) + + for episode_idx in tqdm( + range(completed_trials, plan.trials_per_task), + total=plan.trials_per_task - completed_trials, + ): + if model is None: + model = WebsocketClientPolicy(port=port) + succeeded, prompt = run_one( + model=model, + benchmark_instance=benchmark_instance, + protocol=protocol, + task_idx=task_idx, + video_root=video_root, + episode_idx=episode_idx, + ) + successes += int(succeeded) + completed_trials = episode_idx + 1 + success_rate = successes / completed_trials + task_result = { + # Legacy keys remain for original-LIBERO result consumers. + "succ_num": float(successes), + "total_num": float(completed_trials), + "succ_rate": success_rate, + # Protocol-explicit result schema. + "protocol": protocol, + "suite": libero_benchmark, + "task_index": task_idx, + "task_id": metadata.task_id, + "task_name": metadata.name, + "category": metadata.category, + "difficulty_level": metadata.difficulty_level, + "prompt": prompt, + "successes": successes, + "denominator": completed_trials, + "success_rate": success_rate, + "run_fingerprint": run_fingerprint, + "provenance": { + "manifest": manifest_reference, + "manifest_sha256": manifest_sha256, + "task_classification_sha256": ( + classification.sha256 if classification is not None else None + ), + "task_classification_semantic_sha256": ( + classification_semantic_sha256 + ), + }, + } + write_json_atomic(task_result, result_path) + print( + f"Task {task_idx} [{metadata.category}]: " + f"{successes}/{completed_trials} " + f"({success_rate:.4f})" ) + if task_result is None: + raise AssertionError("Validated evaluation plan produced no trials.") + task_results.append(task_result) + summary = write_summary() + completed = summary["completed"] + observed_rate = completed["observed_success_rate"] + print( + "Observed completed entries: " + f"{completed['successes']}/{completed['denominator']} " + f"({observed_rate:.4f}); planned denominator={plan.denominator}; " + f"score reportable={summary['score']['reportable']}" + ) -def main(): + return summary + + +def build_parser(): parser = argparse.ArgumentParser() parser.add_argument( "--libero-benchmark", type=str, default="libero_10", choices=["libero_10", "libero_goal", "libero_spatial", "libero_object"], - help="Benchmark name", - ) - parser.add_argument( - "--task-range", - type=int, - nargs="+", - default=[0, 10], - help="Task range [start, end) for splitting tasks", + help="Benchmark suite name", ) + add_protocol_arguments(parser) parser.add_argument( "--port", type=int, default=23908, help="WebSocket port", ) - parser.add_argument( - "--test-num", - type=int, - default=50, - help="Number of test episodes", - ) parser.add_argument( "--out-dir", type=str, default="outputs/libero", - help="Output directory for results", + help="Output directory for videos, manifests, and results", + ) + parser.add_argument( + "--task-classification", + type=str, + default=None, + help=( + "Optional LIBERO-Plus task_classification.json override. By default, " + "the file next to the installed benchmark module is used." + ), + ) + parser.add_argument( + "--checkpoint-id", + type=str, + default=None, + help=( + "Checkpoint name/revision recorded as user-declared provenance. " + "Required for LIBERO-Plus; the separate inference server cannot " + "verify it automatically." + ), ) + return parser + + +def main(): + parser = build_parser() args = parser.parse_args() - run(**vars(args)) - print("Finish all process!!!!!!!!!!!!") + try: + run(**vars(args)) + except EvaluationProtocolError as exc: + parser.error(str(exc)) + print("Finished all evaluation tasks.") if __name__ == "__main__": diff --git a/evaluation/libero/eval_protocol.py b/evaluation/libero/eval_protocol.py new file mode 100644 index 00000000..6117d848 --- /dev/null +++ b/evaluation/libero/eval_protocol.py @@ -0,0 +1,674 @@ +"""Protocol planning and result helpers for LIBERO evaluation. + +This module intentionally depends only on the Python standard library so that +the protocol invariants can be tested without installing MuJoCo, LIBERO, or the +LingBot-VA runtime. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable, Mapping, Sequence + + +PROTOCOLS = ("original", "plus") +ORIGINAL_SUITE_TASK_COUNT = 10 +LIBERO_PLUS_SUITE_TASK_COUNTS = { + "libero_spatial": 2402, + "libero_object": 2518, + "libero_goal": 2591, + "libero_10": 2519, +} +# Canonical JSON hash of task_classification.json at LIBERO-Plus commit +# 4976dc30028e805ff8094b55501d532c48fec182. Canonicalization makes this +# independent of whitespace and line-ending conversion. +LIBERO_PLUS_CLASSIFICATION_SEMANTIC_SHA256 = ( + "84b63b9d836146286d62f6d2aafea15a8a68fd197bec1cba36930f22da9143ce" +) + + +class EvaluationProtocolError(ValueError): + """Raised when an evaluation request mixes incompatible protocols.""" + + +@dataclass(frozen=True) +class EvaluationPlan: + """Resolved, validated task and trial selection.""" + + protocol: str + benchmark_total_tasks: int + task_start: int + task_end: int + trials_per_task: int + + @property + def selected_task_count(self) -> int: + return self.task_end - self.task_start + + @property + def denominator(self) -> int: + return self.selected_task_count * self.trials_per_task + + @property + def selection_unit(self) -> str: + return "task" if self.protocol == "original" else "variant" + + @property + def task_indices(self) -> range: + return range(self.task_start, self.task_end) + + def to_dict(self) -> dict[str, Any]: + result = { + "protocol": self.protocol, + "benchmark_total_tasks": self.benchmark_total_tasks, + "task_range": [self.task_start, self.task_end], + "selection_unit": self.selection_unit, + "selected_entries": self.selected_task_count, + "trials_per_entry": self.trials_per_task, + "denominator": self.denominator, + } + result[f"selected_{self.selection_unit}s"] = self.selected_task_count + result[f"trials_per_{self.selection_unit}"] = self.trials_per_task + return result + + +@dataclass(frozen=True) +class TaskMetadata: + """Stable task metadata written next to each evaluation result.""" + + task_id: int + name: str + category: str + difficulty_level: int | None + + +@dataclass(frozen=True) +class ClassificationBundle: + """Validated LIBERO-Plus classification data and its provenance.""" + + tasks: tuple[TaskMetadata, ...] + source: str + sha256: str + semantic_sha256: str + + +def add_protocol_arguments(parser: argparse.ArgumentParser) -> None: + """Add protocol arguments shared by the command line client and tests.""" + + parser.add_argument( + "--protocol", + choices=PROTOCOLS, + default="original", + help=( + "Evaluation protocol. 'original' uses 50 trials per task; 'plus' " + "uses exactly one trial per perturbation variant." + ), + ) + parser.add_argument( + "--task-range", + type=int, + nargs=2, + default=None, + metavar=("START", "END"), + help=( + "Optional half-open task/variant slice [START, END). By default, " + "all tasks or variants in the selected suite are evaluated." + ), + ) + parser.add_argument( + "--test-num", + type=int, + default=None, + help=( + "Trials per task. Defaults to 50 for original LIBERO and 1 for " + "LIBERO-Plus. Values other than 1 are rejected in plus mode." + ), + ) + + +def normalize_checkpoint_id(protocol: str, checkpoint_id: str | None) -> str | None: + """Normalize declared checkpoint identity and require it for Plus resume.""" + + if protocol not in PROTOCOLS: + raise EvaluationProtocolError( + f"Unknown protocol {protocol!r}; expected one of {PROTOCOLS}." + ) + if checkpoint_id is None: + if protocol == "plus": + raise EvaluationProtocolError( + "LIBERO-Plus requires --checkpoint-id NAME@REVISION so an " + "interrupted run cannot resume results from an unidentified " + "checkpoint." + ) + return None + if not isinstance(checkpoint_id, str) or not checkpoint_id.strip(): + raise EvaluationProtocolError( + "--checkpoint-id must be non-empty when it is provided." + ) + return checkpoint_id.strip() + + +def resolve_evaluation_plan( + protocol: str, + benchmark_total_tasks: int, + task_range: Sequence[int] | None = None, + test_num: int | None = None, +) -> EvaluationPlan: + """Resolve defaults and reject protocol/task/trial mismatches.""" + + if protocol not in PROTOCOLS: + raise EvaluationProtocolError( + f"Unknown protocol {protocol!r}; expected one of {PROTOCOLS}." + ) + if benchmark_total_tasks <= 0: + raise EvaluationProtocolError( + f"Benchmark must contain tasks, got {benchmark_total_tasks}." + ) + + trials_per_task = ( + (50 if protocol == "original" else 1) if test_num is None else test_num + ) + if trials_per_task <= 0: + raise EvaluationProtocolError( + f"--test-num must be positive, got {trials_per_task}." + ) + if protocol == "plus" and trials_per_task != 1: + raise EvaluationProtocolError( + "LIBERO-Plus requires exactly one trial per variant. Omit " + "--test-num or set --test-num 1." + ) + + if task_range is None: + task_start, task_end = 0, benchmark_total_tasks + else: + if len(task_range) != 2: + raise EvaluationProtocolError( + "--task-range must contain exactly START END for [START, END)." + ) + task_start, task_end = task_range + + if task_start < 0 or task_start >= task_end or task_end > benchmark_total_tasks: + raise EvaluationProtocolError( + "Invalid --task-range " + f"[{task_start}, {task_end}) for a benchmark with " + f"{benchmark_total_tasks} tasks." + ) + + return EvaluationPlan( + protocol=protocol, + benchmark_total_tasks=benchmark_total_tasks, + task_start=task_start, + task_end=task_end, + trials_per_task=trials_per_task, + ) + + +def validate_benchmark_shape( + protocol: str, + benchmark_total_tasks: int, + suite: str | None = None, +) -> None: + """Fail closed when the installed LIBERO package is for another protocol.""" + + if protocol == "original" and benchmark_total_tasks != ORIGINAL_SUITE_TASK_COUNT: + raise EvaluationProtocolError( + "The original LIBERO protocol expects a 10-task suite, but the " + f"installed benchmark exposes {benchmark_total_tasks} tasks. This " + "looks like LIBERO-Plus; rerun with --protocol plus or install the " + "original LIBERO package." + ) + if protocol == "plus": + if suite is None: + raise EvaluationProtocolError( + "LIBERO-Plus validation requires the benchmark suite name so " + "the exact official variant count can be checked." + ) + expected_count = LIBERO_PLUS_SUITE_TASK_COUNTS.get(suite) + if expected_count is None: + raise EvaluationProtocolError( + f"No canonical LIBERO-Plus task count is registered for {suite!r}." + ) + if benchmark_total_tasks == expected_count: + return + raise EvaluationProtocolError( + f"Canonical LIBERO-Plus {suite} contains exactly {expected_count} " + f"variants, but the installed benchmark exposes {benchmark_total_tasks}. " + "Refusing to label a partial or modified task set as LIBERO-Plus." + ) + + +def sha256_file(path: str | Path) -> str: + """Return a content hash suitable for benchmark provenance.""" + + digest = hashlib.sha256() + with Path(path).open("rb") as file_obj: + for chunk in iter(lambda: file_obj.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def canonical_json_sha256(payload: Any) -> str: + """Hash JSON data independently of whitespace, indentation, and key order.""" + + canonical = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + return hashlib.sha256(canonical).hexdigest() + + +def load_plus_task_metadata( + classification_path: str | Path, + suite: str, + task_names: Sequence[str], + expected_semantic_sha256: str | None = None, +) -> ClassificationBundle: + """Load and exactly align LIBERO-Plus task classification metadata. + + The exact length, 1-based IDs, and task names are checked before any + rollout. This prevents a stale or mismatched classification file from + silently assigning the wrong perturbation categories. + """ + + path = Path(classification_path).resolve() + if not path.is_file(): + raise EvaluationProtocolError( + "LIBERO-Plus task classification file was not found at " + f"{path}. Install the official LIBERO-Plus benchmark or pass the " + "matching --task-classification file." + ) + + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise EvaluationProtocolError( + f"Could not read LIBERO-Plus classification file {path}: {exc}" + ) from exc + + semantic_sha256 = canonical_json_sha256(payload) + if ( + expected_semantic_sha256 is not None + and semantic_sha256 != expected_semantic_sha256 + ): + raise EvaluationProtocolError( + "LIBERO-Plus task metadata does not match the pinned official " + "classification. " + f"Expected semantic SHA-256 {expected_semantic_sha256}, got " + f"{semantic_sha256} from {path}." + ) + + if not isinstance(payload, dict) or suite not in payload: + raise EvaluationProtocolError( + f"Classification file {path} has no entry for suite {suite!r}." + ) + raw_tasks = payload[suite] + if not isinstance(raw_tasks, list): + raise EvaluationProtocolError( + f"Classification entry for {suite!r} must be a list." + ) + if len(raw_tasks) != len(task_names): + raise EvaluationProtocolError( + "LIBERO-Plus classification/benchmark size mismatch for " + f"{suite}: {len(raw_tasks)} metadata entries versus " + f"{len(task_names)} benchmark variants." + ) + + tasks: list[TaskMetadata] = [] + for task_index, (raw_task, benchmark_name) in enumerate( + zip(raw_tasks, task_names, strict=True) + ): + if not isinstance(raw_task, dict): + raise EvaluationProtocolError( + f"Classification entry {task_index} for {suite} is not an object." + ) + + expected_id = task_index + 1 + task_id = raw_task.get("id") + name = raw_task.get("name") + category = raw_task.get("category") + difficulty_level = raw_task.get("difficulty_level") + + if task_id != expected_id: + raise EvaluationProtocolError( + f"Classification entry {task_index} has id {task_id!r}; " + f"expected the 1-based id {expected_id}." + ) + if name != benchmark_name: + raise EvaluationProtocolError( + "LIBERO-Plus classification/benchmark order mismatch at index " + f"{task_index}: metadata={name!r}, benchmark={benchmark_name!r}." + ) + if not isinstance(category, str) or not category.strip(): + raise EvaluationProtocolError( + f"Classification entry {task_index} has no valid category." + ) + if difficulty_level is not None and ( + not isinstance(difficulty_level, int) or isinstance(difficulty_level, bool) + ): + raise EvaluationProtocolError( + "Classification entry " + f"{task_index} has invalid difficulty_level " + f"{difficulty_level!r}." + ) + + tasks.append( + TaskMetadata( + task_id=task_id, + name=name, + category=category, + difficulty_level=difficulty_level, + ) + ) + + return ClassificationBundle( + tasks=tuple(tasks), + source=str(path), + sha256=sha256_file(path), + semantic_sha256=semantic_sha256, + ) + + +def resolve_prompt( + protocol: str, + benchmark_prompt: str, + environment_prompt: str | None, +) -> str: + """Use BDDL language in Plus mode without changing original behavior. + + Some LIBERO-Plus task filenames encode perturbation metadata. The BDDL + ``:language`` field is the task instruction and is exposed by the created + environment, so Plus evaluation uses it rather than filename-derived text. + """ + + if protocol == "original": + return benchmark_prompt + if not isinstance(environment_prompt, str) or not environment_prompt.strip(): + raise EvaluationProtocolError( + "LIBERO-Plus environment did not expose a non-empty BDDL language " + "instruction; refusing to evaluate with filename-derived text." + ) + return environment_prompt.strip() + + +def aggregate_task_results( + task_results: Iterable[Mapping[str, Any]], + selection_unit: str = "variant", +) -> dict[str, Any]: + """Aggregate explicit success numerators and rollout denominators.""" + + if selection_unit not in {"task", "variant"}: + raise EvaluationProtocolError( + f"Unknown selection unit {selection_unit!r}; expected task or variant." + ) + + successes = 0 + denominator = 0 + completed_entries = 0 + by_category: dict[str, dict[str, int]] = {} + + for result in task_results: + task_successes = result.get("successes") + task_denominator = result.get("denominator") + if ( + not isinstance(task_successes, int) + or isinstance(task_successes, bool) + or not isinstance(task_denominator, int) + or isinstance(task_denominator, bool) + or task_successes < 0 + or task_denominator <= 0 + or task_successes > task_denominator + ): + raise EvaluationProtocolError( + "Each task result must have integer 0 <= successes <= " + "denominator and denominator > 0." + ) + + category_value = result.get("category", "Unclassified") + category = ( + category_value + if isinstance(category_value, str) and category_value + else "Unclassified" + ) + category_totals = by_category.setdefault( + category, {"completed_entries": 0, "successes": 0, "denominator": 0} + ) + category_totals["completed_entries"] += 1 + category_totals["successes"] += task_successes + category_totals["denominator"] += task_denominator + + completed_entries += 1 + successes += task_successes + denominator += task_denominator + + category_results: dict[str, dict[str, int | float | None]] = {} + for category, totals in sorted(by_category.items()): + category_denominator = totals["denominator"] + category_results[category] = { + **totals, + f"completed_{selection_unit}s": totals["completed_entries"], + "observed_success_rate": totals["successes"] / category_denominator, + } + + return { + "completed_entries": completed_entries, + f"completed_{selection_unit}s": completed_entries, + "successes": successes, + "denominator": denominator, + "observed_success_rate": successes / denominator if denominator else None, + "by_category": category_results, + } + + +def describe_evaluation_status( + plan: EvaluationPlan, + completed: Mapping[str, Any], +) -> dict[str, Any]: + """Distinguish diagnostics for prefixes/shards from a full-suite score.""" + + completed_entries = completed.get("completed_entries") + completed_denominator = completed.get("denominator") + if ( + not isinstance(completed_entries, int) + or isinstance(completed_entries, bool) + or completed_entries < 0 + or completed_entries > plan.selected_task_count + or not isinstance(completed_denominator, int) + or isinstance(completed_denominator, bool) + or completed_denominator < 0 + or completed_denominator > plan.denominator + ): + raise EvaluationProtocolError( + "Completed entries/denominator must stay within the selected plan." + ) + + is_full_suite = plan.task_start == 0 and plan.task_end == plan.benchmark_total_tasks + is_plan_complete = ( + completed_entries == plan.selected_task_count + and completed_denominator == plan.denominator + ) + canonical_trials_per_entry = 50 if plan.protocol == "original" else 1 + uses_canonical_trial_count = plan.trials_per_task == canonical_trials_per_entry + reportable = is_full_suite and is_plan_complete and uses_canonical_trial_count + official_full_suite_denominator = ( + plan.benchmark_total_tasks * canonical_trials_per_entry + ) + + if reportable: + status = "complete" + score_reason = "complete_full_suite" + elif is_full_suite and is_plan_complete: + status = "complete_noncanonical" + score_reason = "noncanonical_trial_count" + elif is_plan_complete: + status = "complete_shard" + score_reason = "intentional_shard" + else: + status = "in_progress" + score_reason = "incomplete_run" + + return { + "status": status, + "scope": "full_suite" if is_full_suite else "shard", + "is_plan_complete": is_plan_complete, + "is_full_suite": is_full_suite, + "uses_canonical_trial_count": uses_canonical_trial_count, + "canonical_trials_per_entry": canonical_trials_per_entry, + "reportable_as_full_suite_score": reportable, + "score_reason": score_reason, + "official_full_suite_denominator": official_full_suite_denominator, + "planned_coverage_fraction": plan.denominator / official_full_suite_denominator, + "completed_coverage_fraction": ( + completed_denominator / official_full_suite_denominator + ), + } + + +def read_json_object(path: str | Path) -> dict[str, Any]: + """Read a JSON object and convert corruption into a protocol error.""" + + input_path = Path(path) + try: + payload = json.loads(input_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise EvaluationProtocolError( + f"Could not read JSON file {input_path}: {exc}" + ) from exc + if not isinstance(payload, dict): + raise EvaluationProtocolError(f"JSON file {input_path} must contain an object.") + return payload + + +def load_or_create_manifest( + payload: Mapping[str, Any], + path: str | Path, +) -> dict[str, Any]: + """Create a manifest once, or verify an existing resume manifest exactly.""" + + output_path = Path(path) + expected = dict(payload) + if output_path.exists(): + existing = read_json_object(output_path) + existing_stable = { + key: value for key, value in existing.items() if key != "created_at_utc" + } + expected_stable = { + key: value for key, value in expected.items() if key != "created_at_utc" + } + if existing_stable != expected_stable: + raise EvaluationProtocolError( + f"Existing run manifest {output_path} does not match this run; " + "refusing to mix results." + ) + return existing + + write_json_atomic(expected, output_path) + return expected + + +def load_resumable_task_result( + path: str | Path, + *, + run_fingerprint: str, + manifest_reference: str, + manifest_sha256: str, + expected_task_fields: Mapping[str, Any], + trials_per_task: int, +) -> dict[str, Any] | None: + """Load a matching task result; ignore older runs and fail on corruption.""" + + input_path = Path(path) + if not input_path.is_file(): + return None + + result = read_json_object(input_path) + if result.get("run_fingerprint") != run_fingerprint: + return None + + for field, expected_value in expected_task_fields.items(): + if result.get(field) != expected_value: + raise EvaluationProtocolError( + f"Resumable result {input_path} has {field}={result.get(field)!r}; " + f"expected {expected_value!r}." + ) + + provenance = result.get("provenance") + if not isinstance(provenance, dict): + raise EvaluationProtocolError( + f"Resumable result {input_path} has no valid provenance object." + ) + if ( + provenance.get("manifest") != manifest_reference + or provenance.get("manifest_sha256") != manifest_sha256 + ): + raise EvaluationProtocolError( + f"Resumable result {input_path} does not reference the immutable " + "manifest for this run." + ) + + successes = result.get("successes") + denominator = result.get("denominator") + if ( + not isinstance(successes, int) + or isinstance(successes, bool) + or not isinstance(denominator, int) + or isinstance(denominator, bool) + or successes < 0 + or denominator <= 0 + or denominator > trials_per_task + or successes > denominator + ): + raise EvaluationProtocolError( + f"Resumable result {input_path} has an invalid success denominator." + ) + + expected_rate = successes / denominator + reported_rate = result.get("success_rate") + if ( + not isinstance(reported_rate, (int, float)) + or isinstance(reported_rate, bool) + or abs(reported_rate - expected_rate) > 1e-12 + ): + raise EvaluationProtocolError( + f"Resumable result {input_path} has an inconsistent success rate." + ) + prompt = result.get("prompt") + if not isinstance(prompt, str) or not prompt.strip(): + raise EvaluationProtocolError( + f"Resumable result {input_path} has no recorded task instruction." + ) + + return result + + +def write_json_atomic(payload: Mapping[str, Any], path: str | Path) -> None: + """Write JSON without leaving a partially written result file.""" + + output_path = Path(path) + output_path.parent.mkdir(parents=True, exist_ok=True) + file_descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{output_path.name}.", + suffix=".tmp", + dir=output_path.parent, + text=True, + ) + temporary_path = Path(temporary_name) + try: + with os.fdopen( + file_descriptor, "w", encoding="utf-8", newline="\n" + ) as file_obj: + file_obj.write( + json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False) + "\n" + ) + file_obj.flush() + os.fsync(file_obj.fileno()) + os.replace(temporary_path, output_path) + finally: + temporary_path.unlink(missing_ok=True) diff --git a/evaluation/libero/launch_client.sh b/evaluation/libero/launch_client.sh index f22d95c2..27da2438 100644 --- a/evaluation/libero/launch_client.sh +++ b/evaluation/libero/launch_client.sh @@ -2,8 +2,10 @@ START=0 END=10 python evaluation/libero/client.py \ + --protocol original \ --libero-benchmark libero_10 \ --port 29056 \ --test-num 50 \ --task-range $START $END \ - --out-dir outputs/libero + --out-dir outputs/libero \ + "$@" diff --git a/evaluation/libero/launch_client_plus.sh b/evaluation/libero/launch_client_plus.sh new file mode 100644 index 00000000..a1e72882 --- /dev/null +++ b/evaluation/libero/launch_client_plus.sh @@ -0,0 +1,6 @@ +python evaluation/libero/client.py \ + --protocol plus \ + --libero-benchmark libero_10 \ + --port 29056 \ + --out-dir outputs/libero \ + "$@" diff --git a/tests/test_libero_eval_protocol.py b/tests/test_libero_eval_protocol.py new file mode 100644 index 00000000..b871c238 --- /dev/null +++ b/tests/test_libero_eval_protocol.py @@ -0,0 +1,454 @@ +import argparse +import ast +import json +import sys +import tempfile +import unittest +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "evaluation" / "libero")) + +from eval_protocol import ( # noqa: E402 + EvaluationProtocolError, + LIBERO_PLUS_CLASSIFICATION_SEMANTIC_SHA256, + LIBERO_PLUS_SUITE_TASK_COUNTS, + add_protocol_arguments, + aggregate_task_results, + describe_evaluation_status, + load_plus_task_metadata, + load_or_create_manifest, + load_resumable_task_result, + normalize_checkpoint_id, + resolve_evaluation_plan, + resolve_prompt, + sha256_file, + validate_benchmark_shape, + write_json_atomic, +) + + +class EvaluationPlanTest(unittest.TestCase): + def test_plus_requires_declared_checkpoint_for_safe_resume(self): + self.assertEqual( + normalize_checkpoint_id("plus", " model@revision "), + "model@revision", + ) + self.assertIsNone(normalize_checkpoint_id("original", None)) + with self.assertRaisesRegex( + EvaluationProtocolError, "requires --checkpoint-id" + ): + normalize_checkpoint_id("plus", None) + + def test_cli_defaults_preserve_original_protocol(self): + parser = argparse.ArgumentParser() + add_protocol_arguments(parser) + + args = parser.parse_args([]) + plan = resolve_evaluation_plan( + args.protocol, + benchmark_total_tasks=10, + task_range=args.task_range, + test_num=args.test_num, + ) + + self.assertEqual(args.protocol, "original") + self.assertEqual(list(plan.task_indices), list(range(10))) + self.assertEqual(plan.trials_per_task, 50) + self.assertEqual(plan.denominator, 500) + self.assertEqual(plan.to_dict()["selected_tasks"], 10) + self.assertNotIn("selected_variants", plan.to_dict()) + + def test_plus_defaults_to_every_variant_once(self): + plan = resolve_evaluation_plan("plus", benchmark_total_tasks=2519) + + self.assertEqual(plan.task_start, 0) + self.assertEqual(plan.task_end, 2519) + self.assertEqual(plan.selected_task_count, 2519) + self.assertEqual(plan.trials_per_task, 1) + self.assertEqual(plan.denominator, 2519) + self.assertEqual(plan.to_dict()["selected_variants"], 2519) + self.assertEqual(plan.to_dict()["trials_per_variant"], 1) + + def test_plus_rejects_more_than_one_trial(self): + with self.assertRaisesRegex( + EvaluationProtocolError, "exactly one trial per variant" + ): + resolve_evaluation_plan("plus", benchmark_total_tasks=2519, test_num=50) + + def test_plus_task_slice_has_an_explicit_slice_denominator(self): + plan = resolve_evaluation_plan( + "plus", benchmark_total_tasks=2519, task_range=[10, 20] + ) + + self.assertEqual(list(plan.task_indices), list(range(10, 20))) + self.assertEqual(plan.selected_task_count, 10) + self.assertEqual(plan.denominator, 10) + + def test_invalid_task_slices_are_rejected(self): + invalid_ranges = ([-1, 1], [3, 3], [5, 4], [0, 11], [0]) + for task_range in invalid_ranges: + with self.subTest(task_range=task_range): + with self.assertRaises(EvaluationProtocolError): + resolve_evaluation_plan( + "original", + benchmark_total_tasks=10, + task_range=task_range, + ) + + def test_installed_benchmark_shape_cannot_cross_protocols(self): + validate_benchmark_shape("original", 10) + for suite, task_count in LIBERO_PLUS_SUITE_TASK_COUNTS.items(): + with self.subTest(suite=suite): + validate_benchmark_shape("plus", task_count, suite=suite) + + with self.assertRaisesRegex(EvaluationProtocolError, "looks like LIBERO-Plus"): + validate_benchmark_shape("original", 2519) + with self.assertRaisesRegex(EvaluationProtocolError, "exactly 2519"): + validate_benchmark_shape("plus", 11, suite="libero_10") + with self.assertRaisesRegex(EvaluationProtocolError, "suite name"): + validate_benchmark_shape("plus", 2519) + + +class ClassificationTest(unittest.TestCase): + def _write_classification(self, directory: str, tasks: list[dict]) -> Path: + path = Path(directory) / "task_classification.json" + path.write_text( + json.dumps({"libero_10": tasks}), + encoding="utf-8", + ) + return path + + def test_classification_is_loaded_with_content_provenance(self): + tasks = [ + { + "id": 1, + "name": "variant_a", + "category": "Camera Viewpoints", + "difficulty_level": 2, + }, + { + "id": 2, + "name": "variant_b", + "category": "Light Conditions", + "difficulty_level": None, + }, + ] + with tempfile.TemporaryDirectory() as directory: + path = self._write_classification(directory, tasks) + bundle = load_plus_task_metadata( + path, "libero_10", ["variant_a", "variant_b"] + ) + + self.assertEqual(len(bundle.tasks), 2) + self.assertEqual(bundle.tasks[0].category, "Camera Viewpoints") + self.assertIsNone(bundle.tasks[1].difficulty_level) + self.assertEqual(len(bundle.sha256), 64) + self.assertEqual(len(bundle.semantic_sha256), 64) + + def test_noncanonical_plus_metadata_fails_closed(self): + tasks = [ + { + "id": 1, + "name": "variant_a", + "category": "Camera Viewpoints", + "difficulty_level": 2, + } + ] + with tempfile.TemporaryDirectory() as directory: + path = self._write_classification(directory, tasks) + with self.assertRaisesRegex( + EvaluationProtocolError, "pinned official classification" + ): + load_plus_task_metadata( + path, + "libero_10", + ["variant_a"], + expected_semantic_sha256=( + LIBERO_PLUS_CLASSIFICATION_SEMANTIC_SHA256 + ), + ) + + def test_classification_order_mismatch_fails_closed(self): + tasks = [ + { + "id": 1, + "name": "variant_b", + "category": "Camera Viewpoints", + "difficulty_level": 2, + } + ] + with tempfile.TemporaryDirectory() as directory: + path = self._write_classification(directory, tasks) + with self.assertRaisesRegex(EvaluationProtocolError, "order mismatch"): + load_plus_task_metadata(path, "libero_10", ["variant_a"]) + + def test_classification_size_mismatch_fails_closed(self): + with tempfile.TemporaryDirectory() as directory: + path = self._write_classification(directory, []) + with self.assertRaisesRegex(EvaluationProtocolError, "size mismatch"): + load_plus_task_metadata(path, "libero_10", ["variant_a"]) + + +class ReportingTest(unittest.TestCase): + def test_aggregation_preserves_category_numerators_and_denominators(self): + summary = aggregate_task_results( + [ + {"category": "Camera", "successes": 1, "denominator": 1}, + {"category": "Camera", "successes": 0, "denominator": 1}, + {"category": "Language", "successes": 1, "denominator": 1}, + ], + selection_unit="variant", + ) + + self.assertEqual(summary["successes"], 2) + self.assertEqual(summary["denominator"], 3) + self.assertEqual(summary["completed_variants"], 3) + self.assertAlmostEqual(summary["observed_success_rate"], 2 / 3) + self.assertEqual(summary["by_category"]["Camera"]["successes"], 1) + self.assertEqual(summary["by_category"]["Camera"]["denominator"], 2) + + def test_plus_prompt_comes_from_bddl_environment(self): + filename_prompt = "task view 0 0 100 initstate 0" + + self.assertEqual( + resolve_prompt("plus", filename_prompt, "Turn on the stove"), + "Turn on the stove", + ) + self.assertEqual( + resolve_prompt("original", "turn on the stove", "Turn on the stove"), + "turn on the stove", + ) + with self.assertRaisesRegex(EvaluationProtocolError, "BDDL language"): + resolve_prompt("plus", filename_prompt, None) + + +class ScoreEligibilityTest(unittest.TestCase): + @staticmethod + def _results(count: int) -> list[dict]: + return [ + {"category": "Background", "successes": index % 2, "denominator": 1} + for index in range(count) + ] + + def test_incomplete_full_suite_is_not_a_plus_score(self): + plan = resolve_evaluation_plan("plus", benchmark_total_tasks=2519) + completed = aggregate_task_results(self._results(10)) + + status = describe_evaluation_status(plan, completed) + + self.assertEqual(status["status"], "in_progress") + self.assertEqual(status["scope"], "full_suite") + self.assertFalse(status["reportable_as_full_suite_score"]) + self.assertEqual(status["score_reason"], "incomplete_run") + self.assertAlmostEqual(status["completed_coverage_fraction"], 10 / 2519) + + def test_completed_shard_is_not_a_plus_score(self): + plan = resolve_evaluation_plan( + "plus", benchmark_total_tasks=2519, task_range=[0, 10] + ) + completed = aggregate_task_results(self._results(10)) + + status = describe_evaluation_status(plan, completed) + + self.assertEqual(status["status"], "complete_shard") + self.assertEqual(status["scope"], "shard") + self.assertTrue(status["is_plan_complete"]) + self.assertFalse(status["reportable_as_full_suite_score"]) + + def test_only_completed_full_suite_is_reportable(self): + plan = resolve_evaluation_plan("plus", benchmark_total_tasks=2) + completed = aggregate_task_results(self._results(2)) + + status = describe_evaluation_status(plan, completed) + + self.assertEqual(status["status"], "complete") + self.assertTrue(status["reportable_as_full_suite_score"]) + + def test_original_noncanonical_trial_count_is_not_reportable(self): + plan = resolve_evaluation_plan("original", benchmark_total_tasks=10, test_num=1) + completed = aggregate_task_results(self._results(10), selection_unit="task") + + status = describe_evaluation_status(plan, completed) + + self.assertEqual(status["status"], "complete_noncanonical") + self.assertFalse(status["uses_canonical_trial_count"]) + self.assertEqual(status["canonical_trials_per_entry"], 50) + self.assertEqual(status["official_full_suite_denominator"], 500) + self.assertFalse(status["reportable_as_full_suite_score"]) + self.assertEqual(status["score_reason"], "noncanonical_trial_count") + + def test_original_ten_by_fifty_is_reportable(self): + plan = resolve_evaluation_plan("original", benchmark_total_tasks=10) + completed = aggregate_task_results( + [ + { + "category": "Original LIBERO", + "successes": 25, + "denominator": 50, + } + for _ in range(10) + ], + selection_unit="task", + ) + + status = describe_evaluation_status(plan, completed) + + self.assertEqual(status["status"], "complete") + self.assertTrue(status["uses_canonical_trial_count"]) + self.assertEqual(status["official_full_suite_denominator"], 500) + self.assertTrue(status["reportable_as_full_suite_score"]) + + +class ResultLifecycleTest(unittest.TestCase): + def test_manifest_is_immutable_across_resume(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "manifest.json" + manifest = { + "schema_version": 1, + "created_at_utc": "first", + "run_fingerprint": "abc", + "run_identity": {"protocol": "plus"}, + } + first = load_or_create_manifest(manifest, path) + resumed = load_or_create_manifest( + {**manifest, "created_at_utc": "second"}, path + ) + + self.assertEqual(first, resumed) + self.assertEqual(resumed["created_at_utc"], "first") + with self.assertRaisesRegex(EvaluationProtocolError, "refusing to mix"): + load_or_create_manifest( + { + **manifest, + "created_at_utc": "third", + "run_fingerprint": "different", + }, + path, + ) + + def test_matching_task_result_resumes_and_other_run_is_ignored(self): + expected_fields = { + "protocol": "plus", + "suite": "libero_10", + "task_index": 0, + "task_id": 1, + "task_name": "variant_a", + "category": "Background Textures", + "difficulty_level": 1, + } + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "task.json" + result = { + **expected_fields, + "run_fingerprint": "run-a", + "prompt": "Turn on the stove", + "successes": 1, + "denominator": 1, + "success_rate": 1.0, + "provenance": { + "manifest": "plus/runs/run-a/manifest.json", + "manifest_sha256": "manifest-hash", + }, + } + write_json_atomic(result, path) + + resumed = load_resumable_task_result( + path, + run_fingerprint="run-a", + manifest_reference="plus/runs/run-a/manifest.json", + manifest_sha256="manifest-hash", + expected_task_fields=expected_fields, + trials_per_task=1, + ) + ignored = load_resumable_task_result( + path, + run_fingerprint="run-b", + manifest_reference="plus/runs/run-b/manifest.json", + manifest_sha256="other-hash", + expected_task_fields=expected_fields, + trials_per_task=1, + ) + + self.assertEqual(resumed, result) + self.assertIsNone(ignored) + + def test_same_run_with_wrong_manifest_fails_closed(self): + expected_fields = { + "protocol": "plus", + "suite": "libero_10", + "task_index": 0, + } + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "task.json" + write_json_atomic( + { + **expected_fields, + "run_fingerprint": "run-a", + "prompt": "Turn on the stove", + "successes": 0, + "denominator": 1, + "success_rate": 0.0, + "provenance": { + "manifest": "manifest.json", + "manifest_sha256": "stale", + }, + }, + path, + ) + + with self.assertRaisesRegex(EvaluationProtocolError, "immutable manifest"): + load_resumable_task_result( + path, + run_fingerprint="run-a", + manifest_reference="manifest.json", + manifest_sha256="current", + expected_task_fields=expected_fields, + trials_per_task=1, + ) + + def test_atomic_writer_leaves_no_fixed_or_unique_temp_file(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "nested" / "result.json" + write_json_atomic({"value": 1}, path) + write_json_atomic({"value": 2}, path) + + self.assertEqual(json.loads(path.read_text(encoding="utf-8")), {"value": 2}) + self.assertEqual(list(path.parent.glob(f".{path.name}.*.tmp")), []) + self.assertEqual(len(sha256_file(path)), 64) + + +class CompatibilityTest(unittest.TestCase): + def test_old_run_positional_arguments_stay_in_the_same_order(self): + client_path = REPO_ROOT / "evaluation" / "libero" / "client.py" + tree = ast.parse(client_path.read_text(encoding="utf-8")) + run_function = next( + node + for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name == "run" + ) + + self.assertEqual( + [argument.arg for argument in run_function.args.args], + ["libero_benchmark", "port", "out_dir", "test_num", "task_range"], + ) + self.assertEqual( + [argument.arg for argument in run_function.args.kwonlyargs], + ["protocol", "task_classification", "checkpoint_id"], + ) + + def test_plus_launcher_forwards_cli_overrides(self): + launcher = ( + REPO_ROOT / "evaluation" / "libero" / "launch_client_plus.sh" + ).read_text(encoding="utf-8") + original_launcher = ( + REPO_ROOT / "evaluation" / "libero" / "launch_client.sh" + ).read_text(encoding="utf-8") + + self.assertIn('"$@"', launcher) + self.assertIn('"$@"', original_launcher) + + +if __name__ == "__main__": + unittest.main()