Add agent_planner flags to deny/limit its planning simulator

Two CFG knobs let agent_planner run as a model-free or base-sim
baseline against the world-model learner:

- agent_planner_use_simulator (default True): when False, the planner
  gets no option model, so test_option_plan and the scene-rendering
  tools (visualize_state/annotate_scene) are withheld and the prompt
  shifts to open-loop framing -- it must plan from trajectory data and
  LLM reasoning alone.
- agent_planner_use_base_simulator (default False): when a simulator is
  used, wraps the base env (skip_process_dynamics=True) instead of the
  real one, denying the delayed _domain_specific_step dynamics.

create_option_model gains a skip_process_dynamics passthrough (forwarded
only when True, so non-PyBullet analog envs are unaffected).
docker_agent_runner honors the base-sim flag on its in-container
rebuild. agent_bilevel asserts a non-None option model. Defaults
reproduce existing behavior.

Author: yichao-liang

predicators/agent_sdk/docker_agent_runner.py

@@ -271,7 +271,9 @@ def main() -> None:
             CFG as _cfg  # pylint: disable=import-outside-toplevel
         logger.info("Recreating option model (%s) inside Docker...",
                     _cfg.option_model_name)
-        ctx.option_model = create_option_model(_cfg.option_model_name)
+        ctx.option_model = create_option_model(
+            _cfg.option_model_name,
+            skip_process_dynamics=_cfg.agent_planner_use_base_simulator)
         # Sync with all options in context (GT + any previously proposed)
         # after the model has its physics server set up.
         ctx.option_model._name_to_parameterized_option = {  # pylint: disable=protected-access

predicators/approaches/agent_bilevel_approach.py

@@ -169,6 +169,9 @@ def _solve(self, task: Task, timeout: int) -> Callable[[State], Action]:
                 # state-reset noise (see pybullet_env.py:506 warning).
                 # Pass the original sketch so per-step subgoal divergence
                 # is logged with the specific atom that went missing.
+                assert self._option_model is not None, \
+                    "agent_bilevel requires a simulator " \
+                    "(agent_planner_use_simulator=True)."
                 ok, reason = bilevel_sketch.validate_plan_forward(
                     task,
                     plan,
@@ -255,6 +258,9 @@ def _refine_sketch(
         implementation returns ``task`` unchanged.
         """
         task = self._attach_initial_latent(task)
+        assert self._option_model is not None, \
+            "agent_bilevel requires a simulator " \
+            "(agent_planner_use_simulator=True)."
         plan, success, _ = bilevel_sketch.refine_sketch(
             task,
             sketch,

predicators/approaches/agent_planner_approach.py

@@ -55,15 +55,15 @@ def __init__(self,
                          action_space, train_tasks, *args, **kwargs)
         self._offline_dataset = Dataset([])
         self._online_trajectories: List[LowLevelTrajectory] = []
-        if option_model is not None:
-            self._option_model = option_model
-        else:
-            self._option_model = create_option_model(CFG.option_model_name)
+        self._option_model: Optional[_OptionModelBase] = (
+            option_model if option_model is not None else
+            self._create_planner_option_model())
         # Let the option model terminate Wait on atom change using the
         # approach's predicates (which may include invented ones). Looked
         # up lazily so the lambda picks up predicates invented after
         # __init__.
-        if CFG.wait_option_terminate_on_atom_change:
+        if self._option_model is not None and \
+                CFG.wait_option_terminate_on_atom_change:
             cast(  # pylint: disable=protected-access
                 Any, self._option_model)._abstract_function = (
                     lambda s: utils.abstract(s, self._get_all_predicates()))
@@ -119,6 +119,27 @@ def _get_all_trajectories(self) -> List[LowLevelTrajectory]:
         """Return all trajectories (offline + online)."""
         return self._offline_dataset.trajectories + self._online_trajectories
 
+    def _create_planner_option_model(self) -> Optional[_OptionModelBase]:
+        """Build the option model the planner tests plans against.
+
+        Honors two CFG knobs:
+
+        * ``agent_planner_use_simulator`` -- when False, returns ``None``
+          so the agent gets no ``test_option_plan`` rollouts and must
+          plan open-loop from data + LLM reasoning (the model-free
+          baseline).
+        * ``agent_planner_use_base_simulator`` -- when True (and a
+          simulator is used), wraps the *base* env
+          (``skip_process_dynamics=True``) so the planner is denied the
+          delayed ``_domain_specific_step`` dynamics; otherwise wraps the
+          real env.
+        """
+        if not CFG.agent_planner_use_simulator:
+            return None
+        return create_option_model(
+            CFG.option_model_name,
+            skip_process_dynamics=CFG.agent_planner_use_base_simulator)
+
     # ------------------------------------------------------------------ #
     # AgentSessionMixin hooks
     # ------------------------------------------------------------------ #
@@ -217,8 +238,12 @@ def _get_all_trajectories(self) -> List[LowLevelTrajectory]:
 
     def _get_agent_system_prompt(self) -> str:
         use_scratchpad = CFG.agent_planner_use_scratchpad
-        use_visualize = CFG.agent_planner_use_visualize_state
-        use_annotate = CFG.agent_planner_use_annotate_scene
+        # visualize_state / annotate_scene render a live env, so they are
+        # only available when the planner has a simulator.
+        use_visualize = (CFG.agent_planner_use_simulator
+                         and CFG.agent_planner_use_visualize_state)
+        use_annotate = (CFG.agent_planner_use_simulator
+                        and CFG.agent_planner_use_annotate_scene)
 
         sections = [self._SYSTEM_PROMPT_BASE]
 
@@ -317,13 +342,18 @@ def _get_sandbox_reference_files(self) -> Dict[str, str]:
 
     def _get_solve_tool_names(self) -> Optional[List[str]]:
         tools = [
-            "inspect_options", "inspect_trajectories", "inspect_train_tasks",
-            "test_option_plan"
+            "inspect_options", "inspect_trajectories", "inspect_train_tasks"
         ]
-        if CFG.agent_planner_use_annotate_scene:
-            tools.append("annotate_scene")
-        if CFG.agent_planner_use_visualize_state:
-            tools.append("visualize_state")
+        # The remaining tools all require a simulator / live env:
+        # test_option_plan rolls plans out through the option model, and
+        # visualize_state / annotate_scene render env states. None are
+        # offered when the planner has no simulator.
+        if CFG.agent_planner_use_simulator:
+            tools.append("test_option_plan")
+            if CFG.agent_planner_use_annotate_scene:
+                tools.append("annotate_scene")
+            if CFG.agent_planner_use_visualize_state:
+                tools.append("visualize_state")
         return tools
 
     # ------------------------------------------------------------------ #
@@ -524,6 +554,16 @@ def _build_solve_prompt(self, task: Task) -> str:
 {task.goal_nl}
 """
 
+        if CFG.agent_planner_use_simulator:
+            instructions_intro = (
+                "Use your available tools to inspect the environment and "
+                "test your plan before committing to it.")
+        else:
+            instructions_intro = (
+                "You do NOT have a simulator to test plans against. Inspect "
+                "the trajectory data and reason carefully about the dynamics, "
+                "then commit to your best open-loop plan.")
+
         prompt = f"""You are solving a task. \
 Generate an option plan to achieve the goal.
 {goal_nl_section}
@@ -543,7 +583,7 @@ def _build_solve_prompt(self, task: Task) -> str:
 {chr(10).join(option_strs)}
 {traj_summary}{tools_str}
 ## Instructions
-Use your available tools to inspect the environment and test your plan before committing to it.
+{instructions_intro}
 
 Based on the task information and any past trajectory data, output an option plan to achieve the goal.
 

predicators/option_model.py

@@ -8,7 +8,7 @@
 
 import abc
 import logging
-from typing import Callable, Optional, Set, Tuple
+from typing import Any, Callable, Dict, Optional, Set, Tuple
 
 import numpy as np
 import pybullet
@@ -43,23 +43,39 @@ def _check_wait_termination(option: _Option, state: State, last_state: State,
     return False
 
 
-def create_option_model(name: str,
-                        use_gui: Optional[bool] = None) -> _OptionModelBase:
+def create_option_model(
+        name: str,
+        use_gui: Optional[bool] = None,
+        skip_process_dynamics: bool = False) -> _OptionModelBase:
     """Create an option model given its name.
 
     Args:
         name: The name of the option model.
         use_gui: If provided, overrides CFG.option_model_use_gui for the
             environment created by this option model.
+        skip_process_dynamics: If True, the wrapped env runs with its
+            delayed ``_domain_specific_step`` dynamics disabled (the
+            "base" simulator). Forwarded to the env only when True, so
+            non-PyBullet analog envs whose ``__init__`` does not accept
+            the kwarg are unaffected by the default.
     """
     gui = CFG.option_model_use_gui if use_gui is None else use_gui
+    env_kwargs: Dict[str, Any] = {}
+    if skip_process_dynamics:
+        env_kwargs["skip_process_dynamics"] = True
     if name == "oracle":
-        env = create_new_env(CFG.env, do_cache=False, use_gui=gui)
+        env = create_new_env(CFG.env,
+                             do_cache=False,
+                             use_gui=gui,
+                             **env_kwargs)
         options = get_gt_options(env.get_name())
         return _OracleOptionModel(options, env.simulate)
     if name.startswith("oracle"):
         env_name = name[name.index("_") + 1:]
-        env = create_new_env(env_name, do_cache=False, use_gui=gui)
+        env = create_new_env(env_name,
+                             do_cache=False,
+                             use_gui=gui,
+                             **env_kwargs)
         options = get_gt_options(env.get_name())
         return _OracleOptionModel(options, env.simulate)
     raise NotImplementedError(f"Unknown option model: {name}")

predicators/settings.py

@@ -1025,6 +1025,18 @@ class GlobalSettings:
     agent_planner_use_scratchpad = False  # include notes.md scratchpad
     agent_planner_use_visualize_state = False  # include visualize_state tool
     agent_planner_use_annotate_scene = False  # include annotate_scene tool
+    # Whether the planner is given a simulator to test candidate plans with
+    # (the test_option_plan tool / option-model rollouts). When False, the
+    # agent must plan open-loop from trajectory data and LLM reasoning alone
+    # -- the genuinely model-free baseline.
+    agent_planner_use_simulator = True
+    # When a simulator IS given, whether to wrap the *base* env
+    # (skip_process_dynamics=True -- delayed _domain_specific_step effects
+    # such as boiling/heating are disabled) instead of the real env. Lets the
+    # model-free planner be denied the ground-truth delayed dynamics that a
+    # world-model learner has to reconstruct. No effect when
+    # agent_planner_use_simulator is False.
+    agent_planner_use_base_simulator = False
 
     # Agent bilevel approach settings
     agent_bilevel_max_samples_per_step = 50  # param samples per step