diff --git a/docs/algorithms/adding_sandboxed_environments.md b/docs/algorithms/adding_sandboxed_environments.md
new file mode 100644
index 0000000000..b9b35d9278
--- /dev/null
+++ b/docs/algorithms/adding_sandboxed_environments.md
@@ -0,0 +1,539 @@
+# Adding a sandboxed RL environment to open-instruct
+
+A field guide for wiring a new **containerized / sandboxed** environment into the GRPO
+RL pipeline, written from the end-to-end AppWorld integration. Read this before adding
+the next one (terminal variants, web sandboxes, OS-agent worlds, MCP-backed apps, …) —
+it captures the architecture, the exact dataset/launch contract, the two integration
+styles, every trap that cost time, and a copy-paste checklist.
+
+Companion docs: [rl_with_environments.md](rl_with_environments.md) (basic tool/env intro),
+[multi_task_rl.md](multi_task_rl.md) (per-example routing, dataset columns),
+[rollout_loop_internals.md](rollout_loop_internals.md) (token-level rollout),
+[tmax_4b_script_reference.md](tmax_4b_script_reference.md) (annotated swerl launch).
+Worked example: [scripts/general_agent/appworld/](../../scripts/general_agent/appworld/).
+
+> **How to read this doc.** Part 1 is a plain-language mental model — start here if you're
+> new to how this RL pipeline works. Part 2 onward is the engineering reference (exact files,
+> the dataset contract, the traps). You can act on Part 2 alone once Part 1 clicks.
+
+---
+
+# Part 1 — Plain-language mental model (start here)
+
+## What we're actually doing
+
+We're training a model to *do tasks*, not just answer questions. The model takes actions
+(run a shell command, call an API, write code), something **responds**, the model acts again,
+and at the end we **score** how well it did. We then nudge the model toward the behavior that
+scored well. That's reinforcement learning (RL); the specific algorithm here is **GRPO**, but
+you don't need its math to add an environment.
+
+One **attempt at one task** is called a **rollout**: prompt → model acts → world responds →
+… → done → a reward number. Training runs thousands of rollouts and learns from them.
+
+## The "environment" and why it's a sandbox
+
+The **environment** is the little world the model acts in. For agentic tasks that world is
+usually a **sandbox**: an isolated computer (a **container**) created fresh for the task.
+We use containers for three reasons:
+
+1. **Safety** — the model writes arbitrary code/commands; run them in a throwaway box, not on
+   the training machine.
+2. **Reproducibility** — every task starts from an identical clean state.
+3. **Dependency isolation** — the task's software (and its library versions) stay inside the
+   container and can't clash with the trainer's. (This one turned out to matter a lot for AppWorld.)
+
+The environment's job each turn: take the model's action, run it in the container, hand back
+what happened (the "observation"), and — when the task is finished — compute the **reward**.
+
+## The cast of characters (all running at once)
+
+Think of a training step as a small factory, with these workers (each is a separate process;
+**Ray** is the glue that runs and connects them, possibly across several machines):
+
+- **The learner** — holds the model's weights and does the actual learning (gradient updates).
+  Lives on GPUs.
+- **The generator (vLLM)** — runs the current model *fast* to produce the next action/text.
+  Also on GPUs. (Generating and learning are split for speed.)
+- **The environment pool** — a stash of ready-to-go environment instances. Because we run many
+  rollouts in parallel, we keep a *pool* of, say, 128 environments warm; each rollout borrows
+  one, uses it, returns it. For a container env, **one warm environment ≈ one running container**,
+  so the pool size is also how many containers run at once (watch the RAM).
+- **A dataset** — a list of tasks. Each **row** is one task and carries: the prompt, *which*
+  tool/environment to use, the task id, and how to score it.
+
+## How one rollout flows (the narrated version)
+
+1. The pipeline picks a task row and **borrows an environment from the pool**, which **starts
+   (or re-initializes) the container** for that task.
+2. The **generator** produces the model's first action — for a tool-using agent, that's a
+   **tool call** like `execute_python({"code": "..."})`.
+3. The environment **runs that action in the container** and returns an **observation** (stdout,
+   an API response, an error). The observation is appended to the conversation.
+4. Steps 2–3 repeat (multi-turn) until the model signals it's done, or a step limit is hit.
+5. The environment **computes the reward** (e.g. "fraction of the task's tests that pass").
+6. The reward + the conversation go back to the **learner**, which updates the weights.
+
+A few non-obvious facts that follow from this flow (and bite people):
+
+- The **prompt lives in the dataset row**, not in the environment. The environment's "welcome
+  message" from step 1 is actually thrown away by the pipeline — so the task instructions must
+  be written into the dataset.
+- The **environment supplies the tool's definition** (its name + arguments). The dataset just
+  says *which* tool name is allowed; the schema the model sees comes from the environment.
+- The **reward comes from the environment** (we set the dataset's separate "verifier" to a
+  no-op called `passthrough` so it doesn't double-count).
+
+## Where the container actually runs (local vs. the cluster)
+
+- **On your dev box / locally:** a normal Docker daemon. Easy.
+- **On the Beaker cluster:** each node runs a fleet of **podman** services (podman = a
+  Docker-compatible container engine). The trainer container spawns the task containers as
+  **siblings** next to it. There's also a **registry mirror** (a local cache of Docker Hub) so
+  pulling images is fast. There's a standard recipe of environment variables + a startup script
+  that turns all this on — you copy it from an existing script; you don't invent it.
+
+## The two ways an environment can work
+
+Both kinds of environment are *stateful* — the difference is **where the state lives** and
+whether a brand-new process can still see it:
+
+- **Style A — "run a command, get output."** State lives in the container's **files**. Each turn
+  is a fresh shell, but it sees everything previous turns wrote to disk, so a simple "run this
+  command in the box" model works (the repo's ready-made container helper does this). The only
+  thing a fresh shell forgets is the current folder and environment variables, which the terminal
+  envs quietly save to a file and reload each turn. Example: the terminal/coding sandboxes (`swerl`).
+- **Style B — "talk to a long-running program."** State lives in a **live program's memory** — a
+  Python session with variables, a loaded "world", open connections, a browser. You can't save
+  that to a file and reload it the way you can a folder path, so the container has to keep **one
+  long-running program** alive; our environment is a thin **client** that sends it requests over
+  HTTP. Example: **AppWorld**.
+
+AppWorld is style B for two reasons: it's a Python session where the agent's variables must
+persist across turns, **and** its software needs an old library version that clashes with the
+trainer's — so it *has* to live in its own container, talked to over HTTP, never imported into
+the trainer. The container is both the sandbox and the "keep its dependencies away from ours" wall.
+
+(And don't worry that style B is heavy: the container is started **once and reused** for many
+tasks — not per attempt — so the setup cost is amortized, just like the terminal sandboxes.)
+
+## What "adding an environment" therefore means
+
+1. Write a small Python class that knows how to **start the container, send the model's action
+   to it, read back the observation, and compute the reward** (`reset` / `step` / `close`).
+2. **Register** that class under a name.
+3. Build a **dataset** of task rows in the exact format the pipeline expects.
+4. Write a **launch script** that turns on the container machinery and points at your env + dataset.
+5. **Test it in three stages**: the env alone → a tiny local training run → the real cluster run.
+
+Part 2 is the precise, file-by-file version of those five steps, plus every trap we hit.
+
+## Mini-glossary
+
+- **Policy / model** — the thing being trained.
+- **Rollout** — one attempt at one task; ends in a reward.
+- **GRPO** — the RL training algorithm used here (details not needed to add an env).
+- **RLVR data** — the dataset format for RL ("RL with verifiable rewards"): rows of task +
+  how-to-score, *not* example answers.
+- **Environment / env** — the world the model acts in (here, usually a container sandbox).
+- **Tool** — an action the model can take, with a name + arguments (e.g. `execute_python`).
+- **Observation** — what the environment returns after an action.
+- **Reward** — the score for a rollout, produced by the environment.
+- **Pool** — a set of warm environment instances for running rollouts in parallel.
+- **Container / image** — an isolated mini-computer / its frozen template.
+- **Ray** — runs the many pieces (learner, generator, envs) as parallel processes ("actors").
+- **vLLM** — the fast model-serving engine used to generate actions.
+- **Beaker** — Ai2's compute cluster. **podman** — the container engine on cluster nodes.
+- **Registry mirror** — a local cache of Docker Hub on the cluster, for fast image pulls.
+
+---
+
+# Part 2 — Engineering reference
+
+## 0. TL;DR — the two integration styles
+
+There are two ways a sandboxed env runs, and which one you pick decides everything else:
+
+| | **A. Filesystem-state (swerl model)** | **B. Live-session server (AppWorld model)** |
+|---|---|---|
+| Interaction | `docker exec bash -c <cmd>` per turn | long-lived server in the container; env is an HTTP client |
+| State between turns | container **filesystem** (+ cwd/env serialized to files) | a **live process**: REPL vars, imports, in-memory world/handles |
+| Reuses `backends.py` | **Yes** (`DockerBackend.run_command`) | **No** — needs network reachability to the container |
+| In-process possible? | n/a (always container) | only if the env's deps don't conflict with the trainer's |
+| Examples | `swerl_sandbox`, `swerl_vanillux_sandbox`, `generic_sandbox` | `appworld` |
+
+**"Stateful vs stateless" is the wrong axis — both are stateful. The real question is *where the
+state lives* and whether a fresh process can see it:**
+
+- **Style A:** state is the container **filesystem**, which every new `docker exec` shell sees.
+  The only per-shell state a fresh `exec` loses is **cwd + env vars**, and swerl explicitly
+  serializes those to files and reloads them each call (see the bash wrapper in
+  `swerl_vanillux_sandbox.py`). Everything serializable → stateless-per-call `exec` works, no
+  network, no ports.
+- **Style B:** state is a **live in-memory session** — Python REPL variables, imported modules,
+  an `apis` object with open DB handles. You cannot serialize/reload that between fresh
+  processes the way you can cwd/env, so you need **one long-lived process** holding it = a
+  server, reached by a client. (AppWorld persists its *world DB* to disk each step, but the
+  agent's REPL namespace — `x = apis.amazon.search(...)` reused next turn — only lives in the
+  process.)
+
+Decision rule: filesystem (or trivially serializable) state → **A**; live interpreter/browser/
+session state → **B**.
+
+> AppWorld forced style B for two reasons: (1) it's a Python REPL where the agent's in-memory
+> variables persist across turns, and (2) AppWorld pins `pydantic<2` while open-instruct/openenv
+> need `pydantic>=2`, so it **cannot be imported in the trainer process at all**. The container
+> is both the sandbox *and* the dependency quarantine.
+
+**Does style B scale?** Yes — the per-container cost people worry about isn't per-rollout. The
+container (and its port) is created **once per pool actor** and reused across all tasks (each
+`reset()` just re-`/initialize`s over HTTP; each turn is one HTTP call) — exactly how swerl
+reuses its `sleep infinity` container. Allocating a free port is a single `bind(0)`/`getsockname`
+syscall done ~once per actor (a few hundred times over a multi-hour run), and HTTP-over-localhost
+per turn costs about the same as a `docker exec`. The real per-node costs (container RAM, podman
+create throughput) are the same shape as swerl; AppWorld's only extra is its server loading
+apps/DBs at startup. Ephemeral ports (~28k) vastly exceed containers-per-node, and the tiny
+bind→start race self-heals via reset retry.
+
+---
+
+## 1. The environment/tool system (what the repo gives you)
+
+### 1.1 The `RLEnvironment` ABC — `open_instruct/environments/base.py`
+
+Every env subclasses `RLEnvironment` (or `TextRLEnvironment`). The contract:
+
+```python
+class RLEnvironment(ABC):
+    config_name: str = ""            # registry key, e.g. "appworld"
+    response_role: str = "tool"      # role used to inject observations back ("tool" or "user")
+
+    async def setup(self) -> None: ...                 # once per actor at startup (clients, downloads)
+    @classmethod
+    def get_tool_definitions(cls) -> list[dict]: ...   # OpenAI-format tool schemas the model sees
+    async def reset(self, **kwargs) -> tuple[StepResult, list[dict]]: ...  # start episode -> (obs, tools)
+    async def step(self, call: EnvCall) -> StepResult: ...                 # one action -> obs + reward + done
+    def state(self) -> State: ...
+    def get_metrics(self) -> dict[str, float]: ...     # custom wandb metrics
+    async def close(self) / shutdown(self): ...        # teardown
+```
+
+- **`EnvCall`** = a parsed tool call (`id`, `name`, `args`). For tool-style envs the rollout
+  loop parses the model's tool call and dispatches `step(call)`. `call.name` is the **tool
+  function name** (e.g. `execute_python`), not the config_name.
+- **`StepResult`** = `result: str` (the observation text injected back), `reward: float`,
+  `done: bool`, `metadata: dict`.
+- **`TextRLEnvironment`** (subclass) is for envs that consume the model's *entire* generation
+  as text (no parsed tool call) — implement `text_step(text)` and `_reset()`. `swerl_vanillux`
+  is tool-style; `wordle_text_env` is text-style. Tool-style is usually the right default for a
+  sandbox (it composes with the tool-calling rollout machinery and the qwen3_xml parser).
+- **`BaseEnvConfig`** (dataclass) pairs a config with the env via `tool_class: ClassVar`.
+  Config fields become `__init__` kwargs of the env (the pool expands the dataclass).
+
+### 1.2 Registration — `open_instruct/environments/tools/tools.py`
+
+Add one import + one `TOOL_REGISTRY` entry (`~line 652`):
+
+```python
+TOOL_REGISTRY[MyEnvConfig.tool_class.config_name] = MyEnvConfig
+```
+
+`tools/tools.py` is imported at trainer startup, so **the env module must import cleanly
+without the env's heavy/optional deps** — import them lazily inside `setup()` (see §6 trap 6).
+
+### 1.3 The pool — `open_instruct/environments/pool.py`
+
+`EnvironmentPool` is a Ray actor holding `pool_size` env-actor instances. Key behavior:
+
+- **One actor per concurrent rollout.** `pool_size` should be ≥ `num_unique_prompts_rollout
+  * num_samples_per_prompt_rollout` (the max concurrent rollouts). For heavyweight container
+  envs this is also the max concurrent containers — size accordingly (RAM!).
+- **`acquire_reset(reset_kwargs)`** acquires a free actor and calls `actor.reset(**kwargs)`,
+  returning `(actor, tool_defs)`. **The reset `StepResult` observation is dropped** for tool
+  envs — the prompt comes entirely from the dataset `messages` (see §2).
+- **Podman host rotation:** the pool injects a `docker_host` into `reset_kwargs` by leasing
+  from `SWERL_PODMAN_DOCKER_HOSTS` — but **only when the env config has `backend == "docker"`**
+  (`pool.py` ~line 74). Your config dataclass **must** have a `backend: str = "docker"` field
+  or your containers never land on the beaker podman hosts. The env's `reset` should read
+  `kwargs.get("docker_host")` and (re)connect to it.
+- Reset is retried with backoff; host-connectivity failures rotate to the next podman host.
+
+### 1.4 Backends — `open_instruct/environments/backends.py`
+
+`DockerBackend` / `ApptainerBackend` implement `start / run_command(bash) / write_file /
+read_file / put_archive / close`, with `docker_host` support and concurrency semaphores
+(`SWERL_DOCKER_START_CONCURRENCY`, `SWERL_DOCKER_EXEC_CONCURRENCY`). **Reuse these for style A**
+(bash-exec). For style B you typically don't use the backend's `run_command` (you make HTTP
+calls), but you can still mirror its container *lifecycle* (start/stop, `docker_host`).
+
+### 1.5 How envs reach the rollout — `open_instruct/grpo_fast.py`
+
+- `create_tool_pools(parsed_tools, pool_size)` (`~1962`): builds one `EnvironmentPool` per
+  `--tools` entry, **keyed by call name** (`pools[call_name] = ...`; call name = `--tool_call_names`
+  or the config_name if not remapped).
+- `_discover_tools_from_datasets(...)` (`~3050`): scans the dataset `tools` column **and**
+  `env_config.env_name`s and **auto-creates a pool** for any registry name not already
+  configured. ← this is where the duplicate-pool trap lives (§6 trap 3).
+- `initialize_tools_and_envs(...)` (`~3085`): calls the above, collects
+  `tool_definitions = [d for pool in pools for d in pool.get_tool_definitions()]`, returns
+  `(pools, tool_definitions, stop_sequences)`.
+- `_validate_and_log_dataset_tools(...)` (`~1876`): asserts the dataset `tools` names ⊆
+  configured call names. A mismatch is a hard error here.
+
+### 1.6 The rollout loop — `open_instruct/vllm_utils.py`
+
+- `process_request` → `_acquire_and_reset_pools(...)` (`~1042`) runs **before generation**:
+  it acquires + resets one actor from *every configured pool* (so a heavyweight container
+  starts at the top of every rollout — see the "over-acquire" note in
+  [multi_task_rl.md §1.2](multi_task_rl.md)).
+- `allowed_tools = configured_tools(pool keys) & active_tools(dataset tools column)`, then the
+  env's tool function names are added to `allowed_tools` at register time. Tool calls are
+  filtered to `allowed_tools` (`~1231`).
+- Each `step()` result's `.result` is injected back as an observation with the env's
+  `response_role`; `.reward` is appended to `rollout.rewards` (`~1285`). On `done`, the rollout
+  ends.
+- **Hard limit:** only **one text env** per rollout (`~1090`). Tool envs are unrestricted.
+
+### 1.7 Reward flow — `open_instruct/ground_truth_utils.py`
+
+- The env emits reward via `StepResult.reward`. To make the env's reward the *only* reward,
+  set the dataset `dataset` column to **`passthrough`** (the `passthrough` verifier returns 0,
+  `~line 1037`), so total reward = env reward + 0. This is the swerl/appworld pattern.
+- `apply_verifiable_reward` (`~1253`) selects the verifier per-row by the `dataset` column.
+
+---
+
+## 2. The dataset (RLVR) contract — get this exactly right
+
+One row per task. Columns (`open_instruct/dataset_transformation.py`):
+
+| Column | Value for a sandbox env | Notes |
+|--------|------------------------|-------|
+| `messages` | `[{system, …}, {user, instruction}]` | The **prompt**. The env reset observation is **dropped**, so the task instruction must live here. Bake the system prompt in (or use `--system_prompt_override_file`). |
+| `ground_truth` | usually the **task id** (a string) | The env verifies internally; this is just an identifier. |
+| `dataset` | `"passthrough"` | Env emits the reward; no extra verifier. |
+| `tools` | `["<tool FUNCTION name>"]` e.g. `["execute_python"]` | **NOT the config_name, NOT schema dicts.** See below. |
+| `env_config` | `{"env_configs": [{"env_name": "<POOL KEY>", "task_id": …}], "max_steps": N}` | `env_name` must equal the **pool key = call name** (§6 trap 3). |
+
+### The `tools` column is the single most error-prone field
+
+There are **two** tool representations in the codebase and they are NOT interchangeable:
+
+- **RL (RLVR) data → name strings.** `rlvr_tokenize_v1` (`dataset_transformation.py:~1539`)
+  does `active_tool_names = set(row["tools"])` then filters the env-provided global
+  `tool_definitions` by `t["function"]["name"] in active_tool_names`. So the column must hold
+  the tool **function name(s)** that match `get_tool_definitions()`. A schema dict here fails
+  the name match → **no tool is injected** and the model silently never gets the tool.
+- **SFT data → schema dicts.** `_normalize_tools_for_chat_template` (`:~976`) expects the
+  column to *be* OpenAI-style schema dicts. That's the SFT path; don't use it for RL.
+
+Verify empirically (cheap, do this for every new env):
+
+```python
+import open_instruct.dataset_transformation as dt
+from datasets import load_dataset
+out = dt.rlvr_tokenize_v1(dict(load_dataset("parquet", data_files=PARQUET, split="train")[0]),
+                          tokenizer, tool_definitions=[MY_TOOL_SCHEMA])
+assert "my_tool_name" in tokenizer.decode(out[dt.INPUT_IDS_PROMPT_KEY])  # schema actually injected
+```
+
+### Launch flags that pair with the dataset
+
+```
+--tools <config_name>            # e.g. appworld   (registry key; creates the pool)
+--tool_call_names <function>     # e.g. execute_python  (pool is keyed by THIS)
+--tool_configs '{"image": "...", ...}'   # run-level env kwargs (one JSON per --tools entry)
+```
+
+Keep the names consistent: **pool key = call name = dataset `tools` entry = `env_config.env_name`**.
+If `get_tool_definitions()`'s function name differs from the config_name (like AppWorld:
+config_name `appworld`, function `execute_python`), you *must* remap via `--tool_call_names`
+and set `env_config.env_name` to the **call name**, not the config_name (§6 trap 3).
+
+Local datasets: `dataset_mixer_list` accepts a local `.parquet`/`.jsonl` path (train split only)
+— great for debug runs with no HF push.
+
+---
+
+## 3. Containers on Beaker (the podman recipe)
+
+Sandboxes don't run a local docker daemon on the trainer; they use a fleet of **podman
+services** on each node. The canonical recipe is
+[scripts/general_agent/terminal/rl/qwen35_4b_base_tmax_10k.sh](../../scripts/general_agent/terminal/rl/qwen35_4b_base_tmax_10k.sh).
+Copy its mason block. The load-bearing pieces:
+
+- `--env BEAKER_ALLOW_SUBCONTAINERS=1` and `--env BEAKER_SKIP_DOCKER_SOCKET=1` — allow the
+  trainer container to spawn sibling containers.
+- `--env SWERL_PODMAN_SERVICE_COUNT=8` — number of podman service instances (the "docker hosts").
+- `--env MIRROR_URL=jupiter-cs-aus-193.reviz.ai2.in:5000` — the cluster **registry mirror /
+  pull-through cache**. It's consumed by the host podman config (not the Python code), so plain
+  `docker.io/...` image refs are pulled through it automatically — **no prefix needed** in your
+  `--tool_configs` image.
+- `--env DOCKERHUB_USERNAME=<user>` + `--secret DOCKER_PAT=<user>_DOCKER_PAT` — docker login
+  (avoids rate limits; required to pull private images).
+- `--env CONTAINERS_STORAGE_CONF=...`, `PODMAN_NUM_LOCKS=...`, `SWERL_DOCKER_AUTO_REMOVE=1`,
+  `SWERL_RESET_FAILURE_ZERO_REWARD=1` (reset failure → zero reward instead of crashing — keep this on),
+  image-janitor envs (cleanup).
+- **Entry command:** `source scripts/docker/docker_login.sh && source configs/beaker_configs/ray_node_setup.sh && python open_instruct/grpo_fast.py …`
+  - **`scripts/docker/docker_login.sh`** is the one that **starts the podman services and sets
+    `SWERL_PODMAN_DOCKER_HOSTS`** (despite the name; `ray_node_setup.sh` does *not*). The pool
+    then injects those hosts as `docker_host` into env resets (requires the `backend="docker"`
+    config field — §1.3).
+
+Launch with **`scripts/train/build_image_and_launch_dirty.sh <script>`** when the working tree
+is dirty (the non-`_dirty` builder refuses uncommitted changes). It builds the trainer image
+and passes it to the script as `$1` (`BEAKER_IMAGE`).
+
+### Networking to the container (style B only) — the big one
+
+**Beaker's podman services run every container on the _host_ network.** Verified by probe:
+`NetworkSettings.Networks = {"host": …}`, `IPAddress=""`, `Ports={}` even when you ask to
+publish. So:
+
+- There is **no bridge IP** and **port publishing is a no-op** — `docker exec` (style A) doesn't
+  care, but an HTTP client (style B) has nothing to connect to via those paths.
+- The server **is** reachable at **`127.0.0.1:<port>`** from the trainer process, because the
+  podman services run inside the trainer's beaker container, so its host netns *is* the
+  trainer's netns. (Probe: `GET http://127.0.0.1:<port>/ → 200`.)
+- Because all containers share that one netns, **they collide on a single port** → each
+  container must bind a **unique port**. Allocate a free port per container in the trainer
+  process (same netns) and pass it as the server's `--port`.
+
+So for style B on Beaker: **don't publish; give each container a unique free port; reach it at
+`127.0.0.1:<that port>`.** Keep the container **bridge IP** as a fallback candidate for *local
+docker* (default bridge network, where `127.0.0.1` won't reach it but the bridge IP will). See
+`AppWorldEnv._pick_free_port` / `_candidate_base_urls` / `_wait_for_server`.
+
+This was the cause of a 4h "hang": the env only probed the (empty) bridge IP and published
+port, never `127.0.0.1`, so every reset failed → zero-reward no-ops → idle GPUs. Diagnose with
+[scripts/general_agent/appworld/debug/probe_container.py](../../scripts/general_agent/appworld/debug/probe_container.py)
+(a no-training job that starts the container and prints status/exit/logs/`NetworkSettings`/reachability),
+and **keep `auto_remove=False` + capture container logs on early exit** — `auto_remove=True`
+deletes the evidence and you go blind.
+
+---
+
+## 4. Validation ladder (always climb it in order)
+
+1. **Pin the env's real API** in a throwaway venv: `python -m venv /tmp/x && /tmp/x/bin/pip
+   install <pkg>`, then read its source for exact signatures. Never guess an external API.
+2. **Standalone env smoke** (no GRPO): instantiate the env, `setup()`, `reset()`, a few
+   `step()`s, confirm reward. Catches the env logic + container lifecycle + reward path.
+3. **Local 2-GPU GRPO** ([scripts/general_agent/appworld/rl/local_rl_2gpu.sh](../../scripts/general_agent/appworld/rl/local_rl_2gpu.sh)):
+   Qwen3-0.6B, local parquet, data-baked image, no HF/Beaker. Catches the dataset contract,
+   pool wiring, prompt enqueue, the duplicate-pool trap, tokenization. **This is where most
+   integration bugs surface** — run it before ever touching Beaker.
+4. **Beaker** via `build_image_and_launch_dirty.sh`. Catches podman-host injection, image pull
+   via mirror, and (style B) container networking.
+
+A tiny debug model won't reliably emit tool calls in-loop — that's expected; use a capable
+model (4B+) to actually exercise the tool, but the small model still validates the *plumbing*.
+
+---
+
+## 5. AppWorld, concretely (the worked example)
+
+- **What it is:** an interactive coding benchmark; the agent writes Python that calls
+  `apis.<app>.<endpoint>(...)` against a stateful simulated world and submits via
+  `apis.supervisor.complete_task()`. Reward = AppWorld's own unit-test pass fraction.
+- **API (pinned against `appworld==0.1.3.post1`):** `AppWorld(task_id, experiment_name, …)` →
+  `.execute(code)->str`, `.task_completed()->bool`, `.evaluate(suppress_errors=True)->TestTracker`
+  (`.success`, `.pass_count`, `.num_tests`; the HTTP server returns `passes`/`failures` lists +
+  `num_tests`). Per-app login passwords come from `apis.supervisor.show_account_passwords()` at
+  **runtime** (not the prompt). Supervisor name/email/phone are synthetic.
+- **Why container+HTTP:** `pydantic<2` pin (can't share the trainer process) **and** REPL state
+  must persist. AppWorld ships an HTTP **environment server** (`appworld serve environment`,
+  image `ghcr.io/stonybrooknlp/appworld:latest`) built for exactly this. Endpoints:
+  `/initialize`, `/execute`, `/task_completed`, `/evaluate`, `/close`. **One world per server →
+  one container per concurrent rollout** (reused across resets via re-`/initialize`).
+- **Data:** `appworld install` + `appworld download data` (~193 MB). Either bind-mount a weka
+  data root (`data_root=APPWORLD_ROOT`) or bake it into the image (`data_root=""`). The data is
+  the "protected portion" — Apache-2.0 **but public redistribution must be encrypted**, so keep
+  the dataset and any data-baked image **private**.
+- **Implementation:** [open_instruct/environments/appworld_env.py](../../open_instruct/environments/appworld_env.py)
+  (env = thin HTTP client, never imports `appworld`),
+  [scripts/data/convert_appworld_to_rl.py](../../scripts/data/convert_appworld_to_rl.py)
+  (reads `data/tasks/<id>/specs.json` off disk; no `appworld` import),
+  [scripts/general_agent/appworld/](../../scripts/general_agent/appworld/) (launch + smoke + README).
+
+---
+
+## 6. Traps that cost time (check every one for a new env)
+
+1. **`tools` column = function names, not config_name, not schema dicts.** Wrong value →
+   model silently gets no tool. Verify with the tokenize snippet in §2.
+2. **Reset observation is dropped** for tool envs — the prompt must be in the dataset
+   `messages`, not returned from `reset()`.
+3. **`env_config.env_name` must equal the pool key (= call name).** If it's the config_name
+   while the pool is keyed by a different call name, `_discover_tools_from_datasets`
+   **auto-creates a duplicate pool** and per-row `task_id` routing misses → the rollout
+   **hangs at prompt enqueue** (vLLM waits for prompts, data-prep waits for results, env actors
+   idle). Symptom check: log says `Initialized 2 tool pools` for one logical env.
+4. **`backend = "docker"` field on the config** is required for the pool to inject
+   `SWERL_PODMAN_DOCKER_HOSTS` → without it, no `docker_host` on beaker, containers can't start.
+5. **Dependency conflicts** (e.g. `pydantic` major) → you cannot import the env in the trainer
+   process. Go container + HTTP; the env module must not import the conflicting package at all.
+6. **Lazy-import optional deps.** `tools/tools.py` imports your env module at startup; importing
+   a heavy/optional dep at module top breaks every install. Import inside `setup()` with `# noqa: PLC0415`.
+7. **Local `.parquet`/`.jsonl` in `dataset_mixer_list`** now works in
+   `_discover_tools_from_datasets` too (it previously only worked in `DatasetConfig` and threw
+   `FileNotFoundError` in discovery — fixed). Keep both paths in mind if you add more loaders.
+8. **Container networking** (style B): Beaker podman = **host network** (no bridge IP, publish is
+   a no-op). Give each container a **unique free port** and reach it at **`127.0.0.1:<port>`**;
+   keep the bridge IP as a local-docker fallback. (And `auto_remove=False` so you can read a
+   dead container's logs.)
+9. **Dev-box docker has no shared FS with the daemon sometimes** (e.g. this sandbox: daemon root
+   `/media/8TBNVME`, no `/tmp` sharing) → bind mounts silently mount empty dirs. Validate by
+   `docker run --rm -v <src>:/x alpine ls /x`; if empty, **bake data into the image** instead.
+10. **`build_image_and_launch.sh` refuses a dirty tree** ("Uncommitted changes detected") →
+    use `build_image_and_launch_dirty.sh`.
+11. **One container per concurrent rollout.** `pool_size = num_unique_prompts_rollout *
+    num_samples_per_prompt_rollout` (cap concurrency; don't over-provision heavyweight containers).
+12. **Pin the external API in a throwaway venv** before writing code — don't guess signatures.
+
+---
+
+## 7. Copy-paste checklist for a new sandboxed env `myenv`
+
+```
+[ ] Decide style A (bash-exec, reuse DockerBackend) vs B (stateful server + HTTP client).
+[ ] Throwaway venv: install the env's package, read its source, pin exact API.
+[ ] Dependency conflict with pydantic>=2 / openenv? If yes -> style B, container quarantine.
+[ ] open_instruct/environments/myenv_env.py:
+      class MyEnv(RLEnvironment): config_name="myenv"; backend field via config; lazy import deps in setup()
+      get_tool_definitions() -> [ { "type":"function","function":{"name":"<fn>", ...} } ]
+      reset(task_id, docker_host=..., max_steps=...): start/reuse container, init episode, return (StepResult(""), tools)
+      step(call): dispatch by call.name, detect done, compute reward; truncate long observations
+      close()/shutdown(): tear down container
+    @dataclass MyEnvConfig(BaseEnvConfig): tool_class=MyEnv; backend="docker"; image=...; <params>
+[ ] Register in tools/tools.py TOOL_REGISTRY (import + entry).
+[ ] Data converter -> rows: messages(system+user instruction), ground_truth(task_id),
+      dataset="passthrough", tools=["<fn>"], env_config={env_configs:[{env_name:"<fn>",task_id}],max_steps}
+[ ] Verify the tool schema injects (rlvr_tokenize_v1 snippet, §2).
+[ ] tests/test_environments.py: prompt builder, reward mapping, registry, get_tool_definitions, reset-needs-id.
+[ ] local_rl_2gpu.sh: Qwen3-0.6B, local parquet, data-baked image, run end-to-end.
+[ ] Beaker script: copy the podman recipe (§3), set --tools myenv --tool_call_names <fn> --tool_configs {...}.
+[ ] Data/image private if license-restricted.
+[ ] make style && uv run pytest tests/test_environments.py -k MyEnv
+```
+
+---
+
+## 8. Key file/line index (jump straight here)
+
+| Concern | File |
+|---|---|
+| Env ABC, TextRLEnvironment, BaseEnvConfig | `open_instruct/environments/base.py` |
+| Registry | `open_instruct/environments/tools/tools.py` (`TOOL_REGISTRY`, ~652) |
+| Pool, podman host rotation, `backend=="docker"` gate | `open_instruct/environments/pool.py` (~74) |
+| Docker/Apptainer backends (style A) | `open_instruct/environments/backends.py` |
+| Tool discovery, pool creation, tool_definitions | `open_instruct/grpo_fast.py` (`_discover_tools_from_datasets` ~3050, `create_tool_pools` ~1962, `initialize_tools_and_envs` ~3085) |
+| Dataset tools validation | `open_instruct/grpo_fast.py:_validate_and_log_dataset_tools` (~1876) |
+| `tools` column semantics (names vs schemas) | `open_instruct/dataset_transformation.py` (`rlvr_tokenize_v1` ~1518/1539, `_normalize_tools_for_chat_template` ~976, local-file loader ~1793) |
+| Rollout: reset-before-gen, allowed_tools, reward collect, one-text-env | `open_instruct/vllm_utils.py` (`_acquire_and_reset_pools` ~1042, ~1090, ~1231, ~1285) |
+| passthrough verifier, reward selection | `open_instruct/ground_truth_utils.py` (~1037, ~1253) |
+| Podman services + `SWERL_PODMAN_DOCKER_HOSTS` | `scripts/docker/docker_login.sh` |
+| Ray + node setup | `configs/beaker_configs/ray_node_setup.sh` |
+| Dirty-tree launcher | `scripts/train/build_image_and_launch_dirty.sh` |
+| Podman recipe reference | `scripts/general_agent/terminal/rl/qwen35_4b_base_tmax_10k.sh` |
+| Style-B worked example | `open_instruct/environments/appworld_env.py` + `scripts/general_agent/appworld/` |
diff --git a/mkdocs.yml b/mkdocs.yml
index e7cc71bd8a..47c2f94a5b 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -92,6 +92,7 @@ nav:
     - algorithms/grpo_pipeline_overview.md
     - algorithms/grpo_fast_internals.md
     - algorithms/rl_with_environments.md
+    - algorithms/adding_sandboxed_environments.md
     - algorithms/rollout_loop_internals.md
     - algorithms/monitoring_and_debugging_runs.md
     - algorithms/terminal_rl_trajectory_analysis.md
diff --git a/open_instruct/environments/appworld_env.py b/open_instruct/environments/appworld_env.py
new file mode 100644
index 0000000000..ac345ff7f0
--- /dev/null
+++ b/open_instruct/environments/appworld_env.py
@@ -0,0 +1,530 @@
+"""AppWorld code-as-action RL environment (container + HTTP).
+
+Wraps the AppWorld benchmark (https://appworld.dev, ACL'24 best resource paper)
+as an open-instruct RL environment. The policy solves a task by writing Python
+that calls ``apis.<app>.<endpoint>(...)`` against a stateful, simulated world of
+apps (Amazon, Spotify, Venmo, phone, file system, ...) on behalf of a supervisor,
+and submits by calling ``apis.supervisor.complete_task()``. Reward is AppWorld's
+own programmatic evaluation: the fraction of unit tests passed (dense) or whether
+*all* tests pass (binary Task Goal Completion).
+
+Why a container + HTTP (not in-process)
+---------------------------------------
+AppWorld pins ``pydantic<2`` while open-instruct/openenv require ``pydantic>=2``;
+they cannot share a Python process. AppWorld ships an HTTP *environment server*
+for exactly this case, so we mirror the Terminal RL (swerl) podman/docker workflow:
+
+* One AppWorld container per rollout (the server holds a single world at a time,
+  matching swerl's one-container-per-rollout ratio), reused across resets.
+* The container runs ``appworld serve environment``; this env is a thin HTTP client
+  (``/initialize``, ``/execute``, ``/task_completed``, ``/evaluate``, ``/close``) and
+  never imports ``appworld`` — so the trainer process stays pydantic-2-clean.
+* Unlike swerl (stateless ``docker exec bash`` per turn), AppWorld is a stateful
+  Python REPL: the agent's variables and the world object must persist in one
+  long-lived process, hence the in-container server + HTTP rather than exec-per-turn.
+
+The container image (``ghcr.io/stonybrooknlp/appworld:latest``) bundles AppWorld;
+the host provides the task data via a bind mount (``data_root``). Model-generated
+code runs inside the container, isolating it from the trainer.
+"""
+
+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
+
+logger = logger_utils.setup_logger(__name__)
+
+EXECUTE_TOOL = {
+    "type": "function",
+    "function": {
+        "name": "execute_python",
+        "description": (
+            "Execute a snippet of Python code in a persistent IPython session. "
+            "The namespace already contains `apis` (the AppWorld app APIs) and persists "
+            "across calls, so variables defined in one call are available in the next. "
+            "Use `print(...)` to inspect values; only printed output is returned. "
+            "Discover APIs with `print(apis.api_docs.show_app_descriptions())`, "
+            "`print(apis.api_docs.show_api_descriptions(app_name='...'))`, and "
+            "`print(apis.api_docs.show_api_doc(app_name='...', api_name='...'))`. "
+            "Call `apis.supervisor.complete_task()` when the task is finished."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {"code": {"type": "string", "description": "The Python code to execute."}},
+            "required": ["code"],
+        },
+    },
+}
+
+# Canonical code-as-action system prompt, adapted from AppWorld's reference REACT
+# agent for the single-tool ``execute_python`` interface. Kept here (not in the data
+# converter) so the env and the dataset builder share one source of truth.
+SYSTEM_PROMPT = """\
+You are a coding agent who interacts with a digital world of apps by writing and \
+executing Python code. You complete a task on behalf of your supervisor by calling \
+the apps' APIs through the `apis` object.
+
+How you operate:
+- At each step you write a short snippet of Python code. It runs in a persistent \
+IPython session; only what you `print(...)` is shown back to you, and your variables \
+persist across steps.
+- The `apis` object exposes one attribute per app. Read the docs before calling \
+anything you are unsure about:
+  - `print(apis.api_docs.show_app_descriptions())` lists the apps.
+  - `print(apis.api_docs.show_api_descriptions(app_name='amazon'))` lists an app's APIs.
+  - `print(apis.api_docs.show_api_doc(app_name='amazon', api_name='search_products'))` \
+shows one API's arguments and response schema.
+- Most apps require an access token. Get the supervisor's per-app passwords with \
+`print(apis.supervisor.show_account_passwords())`, then log in, e.g. \
+`apis.<app>.login(username=<supervisor email or phone>, password=<that app's password>)`, \
+and pass the returned access token to subsequent calls.
+- Work incrementally: inspect results, handle pagination, and verify before acting. \
+Write robust code (loops, error handling) rather than guessing.
+- When, and only when, the task is fully done, call `apis.supervisor.complete_task()` \
+(pass `answer=...` if the task asks for an answer). Do not call it prematurely.
+
+Your supervisor's details:
+{supervisor_block}
+"""
+
+USER_PROMPT = "Task:\n{instruction}"
+
+# Observation truncation (head/tail) for long execution output.
+_OBS_MAX_CHARS = 8000
+_OBS_HEAD_CHARS = 4000
+_OBS_TAIL_CHARS = 3000
+
+
+def _supervisor_dict(supervisor: Any) -> dict[str, str]:
+    """Extract the supervisor fields an agent needs to log in, robustly."""
+    fields = ("first_name", "last_name", "email", "phone_number", "password")
+    out: dict[str, str] = {}
+    for key in fields:
+        value = getattr(supervisor, key, None)
+        if value is None and isinstance(supervisor, dict):
+            value = supervisor.get(key)
+        if value is not None:
+            out[key] = str(value)
+    return out
+
+
+def format_supervisor_block(supervisor: Any) -> str:
+    info = _supervisor_dict(supervisor)
+    label = {
+        "first_name": "First name",
+        "last_name": "Last name",
+        "email": "Email",
+        "phone_number": "Phone number",
+        "password": "Password",
+    }
+    lines = [f"- {label.get(k, k)}: {v}" for k, v in info.items()]
+    return "\n".join(lines) if lines else "- (no supervisor details provided)"
+
+
+def build_prompt_messages(instruction: str, supervisor: Any) -> list[dict[str, str]]:
+    """Build the chat messages for an AppWorld task.
+
+    The env's :meth:`reset` observation is discarded by the rollout pipeline, so
+    the task instruction and supervisor credentials must live in the dataset
+    ``messages``. The data converter calls this to bake the prompt into each row.
+    """
+    return [
+        {"role": "system", "content": SYSTEM_PROMPT.format(supervisor_block=format_supervisor_block(supervisor))},
+        {"role": "user", "content": USER_PROMPT.format(instruction=instruction)},
+    ]
+
+
+def truncate_observation(output: str) -> str:
+    if len(output) <= _OBS_MAX_CHARS:
+        return output
+    elided = len(output) - _OBS_HEAD_CHARS - _OBS_TAIL_CHARS
+    return f"{output[:_OBS_HEAD_CHARS]}\n---- {elided} chars elided ----\n{output[-_OBS_TAIL_CHARS:]}"
+
+
+def reward_from_evaluation(evaluation: dict, dense: bool) -> tuple[float, dict[str, float]]:
+    """Map the AppWorld ``/evaluate`` payload to a scalar reward + metrics.
+
+    The server returns ``TestTracker.to_dict()``: ``passes`` (list), ``failures``
+    (list), ``num_tests`` (int), ``success`` (bool). Dense reward is the pass
+    fraction; binary reward is all-pass (Task Goal Completion).
+    """
+    num_tests = int(evaluation.get("num_tests", 0) or 0)
+    pass_count = len(evaluation.get("passes", []) or [])
+    success = bool(evaluation.get("success", num_tests > 0 and pass_count == num_tests))
+    dense_reward = pass_count / num_tests if num_tests > 0 else 0.0
+    reward = dense_reward if dense else (1.0 if success else 0.0)
+    metrics = {"num_tests": float(num_tests), "pass_count": float(pass_count), "success": float(success)}
+    return max(0.0, min(1.0, reward)), metrics
+
+
+class AppWorldEnv(RLEnvironment):
+    """AppWorld code-as-action environment: a per-rollout container + HTTP client."""
+
+    config_name = "appworld"
+    response_role = "tool"
+    _tool_definitions = (EXECUTE_TOOL,)
+
+    def __init__(
+        self,
+        image: str = "ghcr.io/stonybrooknlp/appworld:latest",
+        data_root: str = "",
+        in_container_root: str = "/run",
+        server_port: int = 8000,
+        max_interactions: int = 50,
+        timeout_seconds: int = 120,
+        startup_timeout: int = 180,
+        dense_reward: bool = True,
+        mem_limit: str = "4g",
+        experiment_prefix: str = "appworld_rl",
+        backend: str = "docker",
+        publish_port: bool = False,
+        start_retries: int = 5,
+        docker_host: str | None = None,
+        **kwargs: Any,
+    ):
+        # Tolerate generic config kwargs injected by the pool (e.g. call_name, penalty).
+        kwargs.pop("call_name", None)
+        kwargs.pop("penalty", None)
+        # ``backend`` exists so EnvironmentPool engages SWERL_PODMAN_DOCKER_HOSTS
+        # rotation (it keys on backend == "docker") and injects a docker_host into
+        # reset kwargs on beaker; we always use the docker SDK regardless of value.
+        self._backend = backend
+        self._image = image
+        self._data_root = data_root
+        self._in_container_root = in_container_root.rstrip("/")
+        self._server_port = server_port
+        self._max_interactions = max_interactions
+        self._timeout_seconds = timeout_seconds
+        self._startup_timeout = startup_timeout
+        self._dense_reward = dense_reward
+        self._mem_limit = mem_limit
+        self._experiment_prefix = experiment_prefix
+        self._publish_port = publish_port
+        self._start_retries = start_retries
+        self._docker_host = docker_host
+
+        self._client: Any = None  # docker client (lazily imported)
+        self._container: Any = None
+        self._port = server_port  # per-container server port (reassigned in _start_container)
+        self._base_url: str | None = None
+        self._task_id: str | None = None
+        self._experiment_name: str | None = None
+        self._step_count = 0
+        self._completed = False
+        self._last_reward = 0.0
+        self._last_eval: dict[str, float] = {}
+        self._max_steps: int | None = None
+
+    async def setup(self) -> None:
+        """Create the docker client. AppWorld is never imported in this process."""
+        if self._client is not None:
+            return
+        import docker  # noqa: PLC0415 -- optional dep; keep out of module import path
+
+        self._client = (
+            docker.DockerClient(base_url=self._docker_host) if self._docker_host else docker.from_env()
+        )
+
+    @classmethod
+    def get_tool_definitions(cls) -> list[dict]:
+        return list(cls._tool_definitions)
+
+    # ---- container lifecycle -------------------------------------------------
+
+    def _resolve_host(self) -> str:
+        """Host to reach published container ports on (bridge IP for remote podman)."""
+        if self._docker_host:
+            parsed = urlparse(self._docker_host)
+            if parsed.hostname:
+                return parsed.hostname
+        return "127.0.0.1"
+
+    @staticmethod
+    def _pick_free_port() -> int:
+        """Grab a currently-free TCP port in this process's network namespace.
+
+        On Beaker the podman services run containers on the *host* network (shared with this
+        trainer netns), so every AppWorld server binds 127.0.0.1 here and concurrent
+        containers must use distinct ports — hence one free port per container. A bind race
+        (container takes the port between close() and start) just fails startup and the reset
+        retries with a new port.
+        """
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.bind(("127.0.0.1", 0))
+            return s.getsockname()[1]
+
+    def _start_container(self) -> None:
+        # Bind-mount the host data when data_root is set (Beaker/weka nodes); when it
+        # is empty, the image is expected to already contain the data at
+        # {in_container_root}/data (a data-baked image, e.g. for envs without a shared FS).
+        volumes = {}
+        if self._data_root:
+            volumes = {
+                f"{self._data_root}/data": {"bind": f"{self._in_container_root}/data", "mode": "ro"},
+                f"{self._data_root}/experiments/outputs": {
+                    "bind": f"{self._in_container_root}/experiments/outputs",
+                    "mode": "rw",
+                },
+            }
+        # Unique port per container: Beaker's podman defaults containers onto the *host*
+        # network (no bridge IP, publishing is a no-op), so the server is reached at
+        # 127.0.0.1:<port> and concurrent containers cannot share one port. There's an
+        # inherent race between picking a free port and the container binding it (another
+        # container can grab it in between -> "address already in use" -> the server exits),
+        # so retry with a fresh port each attempt. On exhaustion raise with "reset failed
+        # after" so the rollout is zero-rewarded (not fatal) per RESET_FAILURE_ZERO_REWARD.
+        last_error: Exception | None = None
+        for attempt in range(1, self._start_retries + 1):
+            self._port = self._pick_free_port()
+            container = self._client.containers.run(
+                self._image,
+                command=["environment", "--port", str(self._port), "--no-show-usage"],
+                environment={"APPWORLD_ROOT": self._in_container_root},
+                volumes=volumes,
+                mem_limit=self._mem_limit,
+                memswap_limit=self._mem_limit,
+                detach=True,
+                # auto_remove is intentionally OFF: an instantly-exiting container would be
+                # deleted before we could read its logs/exit code. We remove it ourselves.
+                auto_remove=False,
+                labels={"appworld_rl": "1"},
+            )
+            self._container = container
+            try:
+                self._base_url = self._wait_for_server(container)
+                return
+            except Exception as e:
+                last_error = e
+                self._remove_container(container)
+                self._container = None
+                logger.warning(
+                    "AppWorld container start attempt %d/%d failed (port %d): %s",
+                    attempt,
+                    self._start_retries,
+                    self._port,
+                    e,
+                )
+        raise RuntimeError(
+            f"AppWorld reset failed after {self._start_retries} container-start attempts: {last_error}"
+        )
+
+    @staticmethod
+    def _remove_container(container: Any) -> None:
+        with contextlib.suppress(Exception):
+            container.remove(force=True)
+
+    def _container_exit_detail(self, container: Any) -> str:
+        """Exit code + log tail of a container that died — the actual failure reason."""
+        try:
+            container.reload()
+            state = container.attrs.get("State", {}) or {}
+            logs = container.logs(tail=60).decode("utf-8", "replace")
+            return f"exit_code={state.get('ExitCode')} error={state.get('Error')!r}\n--- container logs (tail) ---\n{logs}"
+        except Exception as e:
+            return f"(could not read container logs: {e})"
+
+    def _candidate_base_urls(self, container: Any) -> list[str]:
+        """Candidate base URLs to reach the in-container server, most-portable first.
+
+        Reachability differs by setup, so we probe and use whichever answers:
+        - ``127.0.0.1:<port>`` — Beaker podman defaults containers onto the host network
+          (shared with this netns), so the server is here. Primary path.
+        - container bridge IP — local docker with the default bridge network.
+        - published host port — only if port publishing was enabled and honored.
+        """
+        candidates: list[str] = [f"http://127.0.0.1:{self._port}"]
+        container.reload()
+        net = container.attrs.get("NetworkSettings", {}) or {}
+        ip = net.get("IPAddress") or next(
+            (n.get("IPAddress") for n in (net.get("Networks") or {}).values() if n.get("IPAddress")), None
+        )
+        if ip:
+            candidates.append(f"http://{ip}:{self._port}")
+        mapping = (net.get("Ports") or {}).get(f"{self._port}/tcp")
+        if mapping:
+            candidates.append(f"http://{self._resolve_host()}:{mapping[0]['HostPort']}")
+        return candidates
+
+    def _wait_for_server(self, container: Any) -> str:
+        deadline = time.monotonic() + self._startup_timeout
+        candidates: list[str] = []
+        while time.monotonic() < deadline:
+            try:
+                container.reload()
+            except Exception as e:
+                raise RuntimeError(
+                    f"AppWorld container vanished during startup (cannot inspect): {e}. "
+                    "If this is podman with auto-remove, the container exited instantly."
+                ) from e
+            if container.status not in ("created", "running"):
+                raise RuntimeError(
+                    f"AppWorld container exited early (status={container.status}). "
+                    f"{self._container_exit_detail(container)}"
+                )
+            candidates = self._candidate_base_urls(container)
+            for base_url in candidates:
+                with contextlib.suppress(requests.RequestException):
+                    if requests.get(f"{base_url}/", timeout=5).status_code == 200:
+                        logger.info(f"[{self._task_id}] AppWorld server reachable at {base_url}")
+                        return base_url
+            time.sleep(1.0)
+        raise RuntimeError(
+            f"AppWorld server did not become ready within {self._startup_timeout}s "
+            f"(tried {candidates}). {self._container_exit_detail(container)}"
+        )
+
+    def _http(self, method: str, **body: Any) -> Any:
+        if self._base_url is None:
+            raise RuntimeError("AppWorld container not started.")
+        response = requests.post(f"{self._base_url}/{method}", json=body, timeout=self._timeout_seconds + 60)
+        if response.status_code != 200:
+            raise RuntimeError(f"AppWorld /{method} failed ({response.status_code}): {response.text[:500]}")
+        return response.json()["output"]
+
+    # ---- env API -------------------------------------------------------------
+
+    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"]
+            self._teardown_container()
+            await self.setup()
+        if self._container is None or self._base_url is None:
+            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.
+        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)
+
+    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.")
+
+        output = self._http("execute", task_id=self._task_id, code=code)
+        observation = truncate_observation(output) if output else "(no output)"
+
+        self._completed = bool(self._http("task_completed", task_id=self._task_id))
+        at_step_limit = self._max_steps is not None and self._step_count >= self._max_steps
+        if self._completed or at_step_limit:
+            reward = 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})
+
+    def _compute_reward(self) -> float:
+        try:
+            evaluation = self._http("evaluate", task_id=self._task_id, suppress_errors=True, report=False)
+            reward, self._last_eval = reward_from_evaluation(evaluation, self._dense_reward)
+        except Exception as e:
+            logger.warning(f"[{self._task_id}] AppWorld evaluation failed: {e}")
+            self._last_eval = {"num_tests": 0.0, "pass_count": 0.0, "success": 0.0}
+            reward = 0.0
+        self._last_reward = reward
+        return reward
+
+    def get_metrics(self) -> dict[str, float]:
+        return {
+            "step_count": float(self._step_count),
+            "task_completed": float(self._completed),
+            "reward": float(self._last_reward),
+            **self._last_eval,
+        }
+
+    def state(self) -> State:
+        return State(episode_id=self._task_id, step_count=self._step_count)
+
+    def _teardown_container(self) -> None:
+        if self._task_id is not None and self._base_url is not None:
+            with contextlib.suppress(Exception):
+                self._http("close", task_id=self._task_id)
+        if self._container is not None:
+            with contextlib.suppress(Exception):
+                self._container.stop(timeout=5)
+            with contextlib.suppress(Exception):
+                self._container.remove(force=True)
+        self._container = None
+        self._base_url = None
+
+    async def close(self) -> None:
+        self._teardown_container()
+
+    async def shutdown(self) -> None:
+        await self.close()
+
+
+@dataclass
+class AppWorldEnvConfig(BaseEnvConfig):
+    """Configuration for :class:`AppWorldEnv`."""
+
+    tool_class: ClassVar[type[RLEnvironment]] = AppWorldEnv
+    image: str = "ghcr.io/stonybrooknlp/appworld:latest"
+    data_root: str = ""
+    in_container_root: str = "/run"
+    server_port: int = 8000
+    max_interactions: int = 50
+    timeout_seconds: int = 120
+    startup_timeout: int = 180
+    dense_reward: bool = True
+    mem_limit: str = "4g"
+    experiment_prefix: str = "appworld_rl"
+    backend: str = "docker"
+    publish_port: bool = False
+    start_retries: int = 5
+    docker_host: str | None = None
diff --git a/open_instruct/environments/tools/tools.py b/open_instruct/environments/tools/tools.py
index f26f16c633..c836f0450d 100644
--- a/open_instruct/environments/tools/tools.py
+++ b/open_instruct/environments/tools/tools.py
@@ -10,6 +10,7 @@
 from typing import Any, ClassVar
 
 from open_instruct import logger_utils
+from open_instruct.environments.appworld_env import AppWorldEnvConfig
 from open_instruct.environments.base import BaseEnvConfig, EnvCall, StepResult
 from open_instruct.environments.examples import CounterEnvConfig, GuessNumberEnvConfig, WordleTextEnvConfig
 from open_instruct.environments.generic_sandbox import GenericSandboxEnvConfig
@@ -663,4 +664,5 @@ class DrAgentMCPToolConfig(BaseEnvConfig):
     SWERLSandboxEnvConfig.tool_class.config_name: SWERLSandboxEnvConfig,
     SWERLVanilluxSandboxEnvConfig.tool_class.config_name: SWERLVanilluxSandboxEnvConfig,
     WordleTextEnvConfig.tool_class.config_name: WordleTextEnvConfig,
+    AppWorldEnvConfig.tool_class.config_name: AppWorldEnvConfig,
 }
diff --git a/open_instruct/grpo_fast.py b/open_instruct/grpo_fast.py
index 4745ca2794..f73fad5387 100644
--- a/open_instruct/grpo_fast.py
+++ b/open_instruct/grpo_fast.py
@@ -3059,7 +3059,14 @@ def _discover_tools_from_datasets(dataset_mixer_list: list[str], dataset_mixer_l
     for i in range(0, len(dataset_mixer_list), 2):
         dataset_name = dataset_mixer_list[i]
         split = splits[i // 2]
-        ds = datasets.load_dataset(dataset_name, split=split)
+        # Mirror DatasetConfig's local-file handling so local .jsonl/.parquet datasets
+        # work here too (datasets.load_dataset(path) alone does not resolve a file path).
+        if os.path.exists(dataset_name) and dataset_name.endswith(".jsonl"):
+            ds = datasets.load_dataset("json", data_files=dataset_name, split=split)
+        elif os.path.exists(dataset_name) and dataset_name.endswith(".parquet"):
+            ds = datasets.load_dataset("parquet", data_files=dataset_name, split=split)
+        else:
+            ds = datasets.load_dataset(dataset_name, split=split)
         if TOOLS_COLUMN_KEY in ds.column_names:
             for tools in ds[TOOLS_COLUMN_KEY]:
                 if tools:
diff --git a/scripts/data/convert_appworld_to_rl.py b/scripts/data/convert_appworld_to_rl.py
new file mode 100644
index 0000000000..64bc8a72bc
--- /dev/null
+++ b/scripts/data/convert_appworld_to_rl.py
@@ -0,0 +1,101 @@
+"""Convert an AppWorld split into an open-instruct RL (RLVR) dataset.
+
+Produces one row per AppWorld task with the columns the GRPO pipeline routes on:
+
+    messages     : [system (with supervisor creds), user (task instruction)]
+    ground_truth : task_id (the AppWorldEnv verifies internally via world.evaluate)
+    dataset      : "passthrough" (no extra verifier; the env emits the reward)
+    tools        : ["execute_python"] (the tool *call name*; selects the schema injected
+                   into the prompt and gates dispatch — pair with --tool_call_names execute_python)
+    env_config   : {"env_configs": [{"env_name": "appworld", "task_id": ...}], "max_steps": N}
+
+Reads AppWorld task data straight off disk (split file + per-task ``specs.json``) so it
+does NOT import the (pydantic-1) ``appworld`` package — it runs in the open-instruct venv.
+Point --data_root at an AppWorld data root (the dir containing ``data/tasks`` and
+``data/datasets``), e.g. one produced by ``appworld download data``.
+
+Example:
+    uv run python scripts/data/convert_appworld_to_rl.py \
+        --data_root /weka/.../appworld_root --splits train,dev,test_normal,test_challenge --private \
+        --push_to_hub <org>/appworld-train-rl
+"""
+
+import argparse
+import json
+import os
+
+from datasets import Dataset, DatasetDict
+
+from open_instruct import logger_utils
+from open_instruct.environments.appworld_env import build_prompt_messages
+
+logger = logger_utils.setup_logger(__name__)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--data_root", required=True, help="AppWorld data root (contains data/tasks, data/datasets).")
+    parser.add_argument(
+        "--splits",
+        default="train",
+        help="Comma-separated split files under data/datasets (e.g. train,dev,test_normal,test_challenge). "
+        "Each becomes a named split in the pushed dataset.",
+    )
+    parser.add_argument("--max_steps", type=int, default=50, help="Per-rollout interaction budget (env max_steps).")
+    parser.add_argument("--push_to_hub", default="", help="HF repo id to push to (optional).")
+    parser.add_argument("--private", action="store_true", help="Push as a private dataset (AppWorld license).")
+    parser.add_argument("--output_parquet", default="", help="Local parquet path for the first split (optional).")
+    parser.add_argument("--limit", type=int, default=0, help="Keep only the first N tasks per split (0 = all).")
+    return parser.parse_args()
+
+
+def load_task_ids(data_root: str, split: str) -> list[str]:
+    split_path = os.path.join(data_root, "data", "datasets", f"{split}.txt")
+    with open(split_path, encoding="utf-8") as fh:
+        # Lines may carry an optional "tag:task_id" prefix; keep the task_id.
+        return [line.strip().split(":")[-1] for line in fh if line.strip()]
+
+
+def build_row(data_root: str, task_id: str, max_steps: int) -> dict:
+    specs_path = os.path.join(data_root, "data", "tasks", task_id, "specs.json")
+    with open(specs_path, encoding="utf-8") as fh:
+        specs = json.load(fh)
+    return {
+        "messages": build_prompt_messages(specs["instruction"], specs["supervisor"]),
+        "ground_truth": task_id,
+        "dataset": "passthrough",
+        "tools": ["execute_python"],
+        # env_name must match the pool key, which is the *call name* (execute_python),
+        # not the registry config_name (appworld) — otherwise tool discovery creates a
+        # duplicate pool and per-row task_id routing misses. Pair with:
+        #   --tools appworld --tool_call_names execute_python
+        "env_config": {"env_configs": [{"env_name": "execute_python", "task_id": task_id}], "max_steps": max_steps},
+    }
+
+
+def build_split(data_root: str, split: str, max_steps: int, limit: int) -> Dataset:
+    task_ids = load_task_ids(data_root, split)
+    if limit:
+        task_ids = task_ids[:limit]
+    logger.info(f"Building AppWorld split '{split}': {len(task_ids)} tasks.")
+    return Dataset.from_list([build_row(data_root, task_id, max_steps) for task_id in task_ids])
+
+
+def main() -> None:
+    args = parse_args()
+    splits = [s.strip() for s in args.splits.split(",") if s.strip()]
+    dataset_dict = DatasetDict({split: build_split(args.data_root, split, args.max_steps, args.limit) for split in splits})
+    logger.info(f"Built DatasetDict with splits {list(dataset_dict)}; columns {dataset_dict[splits[0]].column_names}")
+
+    if args.output_parquet:
+        dataset_dict[splits[0]].to_parquet(args.output_parquet)
+        logger.info(f"Wrote {args.output_parquet} (split '{splits[0]}')")
+    if args.push_to_hub:
+        dataset_dict.push_to_hub(args.push_to_hub, private=args.private)
+        logger.info(f"Pushed to hub: {args.push_to_hub} (private={args.private})")
+    if not args.output_parquet and not args.push_to_hub:
+        logger.warning("Neither --output_parquet nor --push_to_hub set; dataset was built but not saved.")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/general_agent/appworld/README.md b/scripts/general_agent/appworld/README.md
new file mode 100644
index 0000000000..3810bbdc1a
--- /dev/null
+++ b/scripts/general_agent/appworld/README.md
@@ -0,0 +1,99 @@
+# AppWorld RL (code-as-action)
+
+Train a policy on [AppWorld](https://appworld.dev) tasks: the model writes Python that
+drives a simulated world of apps (Amazon, Spotify, Venmo, phone, file system, …) via a
+single `execute_python` tool, and is rewarded by AppWorld's own programmatic evaluation
+(fraction of unit tests passed).
+
+## Execution model: per-rollout container + HTTP (mirrors swerl)
+
+AppWorld pins **pydantic <2**; open-instruct/openenv need **pydantic ≥2** — they can't
+share a process. So we use AppWorld's HTTP *environment server* and mirror the Terminal
+RL (swerl) podman/docker workflow:
+
+- One AppWorld container per rollout (`ghcr.io/stonybrooknlp/appworld:latest`), running
+  `appworld serve environment`. The server holds one task world at a time → one container
+  per concurrent rollout, reused across resets (re-`/initialize` per task).
+- `AppWorldEnv` is a thin **HTTP client** (`/initialize`, `/execute`, `/task_completed`,
+  `/evaluate`, `/close`) that **never imports `appworld`** — the trainer stays pydantic-2-clean.
+- Unlike swerl's stateless `docker exec bash`, AppWorld is a stateful Python REPL (agent
+  variables persist across turns), so the container runs a long-lived server and we talk
+  HTTP, rather than exec-per-turn.
+- Networking: locally the env reaches the container on its **bridge IP** + in-container
+  port; for a remote `docker_host` it uses the published host port. (Same sibling-container
+  caveat as the harbor/tmax evals.)
+- The trainer needs a docker/podman socket, exactly like swerl.
+
+## Pieces
+
+| Path | What |
+|------|------|
+| [open_instruct/environments/appworld_env.py](../../../open_instruct/environments/appworld_env.py) | `AppWorldEnv` (config_name `appworld`): container lifecycle + HTTP client + prompt builder. |
+| [scripts/data/convert_appworld_to_rl.py](../../data/convert_appworld_to_rl.py) | Build the RLVR dataset (reads `specs.json` off disk; no `appworld` import). |
+| [smoke_test_appworld_env.py](smoke_test_appworld_env.py) | End-to-end env check (starts a real container; reset → execute → evaluate → reward). |
+| [rl/qwen35_4b_appworld.sh](rl/qwen35_4b_appworld.sh) | GRPO launch script. |
+
+## Setup (image + data)
+
+```bash
+docker pull ghcr.io/stonybrooknlp/appworld:latest
+
+# Get the task data (run once, in a throwaway pydantic-1 venv — NOT the trainer venv):
+pip install appworld && appworld install && appworld download data --root $APPWORLD_ROOT
+```
+
+The **docker daemon of each node** must be able to provide the data to the container. Two
+options:
+
+- **Bind-mount (default)**: stage `$APPWORLD_ROOT` (contains `data/` and
+  `experiments/outputs/`) on a path the daemon sees (e.g. weka) and pass `APPWORLD_ROOT`.
+  The env mounts `data/`→`/run/data` (ro) and `experiments/outputs/`→`/run/experiments/outputs` (rw).
+- **Data-baked image**: build a derivative and set `APPWORLD_ROOT=""` (no mount). Useful
+  where the daemon has no shared FS with the host:
+  ```dockerfile
+  FROM ghcr.io/stonybrooknlp/appworld:latest
+  COPY data /run/data
+  RUN mkdir -p /run/experiments/outputs
+  ```
+  `docker build -t <org>/appworld-data:latest .` then set `APPWORLD_IMAGE` to it.
+
+> `appworld` is intentionally NOT a dependency of this repo (pydantic conflict) — it lives
+> only inside the container image.
+
+## Build the dataset
+
+```bash
+uv run python scripts/data/convert_appworld_to_rl.py \
+    --data_root $APPWORLD_ROOT --splits train,dev,test_normal,test_challenge --private \
+    --push_to_hub <org>/appworld-train-rl
+```
+
+Each row: `messages` (system with supervisor creds + user instruction), `ground_truth`
+(task_id), `dataset="passthrough"` (env emits the reward), `tools=["appworld"]`,
+`env_config` (per-row `task_id` + `max_steps`).
+
+## Smoke-test the env (needs Docker + image + data)
+
+```bash
+uv run python scripts/general_agent/appworld/smoke_test_appworld_env.py --data_root $APPWORLD_ROOT
+# or against a data-baked image:
+#   ... --image <org>/appworld-data:latest --data_root ""   # (edit the script's data_root)
+```
+
+## Launch RL
+
+```bash
+APPWORLD_DATASET=<org>/appworld-train-rl APPWORLD_ROOT=/weka/.../appworld_root \
+  ./scripts/train/build_image_and_launch.sh scripts/general_agent/appworld/rl/qwen35_4b_appworld.sh
+```
+
+## Design notes
+
+- **Reward**: `/evaluate` returns `passes`/`failures`/`num_tests`/`success`. `dense_reward=true`
+  uses `len(passes)/num_tests`; else binary all-pass (Task Goal Completion). Returned as the
+  env step reward.
+- **Completion**: the agent calls `apis.supervisor.complete_task()`; the env detects it via
+  `/task_completed`. On hitting `max_steps` the env evaluates the partial state.
+- **Isolation**: one container per rollout; unique `experiment_name` per rollout; container
+  torn down on close. Model code runs inside the container, away from the trainer.
+- **Pool sizing**: `pool_size` = max concurrent rollouts (one container each), same as swerl.
diff --git a/scripts/general_agent/appworld/debug/probe_container.py b/scripts/general_agent/appworld/debug/probe_container.py
new file mode 100644
index 0000000000..e758ef88ac
--- /dev/null
+++ b/scripts/general_agent/appworld/debug/probe_container.py
@@ -0,0 +1,126 @@
+"""Diagnostic probe: why does the AppWorld container die on Beaker podman, and is the
+container's bridge IP reachable from here?
+
+Runs as a tiny no-training Beaker job. It connects to a podman shard and starts the AppWorld
+server container two ways — WITHOUT and WITH host-port publishing — and for each prints the
+container status, exit code, error, log tail, bridge IP, published ports, and HTTP reachability
+(bridge IP and, if published, the host port). This isolates:
+
+  1. Whether host-port publishing is what makes the container exit instantly (the swerl contrast).
+  2. Whether the trainer can reach the container's bridge IP (the open networking question for
+     the stateful-server style).
+
+Launch: ./scripts/train/build_image_and_launch_dirty.sh \
+            scripts/general_agent/appworld/debug/probe_container_beaker.sh
+"""
+
+import os
+import time
+
+import docker
+import requests
+
+IMAGE = os.environ.get("APPWORLD_IMAGE", "shatu/appworld-data:latest")
+PORT = 8000
+
+
+def first_docker_host() -> str | None:
+    hosts = [h.strip() for h in os.environ.get("SWERL_PODMAN_DOCKER_HOSTS", "").split(",") if h.strip()]
+    if hosts:
+        return hosts[0]
+    # Fallback to the conventional shard-0 socket that docker_login.sh creates.
+    sock = "/tmp/podman-services/0/podman.sock"
+    return f"unix://{sock}" if os.path.exists(sock) else None
+
+
+def make_client() -> docker.DockerClient:
+    dh = first_docker_host()
+    print(f"docker_host = {dh!r}  (SWERL_PODMAN_DOCKER_HOSTS={os.environ.get('SWERL_PODMAN_DOCKER_HOSTS')!r})", flush=True)
+    return docker.DockerClient(base_url=dh) if dh else docker.from_env()
+
+
+def run_test(cl: docker.DockerClient, publish: bool, host_network: bool = False, port: int = PORT) -> None:
+    label = "network_mode=host" if host_network else ("WITH host-port publish" if publish else "NO publish (bridge IP only)")
+    print(f"\n================ TEST: {label} (port {port}) ================", flush=True)
+    kwargs = dict(
+        image=IMAGE,
+        command=["environment", "--port", str(port), "--no-show-usage"],
+        environment={"APPWORLD_ROOT": "/run"},
+        detach=True,
+        auto_remove=False,
+        mem_limit="4g",
+        memswap_limit="4g",
+        labels={"appworld_probe": "1"},
+    )
+    if host_network:
+        kwargs["network_mode"] = "host"
+    elif publish:
+        kwargs["ports"] = {f"{port}/tcp": None}
+
+    try:
+        c = cl.containers.run(**kwargs)
+    except Exception as e:
+        print(f"containers.run FAILED to even create/start: {type(e).__name__}: {e}", flush=True)
+        return
+
+    try:
+        # Watch the container for up to 30s; note if/when it stops running.
+        last_status = None
+        for _ in range(30):
+            c.reload()
+            if c.status != last_status:
+                print(f"  t status -> {c.status}", flush=True)
+                last_status = c.status
+            if c.status != "running":
+                break
+            time.sleep(1)
+        c.reload()
+        state = c.attrs.get("State", {}) or {}
+        print(f"final status: {c.status} | exit_code: {state.get('ExitCode')} | error: {state.get('Error')!r}", flush=True)
+        print("--- container logs (tail 60) ---", flush=True)
+        print(c.logs(tail=60).decode("utf-8", "replace"), flush=True)
+
+        net = c.attrs.get("NetworkSettings", {}) or {}
+        ip = net.get("IPAddress") or next(
+            (n.get("IPAddress") for n in (net.get("Networks") or {}).values() if n.get("IPAddress")), None
+        )
+        ports = net.get("Ports")
+        print(f"bridge IP: {ip} | published ports: {ports}", flush=True)
+        print(f"full NetworkSettings: {net}", flush=True)
+
+        urls = [f"http://127.0.0.1:{port}/"]  # works if host_network (shared netns)
+        if ip:
+            urls.append(f"http://{ip}:{port}/")
+        if ports and ports.get(f"{port}/tcp"):
+            urls.append(f"http://127.0.0.1:{ports[f'{port}/tcp'][0]['HostPort']}/")
+        for url in urls:
+            try:
+                r = requests.get(url, timeout=5)
+                print(f"  GET {url} -> {r.status_code}  {r.text[:80]!r}", flush=True)
+            except Exception as e:
+                print(f"  GET {url} -> ERROR {type(e).__name__}: {e}", flush=True)
+    finally:
+        try:
+            c.remove(force=True)
+        except Exception:
+            pass
+
+
+def main() -> None:
+    cl = make_client()
+    try:
+        cl.images.get(IMAGE)
+        print(f"image {IMAGE} already present", flush=True)
+    except Exception:
+        print(f"pulling {IMAGE} (through the mirror) ...", flush=True)
+        cl.images.pull(IMAGE)
+        print("pull done", flush=True)
+
+    run_test(cl, publish=False)
+    run_test(cl, publish=True)
+    run_test(cl, publish=False, host_network=True, port=8137)  # candidate fix: reach via 127.0.0.1
+    print("\nPROBE DONE", flush=True)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/general_agent/appworld/debug/probe_container_beaker.sh b/scripts/general_agent/appworld/debug/probe_container_beaker.sh
new file mode 100644
index 0000000000..88b4ca56a9
--- /dev/null
+++ b/scripts/general_agent/appworld/debug/probe_container_beaker.sh
@@ -0,0 +1,33 @@
+#!/bin/bash
+
+# Cheap diagnostic (no training): start the AppWorld container on a Beaker podman host and
+# report why it lives/dies + bridge-IP reachability. 1 node, 1 GPU (GPU unused; minimal for
+# fast scheduling). See scripts/general_agent/appworld/debug/probe_container.py.
+#
+# Launch:
+#   ./scripts/train/build_image_and_launch_dirty.sh scripts/general_agent/appworld/debug/probe_container_beaker.sh
+
+BEAKER_IMAGE="${1:?Usage: $0 <beaker-image>}"
+
+uv run python mason.py \
+       --cluster ai2/jupiter \
+       --image "$BEAKER_IMAGE" \
+       --description "AppWorld container probe (podman start + reachability)" \
+       --pure_docker_mode \
+       --workspace ai2/general-tool-use \
+       --priority urgent \
+       --preemptible \
+       --num_nodes 1 \
+       --max_retries 1 \
+       --env BEAKER_ALLOW_SUBCONTAINERS=1 \
+       --env BEAKER_SKIP_DOCKER_SOCKET=1 \
+       --env SWERL_PODMAN_SERVICE_COUNT=1 \
+       --env MIRROR_URL=jupiter-cs-aus-193.reviz.ai2.in:5000 \
+       --env PODMAN_NUM_LOCKS=65536 \
+       --env CONTAINERS_STORAGE_CONF=/etc/containers/storage.conf \
+       --env DOCKERHUB_USERNAME=shashankg209 \
+       --env APPWORLD_IMAGE=shatu/appworld-data:latest \
+       --secret DOCKER_PAT=shashankg_DOCKER_PAT \
+       --gpus 1 \
+       --no_auto_dataset_cache \
+       -- source scripts/docker/docker_login.sh \&\& python scripts/general_agent/appworld/debug/probe_container.py
diff --git a/scripts/general_agent/appworld/rl/local_rl_2gpu.sh b/scripts/general_agent/appworld/rl/local_rl_2gpu.sh
new file mode 100644
index 0000000000..ba6fd241ee
--- /dev/null
+++ b/scripts/general_agent/appworld/rl/local_rl_2gpu.sh
@@ -0,0 +1,68 @@
+#!/bin/bash
+
+# Local 2-GPU debug run for AppWorld RL (GRPO + appworld env).
+# Layout: 1 learner GPU + 1 vLLM engine GPU. Qwen3-0.6B + small dataset slice.
+# Docker must be running locally; each rollout starts an AppWorld container.
+# No Podman services, no Beaker, no mason.py, no HuggingFace.
+#
+# Prereqs (one-time):
+#   1. AppWorld image reachable by the local docker daemon. Either
+#        docker pull ghcr.io/stonybrooknlp/appworld:latest   (+ a daemon-visible data root), or
+#        build a data-baked image (no bind mount needed):
+#          FROM ghcr.io/stonybrooknlp/appworld:latest
+#          COPY data /run/data && RUN mkdir -p /run/experiments/outputs
+#        docker build -t appworld-data:test <appworld_root>
+#   2. A local RL parquet built by scripts/data/convert_appworld_to_rl.py:
+#        uv run python scripts/data/convert_appworld_to_rl.py \
+#          --data_root <appworld_root> --splits train --limit 16 --max_steps 10 \
+#          --output_parquet $APPWORLD_DEBUG_PARQUET
+
+export VLLM_ALLOW_INSECURE_SERIALIZATION=1
+export VLLM_DISABLE_COMPILE_CACHE=1
+export VLLM_USE_V1=1
+
+# Data-baked image (data_root="" => no bind mount) and local debug dataset.
+APPWORLD_IMAGE="${APPWORLD_IMAGE:-appworld-data:test}"
+APPWORLD_DEBUG_PARQUET="${APPWORLD_DEBUG_PARQUET:-/tmp/aw_probe/appworld_debug16.parquet}"
+
+uv run python open_instruct/grpo_fast.py \
+    --exp_name appworld_local_rl_2gpu \
+    --model_name_or_path Qwen/Qwen3-0.6B \
+    --dataset_mixer_list "${APPWORLD_DEBUG_PARQUET}" 16 \
+    --dataset_mixer_list_splits train \
+    --max_prompt_token_length 2048 \
+    --response_length 4096 \
+    --pack_length 8192 \
+    --per_device_train_batch_size 1 \
+    --num_unique_prompts_rollout 2 \
+    --num_samples_per_prompt_rollout 2 \
+    --learning_rate 1e-6 \
+    --lr_scheduler_type constant \
+    --total_episodes 16 \
+    --deepspeed_stage 2 \
+    --num_learners_per_node 1 \
+    --vllm_num_engines 1 \
+    --vllm_tensor_parallel_size 1 \
+    --vllm_gpu_memory_utilization 0.6 \
+    --vllm_enforce_eager \
+    --beta 0.0 \
+    --use_vllm_logprobs true \
+    --truncated_importance_sampling_ratio_cap 0.0 \
+    --advantage_normalization_type centered \
+    --verification_reward 1.0 \
+    --temperature 1.0 \
+    --tools appworld \
+    --tool_call_names execute_python \
+    --tool_configs "{\"image\": \"${APPWORLD_IMAGE}\", \"data_root\": \"\", \"max_interactions\": 10}" \
+    --tool_parser_type vllm_qwen3_xml \
+    --pool_size 4 \
+    --max_steps 10 \
+    --backend_timeout 600 \
+    --gradient_checkpointing \
+    --save_traces \
+    --local_eval_every 8 \
+    --logging_steps 1 \
+    --seed 42 \
+    --report_to null \
+    --output_dir output/appworld_rl_local_2gpu \
+    --push_to_hub false
diff --git a/scripts/general_agent/appworld/rl/qwen35_4b_appworld.sh b/scripts/general_agent/appworld/rl/qwen35_4b_appworld.sh
new file mode 100644
index 0000000000..06b2ca0d5b
--- /dev/null
+++ b/scripts/general_agent/appworld/rl/qwen35_4b_appworld.sh
@@ -0,0 +1,113 @@
+#!/bin/bash
+
+# AppWorld code-as-action RL with Qwen3.5-4B (GRPO + appworld env).
+#
+# Modeled on scripts/general_agent/terminal/rl/qwen35_4b_base_tmax_10k.sh (the swerl
+# podman-services recipe) — AppWorld reuses the same per-rollout container infra, but
+# each rollout runs an AppWorld server container and the env talks to it over HTTP.
+#
+# Reward is AppWorld's own /evaluate (fraction of unit tests passed), emitted by the env;
+# the dataset uses the `passthrough` verifier so no extra reward is added. The prompt
+# (incl. supervisor creds) is baked into the dataset messages, so NO system_prompt_override.
+#
+# Data: the AppWorld task data is baked into the container image (shatu/appworld-data:latest,
+# data_root="" => no bind mount). The image is pulled through the cluster registry mirror.
+#
+# Launch (dirty tree is fine via the _dirty variant):
+#   ./scripts/train/build_image_and_launch_dirty.sh scripts/general_agent/appworld/rl/qwen35_4b_appworld.sh
+
+BEAKER_IMAGE="${1:?Usage: $0 <beaker-image>}"
+MODEL=Qwen/Qwen3.5-4B
+TOKENIZER=Qwen/Qwen3.5-4B
+DATASET=shatu/appworld-rl
+APPWORLD_IMAGE=shatu/appworld-data:latest
+# Shakeout config (smaller/shorter/less-concurrent than a full run). The first run wedged on a
+# weight-sync hang after 1 step, with acquire_reset_pools at 150-260s/rollout (pool contention +
+# heavy AppWorld /initialize) and many rollouts truncating at 24k tokens. So: fewer concurrent
+# containers (8x8=64), pool headroom (96), shorter rollouts (20 turns / 16k tokens / 4k per turn),
+# no active_sampling (bounds concurrency to the batch), and inflight_updates off (sync weights
+# between steps, not during generation) to avoid the weight-sync/engine deadlock.
+MAX_STEPS=20
+
+uv run python mason.py \
+       --cluster ai2/jupiter \
+       --image "$BEAKER_IMAGE" \
+       --description "AppWorld GRPO with Qwen3.5-4B" \
+       --pure_docker_mode \
+       --workspace ai2/general-tool-use \
+       --priority urgent \
+       --preemptible \
+       --num_nodes 2 \
+       --max_retries 5 \
+       --env REPO_PATH=/stage \
+       --env BEAKER_ALLOW_SUBCONTAINERS=1 \
+       --env BEAKER_SKIP_DOCKER_SOCKET=1 \
+       --env VLLM_ALLOW_INSECURE_SERIALIZATION=1 \
+       --env VLLM_DISABLE_COMPILE_CACHE=1 \
+       --env VLLM_USE_V1=1 \
+       --env GIT_COMMIT="$(git rev-parse --short HEAD)" \
+       --env DOCKERHUB_USERNAME=shashankg209 \
+       --env SWERL_SANDBOX_TIMING_LOGS=1 \
+       --env SWERL_DOCKER_AUTO_REMOVE=1 \
+       --env SWERL_RESET_FAILURE_ZERO_REWARD=1 \
+       --env SWERL_PODMAN_SERVICE_COUNT=8 \
+       --env SWERL_DOCKER_START_CONCURRENCY=128 \
+       --env SWERL_SANDBOX_TIMING_LOG_THRESHOLD_S=1.0 \
+       --env MIRROR_URL=jupiter-cs-aus-193.reviz.ai2.in:5000 \
+       --env PODMAN_NUM_LOCKS=65536 \
+       --env CONTAINERS_STORAGE_CONF=/etc/containers/storage.conf \
+       --secret DOCKER_PAT=shashankg_DOCKER_PAT \
+       --gpus 8 \
+       --no_auto_dataset_cache \
+       -- source scripts/docker/docker_login.sh \&\& source configs/beaker_configs/ray_node_setup.sh  \&\& python open_instruct/grpo_fast.py \
+    --dataset_mixer_list $DATASET 1.0 \
+    --dataset_mixer_list_splits train \
+    --dataset_mixer_eval_list $DATASET 8 \
+    --dataset_mixer_eval_list_splits dev \
+    --max_prompt_token_length 2048 \
+    --per_turn_max_tokens 4096 \
+    --response_length 16384 \
+    --pack_length 18432 \
+    --per_device_train_batch_size 1 \
+    --num_unique_prompts_rollout 8 \
+    --num_samples_per_prompt_rollout 8 \
+    --async_steps 4 \
+    --model_name_or_path $MODEL \
+    --tokenizer_name_or_path $TOKENIZER \
+    --temperature 1.0 \
+    --learning_rate 1e-6 \
+    --total_episodes 64000 \
+    --lr_scheduler_type constant \
+    --deepspeed_stage 3 \
+    --sequence_parallel_size 2 \
+    --num_epochs 1 \
+    --num_learners_per_node 8 \
+    --vllm_num_engines 8 \
+    --vllm_tensor_parallel_size 1 \
+    --beta 0.0 \
+    --use_vllm_logprobs true \
+    --truncated_importance_sampling_ratio_cap 0.0 \
+    --seed 42 \
+    --gradient_checkpointing \
+    --vllm_enable_prefix_caching \
+    --push_to_hub false \
+    --with_tracking \
+    --wandb_project oe-general-agents \
+    --save_traces \
+    --tools appworld \
+    --tool_call_names execute_python \
+    --tool_configs "{\"image\": \"${APPWORLD_IMAGE}\", \"data_root\": \"\", \"max_interactions\": ${MAX_STEPS}, \"startup_timeout\": 600, \"dense_reward\": true}" \
+    --pool_size 96 \
+    --max_steps ${MAX_STEPS} \
+    --verification_reward 1.0 \
+    --tool_parser_type vllm_qwen3_xml \
+    --backend_timeout 1200 \
+    --checkpoint_state_freq 10 \
+    --inflight_updates false \
+    --advantage_normalization_type centered \
+    --rollouts_save_path /weka/oe-adapt-default/allennlp/deletable_rollouts/ \
+    --output_dir /output \
+    --exp_name appworld_qwen35_4b_grpo \
+    --local_eval_every 10 \
+    --save_freq 20 \
+    --try_launch_beaker_eval_jobs_on_weka False
diff --git a/scripts/general_agent/appworld/smoke_test_appworld_env.py b/scripts/general_agent/appworld/smoke_test_appworld_env.py
new file mode 100644
index 0000000000..0483d97999
--- /dev/null
+++ b/scripts/general_agent/appworld/smoke_test_appworld_env.py
@@ -0,0 +1,79 @@
+"""End-to-end smoke test for AppWorldEnv (requires Docker + the AppWorld image + data).
+
+Drives a single AppWorld task through the env the way the rollout loop would:
+reset -> a few execute_python steps -> completion -> reward. This starts a real
+AppWorld container and talks to it over HTTP, so it validates the full container +
+HTTP path without importing the (pydantic-1) appworld package in this process.
+
+Prereqs:
+    docker pull ghcr.io/stonybrooknlp/appworld:latest
+    # AppWorld data root on the host (contains data/ and experiments/outputs/):
+    #   pip install appworld && appworld install && appworld download data --root $APPWORLD_ROOT
+
+Example:
+    uv run python scripts/general_agent/appworld/smoke_test_appworld_env.py --data_root $APPWORLD_ROOT
+"""
+
+import argparse
+import asyncio
+import os
+
+from open_instruct.environments.appworld_env import AppWorldEnv
+from open_instruct.environments.base import EnvCall
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--data_root", required=True, help="Host AppWorld root (has data/ and experiments/outputs/).")
+    parser.add_argument("--split", default="train", help="Split file under data/datasets to pick a task from.")
+    parser.add_argument("--task_id", default="", help="Specific task id; default = first id in the split file.")
+    parser.add_argument("--image", default="ghcr.io/stonybrooknlp/appworld:latest", help="AppWorld docker image.")
+    return parser.parse_args()
+
+
+def _first_task_id(data_root: str, split: str) -> str:
+    split_path = os.path.join(data_root, "data", "datasets", f"{split}.txt")
+    with open(split_path, encoding="utf-8") as fh:
+        for line in fh:
+            tid = line.strip().split(":")[-1]  # strip optional tag prefix
+            if tid:
+                return tid
+    raise RuntimeError(f"No task ids found in {split_path}")
+
+
+def _call(code: str) -> EnvCall:
+    return EnvCall(id="0", name="execute_python", args={"code": code})
+
+
+async def run(args: argparse.Namespace) -> None:
+    os.makedirs(os.path.join(args.data_root, "experiments", "outputs"), exist_ok=True)
+    task_id = args.task_id or _first_task_id(args.data_root, args.split)
+    print(f"Using task_id={task_id}")
+
+    env = AppWorldEnv(image=args.image, data_root=args.data_root, max_interactions=10)
+    await env.setup()
+    try:
+        result, tools = await env.reset(task_id=task_id)
+        print(f"reset -> tools={[t['function']['name'] for t in tools]}, base_url={env._base_url}")
+
+        # Probe the API surface, then complete the task (no real solution; this just
+        # exercises execute -> task_completed -> evaluate -> reward end to end).
+        for code in [
+            "print(apis.api_docs.show_app_descriptions())",
+            "print(apis.supervisor.show_profile())",
+            "apis.supervisor.complete_task()",
+        ]:
+            step = await env.step(_call(code))
+            head = step.result if len(step.result) < 600 else step.result[:600] + " ...[truncated]"
+            print(f"\n>>> {code}\n{head}\n[reward={step.reward} done={step.done}]")
+            if step.done:
+                break
+
+        print(f"\nmetrics={env.get_metrics()}")
+    finally:
+        await env.close()
+    print("Smoke test complete.")
+
+
+if __name__ == "__main__":
+    asyncio.run(run(parse_args()))
diff --git a/tests/test_environments.py b/tests/test_environments.py
index 9bd1d072a6..fd27e1dd52 100644
--- a/tests/test_environments.py
+++ b/tests/test_environments.py
@@ -6,8 +6,17 @@
 
 from parameterized import parameterized
 
+from open_instruct.environments.appworld_env import (
+    AppWorldEnv,
+    AppWorldEnvConfig,
+    build_prompt_messages,
+    format_supervisor_block,
+    reward_from_evaluation,
+    truncate_observation,
+)
 from open_instruct.environments.base import EnvCall, RLEnvironment, StepResult, TextRLEnvironment
 from open_instruct.environments.examples import CounterEnv, GuessNumberEnv, WordleTextEnv
+from open_instruct.environments.tools.tools import TOOL_REGISTRY
 
 
 class TestEnvironmentReset(unittest.TestCase):
@@ -32,10 +41,12 @@ def test_reset_returns_result_and_tools(self, _name, env_cls, kwargs, reset_kwar
 
 
 class TestEnvironments(unittest.TestCase):
-    @parameterized.expand([
-        (CounterEnv, {"target": 3}, {"target": "3"}),
-        (GuessNumberEnv, {"min_val": 1, "max_val": 10}, {"number": "3"}),
-    ])
+    @parameterized.expand(
+        [
+            (CounterEnv, {"target": 3}, {"target": "3"}),
+            (GuessNumberEnv, {"min_val": 1, "max_val": 10}, {"number": "3"}),
+        ]
+    )
     def test_20_random_steps(self, env_cls: type[RLEnvironment], kwargs: dict, reset_kwargs: dict):
         env = env_cls(**kwargs)
         result, tools = asyncio.run(env.reset(**reset_kwargs))
@@ -134,6 +145,79 @@ def test_metrics(self):
         self.assertGreater(m["total_greens"], 0)
 
 
+class TestAppWorldEnv(unittest.TestCase):
+    """Tests for AppWorldEnv that do not require the appworld package or its data."""
+
+    SUPERVISOR = {
+        "first_name": "Jane",
+        "last_name": "Doe",
+        "email": "jane@example.com",
+        "phone_number": "555-0100",
+        "password": "hunter2",
+    }
+
+    def test_registered_in_tool_registry(self):
+        self.assertIn("appworld", TOOL_REGISTRY)
+        self.assertIs(TOOL_REGISTRY["appworld"], AppWorldEnvConfig)
+        self.assertEqual(AppWorldEnvConfig.tool_class.config_name, "appworld")
+
+    def test_tool_definition_is_execute_python(self):
+        tools = AppWorldEnv.get_tool_definitions()
+        self.assertEqual(len(tools), 1)
+        self.assertEqual(tools[0]["function"]["name"], "execute_python")
+        self.assertIn("code", tools[0]["function"]["parameters"]["properties"])
+
+    def test_build_prompt_messages_bakes_instruction_and_creds(self):
+        msgs = build_prompt_messages("Pay rent via Venmo.", self.SUPERVISOR)
+        self.assertEqual([m["role"] for m in msgs], ["system", "user"])
+        # Supervisor login details belong in the system prompt (reset obs is dropped).
+        self.assertIn("jane@example.com", msgs[0]["content"])
+        self.assertIn("hunter2", msgs[0]["content"])
+        self.assertIn("Pay rent via Venmo.", msgs[1]["content"])
+
+    def test_format_supervisor_block_from_object(self):
+        class Sup:
+            first_name = "Bob"
+            email = "bob@x.com"
+
+        block = format_supervisor_block(Sup())
+        self.assertIn("Bob", block)
+        self.assertIn("bob@x.com", block)
+
+    def test_truncate_observation(self):
+        short = "ok"
+        self.assertEqual(truncate_observation(short), short)
+        long = "A" * 20000
+        out = truncate_observation(long)
+        self.assertLess(len(out), len(long))
+        self.assertIn("chars elided", out)
+
+    def test_reset_requires_task_id(self):
+        env = AppWorldEnv()
+        env._client = object()  # skip setup()/docker client creation
+        with self.assertRaises(ValueError):
+            asyncio.run(env.reset(task_id=None))
+
+    def test_config_defaults(self):
+        cfg = AppWorldEnvConfig()
+        self.assertEqual(cfg.image, "ghcr.io/stonybrooknlp/appworld:latest")
+        self.assertEqual(cfg.in_container_root, "/run")
+        self.assertTrue(cfg.dense_reward)
+        self.assertEqual(cfg.max_interactions, 50)
+
+    def test_reward_from_evaluation(self):
+        ev = {"num_tests": 4, "passes": [1, 2, 3], "failures": [4], "success": False}
+        dense, m = reward_from_evaluation(ev, dense=True)
+        self.assertAlmostEqual(dense, 0.75)
+        self.assertEqual(m["pass_count"], 3.0)
+        binary, _ = reward_from_evaluation(ev, dense=False)
+        self.assertEqual(binary, 0.0)
+        all_pass = {"num_tests": 2, "passes": [1, 2], "failures": [], "success": True}
+        self.assertEqual(reward_from_evaluation(all_pass, dense=False)[0], 1.0)
+        # no tests -> zero, no division error
+        self.assertEqual(reward_from_evaluation({"num_tests": 0, "passes": []}, dense=True)[0], 0.0)
+
+
 def _random_env_call(tools: list[dict]) -> EnvCall:
     """Pick a random tool and fill required args with random values."""
     tool = random.choice(tools)