Checking appworld specific changes compared to agents base

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

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,
 }

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:

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()

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.

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()