diff --git a/env/minerl_sandbox.py b/env/minerl_sandbox.py index b74f0e8..b3f3860 100644 --- a/env/minerl_sandbox.py +++ b/env/minerl_sandbox.py @@ -194,7 +194,8 @@ class MineRLSandboxEnv(gym.Env): for attempt in range(max_retries): try: logger.info(f"Resetting environment (attempt {attempt + 1}/{max_retries}) session_id={self.session_id}...") - response = self._post("reset_env", timeout=180) + # a client timeout must not trigger a retry while the server is still resetting (two resets at once) + response = self._post("reset_env", timeout=int(os.getenv("MC_RESET_TIMEOUT", "180"))) if response.get("status") != 0: raise RuntimeError(f"Server error during reset: {response.get('msg')}") return self._get_obs_from_response(response), {} diff --git a/eval_benchmark.py b/eval_benchmark.py index 54487bd..a579b48 100644 --- a/eval_benchmark.py +++ b/eval_benchmark.py @@ -22,6 +22,10 @@ if str(_ROOT_DIR) not in sys.path: from env.minerl_sandbox import MineRLSandboxEnv from env.render import RenderWrapper from mc_agent import DefaultAgent, MinerRLActionSpace, OpenAIProvider, VLLMProvider, DefaultContextBuilder +from mc_agent.utils import convert_buffer_to_base64_images + +sys.path.append(str(_ROOT_DIR.parent / "train")) +from ctrl_policy import make_policy # noqa: E402 FRAME_BUFFER_SIZE = 20 # MAX_STEPS = 50 @@ -45,13 +49,22 @@ class MineRLBenchmarkEnv(MineRLSandboxEnv): raise FileNotFoundError(f"Metadata not found: {meta_path}") with open(meta_path, "r", encoding="utf-8") as f: self._metadata = json.load(f) + self._scenario_id = meta_path.parent.parent.name self._parse_metadata() super().__init__(env_id=self._scene_name, use_friday=use_friday) def _parse_metadata(self) -> None: - self._commands_list = self._metadata.get("commands", []) + # command feedback ("Changed the block at ...") covered the lower-left of the first ~150 frames + self._commands_list = ["/gamerule sendCommandFeedback false"] + self._metadata.get("commands", []) self._task_text = self._metadata.get("task_text", "") self._scene_name = self._metadata.get("scene_name", "benchmark_scene") + self._world_seed = 0 + if os.getenv("MC_EXPLORE_TASK"): # goal-free exploration: bare world, one fixed seed per scenario id + self._commands_list = ["/gamerule sendCommandFeedback false", "/time set day"] + self._task_text = os.environ["MC_EXPLORE_TASK"] + self._world_seed = int(self._scenario_id) + 1 + if os.getenv("MC_START_YAW"): # start facing this yaw (the reversed-start find25 variant: 180) + self._commands_list.append(f"/tp @p ~ ~ ~ {os.environ['MC_START_YAW']} 0") logger.info(f"[BenchmarkEnv] scene={self._scene_name} commands={len(self._commands_list)}") logger.info(f"[BenchmarkEnv] task: {self._task_text[:100]}") @@ -61,7 +74,7 @@ class MineRLBenchmarkEnv(MineRLSandboxEnv): env='MinecraftSim', obs_size=[128, 128], render_size=[640, 360], - seed=0, + seed=self._world_seed, record=False, record_path='./output/', yaml_config=None, @@ -118,6 +131,7 @@ def _run_benchmark( vllm_url: str = "http://localhost:8000/v1", frame_size: int = FRAME_BUFFER_SIZE, use_friday: bool = False, + end_on_esc: bool = False, ) -> Dict[str, Any]: """Run one benchmark scenario and save results.""" global _shutdown_requested @@ -129,8 +143,15 @@ def _run_benchmark( scene_id = meta_path.parent.parent.name task_desc = metadata_dict.get("task_text", f"Complete the Minecraft task for {scene_id}.") safe_model = model.replace("/", "_") - - if _HAS_MILESTONE_CHECKER: + explore_task = os.getenv("MC_EXPLORE_TASK") # goal-free: no milestones, always runs max_steps + # MC_POLICY (JSON: npz, perception gt|vlm, url and model of the perception server for vlm, seed) runs the + # controller of train/ctrl_policy.py instead of agent.get_action; it samples with the seed and the scene id + policy_cfg = json.loads(os.getenv("MC_POLICY", "null")) + policy = policy_cfg and make_policy(**{**policy_cfg, "seed": [policy_cfg["seed"], int(scene_id)]}) + if explore_task: + task_desc = explore_task + checker = MilestoneChecker([]) + elif _HAS_MILESTONE_CHECKER: try: checker = MilestoneChecker.from_metadata(metadata_path) logger.info(f"Loaded {len(checker._milestones)} milestones") @@ -159,13 +180,16 @@ def _run_benchmark( env = RenderWrapper(_base_env, save_messages=True, save_path=str(output_dir)) - frame_buffer = deque(maxlen=frame_size) + # MC_FRAME_STRIDE=k feeds frame_size frames spaced k steps apart (default 1: consecutive). + frame_stride = int(os.getenv("MC_FRAME_STRIDE", "1")) + agent.frame_stride = frame_stride + frame_buffer = deque(maxlen=(frame_size - 1) * frame_stride + 1) thought_history: List[str] = [] action_history: List[Dict] = [] _frame_completed: Dict[str, Optional[int]] = {} _presatisfied_ids: set = set() _frame_offset: int = 0 - _all_done_logged: bool = False + initial_player_pos: Dict[str, Any] = {} long_term_memory: str = "" original_sigint_handler = signal.signal(signal.SIGINT, _signal_handler) @@ -173,7 +197,6 @@ def _run_benchmark( try: has_loading = loading_command_steps > 0 obs, info = env.reset(save_frame=not has_loading) - frame_buffer.append(obs["pov"]) for step in range(loading_command_steps): if _shutdown_requested: @@ -190,8 +213,9 @@ def _run_benchmark( if not info.get("player_pos"): _, noop_action = agent.get_default_action(is_call_failed=False) obs, reward, terminated, truncated, info = env.step(noop_action, save_frame=True) - frame_buffer.append(obs["pov"]) + frame_buffer.append(obs["pov"]) + initial_player_pos = dict(info.get("player_pos", {})) checker.reset(info) agent.load_system_prompt(task_desc) @@ -205,7 +229,8 @@ def _run_benchmark( # agent takes any action). These are treated as "pre-satisfied" and # will NOT be counted as completed — they are excluded via a blocklist # so that _frame_completed is only set when the agent genuinely - # triggers the condition during gameplay. + # triggers the condition during gameplay. They must also be excluded + # from the required targets, otherwise the episode can never succeed. _baseline_check = checker.check(info) _presatisfied_ids: set = set() for _bms in _baseline_check: @@ -217,11 +242,17 @@ def _run_benchmark( ) # Reset the checker state so those pre-satisfied flags don't persist checker.reset(info) + _trackable = [ + m["milestone_id"] for m in checker._milestones + if m.get("rules") and m["milestone_id"] not in _presatisfied_ids + ] + if not _trackable: + logger.info(f"[{run_id}] No unsatisfied trackable targets at spawn; excluding episode from scores.") step_error: Optional[str] = None step = -1 - for step in range(max_steps): + for step in range(max_steps if _trackable or explore_task else 0): if _shutdown_requested: logger.info(f"[{run_id}] Shutdown requested, exiting after step {step}...") step_error = "interrupted_by_user" @@ -230,10 +261,17 @@ def _run_benchmark( logger.info(f"[{run_id}] --- Step {step + 1}/{max_steps} ---") try: - thought, action, memory_update = agent.get_action( - list(frame_buffer), list(thought_history), list(action_history), step + 1, - long_term_memory=long_term_memory - ) + if policy: + # a policy with raw_frame (train/optimus_policy.py) gets the env's RGB frame, not the 320x180 PNG + frame = obs["pov"] if getattr(policy, "raw_frame", False) else \ + convert_buffer_to_base64_images([obs["pov"]])[0] + action, policy_log = policy.act(frame, info, checker, _presatisfied_ids) + thought = memory_update = "" + else: + thought, action, memory_update = agent.get_action( + list(frame_buffer)[::-1][::frame_stride][::-1], list(thought_history), list(action_history), + step + 1, long_term_memory=long_term_memory + ) except Exception as agent_err: logger.error(f"[{run_id}] Agent call failed: {agent_err}. Retrying in 10s...") time.sleep(10) @@ -249,6 +287,7 @@ def _run_benchmark( thought_history.append(thought) action_history.append(action) + player_pos_before = dict(info.get("player_pos", {})) try: augmented_action = checker.augment_action_with_queries(action, info) @@ -258,6 +297,13 @@ def _run_benchmark( time.sleep(10) continue + with open(output_dir / "steps.jsonl", "a", encoding="utf-8") as _f: + _f.write(json.dumps({"step": step + 1, "thought": thought, "memory": long_term_memory, + "action": {k: v for k, v in action.items() if v not in (0, [0.0, 0.0], [0, 0])}, + "player_pos_before": player_pos_before, + "player_pos_after": info.get("player_pos", {}), + **({"policy": policy_log} if policy else {})}, + ensure_ascii=False) + "\n") done = terminated or truncated frame_buffer.append(obs["pov"]) @@ -272,11 +318,17 @@ def _run_benchmark( _frame_completed[mid] = env.frame_count - _frame_offset logger.success(f"[{run_id}] Milestone '{mid}' completed at step {step + 1} (frame {_frame_completed[mid]})") - if not _all_done_logged and checker.num_completed() > 0 and checker.all_done(): - _all_done_logged = True - logger.success(f"[{run_id}] All milestones completed at step {step + 1}!") + if _trackable and all(m in _frame_completed for m in _trackable): + logger.success(f"[{run_id}] Every trackable milestone done at step {step + 1}, stopping early.") + break + + # the prompt says ESC ends the episode, but the released sandbox never reports done for it and + # agents then press ESC for the rest of the budget; --end-on-esc ends it as the prompt promises + if end_on_esc and action.get("ESC"): + logger.info(f"[{run_id}] Agent pressed ESC at step {step + 1}, ending the episode.") + break - if (step + 1) % 10 == 0: + if (step + 1) % 300 == 0: env.save_video_checkpoint() if done: @@ -314,11 +366,12 @@ def _run_benchmark( }) # Compute milestones_completed / all_milestones_done using only the - # corrected _frame_completed dict (excludes pre-satisfied milestones). + # corrected _frame_completed dict. Pre-satisfied milestones are excluded + # from both the completed count and the number of required targets. trackable_mids = { ms.get("milestone_id", "") for ms in checker._milestones - if len(ms.get("rules", [])) > 0 + if len(ms.get("rules", [])) > 0 and ms.get("milestone_id", "") not in _presatisfied_ids } corrected_completed = sum( 1 for mid in trackable_mids @@ -338,6 +391,9 @@ def _run_benchmark( "mode": "multi-agent", "metadata_path": str(meta_path), "total_steps": total_steps, + "frame_size": frame_size, + "frame_stride": frame_stride, + "initial_player_pos": initial_player_pos, "milestones_completed": corrected_completed, "milestones_trackable": corrected_trackable, "milestones_total": len(checker._milestones), @@ -358,6 +414,8 @@ def _run_benchmark( def _worker_eval(worker_args: dict) -> dict: import signal as _signal _signal.signal(_signal.SIGINT, _signal.SIG_IGN) + if worker_args.get("sandbox_url"): + os.environ["MC_SANDBOX_URL"] = worker_args["sandbox_url"] _root = Path(__file__).resolve().parent sys.path.append(str(_root)) @@ -372,6 +430,7 @@ def _worker_eval(worker_args: dict) -> dict: vllm_url=worker_args.get("vllm_url", "http://localhost:8000/v1"), frame_size=worker_args.get("frame_size", FRAME_BUFFER_SIZE), use_friday=worker_args.get("use_friday", False), + end_on_esc=worker_args.get("end_on_esc", False), ) @@ -382,10 +441,12 @@ def eval_benchmark( help="Path to benchmark directory"), output_dir: str = typer.Option("outputs", "--output-dir", "-o", help="Output directory"), - loading_command_steps: int = typer.Option(20, "--loading-command-steps", + loading_command_steps: int = typer.Option(200, "--loading-command-steps", # failed-command chat lines fade after 200 ticks help="No-op steps after reset"), max_steps: int = typer.Option(MAX_STEPS, "--max-steps", help="Maximum agent steps per episode"), + frame_size: int = typer.Option(FRAME_BUFFER_SIZE, "--frame-size", min=1, + help="Number of recent observation frames"), resume: bool = typer.Option(False, "--resume", help="Resume evaluation (skip completed)"), use_vllm: bool = typer.Option(False, "--use-vllm", @@ -398,6 +459,10 @@ def eval_benchmark( help="Limit number of scenarios to evaluate"), use_friday: bool = typer.Option(False, "--use-friday", help="Use Friday platform sandbox instead of local Docker"), + tasks: Optional[str] = typer.Option(None, "--tasks", + help="Comma-separated scenario ids to evaluate (default: all)"), + end_on_esc: bool = typer.Option(False, "--end-on-esc", + help="End the episode when the agent presses ESC"), ): """Evaluate all benchmark scenarios in benchmark_dir.""" logger.info(f"--- Starting evaluation (model={model}) ---") @@ -418,6 +483,10 @@ def eval_benchmark( logger.warning(f"No metadata.json found under {bench_path}") return + if tasks: + wanted = set(tasks.split(",")) + metadata_entries = [e for e in metadata_entries if e[0] in wanted] + if limit and limit > 0: metadata_entries = metadata_entries[:limit] @@ -433,7 +502,18 @@ def eval_benchmark( if resume and result_path.exists(): logger.info(f"[SKIP] {scene_num} - result.json exists") with open(result_path, "r", encoding="utf-8") as f: - all_results.append(json.load(f)) + summary = json.load(f) + if not summary.get("error"): + milestones = MilestoneChecker.from_metadata(str(meta_path))._milestones + trackable = {m["milestone_id"] for m in milestones if m.get("rules")} + statuses = summary["milestone_status"] + trackable -= {m["milestone_id"] for m in statuses if m.get("presatisfied_at_spawn")} + summary["milestones_trackable"] = len(trackable) + summary["milestones_completed"] = sum( + m["completed"] for m in statuses if m["milestone_id"] in trackable + ) + summary["all_milestones_done"] = bool(trackable) and summary["milestones_completed"] == len(trackable) + all_results.append(summary) else: pending.append((scene_num, meta_path)) @@ -466,9 +546,11 @@ def eval_benchmark( output_dir=scene_out_dir, loading_command_steps=loading_command_steps, max_steps=max_steps, + frame_size=frame_size, use_vllm=use_vllm, vllm_url=vllm_url, use_friday=use_friday, + end_on_esc=end_on_esc, ) except Exception as e: logger.error(f"[ERROR] {scene_num}: {e}") @@ -485,51 +567,44 @@ def eval_benchmark( else: import multiprocessing as _mp - effective_workers = min(num_workers, len(pending)) mp_ctx = _mp.get_context("spawn") - - remaining = list(pending) - batch_num = 0 - - while remaining: - batch = remaining[:effective_workers] - remaining = remaining[effective_workers:] - batch_num += 1 - - logger.info(f"\n[Batch {batch_num}] Launching {len(batch)} worker(s)") - - batch_jobs: List[Dict[str, Any]] = [] - for scene_num, meta_path in batch: - scene_out_dir = out_root / scene_num - scene_out_dir.mkdir(parents=True, exist_ok=True) - batch_jobs.append({ - "metadata_path": str(meta_path), - "model": model, - "output_dir": str(scene_out_dir), - "loading_command_steps": loading_command_steps, - "max_steps": max_steps, - "use_vllm": use_vllm, - "vllm_url": vllm_url, - "use_friday": use_friday, - "_scene_num": scene_num, - }) - - with ProcessPoolExecutor(max_workers=len(batch_jobs), mp_context=mp_ctx) as pool: - futures = {pool.submit(_worker_eval, job): job for job in batch_jobs} - for future in as_completed(futures): - job = futures[future] - scene_num = job["_scene_num"] - try: - summary = future.result() - all_results.append(summary) - task_icon = "✅" if summary.get("all_milestones_done") else "❌" - logger.info(f"[DONE] {scene_num} {task_icon} | milestones {summary.get('milestones_completed', 0)}/{summary.get('milestones_trackable', 0)}") - except Exception as exc: - logger.error(f"[ERROR] {scene_num}: {exc}") - summary = _make_error_summary(scene_num, exc) - all_results.append(summary) - - total = len(all_results) + urls = [u for u in os.getenv("MC_SANDBOX_URLS", "").split(",") if u] + jobs: List[Dict[str, Any]] = [] + for i, (scene_num, meta_path) in enumerate(pending): + scene_out_dir = out_root / scene_num + scene_out_dir.mkdir(parents=True, exist_ok=True) + jobs.append({ + "metadata_path": str(meta_path), + "model": model, + "output_dir": str(scene_out_dir), + "loading_command_steps": loading_command_steps, + "max_steps": max_steps, + "frame_size": frame_size, + "use_vllm": use_vllm, + "vllm_url": vllm_url, + "use_friday": use_friday, + "end_on_esc": end_on_esc, + "sandbox_url": urls[i % len(urls)] if urls else None, + "_scene_num": scene_num, + }) + + with ProcessPoolExecutor(max_workers=min(num_workers, len(jobs)), mp_context=mp_ctx) as pool: + futures = {pool.submit(_worker_eval, job): job for job in jobs} + for future in as_completed(futures): + job = futures[future] + scene_num = job["_scene_num"] + try: + summary = future.result() + all_results.append(summary) + task_icon = "✅" if summary.get("all_milestones_done") else "❌" + logger.info(f"[DONE] {scene_num} {task_icon} | milestones {summary.get('milestones_completed', 0)}/{summary.get('milestones_trackable', 0)}") + except Exception as exc: + logger.error(f"[ERROR] {scene_num}: {exc}") + summary = _make_error_summary(scene_num, exc) + all_results.append(summary) + + scored_results = [r for r in all_results if r.get("milestones_trackable", 0) > 0 or r.get("error")] + total = len(scored_results) done = sum(1 for r in all_results if r.get("all_milestones_done")) ms_done = sum(r.get("milestones_completed", 0) for r in all_results) ms_total = sum(r.get("milestones_trackable", 0) for r in all_results) @@ -539,7 +614,7 @@ def eval_benchmark( logger.info("-" * 60) task_rate = done / total * 100 if total > 0 else 0.0 ms_rate = ms_done / ms_total * 100 if ms_total > 0 else 0.0 - logger.info(f"Scenes: {total} | Tasks: {done}/{total} ({task_rate:.1f}%) | Milestones: {ms_done}/{ms_total} ({ms_rate:.1f}%)") + logger.info(f"Scenes: {len(all_results)} | Eligible tasks: {done}/{total} ({task_rate:.1f}%) | Milestones: {ms_done}/{ms_total} ({ms_rate:.1f}%)") logger.info(f"{'='*60}") agg_path = out_root / "eval_summary.json" @@ -548,6 +623,7 @@ def eval_benchmark( "model": model, "benchmark": str(bench_path), "total": total, + "excluded": len(all_results) - total, "done": done, "ms_done": ms_done, "ms_total": ms_total,