Evals and RL environments

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.

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 key for the model:

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; the Python agent calls the same model through the gateway’s OpenAI-compatible API.

npm init -y && npm pkg set type=module
npm install disk ai zod tsx

pip install archil openai

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.

setup.sh

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.

grade.py

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.

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

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

Run the rollouts

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

Forks the template. A fork of a stopped sandbox comes up running, with a copy of its root filesystem, in about a second.
Runs an agent with a bash tool bound to its fork, capped at 10 steps.
Uploads grade.py, runs it, and uses the fraction of checks passed as the reward.
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.
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.

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}`);

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}")

npx tsx prepare.ts && npx tsx rollouts.ts

python prepare.py && python rollouts.py

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:

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 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.

score.jq

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.

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

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)

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.

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.

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

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

Next steps

Parallel agents

Race several agents on forks of one sandbox and keep the best result.

Persistent sandboxes

Lifecycle, services, network access, and forking in detail.

Serverless execution

Run commands directly on a disk without managing a sandbox.

S3-compatible API

Append to objects and read them from any S3 client.

Agents

Analytics

Training

Apps & databases

Use Archil on other platforms

Evals and RL environments

Prerequisites

Define the tasks

Build the environment template

Run the rollouts

Score the run

Scale it up

Clean up

Next steps

Parallel agents

Persistent sandboxes

Serverless execution

S3-compatible API

​Prerequisites

​Define the tasks

​Build the environment template

​Run the rollouts

​Score the run

​Scale it up

​Clean up

​Next steps

Parallel agents

Persistent sandboxes

Serverless execution

S3-compatible API

Prerequisites

Define the tasks

Build the environment template

Run the rollouts

Score the run

Scale it up

Clean up

Next steps