Checking appworld specific changes compared to agents base

[shatu]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces a new containerized, stateful RL environment for the AppWorld benchmark, along with comprehensive documentation, dataset conversion scripts, diagnostic tools, and unit tests. The reviewer's feedback highlights a critical performance and stability issue: the asynchronous methods of AppWorldEnv (reset, step, and close) perform synchronous, blocking Docker and HTTP operations directly on the main event loop. To prevent blocking the Ray actor's event loop and causing potential heartbeat timeouts, the reviewer recommends wrapping these blocking calls in asyncio.to_thread and adding robust error handling in step to gracefully handle container or network failures.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Import asyncio to allow running synchronous, blocking Docker and HTTP requests in a separate thread using asyncio.to_thread. This prevents blocking the main event loop of the Ray actor.

Suggested change

	import contextlib
	import socket
	import time
	import uuid
	from dataclasses import dataclass
	from typing import Any, ClassVar
	from urllib.parse import urlparse
	import requests
	from openenv.core.env_server.types import State
	from open_instruct import logger_utils
	from .base import BaseEnvConfig, EnvCall, RLEnvironment, StepResult
	from .tools.utils import coerce_args
	import asyncio
	import contextlib
	import socket
	import time
	import uuid
	from dataclasses import dataclass
	from typing import Any, ClassVar
	from urllib.parse import urlparse
	import requests
	from openenv.core.env_server.types import State
	from open_instruct import logger_utils
	from .base import BaseEnvConfig, EnvCall, RLEnvironment, StepResult
	from .tools.utils import coerce_args

[gemini-code-assist]

The reset method is asynchronous, but it calls synchronous, blocking operations like _teardown_container, _start_container, and _http directly on the event loop. This can block the event loop for up to 180 seconds (the startup timeout), which will likely cause Ray actor heartbeat timeouts and performance degradation. Wrap these blocking calls in asyncio.to_thread to run them in a separate thread.

    async def reset(self, task_id: str | None = None, **kwargs: Any) -> tuple[StepResult, list[dict]]:
        if self._client is None:
            await self.setup()
        if task_id is None:
            raise ValueError("AppWorldEnv.reset requires a task_id (set it via env_config).")
        # Allow the pool to rotate docker hosts on retry (mirrors swerl).
        if kwargs.get("docker_host") and kwargs["docker_host"] != self._docker_host:
            self._docker_host = kwargs["docker_host"]
            await asyncio.to_thread(self._teardown_container)
            await self.setup()
        if self._container is None or self._base_url is None:
            await asyncio.to_thread(self._start_container)

        self._step_count = 0
        self._completed = False
        self._last_reward = 0.0
        self._last_eval = {}
        self._task_id = task_id
        self._max_steps = kwargs.get("max_steps", self._max_interactions)
        episode = kwargs.get("episode_id") or kwargs.get("instance_id") or uuid.uuid4().hex
        self._experiment_name = f"{self._experiment_prefix}_{task_id}_{episode}"

        # Re-initialize the container's single world to this task.
        await asyncio.to_thread(
            self._http,
            "initialize",
            task_id=task_id,
            experiment_name=self._experiment_name,
            max_interactions=self._max_interactions,
            timeout_seconds=self._timeout_seconds,
            load_ground_truth=True,  # required for /evaluate
            raise_on_unsafe_syntax=True,
            null_patch_unsafe_execution=True,
            raise_on_failure=False,
        )
        # The reset observation is discarded by the rollout pipeline; the prompt
        # comes from the dataset messages (see build_prompt_messages).
        return StepResult(result="", metadata={"task_id": task_id}), list(self._tool_definitions)

[gemini-code-assist]

The step method is asynchronous, but it calls synchronous, blocking HTTP requests directly on the event loop. Wrap these calls in asyncio.to_thread to prevent blocking the event loop. Additionally, wrap the HTTP calls in a try...except block to gracefully handle container crashes, OOMs, or network timeouts by returning an error observation and ending the episode with 0 reward, rather than crashing the entire training process.

    async def step(self, call: EnvCall) -> StepResult:
        if self._base_url is None or self._task_id is None:
            raise RuntimeError("Environment not reset. Call reset() first.")
        self._step_count += 1

        if call.name != "execute_python":
            return StepResult(
                result=f"Unknown tool '{call.name}'. The only available tool is `execute_python`."
            )

        args = coerce_args(EXECUTE_TOOL["function"]["parameters"], call.args)
        code = args.get("code", "")
        if not code or not code.strip():
            return StepResult(result="Error: the `code` parameter is required and must be non-empty.")

        try:
            output = await asyncio.to_thread(self._http, "execute", task_id=self._task_id, code=code)
            observation = truncate_observation(output) if output else "(no output)"
            self._completed = bool(await asyncio.to_thread(self._http, "task_completed", task_id=self._task_id))
        except Exception as e:
            logger.error(f"[{self._task_id}] AppWorld HTTP call failed: {e}")
            return StepResult(
                result=f"Error: Failed to communicate with the AppWorld environment server: {e}",
                reward=0.0,
                done=True,
                metadata={"task_id": self._task_id, "error": str(e)},
            )

        at_step_limit = self._max_steps is not None and self._step_count >= self._max_steps
        if self._completed or at_step_limit:
            reward = await asyncio.to_thread(self._compute_reward)
            summary = (
                f"\n\n[Task {'completed' if self._completed else 'truncated at step limit'}. "
                f"Reward {reward:.3f} "
                f"({self._last_eval.get('pass_count', 0):.0f}/{self._last_eval.get('num_tests', 0):.0f} tests passed).]"
            )
            return StepResult(
                result=observation + summary,
                reward=reward,
                done=True,
                metadata={"task_id": self._task_id, "completed": self._completed, **self._last_eval},
            )
        return StepResult(result=observation, metadata={"task_id": self._task_id, "completed": False})

[gemini-code-assist]

The close method is asynchronous, but it calls synchronous, blocking operations like _teardown_container directly on the event loop. Wrap this call in asyncio.to_thread to prevent blocking the event loop.

Suggested change

	async def close(self) -> None:
	self._teardown_container()
	async def close(self) -> None:
	await asyncio.to_thread(self._teardown_container)