| |
| |
| |
| |
| @@ -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), {} |
| |
| |
| |
| |
| @@ -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, |
|
|
