Evaluate a model on OSWorld tasks

Pick the tasks

client.tasks() returns every task of the host's pinned OSWorld checkout with its domain and whether it is supported on this host. Start from the supported ones, and read a task's JSON (client.task(id)) before relying on it: tasks with "evaluator": {"func": "infeasible"} have no solution, and some tasks download files during setup.

The harness

The harness below runs one episode per task, writes one JSON line per attempt, and reports success over valid episodes with its denominators. policy() is a stand-in that moves the mouse; replace it with a call to your model (the screenshot is obs.png, the instruction obs.meta["instruction"]).

import json
import time
import drums

MAX_STEPS = 5
WORKERS = 2            # desktops at once; your workspace and host limits apply


def policy(obs, step):
    """Stand-in for your model.

    Return pyautogui code (str) or an action dict, or None to stop."""
    if step >= 2:
        return None
    return f"pyautogui.moveTo({200 + 100 * step}, 300)"


def run_task(client, task_id):
    t0 = time.time()
    row = {"task_id": task_id}
    try:
        with client.create(task=task_id, ttl=1800) as env:
            env.wait_valid()
            for step in range(MAX_STEPS):
                action = policy(env.observe(), step)
                if action is None:
                    break
                env.step(action, check=False)
                row["steps"] = step + 1
            r = env.verify(strict=False)
            row.update(status="done", valid=r.valid, score=r.score,
                       invalid_reasons=r.invalid_reasons, checker_error=r.checker_error)
    except drums.DrumsError as e:
        # setup failed, host full, quota: an error, not a model failure
        row.update(status="error", error=f"{type(e).__name__}: {e}")
    row["seconds"] = round(time.time() - t0, 1)
    return row


with drums.connect() as client:
    tasks = [t["id"] for t in client.tasks(domain="os") if t["supported"]][:2]
    rows = client.map(lambda tid: run_task(client, tid), tasks, workers=WORKERS)
    with open("results.jsonl", "w") as f:
        for row in rows:
            f.write(json.dumps(row) + "\n")

done = [r for r in rows if r.get("status") == "done"]
valid = [r for r in done if r["valid"]]
solved = [r for r in valid if (r["score"] or 0) >= 1.0]
print(f"attempted {len(rows)}, errors {len(rows) - len(done)}, "
      f"valid {len(valid)}, solved {len(solved)}")
if valid:
    rate = len(solved) / len(valid)
    print(f"success rate {rate:.2f} over {len(valid)} valid episodes")

Report it honestly

• Three outcomes, not two. An attempt is an error (no episode, or the checker failed), invalid (an episode whose gates failed: its score means nothing), or valid. Only valid attempts have a success or failure. Report all three counts with the success rate, e.g. "12 solved of 30 valid, 32 attempted".
• Do not retry silently. If you retry errors, say so and keep the first attempt's row.
• Same tasks, same settings. Record the task ids, the step limit, your model and sampling settings, and the runtime version (client.health()["version"]) next to the results.
• Screen size. Desktops are 1920 x 1080. If your model answers in another coordinate system (e.g. 0..1000), scale in your code before building the pyautogui call.
• No network inside a desktop. Tasks whose setup or checker needs to download files can fail at setup on this host; the harness records them as errors.

Concurrency

client.map(fn, items, workers=n) runs n episodes at once in threads. For many concurrent episodes use drums.aio (asyncio, with its own thread pool): see drums.aio. Every desktop counts against your live-computer limit; a create over the limit raises DrumsError with status 403 (quota_exceeded), and a full host raises Overloaded (429), in which case nothing started and you can retry later.

With prime-rl or verifiers

If you already evaluate with verifiers (vf-eval), the drums-verifiers adapter runs the same loop with a tool-calling model and the same validity rules.