> ## Documentation Index
> Fetch the complete documentation index at: https://docs.archil.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Evals and RL environments

> Fork a task environment once per rollout, grade each agent run with a hidden grader, and collect the graded trajectories as a JSONL dataset on an Archil disk.

Evals and reinforcement learning for agents run the same loop: put an agent in an environment, let it act, and grade the result. To measure a model or train one, you run that loop many times per task and keep every trajectory with its reward. That takes an environment that starts from the same state every time, a grader the agent can't tamper with, and a place to collect the results.

This guide builds all three on Archil:

* **Environment:** a stopped sandbox with a small task suite installed, kept as a template. Each rollout forks it, so every agent starts on an identical, isolated machine. The fork is the environment's `reset()`.
* **Grader:** hidden checks that enter the fork only after the agent finishes, and turn the result into a reward.
* **Dataset:** one JSONL record per rollout, appended to an Archil disk, then scored where it lives with a [serverless command](/compute/serverless-execution).

[Parallel agents](/guides/agents/parallel-agents) races forks and keeps the best one. Here you keep every rollout, pass or fail: the output is a graded trajectory dataset for pass\@k reports, rejection sampling, or RL fine-tuning.

## Prerequisites

Export your Archil API key, a region where sandboxes are available (`aws-us-east-1`, `aws-us-west-2`, or `aws-eu-west-1`), and an [AI Gateway](https://vercel.com/ai-gateway) key for the model:

```bash theme={null}
export ARCHIL_API_KEY=key-...
export ARCHIL_REGION=aws-us-east-1
export AI_GATEWAY_API_KEY=...
```

Install the dependencies. The TypeScript agent uses the [AI SDK](https://ai-sdk.dev); the Python agent calls the same model through the gateway's OpenAI-compatible API.

<CodeGroup>
  ```bash TypeScript theme={null}
  npm init -y && npm pkg set type=module
  npm install disk ai zod tsx
  ```

  ```bash Python theme={null}
  pip install archil openai
  ```
</CodeGroup>

## Define the tasks

The suite has three small, deterministic tasks: implement an expression evaluator from a spec, write a report from a messy CSV, and fix a buggy parser. This script runs once inside the template and creates a directory per task under `/work`, each with a `TASK.md` for the agent and the starting files.

```bash setup.sh theme={null}
set -euo pipefail
mkdir -p /work/calc /work/sales-report /work/duration

cd /work/calc
cat > TASK.md <<'EOF'
Implement evaluate(expr) in calc.py. It takes an arithmetic expression with numbers,
+ - * / // % **, and parentheses, and returns the value Python would compute for it, with
Python's precedence and associativity. Don't use eval, exec, compile, or the ast module.
EOF
printf 'def evaluate(expr):\n    raise NotImplementedError\n' > calc.py

cd /work/sales-report
cat > TASK.md <<'EOF'
Write report.py that reads sales.csv and prints one `region,total` line per region, where
total is the sum of quantity * unit_price formatted with 2 decimals, sorted by total descending.
Region names are case-insensitive; print them in lowercase. Skip rows with no quantity.
EOF
cat > sales.csv <<'EOF'
region,quantity,unit_price
West,3,$19.99
east,10,$4.50
 West ,1,"$1,200.00"
North,,$15.00
East,2,$120.00
north,4,$15.00
EOF

cd /work/duration
cat > TASK.md <<'EOF'
Fix parse_duration in duration.py. It takes strings like "1h30m" and returns seconds.
Units are d, h, m, s, in that order, each at most once, and at least one is required.
Raise ValueError for anything else, including whitespace.
EOF
cat > duration.py <<'EOF'
import re

UNITS = {"d": 86400, "h": 3600, "m": 60, "s": 1}


def parse_duration(text):
    total = 0
    for amount, unit in re.findall(r"(\d+)([dhms])", text):
        total += int(amount) * UNITS[unit]
    return total
EOF
```

The grader stays on your machine, so the agent can't read the checks or edit them to earn a reward it didn't. Like most real evals, it tests cases the spec implies but doesn't list: Python number literals such as `1e3`, and a CSV the agent never saw. It prints how many checks passed for one task.

```python grade.py theme={null}
import builtins
import importlib
import json
import subprocess
import sys

CALC = {"1 + 2 * 3": 7, "(1 + 2) * 3": 9, "2 ** 3 ** 2": 512, "-2 ** 2": -4, "2 ** -1": 0.5, "7 // -2": -4,
        "-7 % 3": 2, "10 / 4": 2.5, "2 * -3": -6, "--3": 3, ".5 + 1": 1.5, "1e3 / 8": 125.0, "1_000 // 7": 142}
DURATIONS = {"1h30m": 5400, "45s": 45, "2d": 172800, "1d2h3m4s": 93784}
BAD_DURATIONS = ["", "1x", "30m1h", "1h 30m", "h", "1h1h", "1H", "1.5h", "1h30", "-5m"]
HIDDEN_CSV = 'region,quantity,unit_price\nSouth,2,"$1,000.50"\nsouth ,1,$0.99\nEast,,$5.00\nEast,3,$3.33\n'


def load(module):
    try:
        return importlib.import_module(module)
    except Exception:
        return None


def ok(check, raises=None):
    """True if check() returns True, or raises the expected exception."""
    try:
        return check() is True and raises is None
    except Exception as err:
        return raises is not None and isinstance(err, raises)


def report(csv):
    with open("sales.csv", "w") as f:
        f.write(csv)
    return subprocess.run([sys.executable, "report.py"], capture_output=True, text=True, timeout=30).stdout.split()


def forbidden(*args, **kwargs):
    raise RuntimeError("not allowed")


task = sys.argv[1]
sys.path.insert(0, ".")
if task == "calc":
    calc = load("calc")
    builtins.eval = builtins.exec = builtins.compile = forbidden  # also blocks ast.parse
    results = [ok(lambda: calc.evaluate(expr) == value) for expr, value in CALC.items()]
elif task == "duration":
    duration = load("duration")
    results = [ok(lambda: duration.parse_duration(text) == secs) for text, secs in DURATIONS.items()]
    results += [ok(lambda: duration.parse_duration(text), raises=ValueError) for text in BAD_DURATIONS]
else:
    original = open("sales.csv").read()
    results = [ok(lambda: report(original) == ["west,1259.97", "east,285.00", "north,60.00"]),
               ok(lambda: report(HIDDEN_CSV) == ["south,2001.99", "east,9.99"])]
print(json.dumps({"passed": sum(results), "total": len(results)}))
```

## Build the environment template

Create the disk for the trajectories, then a sandbox that runs `setup.sh`. The tasks live on the sandbox's root filesystem, which every fork copies, and the template doesn't mount the disk: the orchestrator writes to it through the API. Once the tasks are in place, the template is stopped, so it doesn't use compute while it waits.

<CodeGroup>
  ```typescript prepare.ts theme={null}
  import { readFile } from "node:fs/promises";
  import * as archil from "disk";

  await archil.createDisk({ name: "rl-trajectories" });

  const template = await archil.createSandbox({
    name: "rl-env-template",
    baseImage: "python:3.13",
    vcpuCount: 2,
    memSizeMiB: 2048,
  });
  await template.files.uploadFile(await readFile("setup.sh"), "/root/setup.sh");
  const setup = await template.exec("bash /root/setup.sh");
  if (setup.exitCode !== 0) throw new Error(setup.stderr);
  await template.stop();
  ```

  ```python prepare.py theme={null}
  import archil

  archil.create_disk(name="rl-trajectories")

  template = archil.create_sandbox(
      name="rl-env-template",
      base_image="python:3.13",
      vcpu_count=2,
      mem_size_mib=2048,
  )
  template.files.upload_file("setup.sh", "/root/setup.sh")
  setup = template.exec("bash /root/setup.sh")
  if setup.exit_code != 0:
      raise RuntimeError(setup.stderr)
  template.stop()
  ```
</CodeGroup>

## Run the rollouts

The rollout script runs each task twice, at most four rollouts at a time. Each rollout:

1. Forks the template. A fork of a stopped sandbox comes up running, with a copy of its root filesystem, in about a second.
2. Runs an agent with a `bash` tool bound to its fork, capped at 10 steps.
3. Uploads `grade.py`, runs it, and uses the fraction of checks passed as the reward.
4. Appends one JSONL record to `trajectories/<run>.jsonl` on the disk: task, sample, model, reward, steps, token usage, timings, and every message and tool call, with tool output truncated to 2,000 characters.
5. Deletes the fork, in a `finally` block so a failed rollout doesn't leave a sandbox behind.

Concurrent appends to an existing file are safe, but two first appends that both create the file conflict, so the script creates the file before the rollouts start.

<CodeGroup>
  ```typescript rollouts.ts theme={null}
  import { readFile } from "node:fs/promises";
  import { ToolLoopAgent, isStepCount, tool } from "ai";
  import * as archil from "disk";
  import type { Sandbox } from "disk";
  import { z } from "zod";

  const TASKS = ["calc", "sales-report", "duration"];
  const SAMPLES = 2; // rollouts per task
  const CONCURRENCY = 4;
  const MODEL = "anthropic/claude-haiku-5.5";

  const template = (await archil.listSandboxes()).find((s) => s.name === "rl-env-template")!;
  const [disk] = await archil.listDisks({ name: "rl-trajectories" });
  const grader = await readFile("grade.py");
  const run = `run-${Date.now()}`;
  const dataset = `trajectories/${run}.jsonl`;
  await disk.putObject(dataset, "");

  function bashTool(env: Sandbox, cwd: string) {
    return tool({
      description: "Run a bash command. Returns the exit code and output.",
      inputSchema: z.object({ command: z.string() }),
      execute: async ({ command }) => {
        const r = await env.exec('bash -c "$COMMAND"', { cwd, env: { COMMAND: command }, timeoutSeconds: 60 });
        return `exit ${r.exitCode ?? r.status}\n${r.stdout}${r.stderr}`.slice(-2000);
      },
    });
  }

  async function rollout(task: string, sample: number) {
    const t0 = Date.now();
    const env = await template.fork({ name: `rl-rollout-${task}-${sample}` });
    try {
      const t1 = Date.now();
      const cwd = `/work/${task}`;
      const prompt =
        `You are in ${cwd}. Read TASK.md and complete the task. Hidden tests will grade your files, ` +
        "so test your work. Finish with a one-sentence summary.";
      const agent = new ToolLoopAgent({ model: MODEL, tools: { bash: bashTool(env, cwd) }, stopWhen: isStepCount(10) });
      const result = await agent.generate({ prompt });
      const t2 = Date.now();

      await env.files.uploadFile(grader, "/grader/grade.py");
      const { passed, total } = JSON.parse((await env.exec(`python3 /grader/grade.py ${task}`, { cwd })).stdout);
      const record = {
        run, task, sample, model: MODEL, reward: passed / total, passed, total, steps: result.steps.length,
        usage: { input: result.totalUsage.inputTokens, output: result.totalUsage.outputTokens },
        fork_ms: t1 - t0, agent_ms: t2 - t1,
        messages: [{ role: "user", content: prompt }, ...result.responseMessages],
      };
      await disk.appendObject(dataset, `${JSON.stringify(record)}\n`);
      console.log(`${task} #${sample}: reward ${record.reward.toFixed(2)} (${passed}/${total}) in ${record.steps} steps`);
    } finally {
      await env.stop();
      await env.delete();
    }
  }

  const jobs = TASKS.flatMap((task) => Array.from({ length: SAMPLES }, (_, sample) => ({ task, sample })));
  const count = jobs.length;
  const started = Date.now();
  await Promise.all(
    Array.from({ length: CONCURRENCY }, async () => {
      for (let job = jobs.shift(); job; job = jobs.shift()) {
        const { task, sample } = job;
        await rollout(task, sample).catch((err) => console.error(`${task} #${sample} failed: ${err}`));
      }
    }),
  );
  console.log(`${count} rollouts in ${Math.round((Date.now() - started) / 1000)}s -> ${dataset}`);
  ```

  ```python rollouts.py theme={null}
  import json
  import os
  import time
  from concurrent.futures import ThreadPoolExecutor

  import archil
  from openai import OpenAI

  TASKS = ["calc", "sales-report", "duration"]
  SAMPLES = 2  # rollouts per task
  CONCURRENCY = 4
  MODEL = "anthropic/claude-haiku-5.5"
  MAX_STEPS = 10
  PROMPT = (
      "You are in {cwd}. Read TASK.md and complete the task. Hidden tests will grade your files, "
      "so test your work. Finish with a one-sentence summary."
  )
  BASH_TOOL = {"type": "function", "function": {
      "name": "bash", "description": "Run a bash command. Returns the exit code and output.",
      "parameters": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]},
  }}

  client = OpenAI(base_url="https://ai-gateway.vercel.sh/v1", api_key=os.environ["AI_GATEWAY_API_KEY"])
  template = next(s for s in archil.list_sandboxes() if s.name == "rl-env-template")
  disk = archil.list_disks(name="rl-trajectories")[0]
  run = f"run-{int(time.time() * 1000)}"
  dataset = f"trajectories/{run}.jsonl"
  disk.put_object(dataset, b"")


  def bash(env, cwd, command):
      r = env.exec('bash -c "$COMMAND"', cwd=cwd, env={"COMMAND": command}, timeout_seconds=60)
      status = r.status if r.exit_code is None else r.exit_code
      return f"exit {status}\n{r.stdout}{r.stderr}"[-2000:]


  def run_agent(env, cwd):
      messages = [{"role": "user", "content": PROMPT.format(cwd=cwd)}]
      usage = {"input": 0, "output": 0}
      for step in range(1, MAX_STEPS + 1):
          response = client.chat.completions.create(model=MODEL, messages=messages, tools=[BASH_TOOL])
          usage["input"] += response.usage.prompt_tokens
          usage["output"] += response.usage.completion_tokens
          reply = response.choices[0].message
          messages.append(reply.model_dump(exclude_none=True))
          if not reply.tool_calls:
              break
          for call in reply.tool_calls:
              command = json.loads(call.function.arguments)["command"]
              messages.append({"role": "tool", "tool_call_id": call.id, "content": bash(env, cwd, command)})
      return messages, step, usage


  def rollout(job):
      task, sample = job
      t0 = time.time()
      env = template.fork(name=f"rl-rollout-{task}-{sample}")
      try:
          t1 = time.time()
          cwd = f"/work/{task}"
          messages, steps, usage = run_agent(env, cwd)
          t2 = time.time()

          env.files.upload_file("grade.py", "/grader/grade.py")
          graded = json.loads(env.exec(f"python3 /grader/grade.py {task}", cwd=cwd).stdout)
          passed, total = graded["passed"], graded["total"]
          record = {
              "run": run, "task": task, "sample": sample, "model": MODEL, "reward": passed / total,
              "passed": passed, "total": total, "steps": steps, "usage": usage,
              "fork_ms": round((t1 - t0) * 1000), "agent_ms": round((t2 - t1) * 1000),
              "messages": messages,
          }
          disk.append_object(dataset, (json.dumps(record) + "\n").encode())
          print(f"{task} #{sample}: reward {record['reward']:.2f} ({passed}/{total}) in {steps} steps")
      except Exception as err:
          print(f"{task} #{sample} failed: {err}")
      finally:
          env.stop()
          env.delete()


  jobs = [(task, sample) for task in TASKS for sample in range(SAMPLES)]
  started = time.time()
  with ThreadPoolExecutor(max_workers=CONCURRENCY) as pool:
      list(pool.map(rollout, jobs))
  print(f"{len(jobs)} rollouts in {round(time.time() - started)}s -> {dataset}")
  ```
</CodeGroup>

<CodeGroup>
  ```bash TypeScript theme={null}
  npx tsx prepare.ts && npx tsx rollouts.ts
  ```

  ```bash Python theme={null}
  python prepare.py && python rollouts.py
  ```
</CodeGroup>

Rollouts finish in any order. Most of the run time is the agents working; each `fork_ms` in the dataset was around a second. One run looked like this:

```text theme={null}
sales-report #1: reward 1.00 (2/2) in 4 steps
sales-report #0: reward 1.00 (2/2) in 4 steps
duration #0: reward 1.00 (14/14) in 4 steps
duration #1: reward 1.00 (14/14) in 5 steps
calc #0: reward 0.92 (12/13) in 6 steps
calc #1: reward 0.92 (12/13) in 7 steps
6 rollouts in 151s -> trajectories/run-1792249424509.jsonl
```

## Score the run

The dataset is plain JSONL on the disk, so you can score it without downloading it. This [jq](https://jqlang.org) program groups the records by task and computes the number solved (every check passed), pass\@1, pass\@2, and the mean reward. pass\@2 uses the unbiased estimator, so it needs at least two samples per task and stays correct when you collect more.

```jq score.jq theme={null}
def r: . * 100 | round / 100;

group_by(.task)
| map({task: .[0].task, n: length, c: (map(select(.reward == 1)) | length), reward: (map(.reward) | add / length)})
| map(. + {pass1: (.c / .n), pass2: (1 - ((.n - .c) * (.n - .c - 1)) / (.n * (.n - 1)))})
| (["task", "rollouts", "solved", "pass@1", "pass@2", "mean_reward"] | @tsv),
  (.[] | [.task, .n, .c, (.pass1 | r), (.pass2 | r), (.reward | r)] | @tsv),
  (["overall", (map(.n) | add), (map(.c) | add), (map(.pass1) | add / length | r),
    (map(.pass2) | add / length | r), (map(.reward) | add / length | r)] | @tsv)
```

The score script uploads the program next to the data and runs it with `disk.exec`, in a short-lived VM with the disk mounted. It scores every run on the disk, so samples from later runs add to the totals.

<CodeGroup>
  ```typescript score.ts theme={null}
  import { readFile } from "node:fs/promises";
  import * as archil from "disk";

  const [disk] = await archil.listDisks({ name: "rl-trajectories" });
  await disk.putObject("score.jq", await readFile("score.jq"));
  const { stdout } = await disk.exec("jq -rsf score.jq trajectories/*.jsonl | column -t");
  console.log(stdout);
  ```

  ```python score.py theme={null}
  import archil

  disk = archil.list_disks(name="rl-trajectories")[0]
  with open("score.jq", "rb") as f:
      disk.put_object("score.jq", f.read())
  print(disk.exec("jq -rsf score.jq trajectories/*.jsonl | column -t").stdout)
  ```
</CodeGroup>

```text theme={null}
task          rollouts  solved  pass@1  pass@2  mean_reward
calc          2         0       0       0       0.92
duration      2         2       1       1       1
sales-report  2         2       1       1       1
overall       6         4       0.67    0.67    0.97
```

Results vary from run to run. In this one, every `duration` and `sales-report` rollout passed all its checks, including the CSV the agent never saw. Both `calc` rollouts passed 12 of 13 checks: they handled precedence and `.5` but rejected `1_000`, a number Python accepts. pass\@1 counts those as failures, while the mean reward shows they were close. That partial credit is what lets RL tell a near miss from a wrong answer, and the trajectories show exactly where each agent went wrong.

## Scale it up

* **More samples:** raise `SAMPLES` and `CONCURRENCY`. Each fork is an independent machine, so rollouts don't share state no matter how many run at once.
* **More tasks:** add directories to `setup.sh` and checks to `grade.py`. Build a new template, such as `rl-env-template-v2`, instead of changing the old one, so earlier results stay reproducible.
* **Use the data:** the records are ordinary JSONL. Filter for `reward == 1` to build a fine-tuning set, compare models by grouping on `.model` as well as `.task`, or read the files from any S3 client through the [S3-compatible API](/protocols/s3-api).

## Clean up

Rollout forks delete themselves, so the template and the disk are what's left. The script also deletes forks left behind by an interrupted run; they have to go before the template they came from.

<CodeGroup>
  ```typescript cleanup.ts theme={null}
  import * as archil from "disk";

  const sandboxes = (await archil.listSandboxes()).filter((s) => s.name.startsWith("rl-rollout-"));
  sandboxes.push(...(await archil.listSandboxes()).filter((s) => s.name === "rl-env-template"));
  for (const sandbox of sandboxes) {
    await sandbox.stop();
    await sandbox.delete();
  }

  const [disk] = await archil.listDisks({ name: "rl-trajectories" });
  await disk.delete();
  ```

  ```python cleanup.py theme={null}
  import archil

  sandboxes = [s for s in archil.list_sandboxes() if s.name.startswith("rl-rollout-")]
  sandboxes += [s for s in archil.list_sandboxes() if s.name == "rl-env-template"]
  for sandbox in sandboxes:
      sandbox.stop()
      sandbox.delete()

  archil.list_disks(name="rl-trajectories")[0].delete()
  ```
</CodeGroup>

## Next steps

<CardGroup cols={2}>
  <Card title="Parallel agents" icon="code-fork" href="/guides/agents/parallel-agents">
    Race several agents on forks of one sandbox and keep the best result.
  </Card>

  <Card title="Persistent sandboxes" icon="box" href="/compute/sandboxes/introduction">
    Lifecycle, services, network access, and forking in detail.
  </Card>

  <Card title="Serverless execution" icon="terminal" href="/compute/serverless-execution">
    Run commands directly on a disk without managing a sandbox.
  </Card>

  <Card title="S3-compatible API" icon="bucket" href="/protocols/s3-api#appending">
    Append to objects and read them from any S3 client.
  </Card>
</CardGroup>


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.