"""UI-TARS-1.5 and MolmoWeb as released, each in its own agent loop, on MiniWeb human tasks (EXPERIMENT_PLAN, RQ1; 2026-10-03). The counterpart of src/miniweb/evaluation/fara_runner.py for the two other own-loop baselines. uitars UI-TARS-1.5-7B (ByteDance-Seed, Apache-2.0). Prompt and action parser are the released ones (pip `ui-tars`: ui_tars.prompt.COMPUTER_USE_DOUBAO, ui_tars.action_parser.parse_action_to_structure_output). The loop is OSWorld's UI-TARS-1.5 agent (mm_agents/uitars15_v1.py, run_multienv_uitars15_v1.py): system "You are a helpful assistant.", the prompt, then the full response history with the last 5 screenshots, temperature 0, top_p 0.9, max_tokens 500 (OSWorld's frequency_penalty 1 is off by default: see --frequency-penalty). A failed parse is retried up to 3 times at temperature 1, then the episode ends. Coordinates are in the Qwen2.5-VL smart_resize space of the screenshot and are mapped back as parsing_response_to_pyautogui_code does. The pyautogui calls become Playwright mouse and keyboard calls: type -> keyboard.type, plus Enter for a trailing \\n; scroll -> 5 wheel notches of 100 px at the point; wait -> 5 s. The answer is finished(content=...). molmoweb MolmoWeb-4B / -8B (AllenAI, Apache-2.0): the released agent code itself (github.com/allenai/molmoweb: agent.multimodal_agent.MultimodalAgent, utils.envs SimpleEnv and its action executor, utils.eval_utils.episode.Episode) with its benchmark settings: "molmo_web_think" prompts, the last 10 steps as text, no past images, temperature 0.7, top_p 0.8, 1024 new tokens, a 1280x720 viewport. Only the predictor is swapped: their HF predictor's single user turn (prompt text + screenshot), sent to a vLLM OpenAI endpoint (the repo leaves vLLM to users). Serve it with causal attention on image tokens, as their HF predictor runs it (vLLM's Molmo2 default is a bidirectional prefix LM): --hf-overrides {"is_mm_prefix_lm":false}, which granite's ownloop.sh passes. The answer is read as their benchmarks.evaluate does: the last "[ANSWER] ..." message, else the last other message, else the last thought. fara7b Fara-7B (Microsoft, MIT): the released previous-generation agent itself (github.com/microsoft/fara src/, fara.fara_7b: FaraAgent, its BrowserBB browser manager and PlaywrightController), as `fara-cli --fara-7b` and webeval's WebSurferSystem run it: 1440x900, the last 3 screenshots, temperature 0, single-tab mode, their page script. The browser manager is subclassed only to launch our Chromium with our context, in place of Firefox with an Edge user agent. The answer is theirs: the terminate step's thoughts. A parse or action error ends the episode, as in the released loop, and the task is re-run up to 5 times, as webeval (the harness of their published numbers) does by default. opencua OpenCUA-7B (XLANG, MIT): OSWorld's OpenCUAAgent itself (xlang-ai/OSWorld mm_agents/opencua) with the OpenCUA-7B command of run_multienv_opencua.py (--coordinate_type qwen25 --use_old_sys_prompt; l2 CoT, action history, 3 images, temperature 0, top_p 0.9, max_tokens 2048) on a 1920x1080 page, and the loop of lib_run_single.run_single_example_opencua. Only call_llm is swapped (a vLLM endpoint), and the VM's `python -c` is a pyautogui-to-Playwright translation (run_pyautogui, literal calls only). The V1 prompt has no answer field: the answer is the terminate step's answer= if given, else its Thought + Action text. Serve with --trust-remote-code and a tokenizer dir whose tokenizer_config.json drops unk_token/pad_token (the remote TikTokenV3 code rejects them under transformers 5; it derives the same [UNK]/[PAD] itself). Both run in the MiniWeb env (Playwright sync API, DATAGEN_CHROME) and start their own MiniWeb servers on --ports, as miniwebagent.evaluate does. The released code goes on sys.path through --path (default $OWNLOOP_PATH): a directory holding `pip install --no-deps --target ui-tars==0.5.1`, and a clone of allenai/molmoweb. Setup per task is miniwebagent.evaluate.prepare_human's: /_admin/session-flags (logged out for a login task), then the start page, clipboard allowed, with the task wrapper of miniwebagent.harness.instruction. Tasks come from `python -m miniweb.evaluation.fara_rq1 prep` ({BASE} placeholders); grading is `python -m miniweb.evaluation.fara_rq1 grade --arm `, from what this saves per task: //session.json {"base", "answer", "status", "rounds", "record": [...], "log": [...], "error", ...} //steps.jsonl one line per model call: raw output, parsed action, url, errors //screens/ step_NN.png: the screenshot each step saw, plus the final one python -m miniweb.evaluation.ownloop_runner uitars --tasks /tasks.json --out --ports 8312,8313 \\ --endpoint http://127.0.0.1:8694/v1 --model uitars15-7b --path $W/ownloop_pkgs python -m miniweb.evaluation.ownloop_runner molmoweb --tasks /tasks.json --out --ports 8314,8315 \\ --endpoint http://127.0.0.1:8695/v1 --model molmoweb-4b --path $W/ownloop_pkgs:$W/molmoweb python -m miniweb.evaluation.ownloop_runner fara7b --tasks /tasks.json --out --ports 8312,8313 \\ --endpoint http://127.0.0.1:8696/v1 --model fara-7b --path $W/fara7b_pkgs:$W/fara/src python -m miniweb.evaluation.ownloop_runner opencua --tasks /tasks.json --out --ports 8314,8315 \\ --endpoint http://127.0.0.1:8697/v1 --model opencua-7b --path $W/opencua_pkgs:$W/osworld """ import argparse import base64 import io import json import math import os import re import sys import threading import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path LAUNCH_ARGS = ["--disable-dev-shm-usage", "--no-first-run", "--disable-blink-features=AutomationControlled"] def instruction(task_text, start_url): """Same wrapper as miniwebagent.harness.instruction (our agents' task text) and src/miniweb/evaluation/fara_runner.py.""" return f"You are interacting with a web application at {start_url}. Your task: {task_text}" def launch_browser(pw): from miniwebgen import config exe = config.CHROME if os.path.exists(config.CHROME) else None return pw.chromium.launch(headless=True, executable_path=exe, args=LAUNCH_ARGS) def new_context(browser, width, height): """miniwebgen.browser.new_context (clipboard granted, focus emulated, en-US / UTC) at this agent's viewport.""" from miniwebgen.browser import new_context as nc return nc(browser, viewport=(width, height)) def admin_entries(ctx, base, path): """A privileged /_admin endpoint, read as this browser's session (the context's cookies).""" r = ctx.request.get(base + path, timeout=30000) return (r.json() or {}).get("entries", []) or [] def png_b64(png): return "data:image/png;base64," + base64.b64encode(png).decode() class EndpointError(RuntimeError): """The model server failed (not the model): the task is left unsaved, for a rerun.""" def chat(client, model, messages, **kw): """One chat completion; transport errors are retried (harness-level, not the agent's own retry).""" for attempt in range(3): try: r = client.chat.completions.create(model=model, messages=messages, **kw) return (r.choices[0].message.content or "").strip() except Exception as e: if attempt == 2: raise EndpointError(f"{type(e).__name__}: {str(e)[:200]}") from e time.sleep(5) class Rec: """Per-task output: steps.jsonl lines and screenshots.""" def __init__(self, ep): self.ep = ep (ep / "screens").mkdir(parents=True, exist_ok=True) self.f = open(ep / "steps.jsonl", "w") def step(self, **row): self.f.write(json.dumps(row, default=str) + "\n") self.f.flush() def shot(self, i, png): (self.ep / "screens" / f"step_{i:02d}.png").write_bytes(png) def close(self): self.f.close() # ── UI-TARS-1.5 ─────────────────────────────────────────────────────────────────────────────────────────────────────── PW_KEYS = {"ctrl": "Control", "control": "Control", "shift": "Shift", "alt": "Alt", "option": "Alt", "cmd": "Meta", "command": "Meta", "win": "Meta", "meta": "Meta", "super": "Meta", "enter": "Enter", "return": "Enter", "esc": "Escape", "escape": "Escape", "tab": "Tab", "backspace": "Backspace", "delete": "Delete", "del": "Delete", "space": "Space", "up": "ArrowUp", "down": "ArrowDown", "left": "ArrowLeft", "right": "ArrowRight", "arrowup": "ArrowUp", "arrowdown": "ArrowDown", "arrowleft": "ArrowLeft", "arrowright": "ArrowRight", "pageup": "PageUp", "pagedown": "PageDown", "home": "Home", "end": "End", "insert": "Insert", "capslock": "CapsLock"} def pw_key(k): k = k.strip() if k.lower() in PW_KEYS: return PW_KEYS[k.lower()] if re.fullmatch(r"[fF]\d{1,2}", k): return k.upper() return k class UITars: """OSWorld mm_agents/uitars15_v1.py UITARSAgent.predict (infer_mode qwen25vl_normal), with the repo's computer-use prompt and an OpenAI-compatible endpoint.""" MAX_PIXELS, MIN_PIXELS = 16384 * 28 * 28, 100 * 28 * 28 def __init__(self, client, model, task, history_n=5, max_tokens=500, temperature=0.0, top_p=0.9, frequency_penalty=0.0, language="English"): from ui_tars.prompt import COMPUTER_USE_DOUBAO self.client, self.model = client, model self.prompt = COMPUTER_USE_DOUBAO.format(language=language, instruction=task) self.history_n, self.max_tokens = history_n, max_tokens self.temperature, self.top_p, self.frequency_penalty = temperature, top_p, frequency_penalty self.images, self.responses = [], [] def _image(self, png): """The agent's own resize to [min_pixels, max_pixels] (a no-op at our viewports) -> (png, height, width).""" from PIL import Image im = Image.open(io.BytesIO(png)) if im.width * im.height > self.MAX_PIXELS: f = math.sqrt(self.MAX_PIXELS / (im.width * im.height)) im = im.resize((int(im.width * f), int(im.height * f))) if im.width * im.height < self.MIN_PIXELS: f = math.sqrt(self.MIN_PIXELS / (im.width * im.height)) im = im.resize((math.ceil(im.width * f), math.ceil(im.height * f))) if im.mode != "RGB": im = im.convert("RGB") buf = io.BytesIO() im.save(buf, format="PNG") return buf.getvalue(), im.height, im.width def predict(self, png): """-> (response text or None, parsed actions or None, [failed attempts]).""" from ui_tars.action_parser import add_box_token, parse_action_to_structure_output self.images = (self.images + [png])[-self.history_n:] imgs = [self._image(b) for b in self.images] messages = [{"role": "system", "content": [{"type": "text", "text": "You are a helpful assistant."}]}, {"role": "user", "content": [{"type": "text", "text": self.prompt}]}] n = 0 for i, resp in enumerate(self.responses): if i + self.history_n > len(self.responses): # at most history_n screenshots in all messages.append({"role": "user", "content": [{"type": "image_url", "image_url": {"url": png_b64(imgs[n][0])}}]}) n += 1 messages.append({"role": "assistant", "content": [{"type": "text", "text": add_box_token(resp)}]}) messages.append({"role": "user", "content": [{"type": "image_url", "image_url": {"url": png_b64(imgs[n][0])}}]}) _, h, w = imgs[-1] temperature, fails = self.temperature, [] for _ in range(3): text = chat(self.client, self.model, messages, max_tokens=self.max_tokens, temperature=temperature, top_p=self.top_p, frequency_penalty=self.frequency_penalty) try: parsed = parse_action_to_structure_output(text, 1000, h, w, "qwen25vl", self.MAX_PIXELS, self.MIN_PIXELS) except Exception as e: # as OSWorld: resample at temperature 1 fails.append({"raw": text, "parse_error": f"{type(e).__name__}: {str(e)[:200]}"}) temperature = 1.0 continue self.responses.append(text) return text, parsed, fails return None, None, fails def _box_center(box, w, h): x1, y1, x2, y2 = eval(box) if isinstance(box, str) else box # the parser's own "[x1, y1, x2, y2]" fractions return round((x1 + x2) / 2 * w, 3), round((y1 + y2) / 2 * h, 3) def uitars_execute(page, act, w, h): """One parsed UI-TARS action on the page, as parsing_response_to_pyautogui_code would run it. -> (stop, answer, note).""" t, a = act.get("action_type"), act.get("action_inputs") or {} if t == "finished": return True, a.get("content", ""), "" if t == "call_user": return True, None, "call_user" if t in ("click", "left_single", "left_double", "right_single", "hover"): if not a.get("start_box"): return False, None, "no start_box" x, y = _box_center(a["start_box"], w, h) if t == "left_double": page.mouse.dblclick(x, y) elif t == "right_single": page.mouse.click(x, y, button="right") elif t == "hover": page.mouse.move(x, y) else: page.mouse.click(x, y) elif t in ("drag", "select"): if not (a.get("start_box") and a.get("end_box")): return False, None, "drag without both boxes" sx, sy = _box_center(a["start_box"], w, h) ex, ey = _box_center(a["end_box"], w, h) page.mouse.move(sx, sy) page.mouse.down() page.mouse.move(ex, ey, steps=20) # pyautogui.dragTo(duration=1.0) page.mouse.up() elif t == "hotkey": keys = (a.get("key") or a.get("hotkey") or "").split() if keys: page.keyboard.press("+".join(pw_key(k) for k in keys)) elif t in ("press", "keydown", "keyup"): k = pw_key(a.get("key") or a.get("press") or "") if k: (page.keyboard.up if t == "keyup" else page.keyboard.down)(k) if t == "press": page.keyboard.up(k) elif t == "type": content = a.get("content", "") enter = content.endswith("\n") or content.endswith("\\n") text = content.rstrip("\\n").rstrip("\n") if enter else content if text: page.keyboard.type(text) if enter: page.keyboard.press("Enter") elif t == "scroll": d = (a.get("direction") or "").lower() if a.get("start_box"): page.mouse.move(*_box_center(a["start_box"], w, h)) dx = -500 if "left" in d else 500 if "right" in d else 0 dy = -500 if "up" in d else 500 if "down" in d else 0 page.mouse.wheel(dx, dy) elif t == "wait": time.sleep(5) else: return False, None, f"unrecognized action {t}" return False, None, "" def settle(ctx, page): """-> the active page (a click may open a tab: follow the newest) after its DOM is loaded, plus a short buffer.""" pages = [p for p in ctx.pages if not p.is_closed()] page = pages[-1] if pages else page try: page.wait_for_load_state("domcontentloaded", timeout=5000) except Exception: pass time.sleep(1.0) return page def run_uitars(t, base, args, ep): from openai import OpenAI from playwright.sync_api import sync_playwright start = t["start"].replace("{BASE}", base) prompt = t["prompt"].replace("{BASE}", base) flags = base + "/_admin/session-flags" + ("?logout=1" if t["needs_login"] else "") res = {"task_id": t["task_id"], "base": base, "answer": None, "status": None, "rounds": 0, "error": None, "record": [], "log": [], "seconds": None, "parse_fails": 0, "action_errors": 0} t0, rec = time.time(), Rec(ep) client = OpenAI(base_url=args.endpoint, api_key="not-needed", timeout=300) agent = UITars(client, args.model, instruction(prompt, start), frequency_penalty=args.frequency_penalty) with sync_playwright() as pw: browser = launch_browser(pw) ctx = new_context(browser, args.width, args.height) try: page = ctx.new_page() page.goto(flags, wait_until="domcontentloaded", timeout=60000) page.goto(start, wait_until="domcontentloaded", timeout=60000) time.sleep(1.0) for i in range(1, args.max_steps + 1): if time.time() - t0 > args.timeout: res["status"] = "timeout" break png = page.screenshot(type="png") rec.shot(i, png) text, parsed, fails = agent.predict(png) res["rounds"] = i res["parse_fails"] += sum("parse_error" in f for f in fails) if parsed is None: rec.step(step=i, url=page.url, raw=None, fails=fails) res["status"] = "parse_error" # OSWorld: no parsable response in 3 tries ends it break stop, notes = False, [] for act in parsed: try: stop, answer, note = uitars_execute(page, act, args.width, args.height) except Exception as e: stop, answer, note = False, None, f"{type(e).__name__}: {str(e)[:200]}" res["action_errors"] += bool(note) and note != "call_user" notes.append(note) if stop: res["answer"], res["status"] = answer, "finished" if note != "call_user" else "call_user" break rec.step(step=i, url=page.url, raw=text, actions=[{k: v for k, v in a.items() if k != "text"} for a in parsed], notes=notes, fails=fails) if stop: break page = settle(ctx, page) else: res["status"] = "max_steps" try: rec.shot(res["rounds"] + 1, settle(ctx, page).screenshot(type="png")) except Exception: pass except Exception as e: # a harness error still keeps whatever the session did res["error"] = f"{type(e).__name__}: {str(e)[:300]}" if isinstance(e, EndpointError): res["status"] = "endpoint_error" try: res["record"] = admin_entries(ctx, base, "/_admin/record") res["log"] = admin_entries(ctx, base, "/_admin/log") except Exception as e: res["error"] = (res["error"] or "") + f" | backend read: {type(e).__name__}: {str(e)[:200]}" browser.close() rec.close() res["seconds"] = round(time.time() - t0, 1) return res # ── MolmoWeb ────────────────────────────────────────────────────────────────────────────────────────────────────────── class VllmPredictor: """Their HFActionPredictor's request (one user turn: prompt text, then the screenshot(s); sampling with temperature / top_p; max_new_tokens 1024), sent to an OpenAI-compatible endpoint. Stops the episode at the task deadline: an exception here is the Episode's ACTION_PREDICTION_ERROR, which ends it.""" def __init__(self, client, model, temperature, top_p, deadline, max_tokens=1024): self.client, self.model, self.deadline = client, model, deadline self.temperature, self.top_p, self.max_tokens = temperature, top_p, max_tokens self.failed = False def predict(self, prompt, image_np, past_actions=None, **kw): from PIL import Image if time.time() > self.deadline: raise TimeoutError("task timeout") content = [{"type": "text", "text": prompt}] for im in (image_np if isinstance(image_np, list) else [image_np]): buf = io.BytesIO() Image.fromarray(im.astype("uint8")).convert("RGB").save(buf, format="PNG") content.append({"type": "image_url", "image_url": {"url": png_b64(buf.getvalue())}}) try: text = chat(self.client, self.model, [{"role": "user", "content": content}], max_tokens=self.max_tokens, temperature=self.temperature, top_p=self.top_p) except EndpointError: self.failed = True raise return text or None def molmoweb_answer(interactions): """benchmarks.evaluate.read_trajectory's answer: the last "[ANSWER] ..." message, else the last message that is not [EXIT], else the last thought.""" last_answer = last_msg = last_thought = None for it in interactions: if not it.action: continue act = it.action["action_output"].action msg = getattr(act, "msg", None) if msg is not None: if "[ANSWER]" in msg: last_answer = msg[9:] elif "[EXIT]" not in msg: last_msg = msg if it.action.get("thought"): last_thought = it.action["thought"] return last_answer if last_answer is not None else last_msg if last_msg is not None else last_thought def run_molmoweb(t, base, args, ep): import numpy as np from openai import OpenAI from PIL import Image from agent.actions import SendMsgToUser from agent.multimodal_agent import MultimodalAgent from utils.envs.browser_env import SimpleEnv, _start_playwright from utils.eval_utils.episode import Episode start = t["start"].replace("{BASE}", base) prompt = t["prompt"].replace("{BASE}", base) flags = base + "/_admin/session-flags" + ("?logout=1" if t["needs_login"] else "") class MiniWebEnv(SimpleEnv): """Their local-Chromium env with our Chromium build and context, opening the session flags first.""" def _launch(self): self.playwright = _start_playwright() self.browser = launch_browser(self.playwright) self.context = new_context(self.browser, self.viewport_width, self.viewport_height) self.page = self.context.new_page() def _navigate_to_start(self, url, goal): self.page.goto(flags, wait_until="domcontentloaded", timeout=60000) self.page.goto(url, wait_until="domcontentloaded", timeout=60000) time.sleep(1.0) return goal res = {"task_id": t["task_id"], "base": base, "answer": None, "status": None, "rounds": 0, "error": None, "record": [], "log": [], "seconds": None, "parse_fails": 0, "action_errors": 0} t0, rec = time.time(), Rec(ep) env = MiniWebEnv(start_url=start, goal=instruction(prompt, start), viewport_width=args.width, viewport_height=args.height) agent = MultimodalAgent(endpoint_or_checkpoint=args.endpoint, inference_mode="fastapi", system_message="molmo_web_think", max_past_steps=args.max_past_steps, max_past_images=0, sampling_temperature=args.temperature, sampling_top_p=args.top_p) agent.predictor = VllmPredictor(OpenAI(base_url=args.endpoint, api_key="not-needed", timeout=300), args.model, args.temperature, args.top_p, t0 + args.timeout) episode = Episode(env=env, agent=agent, eps_name=t["task_id"]) try: inters, _ = episode.run_episode(max_steps=args.max_steps) except Exception as e: inters = episode.interactions res["error"] = f"{type(e).__name__}: {str(e)[:300]}" def png(arr): buf = io.BytesIO() Image.fromarray(np.asarray(arr).astype("uint8")).save(buf, format="PNG") return buf.getvalue() for i, it in enumerate(inters, 1): obs = it.state.get("obs") or {} if obs.get("screenshot") is not None: rec.shot(i, png(obs["screenshot"])) a = it.action or {} thought = a.get("thought") or "" res["parse_fails"] += thought.startswith("Could not parse predicted action") nxt = (it.next_state or {}).get("obs") or {} res["action_errors"] += bool(nxt.get("last_action_error")) rec.step(step=i, url=obs.get("url"), raw=it.raw_output, action=a.get("action"), action_str=a.get("action_str"), thought=thought, error=it.error, action_error=nxt.get("last_action_error")) if inters and (inters[-1].next_state or {}).get("obs", {}).get("screenshot") is not None: rec.shot(len(inters) + 1, png(inters[-1].next_state["obs"]["screenshot"])) res["rounds"] = len(inters) res["answer"] = molmoweb_answer(inters) last = inters[-1] if inters else None if last is None: res["status"] = "no_steps" elif last.error: err = json.dumps(last.error) res["status"] = "timeout" if "task timeout" in err else "endpoint_error" if "EndpointError" in err or \ agent.predictor.failed else "error" res["error"] = res["error"] or json.dumps(last.error)[:300] elif last.action and isinstance(last.action["action_output"].action, SendMsgToUser) \ and (last.action["action_output"].action.msg or "").startswith("[EXIT]"): res["status"] = "exit" else: res["status"] = "max_steps" try: if env.context is not None: res["record"] = admin_entries(env.context, base, "/_admin/record") res["log"] = admin_entries(env.context, base, "/_admin/log") except Exception as e: res["error"] = (res["error"] or "") + f" | backend read: {type(e).__name__}: {str(e)[:200]}" try: env.close() except Exception: pass rec.close() res["seconds"] = round(time.time() - t0, 1) return res # ── Fara-7B (added 2026-10-03) ──────────────────────────────────────────────────────────────────────────────────────── _IMPORT_LOCK = threading.Lock() def _fara_package(): """Register the `fara` package of the --path clone (github.com/microsoft/fara src/) without running its __init__.py, which imports the Fara1.5 agent and environment stacks and their extra dependencies. fara.fara_7b and fara.qwen_helpers are then imported unchanged.""" import importlib.machinery import importlib.util with _IMPORT_LOCK: if "fara" in sys.modules: return for p in sys.path: d = Path(p or ".") / "fara" if (d / "fara_7b" / "fara_agent.py").is_file(): spec = importlib.machinery.ModuleSpec("fara", None, is_package=True) spec.submodule_search_locations = [str(d)] sys.modules["fara"] = importlib.util.module_from_spec(spec) return raise ImportError("no fara/fara_7b on sys.path: put the microsoft/fara src/ dir on --path") class _ThreadStdout: """sys.stdout that sends a worker thread's writes to that thread's task file when one is set. The Fara-7B agent print()s every thought, action and observation. Other writes go to the real stdout.""" def __init__(self, base): self.base, self.local = base, threading.local() def _f(self): return getattr(self.local, "f", None) or self.base def write(self, s): return self._f().write(s) def flush(self): return self._f().flush() def __getattr__(self, k): return getattr(self.base, k) def _thread_stdout(): with _IMPORT_LOCK: if not isinstance(sys.stdout, _ThreadStdout): sys.stdout = _ThreadStdout(sys.stdout) return sys.stdout def fara7b_parse(agent, text): """-> (ok, tool-call dict): their _parse_thoughts_and_action, plus the keys their generate_model_call and run() read next (a response without them ends the released loop with an exception, as a parse failure does).""" try: _, action = agent._parse_thoughts_and_action(text) return isinstance(action.get("arguments"), dict) and "action" in action["arguments"], action except Exception: return False, None def run_fara7b(t, base, args, ep): """One task, with webeval's error retries. Their eval loop (webeval core.run_eval_single_example) re-runs a task whose episode raised, which leaves no evaluable trajectory, up to --max_error_task_retries times (default 5). The same here: an attempt that ended with status error / parse_error is moved to /error_attempt_N/, and the task runs again in a fresh browser context (a fresh MiniWeb session). The last attempt is the one graded.""" import asyncio import shutil t0, failed = time.time(), [] while True: res = asyncio.run(_fara7b(t, base, args, ep)) if res["status"] not in ("error", "parse_error") or len(failed) >= args.fara7b_error_retries: break keep = ep / f"error_attempt_{len(failed) + 1}" keep.mkdir() for p in list(ep.iterdir()): if not p.name.startswith("error_attempt_"): shutil.move(str(p), str(keep / p.name)) (keep / "attempt.json").write_text(json.dumps(res, default=str)) failed.append({k: res[k] for k in ("status", "rounds", "error", "seconds", "parse_fails", "action_errors")}) res["error_attempts"] = failed res["parse_fails"] += sum(f["parse_fails"] for f in failed) res["action_errors"] += sum(f["action_errors"] for f in failed) res["seconds"] = round(time.time() - t0, 1) return res async def _fara7b(t, base, args, ep): """Fara-7B's own agent (fara.fara_7b.FaraAgent) and browser manager (BrowserBB, PlaywrightController) as `fara-cli --fara-7b` and webeval's WebSurferSystem run it: 1440x900, the last 3 screenshots, temperature 0. Their browser manager is subclassed only to launch our Chromium build with our context (clipboard granted, focus emulated, en-US/UTC), in place of Firefox with an Edge user agent.""" import asyncio import logging import openai from miniwebgen import config _fara_package() from fara.fara_7b.browser.browser_bb import BrowserBB from fara.fara_7b.fara_agent import FaraAgent from fara.fara_7b.fara_types import AssistantMessage if (args.width, args.height) != (1440, 900): raise ValueError("FaraAgent maps coordinates to its fixed 1440x900 viewport") start = t["start"].replace("{BASE}", base) prompt = t["prompt"].replace("{BASE}", base) flags = base + "/_admin/session-flags" + ("?logout=1" if t["needs_login"] else "") res = {"task_id": t["task_id"], "base": base, "answer": None, "status": None, "rounds": 0, "error": None, "record": [], "log": [], "seconds": None, "parse_fails": 0, "action_errors": 0} t0, rec = time.time(), Rec(ep) fdir = ep / "fara7b" # their downloads folder: screenshot{n}.png, web_surfer.log, prints fdir.mkdir(parents=True, exist_ok=True) log = logging.getLogger(f"ownloop.fara7b.{t['task_id']}") log.setLevel(logging.DEBUG) # webeval's level: their Thought/Action/Observation traces are DEBUG log.propagate = False handler = logging.FileHandler(fdir / "web_surfer.log") log.addHandler(handler) out = _thread_stdout() out.local.f = open(fdir / "stdout.log", "w") class MiniWebBB(BrowserBB): async def _init_regular_browser(self, channel="chromium"): exe = config.CHROME if os.path.exists(config.CHROME) else None self.browser = await self._playwright.chromium.launch(headless=True, executable_path=exe, args=LAUNCH_ARGS) ctx = self._context = await self.browser.new_context( viewport={"width": self._viewport_width, "height": self._viewport_height}, device_scale_factor=1, locale="en-US", timezone_id="UTC", accept_downloads=True) try: await ctx.grant_permissions(["clipboard-read", "clipboard-write"]) except Exception: pass async def focus(page): # miniwebgen.browser._focus try: s = await ctx.new_cdp_session(page) await s.send("Emulation.setFocusEmulationEnabled", {"enabled": True}) except Exception: pass ctx.on("page", focus) self._page = await ctx.new_page() bm = MiniWebBB(viewport_height=args.height, viewport_width=args.width, headless=True, page_script_path=None, browser_channel="chromium", browser_data_dir=None, downloads_folder=str(fdir), to_resize_viewport=True, single_tab_mode=True, animate_actions=False, use_browser_base=False, logger=log) agent = FaraAgent(browser_manager=bm, client_config={"model": args.model, "base_url": args.endpoint, "api_key": "not-needed"}, start_page=flags, downloads_folder=str(fdir), save_screenshots=True, max_rounds=args.max_steps, logger=log) answer = None try: await agent.initialize() # browser up at the session-flags page (their start_page) await agent._page.goto(start, wait_until="domcontentloaded", timeout=60000) await asyncio.sleep(1.0) answer, _, _ = await asyncio.wait_for(agent.run(instruction(prompt, start)), timeout=max(60.0, args.timeout - (time.time() - t0))) except (asyncio.TimeoutError, TimeoutError) as e: res["status"], res["error"] = "timeout", f"{type(e).__name__}: {str(e)[:200]}" except (openai.APIConnectionError, openai.InternalServerError, openai.NotFoundError) as e: res["status"], res["error"] = "endpoint_error", f"{type(e).__name__}: {str(e)[:300]}" except Exception as e: # the released loop has no recovery: a parse or action error ends it res["status"], res["error"] = "error", f"{type(e).__name__}: {str(e)[:300]}" responses = [m.content for m in agent._chat_history if isinstance(m, AssistantMessage)] res["rounds"] = len(responses) parsed = [fara7b_parse(agent, r) for r in responses] for i, (text, (ok, action)) in enumerate(zip(responses, parsed), 1): rec.step(step=i, raw=text, action=(action or {}).get("arguments") if ok else None, parse_error=not ok) last_ok, last = parsed[-1] if parsed else (True, None) if res["status"] is None: if last_ok and last and last["arguments"]["action"] in ("terminate", "stop"): res["status"], res["answer"] = "terminate", answer # their final answer: the terminate step's thoughts else: res["status"] = "max_steps" elif res["status"] == "error" and responses and not last_ok: res["status"], res["parse_fails"] = "parse_error", 1 elif res["status"] == "error": res["action_errors"] = 1 for png in fdir.glob("screenshot*.png"): # screenshot{n}: after n actions = what step n+1 saw m = re.fullmatch(r"screenshot(\d+)\.png", png.name) if m: os.replace(png, ep / "screens" / f"step_{int(m.group(1)) + 1:02d}.png") try: if bm.context is not None: for key, path in (("record", "/_admin/record"), ("log", "/_admin/log")): r = await bm.context.request.get(base + path, timeout=30000) res[key] = ((await r.json()) or {}).get("entries", []) or [] except Exception as e: res["error"] = (res["error"] or "") + f" | backend read: {type(e).__name__}: {str(e)[:200]}" for close in (agent.close, getattr(bm.browser, "close", None), # their client is left open: closed here so getattr(agent._openai_client, "close", None)): # GC does not close it after the loop ends try: if close is not None: await close() except Exception: pass out.local.f.close() out.local.f = None log.removeHandler(handler) handler.close() rec.close() res["seconds"] = round(time.time() - t0, 1) return res # ── OpenCUA (added 2026-10-03) ──────────────────────────────────────────────────────────────────────────────────────── PY_KEYS = {**PW_KEYS, "ctrlleft": "Control", "ctrlright": "Control", "shiftleft": "Shift", "shiftright": "Shift", "altleft": "Alt", "altright": "Alt", "winleft": "Meta", "winright": "Meta", "pgup": "PageUp", "pgdn": "PageDown", "prtsc": "PrintScreen", "prtscr": "PrintScreen", "printscreen": "PrintScreen", "\n": "Enter", "\r": "Enter", "\t": "Tab", " ": "Space"} def py_key(k): """A pyautogui key name -> a Playwright key.""" k = str(k) if k.lower() in PY_KEYS: return PY_KEYS[k.lower()] if re.fullmatch(r"[fF]\d{1,2}", k): return k.upper() return k class PyAutoGUI: """The pyautogui functions OSWorld's VM runs (`python -c "import pyautogui; ...; "`), on a Playwright page of the same screen size, with the mouse position tracked as pyautogui tracks it. One scroll click is 100 px, as in the UI-TARS translation above.""" SCROLL_PX = 100 API = {"click", "doubleClick", "tripleClick", "rightClick", "middleClick", "moveTo", "moveRel", "move", "dragTo", "dragRel", "drag", "mouseDown", "mouseUp", "scroll", "vscroll", "hscroll", "write", "typewrite", "press", "hotkey", "keyDown", "keyUp", "sleep"} def __init__(self, page): self.page, self.pos = page, (0.0, 0.0) # Playwright's mouse starts at (0, 0) def _xy(self, x, y): if isinstance(x, (tuple, list)): x, y = x[0], x[1] if x is None or y is None: return self.pos return float(x), float(y) @staticmethod def _button(b): return {"right": "right", "secondary": "right", "middle": "middle"}.get(str(b).lower(), "left") def moveTo(self, x=None, y=None, duration=0.0, *a, **kw): self.pos = self._xy(x, y) self.page.mouse.move(*self.pos, steps=max(1, int(float(duration or 0) * 20))) def moveRel(self, xOffset=0, yOffset=0, *a, **kw): self.moveTo(self.pos[0] + float(xOffset or 0), self.pos[1] + float(yOffset or 0)) move = moveRel def click(self, x=None, y=None, clicks=1, interval=0.0, button="left", *a, **kw): self.moveTo(x, y) self.page.mouse.click(*self.pos, button=self._button(button), click_count=max(1, int(clicks))) def doubleClick(self, x=None, y=None, interval=0.0, button="left", *a, **kw): self.click(x, y, 2, button=button) def tripleClick(self, x=None, y=None, interval=0.0, button="left", *a, **kw): self.click(x, y, 3, button=button) def rightClick(self, x=None, y=None, *a, **kw): self.click(x, y, button="right") def middleClick(self, x=None, y=None, *a, **kw): self.click(x, y, button="middle") def mouseDown(self, x=None, y=None, button="left", *a, **kw): if x is not None: self.moveTo(x, y) self.page.mouse.down(button=self._button(button)) def mouseUp(self, x=None, y=None, button="left", *a, **kw): if x is not None: self.moveTo(x, y) self.page.mouse.up(button=self._button(button)) def dragTo(self, x=None, y=None, duration=0.0, button="left", *a, **kw): self.page.mouse.down(button=self._button(button)) self.pos = self._xy(x, y) self.page.mouse.move(*self.pos, steps=20) self.page.mouse.up(button=self._button(button)) def dragRel(self, xOffset=0, yOffset=0, duration=0.0, button="left", *a, **kw): self.dragTo(self.pos[0] + float(xOffset or 0), self.pos[1] + float(yOffset or 0), button=button) drag = dragRel def scroll(self, clicks, x=None, y=None, *a, **kw): # positive clicks scroll up if x is not None and y is not None: self.moveTo(x, y) self.page.mouse.wheel(0, -float(clicks) * self.SCROLL_PX) vscroll = scroll def hscroll(self, clicks, x=None, y=None, *a, **kw): # positive clicks scroll right if x is not None and y is not None: self.moveTo(x, y) self.page.mouse.wheel(float(clicks) * self.SCROLL_PX, 0) def write(self, message, interval=0.0, *a, **kw): if isinstance(message, (list, tuple)): # typewrite(["a", "enter"]): key names for k in message: self.press(k) return parts = str(message).split("\n") # pyautogui types "\n" as Enter for i, part in enumerate(parts): if part: self.page.keyboard.type(part) if i < len(parts) - 1: self.page.keyboard.press("Enter") typewrite = write def press(self, keys, presses=1, interval=0.0, *a, **kw): for _ in range(max(1, int(presses))): for k in (keys if isinstance(keys, (list, tuple)) else [keys]): self.page.keyboard.press(py_key(k)) def hotkey(self, *keys, **kw): keys = [py_key(k) for k in (keys[0] if len(keys) == 1 and isinstance(keys[0], (list, tuple)) else keys)] for k in keys: self.page.keyboard.down(k) for k in reversed(keys): self.page.keyboard.up(k) def keyDown(self, key, *a, **kw): self.page.keyboard.down(py_key(key)) def keyUp(self, key, *a, **kw): self.page.keyboard.up(py_key(key)) def sleep(self, seconds=0, *a, **kw): time.sleep(min(max(float(seconds), 0.0), 60.0)) class UnsupportedCode(Exception): """A statement that the `pyautogui` + `time` namespace of OSWorld's VM would not run (e.g. computer.triple_click: their prompt lists it, but their loop passes it to the VM's python, where it is a NameError).""" def run_pyautogui(code, gui): """OpenCUA's code, statement by statement, as `python -c` runs it: the first failing statement stops the rest. Model output is never exec'd: only literal-argument calls of pyautogui.* and time.sleep, `import pyautogui|time`, `pass`, and `for _ in range():` loops over those statements run.""" import ast def call(c): f = c.func if not (isinstance(f, ast.Attribute) and isinstance(f.value, ast.Name)): raise UnsupportedCode(f"unsupported call: {ast.unparse(c)[:120]}") mod, name = f.value.id, f.attr args = [ast.literal_eval(a) for a in c.args] kwargs = {k.arg: ast.literal_eval(k.value) for k in c.keywords if k.arg} if mod == "time" and name == "sleep": return gui.sleep(*args, **kwargs) if mod == "pyautogui" and name in gui.API: return getattr(gui, name)(*args, **kwargs) raise UnsupportedCode(f"{mod}.{name} is not defined") def stmt(node): if isinstance(node, ast.Import) and {a.name for a in node.names} <= {"pyautogui", "time"}: return if isinstance(node, ast.Pass): return if isinstance(node, ast.Expr) and isinstance(node.value, ast.Call): return call(node.value) if (isinstance(node, ast.For) and not node.orelse and isinstance(node.iter, ast.Call) and isinstance(node.iter.func, ast.Name) and node.iter.func.id == "range"): n = range(*[int(ast.literal_eval(a)) for a in node.iter.args]) for _ in range(min(len(n), 100)): for s in node.body: stmt(s) return raise UnsupportedCode(f"unsupported statement: {ast.unparse(node)[:120]}") for node in ast.parse(code).body: stmt(node) def chat_full(client, model, messages, **kw): """chat() that also returns the finish reason; a 400 (e.g. a prompt over the context) is the request's own error, not an endpoint failure.""" import openai for attempt in range(3): try: r = client.chat.completions.create(model=model, messages=messages, **kw) return r.choices[0].message.content or "", r.choices[0].finish_reason except openai.BadRequestError: raise except Exception as e: if attempt == 2: raise EndpointError(f"{type(e).__name__}: {str(e)[:200]}") from e time.sleep(5) _OPENCUA = {} def opencua_agent_class(): """Their OpenCUAAgent (xlang-ai/OSWorld mm_agents/opencua) with call_llm pointed at a vLLM endpoint. Everything else is theirs: the prompt, the history, the retries, the parsing and the coordinate projection.""" with _IMPORT_LOCK: if "cls" in _OPENCUA: return _OPENCUA["cls"] from loguru import logger logger.remove() # their module logs every prompt and response through loguru from mm_agents.opencua.opencua_agent import OpenCUAAgent class VllmOpenCUAAgent(OpenCUAAgent): """Their call_llm posts to their hosted endpoint and re-posts the same payload (up to 20 times) until finish_reason == "stop". At temperature 0 a re-post repeats the same output, so here a non-"stop" finish returns None at once; their predict() then retries (5 calls in all, temperature >= 0.2 after the first).""" def __init__(self, client, deadline, **kw): super().__init__(**kw) self.client, self.deadline = client, deadline self.calls, self.endpoint_failed, self.timed_out = [], None, False def call_llm(self, payload, model): if self.endpoint_failed: raise EndpointError(self.endpoint_failed) if time.time() > self.deadline: self.timed_out = True raise TimeoutError("task timeout") try: text, finish = chat_full(self.client, **payload) except EndpointError as e: self.endpoint_failed = str(e) raise except Exception as e: self.calls.append({"error": f"{type(e).__name__}: {str(e)[:300]}", "temperature": payload.get("temperature")}) raise self.calls.append({"raw": text, "finish_reason": finish, "temperature": payload.get("temperature")}) return text if finish == "stop" else None _OPENCUA["cls"] = VllmOpenCUAAgent return VllmOpenCUAAgent def opencua_answer(cot): """The answer of a terminate step: its answer= argument (their V2 prompt's terminate has one) if given, else the step's Thought and Action text. OpenCUA-7B's V1 prompt has no answer field, and OSWorld grades no text answers.""" import ast code = cot.get("original_code") or "" try: for node in ast.walk(ast.parse(code)): if isinstance(node, ast.Call) and getattr(node.func, "attr", None) == "terminate": for k in node.keywords: if k.arg == "answer": return str(ast.literal_eval(k.value)) except Exception: m = re.search(r"answer\s*=\s*('''|\"\"\"|'|\")(.*?)\1", code, re.S) if m: return m.group(2) return "\n".join(p for p in (cot.get("thought"), cot.get("action")) if p) or None def front_page(ctx, page): """The page on screen: a click may open a tab, which a desktop browser brings to the front.""" pages = [p for p in ctx.pages if not p.is_closed()] page = pages[-1] if pages else page try: page.wait_for_load_state("domcontentloaded", timeout=5000) except Exception: pass return page def run_opencua(t, base, args, ep): """OSWorld's OpenCUA-7B command (run_multienv_opencua.py --coordinate_type qwen25 --use_old_sys_prompt, the defaults: cot_level l2, action_history, 3 images, temperature 0, top_p 0.9, max_tokens 2048) and loop (lib_run_single.run_single_example_opencua) on a 1920x1080 page: predict, then each returned action through env.step semantics (DONE / FAIL end the episode, WAIT sleeps, pyautogui code runs), each step followed by sleep_after_execution (5 s).""" from openai import OpenAI from playwright.sync_api import sync_playwright start = t["start"].replace("{BASE}", base) prompt = t["prompt"].replace("{BASE}", base) flags = base + "/_admin/session-flags" + ("?logout=1" if t["needs_login"] else "") res = {"task_id": t["task_id"], "base": base, "answer": None, "status": None, "rounds": 0, "error": None, "record": [], "log": [], "seconds": None, "parse_fails": 0, "action_errors": 0} t0, rec = time.time(), Rec(ep) agent = opencua_agent_class()( OpenAI(base_url=args.endpoint, api_key="not-needed", timeout=300), t0 + args.timeout, model=args.model, history_type="action_history", max_steps=args.max_steps, max_image_history_length=3, platform="ubuntu", max_tokens=args.opencua_max_tokens, top_p=0.9, temperature=0.0, action_space="pyautogui", observation_type="screenshot", cot_level="l2", screen_size=(args.width, args.height), coordinate_type="qwen25", use_old_sys_prompt=True) task = instruction(prompt, start) pause = args.opencua_pause with sync_playwright() as pw: browser = launch_browser(pw) ctx = new_context(browser, args.width, args.height) try: page = ctx.new_page() page.goto(flags, wait_until="domcontentloaded", timeout=60000) page.goto(start, wait_until="domcontentloaded", timeout=60000) time.sleep(1.0) gui = PyAutoGUI(page) for i in range(1, args.max_steps + 1): if time.time() - t0 > args.timeout: res["status"] = "timeout" break png = page.screenshot(type="png") rec.shot(i, png) n_actions, n_calls = len(agent.actions), len(agent.calls) response, actions, cot = agent.predict(task, {"screenshot": png}) calls, accepted = agent.calls[n_calls:], len(agent.actions) > n_actions res["rounds"] = i res["parse_fails"] += len(calls) - accepted row = {"step": i, "url": page.url, "raw": response if accepted else None, "attempts": calls if len(calls) != 1 or not accepted else None, "action": cot.get("action"), "code": cot.get("original_code"), "executed": actions} if agent.endpoint_failed or agent.timed_out or not accepted: rec.step(**row, error=response) res["status"] = "endpoint_error" if agent.endpoint_failed else "timeout" if agent.timed_out \ else "parse_error" # their predict: 5 failed calls -> FAIL, the episode ends res["error"] = agent.endpoint_failed or str(response)[:300] break if not actions or actions[0] == "" or actions[0].lower().startswith("error"): rec.step(**row) res["status"] = "no_action" break stop, notes = False, [] for a in actions: note = "" if a in ("DONE", "FAIL"): stop = True elif a == "WAIT": time.sleep(pause) # env.step sleeps `pause` for WAIT, then once more else: try: run_pyautogui(a, gui) except Exception as e: note = f"{type(e).__name__}: {str(e)[:200]}" res["action_errors"] += 1 notes.append(note) if stop: break time.sleep(pause) gui.page = page = front_page(ctx, page) rec.step(**row, notes=notes) if stop: code = (cot.get("original_code") or "").lower() if "computer.terminate" in code: # theirs: "fail" in the code -> FAIL, else "success" res["status"] = "fail" if "fail" in code else "done" res["answer"] = opencua_answer(cot) else: # predict() forces FAIL at the step budget res["status"] = "max_steps" break else: res["status"] = "max_steps" try: rec.shot(res["rounds"] + 1, front_page(ctx, page).screenshot(type="png")) except Exception: pass except Exception as e: # a harness error still keeps whatever the session did res["error"] = f"{type(e).__name__}: {str(e)[:300]}" res["status"] = res["status"] or "error" try: res["record"] = admin_entries(ctx, base, "/_admin/record") res["log"] = admin_entries(ctx, base, "/_admin/log") except Exception as e: res["error"] = (res["error"] or "") + f" | backend read: {type(e).__name__}: {str(e)[:200]}" browser.close() rec.close() res["seconds"] = round(time.time() - t0, 1) return res # ── driver ──────────────────────────────────────────────────────────────────────────────────────────────────────────── RUNNERS = {"uitars": (run_uitars, (1280, 800)), "molmoweb": (run_molmoweb, (1280, 720)), "fara7b": (run_fara7b, (1440, 900)), "opencua": (run_opencua, (1920, 1080))} def main(argv=None): ap = argparse.ArgumentParser() ap.add_argument("agent", choices=sorted(RUNNERS)) ap.add_argument("--tasks", required=True, help="tasks.json of `python -m miniweb.evaluation.fara_rq1 prep`") ap.add_argument("--out", required=True) ap.add_argument("--ports", required=True, help="MiniWeb server ports to start (8300-8320), e.g. 8312,8313") ap.add_argument("--endpoint", required=True, help="vLLM OpenAI endpoint, e.g. http://127.0.0.1:8694/v1") ap.add_argument("--model", required=True, help="served model name") ap.add_argument("--path", default=os.environ.get("OWNLOOP_PATH", ""), help="colon list prepended to sys.path: the ui-tars --target dir, the molmoweb clone") ap.add_argument("--workers", type=int, default=8) ap.add_argument("--max-steps", type=int, default=40, help="model actions per task (our MiniWeb budget: 40)") ap.add_argument("--timeout", type=int, default=1800) ap.add_argument("--width", type=int, default=0, help="viewport; default uitars 1280x800, molmoweb 1280x720") ap.add_argument("--height", type=int, default=0) ap.add_argument("--frequency-penalty", type=float, default=0.0, help="uitars: 0 as the UI-TARS README's deployment example; OSWorld's request sends 1, which turned " "answers that repeat the thought's words into Chinese in the smoke test") ap.add_argument("--temperature", type=float, default=0.7, help="molmoweb sampling (their default)") ap.add_argument("--top-p", type=float, default=0.8) ap.add_argument("--max-past-steps", type=int, default=10, help="molmoweb text history (their benchmark default)") ap.add_argument("--fara7b-error-retries", type=int, default=5, help="fara7b: webeval's --max_error_task_retries (an episode that raised is re-run)") ap.add_argument("--opencua-max-tokens", type=int, default=2048, help="opencua: run_multienv_opencua.py's --max_tokens (the agent class's own default is 1500)") ap.add_argument("--opencua-pause", type=float, default=5.0, help="opencua: run_multienv_opencua.py's --sleep_after_execution, slept after every action") ap.add_argument("--limit", type=int, default=0) ap.add_argument("--task-ids", default="") args = ap.parse_args(argv) for p in reversed([p for p in args.path.split(":") if p]): sys.path.insert(0, p) run, (w, h) = RUNNERS[args.agent] args.width, args.height = args.width or w, args.height or h out = Path(args.out) tasks = json.loads(Path(args.tasks).read_text()) if args.task_ids: keep = set(args.task_ids.split(",")) tasks = [t for t in tasks if t["task_id"] in keep] done = {p.parent.name for p in out.glob("*/session.json")} tasks = [t for t in tasks if t["task_id"] not in done] # resume if args.limit: tasks = tasks[:args.limit] from miniwebgen import browser as B lock, queue = threading.Lock(), list(tasks) ports = [int(p) for p in args.ports.split(",")] with B.Servers(n=len(ports), ports=ports) as srv: bases = srv.bases print(f"{args.agent}: {len(tasks)} tasks, {args.workers} workers on {bases}", flush=True) def worker(i): while True: with lock: if not queue: return t = queue.pop(0) ep = out / t["task_id"] ep.mkdir(parents=True, exist_ok=True) try: r = run(t, bases[i % len(bases)], args, ep) except Exception as e: r = {"task_id": t["task_id"], "base": bases[i % len(bases)], "answer": None, "status": "crash", "rounds": 0, "error": f"{type(e).__name__}: {str(e)[:300]}", "record": [], "log": []} # a harness or model-server failure is kept aside, unsaved, so a rerun (resume) repeats the task name = "session_failed.json" if r["status"] in ("crash", "endpoint_error") else "session.json" (ep / name).write_text(json.dumps(r, default=str)) print(f"{t['task_id']}: status={r['status']} rounds={r['rounds']} {r.get('seconds')}s " f"parse_fails={r.get('parse_fails')} action_errors={r.get('action_errors')} err={r['error']} " f"answer={str(r['answer'])[:80]!r}", flush=True) with ThreadPoolExecutor(args.workers) as ex: list(ex.map(worker, range(args.workers))) if __name__ == "__main__": main()