elizaos--eliza/packages/benchmarks/visualwebbench/cli.py

"""CLI for VisualWebBench."""

from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import sys
from datetime import datetime
from pathlib import Path

from benchmarks.visualwebbench.runner import VisualWebBenchRunner
from benchmarks.visualwebbench.dataset import VisualWebBenchDataset
from benchmarks.visualwebbench.types import (
    VISUALWEBBENCH_TASK_TYPES,
    VisualWebBenchConfig,
    VisualWebBenchTaskType,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger(__name__)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="VisualWebBench benchmark for ElizaOS")
    # Source ------------------------------------------------------------------
    parser.add_argument(
        "--use-sample-tasks",
        action="store_true",
        help=(
            "Use the bundled labeled JSONL helper (one row per subtask, no images). "
            "Offline-CI only — scores are not comparable to upstream."
        ),
    )
    parser.add_argument("--fixture-path", type=str, default=None)
    parser.add_argument("--hf-repo", type=str, default="visualwebbench/VisualWebBench")
    parser.add_argument("--split", type=str, default="test")
    parser.add_argument(
        "--image-cache-dir",
        type=str,
        default=None,
        help="Where to materialise HF screenshots as PNG (default: ~/.cache/elizaos/visualwebbench/images).",
    )
    parser.add_argument(
        "--no-image-cache",
        dest="cache_images_to_disk",
        action="store_false",
        default=True,
        help="Keep image bytes in memory instead of caching to disk.",
    )

    # Tasks -------------------------------------------------------------------
    parser.add_argument(
        "--task-types",
        type=str,
        default=",".join(t.value for t in VISUALWEBBENCH_TASK_TYPES),
    )
    parser.add_argument("--max-tasks", type=int, default=None)
    parser.add_argument("--output", type=str, default=None)

    # Agent -------------------------------------------------------------------
    parser.add_argument(
        "--mock",
        action="store_true",
        help=(
            "Run the offline oracle agent that reads task.answer directly. "
            "Use ONLY for smoke tests and CI — guarantees 100% score, so it is "
            "never appropriate for real measurements."
        ),
    )
    parser.add_argument(
        "--provider",
        type=str,
        choices=[
            "eliza",
            "eliza-bridge",
            "eliza-ts",
            "eliza-app-harness",
            "eliza-app",
            "eliza-browser-app",
            "app-harness",
            "local-eliza",
            "local_eliza",
            "eliza-local",
            "eliza_local",
        ],
        default="eliza",
        help=(
            "Eliza integration mode (ignored when --mock is set). "
            "Use local-eliza to drive the on-device eliza-1 VLM via llama-mtmd-cli."
        ),
    )
    parser.add_argument("--model", type=str, default=None)
    parser.add_argument("--temperature", type=float, default=0.0)
    parser.add_argument("--timeout", type=int, default=120000)
    parser.add_argument("--bbox-iou-threshold", type=float, default=0.5)

    # App harness passthrough (unchanged) ------------------------------------
    parser.add_argument("--app-harness-script", type=str, default=None)
    parser.add_argument("--app-harness-runtime", type=str, default="bun")
    parser.add_argument(
        "--app-harness-no-launch",
        dest="app_harness_no_launch",
        action="store_true",
        default=True,
    )
    parser.add_argument(
        "--app-harness-launch",
        dest="app_harness_no_launch",
        action="store_false",
    )
    parser.add_argument(
        "--app-harness-prompt-via-ui",
        dest="app_harness_prompt_via_ui",
        action="store_true",
        default=True,
    )
    parser.add_argument(
        "--app-harness-prompt-via-api",
        dest="app_harness_prompt_via_ui",
        action="store_false",
    )
    parser.add_argument("--app-harness-dry-run", action="store_true")
    parser.add_argument("--app-harness-api-base", type=str, default=None)
    parser.add_argument("--app-harness-ui-url", type=str, default=None)
    parser.add_argument("--app-harness-poll-interval", type=int, default=None)

    parser.add_argument("--no-traces", action="store_true")
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--verbose", action="store_true")
    parser.add_argument(
        "--expand-scenarios",
        action="store_true",
        help="Add 10 realistic edge variants for every loaded VisualWebBench task.",
    )
    parser.add_argument(
        "--count-scenarios",
        action="store_true",
        help="Print loaded task counts and exit.",
    )
    parser.add_argument(
        "--validate-scenarios",
        action="store_true",
        help="Validate loaded tasks and exit.",
    )
    return parser.parse_args()


def create_config(args: argparse.Namespace) -> VisualWebBenchConfig:
    if args.output:
        output_dir = args.output
    else:
        ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        output_dir = f"./benchmark_results/visualwebbench/{ts}"

    task_types = _parse_task_types(args.task_types)
    use_sample = bool(args.use_sample_tasks)

    return VisualWebBenchConfig(
        output_dir=output_dir,
        fixture_path=Path(args.fixture_path).resolve() if args.fixture_path else None,
        hf_repo=args.hf_repo,
        split=args.split,
        task_types=task_types,
        max_tasks=args.max_tasks,
        mock=bool(args.mock),
        use_huggingface=not use_sample,
        use_sample_tasks=use_sample,
        cache_images_to_disk=bool(args.cache_images_to_disk),
        image_cache_dir=Path(args.image_cache_dir).resolve() if args.image_cache_dir else None,
        provider=args.provider,
        model=args.model,
        temperature=args.temperature,
        timeout_ms=max(1000, args.timeout),
        bbox_iou_threshold=args.bbox_iou_threshold,
        save_traces=not args.no_traces,
        app_harness_script=Path(args.app_harness_script).resolve()
        if args.app_harness_script
        else None,
        app_harness_runtime=args.app_harness_runtime,
        app_harness_no_launch=args.app_harness_no_launch,
        app_harness_prompt_via_ui=args.app_harness_prompt_via_ui,
        app_harness_dry_run=args.app_harness_dry_run,
        app_harness_api_base=args.app_harness_api_base,
        app_harness_ui_url=args.app_harness_ui_url,
        app_harness_poll_interval_ms=args.app_harness_poll_interval,
        verbose=args.verbose,
        include_edge_scenarios=args.expand_scenarios,
    )


async def load_dataset_for_inventory(config: VisualWebBenchConfig) -> VisualWebBenchDataset:
    dataset = VisualWebBenchDataset(
        fixture_path=config.fixture_path,
        hf_repo=config.hf_repo,
        split=config.split,
        task_types=config.task_types,
        image_cache_dir=config.image_cache_dir,
        cache_images_to_disk=config.cache_images_to_disk,
    )
    await dataset.load(
        use_huggingface=config.use_huggingface,
        use_sample_tasks=config.use_sample_tasks,
        max_tasks=config.max_tasks,
        include_edge_scenarios=config.include_edge_scenarios,
    )
    return dataset


async def run(config: VisualWebBenchConfig) -> dict[str, object]:
    runner = VisualWebBenchRunner(config)
    report = await runner.run_benchmark()
    return {
        "total_tasks": report.total_tasks,
        "overall_accuracy": report.overall_accuracy,
        "by_task_type": report.by_task_type,
        "average_latency_ms": report.average_latency_ms,
        "summary": report.summary,
        "output_dir": config.output_dir,
    }


def main() -> int:
    args = parse_args()
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
    config = create_config(args)
    if args.count_scenarios or args.validate_scenarios:
        try:
            dataset = asyncio.run(load_dataset_for_inventory(config))
        except Exception as exc:
            logger.error("VisualWebBench inventory failed: %s", exc)
            if args.json:
                print(json.dumps({"error": str(exc)}, indent=2))
            return 1
        counts = dataset.count_scenarios()
        if args.validate_scenarios:
            errors = dataset.validate_scenarios()
            payload = {"ok": not errors, **counts}
            if errors:
                payload["errors"] = errors[:50]
                payload["error_count"] = len(errors)
            print(json.dumps(payload, indent=2))
            return 0 if not errors else 1
        print(json.dumps(counts, indent=2))
        return 0
    server_mgr = None
    provider = (config.provider or "").strip().lower()
    needs_eliza_bridge = (
        not config.mock
        and provider in {"eliza", "eliza-bridge", "eliza-ts"}
        and os.environ.get("BENCHMARK_HARNESS", "").strip().lower()
        not in {"hermes", "openclaw"}
    )

    try:
        if needs_eliza_bridge and (
            not os.environ.get("ELIZA_BENCH_URL") or not os.environ.get("ELIZA_BENCH_TOKEN")
        ):
            from eliza_adapter.server_manager import ElizaServerManager

            server_mgr = ElizaServerManager()
            server_mgr.start()
            os.environ["ELIZA_BENCH_TOKEN"] = server_mgr.token
            os.environ["ELIZA_BENCH_URL"] = f"http://localhost:{server_mgr.port}"
        results = asyncio.run(run(config))
    except KeyboardInterrupt:
        logger.info("VisualWebBench interrupted")
        return 130
    except Exception as exc:
        logger.error("VisualWebBench failed: %s", exc)
        if args.json:
            print(json.dumps({"error": str(exc)}, indent=2))
        return 1
    finally:
        if server_mgr is not None:
            server_mgr.stop()

    if args.json:
        print(json.dumps(results, indent=2, default=str))
    else:
        print("\n" + "=" * 60)
        print("VisualWebBench Results")
        print("=" * 60)
        print(f"Tasks: {results['total_tasks']}")
        print(f"Overall Score: {float(results['overall_accuracy']) * 100:.1f}")
        summary = results.get("summary", {}) or {}
        if isinstance(summary, dict):
            print(f"ROUGE mean: {float(summary.get('rouge_score', 0)) * 100:.1f}")
            print(f"F1 (webqa): {float(summary.get('f1_score', 0)) * 100:.1f}")
            print(f"Choice accuracy: {float(summary.get('choice_accuracy', 0)) * 100:.1f}")
        print(f"Results saved to: {config.output_dir}")
        print("=" * 60)
    return 0


def _parse_task_types(raw: str) -> tuple[VisualWebBenchTaskType, ...]:
    values: list[VisualWebBenchTaskType] = []
    for part in raw.split(","):
        value = part.strip()
        if not value:
            continue
        try:
            values.append(VisualWebBenchTaskType(value))
        except ValueError as exc:
            allowed = ", ".join(t.value for t in VISUALWEBBENCH_TASK_TYPES)
            raise argparse.ArgumentTypeError(
                f"Unknown VisualWebBench task type {value!r}; expected one of {allowed}"
            ) from exc
    return tuple(values) or VISUALWEBBENCH_TASK_TYPES


if __name__ == "__main__":
    raise SystemExit(main())