#!/usr/bin/env python3 """ Run OSWorld benchmark with the Eliza bridge agent. Routes ALL decision-making through the elizaOS TypeScript benchmark bridge (``packages/lifeops-bench/src/server.ts``); the legacy Python ``AgentRuntime`` path has been removed. Usage: # Single task (Chrome - Enable Do Not Track) python scripts/python/run_multienv_eliza.py \ --provider_name docker \ --observation_type screenshot_a11y_tree \ --model gemma-4-31b \ --max_steps 15 \ --result_dir ./results/eliza \ --task_id 030eeff7-b492-4218-b312-701ec99ee0cc # All tasks python scripts/python/run_multienv_eliza.py \ --provider_name docker \ --observation_type screenshot_a11y_tree \ --model gemma-4-31b \ --max_steps 15 \ --num_envs 5 \ --result_dir ./results/eliza # VMware on macOS python scripts/python/run_multienv_eliza.py \ --provider_name vmware \ --path_to_vm ~/Virtual\\ Machines.localized/Ubuntu.vmwarevm/Ubuntu.vmx \ --observation_type screenshot_a11y_tree \ --model gemma-4-31b \ --max_steps 15 \ --result_dir ./results/eliza """ from __future__ import annotations import argparse import asyncio import copy import datetime import json import logging import os import sys from typing import Any # Ensure the OSWorld root is on the Python path OSWORLD_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) if OSWORLD_ROOT not in sys.path: sys.path.insert(0, OSWORLD_ROOT) BENCHMARKS_ROOT = os.path.dirname(OSWORLD_ROOT) ELIZA_ADAPTER_ROOT = os.path.join(BENCHMARKS_ROOT, "eliza-adapter") if ELIZA_ADAPTER_ROOT not in sys.path: sys.path.insert(0, ELIZA_ADAPTER_ROOT) try: from desktop_env.desktop_env import DesktopEnv except ModuleNotFoundError as exc: DesktopEnv = None # type: ignore[assignment] DESKTOP_ENV_IMPORT_ERROR = exc else: DESKTOP_ENV_IMPORT_ERROR = None from lib_run_single import run_single_example from lib_results_logger import log_task_error logger = logging.getLogger("osworld.eliza.runner") _DELEGATE_HARNESSES = {"hermes", "openclaw", "smithers"} EDGE_VARIANTS = ( "Before acting, verify the visible target state once and avoid changing unrelated settings.", "If several windows are open, identify the correct application first before completing the task.", "Preserve existing user files and preferences except for the exact change requested.", "Use the most direct GUI route available and stop immediately once the requested state is reached.", "Check for modal dialogs, popups, or permission prompts before assuming the task is blocked.", "Prefer stable menu labels and visible controls over keyboard shortcuts when both are available.", "After completing the action, inspect the relevant screen area to confirm the final state.", "If a required file or page is already open, continue from that state instead of reopening it.", "Handle slow UI updates by waiting for the visible result before issuing the next action.", "Do not perform cleanup, browsing, or formatting beyond what the instruction explicitly requires.", ) def _selected_delegate_harness() -> str: return ( os.environ.get("ELIZA_BENCH_HARNESS") or os.environ.get("BENCHMARK_HARNESS") or "" ).strip().lower() def _effective_harness_label() -> str: return _selected_delegate_harness() or "eliza" def _run_mode_label(args: argparse.Namespace) -> str: return "smoke_dry_run" if args.dry_run else "real_vm" def _summary_agent_label(args: argparse.Namespace) -> str: harness = _effective_harness_label() return f"{harness}-dry-run-smoke" if args.dry_run else harness def _task_agent_artifact_fields() -> dict[str, str]: return { "benchmark_task_agent": os.environ.get("BENCHMARK_TASK_AGENT", ""), "acp_default_agent": os.environ.get("ELIZA_ACP_DEFAULT_AGENT", ""), "default_agent_type": os.environ.get("ELIZA_DEFAULT_AGENT_TYPE", ""), "agent_selection_strategy": os.environ.get( "ELIZA_AGENT_SELECTION_STRATEGY", "" ), } def _configure_bridge_model_env(model: str) -> None: model_name = (model or "").strip() if not model_name: return os.environ["BENCHMARK_MODEL_NAME"] = model_name for key in ( "MODEL_NAME", "OPENAI_LARGE_MODEL", "OPENAI_SMALL_MODEL", "GROQ_LARGE_MODEL", "GROQ_SMALL_MODEL", "OPENROUTER_LARGE_MODEL", "OPENROUTER_SMALL_MODEL", "CEREBRAS_LARGE_MODEL", "CEREBRAS_SMALL_MODEL", ): os.environ.setdefault(key, model_name) def should_start_eliza_server() -> bool: """True when this run needs the TS Eliza benchmark server.""" return ( _selected_delegate_harness() not in _DELEGATE_HARNESSES and not os.environ.get("ELIZA_BENCH_URL") ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run OSWorld with Eliza agent") # VM / Environment parser.add_argument("--provider_name", type=str, default="docker", choices=["vmware", "docker", "virtualbox", "aws"], help="VM provider") parser.add_argument("--path_to_vm", type=str, default=None, help="Path to VMware .vmx file (VMware provider only)") parser.add_argument("--region", type=str, default=None, help="Cloud region (AWS/Azure)") parser.add_argument("--headless", action="store_true", help="Run VMs in headless mode") parser.add_argument("--snapshot_name", type=str, default="init_state", help="VM snapshot to revert to") # Agent configuration parser.add_argument("--model", type=str, default="gemma-4-31b", help="LLM model to use (e.g., gemma-4-31b)") parser.add_argument("--observation_type", type=str, default="screenshot_a11y_tree", choices=["screenshot", "a11y_tree", "screenshot_a11y_tree"], help="Observation type") parser.add_argument("--action_space", type=str, default="pyautogui", choices=["pyautogui", "computer_13"], help="Action space format") parser.add_argument("--max_steps", type=int, default=15, help="Max steps per task") parser.add_argument("--temperature", type=float, default=0.5) parser.add_argument("--max_tokens", type=int, default=2048) parser.add_argument("--max_trajectory_length", type=int, default=5) parser.add_argument("--a11y_tree_max_tokens", type=int, default=500) # Execution parser.add_argument("--result_dir", type=str, default="./results/eliza", help="Directory to store results") parser.add_argument("--task_id", type=str, default=None, help="Run a specific task by ID") parser.add_argument("--domain", type=str, default=None, help="Run tasks from a specific domain (chrome, gimp, etc.)") parser.add_argument("--max_tasks", type=int, default=None, help="Limit number of tasks to run") parser.add_argument("--num_envs", type=int, default=1, help="Number of parallel VMs") parser.add_argument("--sleep_after_execution", type=float, default=3.0, help="Sleep after each action execution") parser.add_argument("--dry_run", action="store_true", help="Run one cheap in-process smoke task without starting VMs or the Eliza server") parser.add_argument("--expand-scenarios", action="store_true", help="Run ten deterministic instruction-pressure variants per selected task") parser.add_argument("--count-scenarios", action="store_true", help="Print base/edge/total task counts for the selected task set") parser.add_argument("--validate-scenarios", action="store_true", help="Validate selected task set and optional expansion before running") return parser.parse_args() def load_tasks(args: argparse.Namespace) -> list[dict[str, object]]: """Load task definitions from OSWorld evaluation examples.""" test_all_path = os.path.join(OSWORLD_ROOT, "evaluation_examples", "test_all.json") with open(test_all_path) as f: test_all = json.load(f) tasks: list[dict[str, object]] = [] if args.task_id: # Load a specific task for domain, task_ids in test_all.items(): if args.task_id in task_ids: task_path = os.path.join( OSWORLD_ROOT, "evaluation_examples", "examples", domain, f"{args.task_id}.json" ) with open(task_path) as f: task = json.load(f) tasks.append(task) break if not tasks: raise ValueError(f"Task {args.task_id} not found in any domain") return tasks # Load by domain or all domains = [args.domain] if args.domain else list(test_all.keys()) for domain in domains: if domain not in test_all: logger.warning("Domain '%s' not found in test_all.json", domain) continue for task_id in test_all[domain]: task_path = os.path.join( OSWORLD_ROOT, "evaluation_examples", "examples", domain, f"{task_id}.json" ) if not os.path.exists(task_path): logger.warning("Task file not found: %s", task_path) continue with open(task_path) as f: task = json.load(f) tasks.append(task) if args.max_tasks: tasks = tasks[: args.max_tasks] return tasks def load_tasks_for_run(args: argparse.Namespace) -> list[dict[str, object]]: """Load benchmark tasks, or a synthetic task for dry-run smoke tests.""" if args.dry_run: total = max(1, int(args.max_tasks or 1)) return [ { "id": f"osworld_eliza_dry_run_{index + 1}", "snapshot": "dry_run", "instruction": ( "Dry-run smoke task for OSWorld harness wiring " f"({index + 1}/{total})." ), } for index in range(total) ] return load_tasks(args) def expand_tasks(tasks: list[dict[str, object]]) -> list[dict[str, object]]: expanded: list[dict[str, object]] = list(tasks) for task_index, task in enumerate(tasks): for variant_index, variant in enumerate(EDGE_VARIANTS): edge = copy.deepcopy(task) base_id = str(edge.get("id") or f"osworld_task_{task_index + 1}") edge["id"] = f"{base_id}__edge_{variant_index + 1:02d}" instruction = str(edge.get("instruction") or "") edge["instruction"] = f"{instruction}\n\nEdge condition: {variant}" edge["edge_variant"] = { "base_task_id": base_id, "variant_index": variant_index + 1, "description": variant, } expanded.append(edge) return expanded def count_tasks(tasks: list[dict[str, object]], include_edge_scenarios: bool = False) -> dict[str, int]: base = len(tasks) edge = base * len(EDGE_VARIANTS) if include_edge_scenarios else 0 return { "base": base, "edge": edge, "edge_multiplier": len(EDGE_VARIANTS) if include_edge_scenarios else 0, "total": base + edge, } def validate_tasks(tasks: list[dict[str, object]], include_edge_scenarios: bool = False) -> None: if not tasks: raise ValueError("OSWorld selected task set is empty") for index, task in enumerate(tasks): if not isinstance(task.get("id"), str) or not str(task.get("id")).strip(): raise ValueError(f"OSWorld task {index} missing id") if not isinstance(task.get("instruction"), str) or not str(task.get("instruction")).strip(): raise ValueError(f"OSWorld task {index} missing instruction") if include_edge_scenarios: expanded = expand_tasks(tasks) expected = len(tasks) * (len(EDGE_VARIANTS) + 1) if len(expanded) != expected: raise ValueError(f"expanded OSWorld tasks has {len(expanded)} tasks, expected {expected}") ids = [str(task.get("id")) for task in expanded] if len(ids) != len(set(ids)): raise ValueError("expanded OSWorld tasks has duplicate ids") def create_eliza_agent(args: argparse.Namespace) -> object: """Create and initialize the elizaOS bridge OSWorld agent. All decision-making is routed through the elizaOS TypeScript benchmark bridge — the legacy Python ``AgentRuntime`` path has been removed. """ _configure_bridge_model_env(args.model) server_manager = None if should_start_eliza_server(): from eliza_adapter.server_manager import ElizaServerManager server_manager = ElizaServerManager() server_manager.start() os.environ["ELIZA_BENCH_TOKEN"] = server_manager.token os.environ.setdefault( "ELIZA_BENCH_URL", f"http://localhost:{server_manager.port}" ) from eliza_adapter.osworld import ElizaBridgeOSWorldAgent agent = ElizaBridgeOSWorldAgent( platform="ubuntu", model=args.model, max_tokens=args.max_tokens, temperature=args.temperature, action_space=args.action_space, observation_type=args.observation_type, max_trajectory_length=args.max_trajectory_length, a11y_tree_max_tokens=args.a11y_tree_max_tokens, max_steps=args.max_steps, client_password="password", screen_width=1920, screen_height=1080, ) if server_manager is not None: setattr(agent, "_eliza_server_manager", server_manager) try: asyncio.run(agent.async_init()) except Exception: if server_manager is not None: server_manager.stop() raise return agent class _DryRunController: def start_recording(self) -> None: pass def end_recording(self, _path: str) -> None: pass class _DryRunEnv: vm_ip = "127.0.0.1" def __init__(self) -> None: self.controller = _DryRunController() self._step_count = 0 def reset(self, task_config: dict[str, object] | None = None) -> dict[str, object]: self._step_count = 0 return self._get_obs() def _get_obs(self) -> dict[str, object]: return { "screenshot": b"dry-run-screenshot", "accessibility_tree": "role\tname\nbutton\tOK", "terminal": None, "instruction": "dry run", } def step(self, action: str, sleep_after_execution: float = 0.0) -> tuple[dict[str, object], float, bool, dict[str, object]]: self._step_count += 1 return self._get_obs(), 0.0, True, { "dry_run": True, "action": action, "sleep_after_execution": sleep_after_execution, } def evaluate(self) -> float: return 1.0 def close(self) -> None: pass class _DryRunAgent: def reset(self, *_args: object, **_kwargs: object) -> None: pass def predict(self, instruction: str, obs: dict[str, object]) -> tuple[str, list[str]]: return ( f"Dry run handled instruction: {instruction}", ["pyautogui.press('esc')"], ) def run_benchmark(args: argparse.Namespace) -> dict[str, object]: """Run the OSWorld benchmark with the Eliza agent.""" base_tasks = load_tasks_for_run(args) if getattr(args, "validate_scenarios", False): validate_tasks(base_tasks, include_edge_scenarios=bool(args.expand_scenarios)) scenario_counts = count_tasks(base_tasks, include_edge_scenarios=bool(args.expand_scenarios)) if getattr(args, "count_scenarios", False): print(json.dumps(scenario_counts, sort_keys=True)) tasks = expand_tasks(base_tasks) if args.expand_scenarios else base_tasks logger.info("Loaded %d tasks", len(tasks)) # Create agent agent = _DryRunAgent() if args.dry_run else create_eliza_agent(args) server_manager = getattr(agent, "_eliza_server_manager", None) # Create environment env_kwargs = { "provider_name": args.provider_name, "action_space": args.action_space, "headless": args.headless, "require_a11y_tree": args.observation_type in ("a11y_tree", "screenshot_a11y_tree"), "require_terminal": False, } if args.path_to_vm: env_kwargs["path_to_vm"] = args.path_to_vm if args.region: env_kwargs["region"] = args.region env: Any | None = None original_sleep: Any | None = None # Results scores: list[float] = [] results: list[dict[str, object]] = [] timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") try: if args.dry_run: env = _DryRunEnv() import lib_run_single original_sleep = lib_run_single.time.sleep lib_run_single.time.sleep = lambda _seconds: None else: if DesktopEnv is None: raise RuntimeError( f"OSWorld dependencies are not installed: {DESKTOP_ENV_IMPORT_ERROR}" ) env = DesktopEnv(**env_kwargs) for i, task in enumerate(tasks): task_id = task["id"] domain = task.get("snapshot", "unknown") instruction = task.get("instruction", "") logger.info("=" * 60) logger.info("Task %d/%d: %s (%s)", i + 1, len(tasks), task_id, domain) logger.info("Instruction: %s", instruction) logger.info("=" * 60) # Create result directory example_result_dir = os.path.join( args.result_dir, args.action_space, args.observation_type, args.model.replace("/", "_"), str(domain), str(task_id), ) os.makedirs(example_result_dir, exist_ok=True) try: run_single_example( agent=agent, env=env, example=task, max_steps=args.max_steps, instruction=instruction, args=args, example_result_dir=example_result_dir, scores=scores, ) result_val = scores[-1] if scores else 0.0 results.append({ "task_id": task_id, "domain": domain, "instruction": instruction, "score": result_val, "result_dir": example_result_dir, }) logger.info("Task %s: score=%.2f", task_id, result_val) except Exception as e: logger.error("Task %s failed with error: %s", task_id, e, exc_info=True) with open(os.path.join(example_result_dir, "result.txt"), "w", encoding="utf-8") as f: f.write("0.0\n") with open(os.path.join(example_result_dir, "traj.jsonl"), "a", encoding="utf-8") as f: f.write(json.dumps({"Error": str(e)})) f.write("\n") log_task_error(task, str(e), example_result_dir, args) results.append({ "task_id": task_id, "domain": domain, "instruction": instruction, "score": 0.0, "error": str(e), }) scores.append(0.0) # Summary total = len(scores) passed = sum(1 for s in scores if s > 0) avg_score = sum(scores) / total if total > 0 else 0 summary = { "model": args.model, "agent": _summary_agent_label(args), "harness": _effective_harness_label(), **_task_agent_artifact_fields(), "run_mode": _run_mode_label(args), "smoke": bool(args.dry_run), "observation_type": args.observation_type, "action_space": args.action_space, "total_tasks": total, "scenario_counts": scenario_counts, "include_edge_scenarios": bool(args.expand_scenarios), "passed_tasks": passed, "overall_success_rate": avg_score, "timestamp": timestamp, "results": results, } # Save summary summary_path = os.path.join(args.result_dir, f"osworld-eliza-results-{timestamp}.json") os.makedirs(os.path.dirname(summary_path), exist_ok=True) with open(summary_path, "w") as f: json.dump(summary, f, indent=2, default=str) logger.info("=" * 60) logger.info("BENCHMARK COMPLETE") logger.info(" Total tasks: %d", total) logger.info(" Passed: %d", passed) logger.info(" Success rate: %.2f%%", avg_score * 100) logger.info(" Results: %s", summary_path) logger.info("=" * 60) return summary finally: if original_sleep is not None: import lib_run_single lib_run_single.time.sleep = original_sleep if env is not None: env.close() if server_manager is not None: server_manager.stop() def main() -> None: logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", ) args = parse_args() # Validate if args.provider_name == "vmware" and not args.path_to_vm: logger.error("VMware provider requires --path_to_vm") sys.exit(1) run_benchmark(args) if __name__ == "__main__": main()