patches/MineExplorer.patch

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,