Blame for src/miniwebagent/evaluate.py

"""Run a WebMix arm on datagen-style tasks through the browser-use harness, served by vLLM.

    python -m miniwebagent.evaluate --tasks data/webmix/valgen/runs/val1/tasks.jsonl --arm base --out data/webmix/eval/base
    python -m miniwebagent.evaluate --tasks ... --arm oracle --adapters drag,io,select,rest

Arms: `base` (no adapter), `pooled` (one adapter for every step), `oracle` (the adapter of the task's own family;
for a task with a prefix macro, the prefix family first, switched at the model's macro_done; data validation),
`agent_planner` (the planner agent, miniwebagent.planner_agent, delegating each step to a family specialist). The model
name of each request picks the adapter on the vLLM server (src/miniwebagent/scripts/serve.sh). Each episode starts like a replay (miniwebagent.harness.prepare) and is graded by the same
backend gate the datagen trajectories passed (miniwebgen.checks.gate); the flash macro judge can be run on
the saved episodes later.

`--human` evaluates the human-written tasks (data/annotations/*/*/task.json) on the held-out sites instead: the
same harness and arms, started like evaluation.run_agent_verify (recorded start page; logged out for a login
macro; the portal home + an app directory for a cross-site task), graded by the task's own verifier.json over
this session's recorder stream + server log (a task the generator never wrote, unlike validation v1/v2).

`--record-rows` also writes each passing episode's steps as training rows (the exact request browser-use sent and
the model's reply; <out>/<episode>/rows.jsonl + screenshots, the miniwebagent.train layout): rejection sampling on
human tasks (`--human --human-split train --repeats N --temperature 0.7`) for the reasoning adapter.

Each episode also saves its grading inputs (grade_inputs.json: the session's request log, data changes,
clipboard, action events, answer, seen text). `--regrade DIR` recomputes every grade from them: granite
runs grade without the QA chain's LLM tier (no credentials there), and are re-graded here.
"""
from __future__ import annotations

import argparse
import base64
import asyncio
import json
import os
import time
from collections import Counter
from pathlib import Path
from types import SimpleNamespace

from miniwebgen import browser as B
from miniwebgen import checks
from miniwebgen.driver import _VISIBLE_TEXT
from miniwebgen.viewer import FAMILY_OF
from miniwebagent import harness as H
from miniwebagent.eval_viewer import copy_screenshots

ADAPTER_OF_FAMILY = {"Drag & gesture": "drag", "Out-of-page I/O": "io", "Discrete selection": "select",
                     "Text entry": "rest", "Form transaction": "rest", "Navigation": "rest", "Reasoning base": "reasoning"}
# Reasoning steps (and the final answer they end in) go to their own specialist: a `reasoning` adapter when one is
# served, else the untouched base model (user, 2026-09-30: an adapter trained without reasoning rows lost question
# answering, 2/12 on validation v2 vs base's 4/12; routing reasoning away from it avoids that forgetting).
REASONING = "Reasoning base"


def family_plan(task):
    """Macro sequence of a task: a human task's annotated macros in order; a datagen task's prefix macro
    (mid-chain start) first, then its own."""
    if task.get("kind") == "human":
        return list(task["macros"])
    pre = ((task.get("start") or {}).get("prefix") or {}).get("macro")
    return ([pre] if pre else []) + [task["macro"]]


MAX_GATES = 3


def premature(plan_left, seg_actions, fam):
    """A `done` is premature when planned steps remain after the current one, or when the current step has not been
    acted on at all (the specialists' trained habit is macro_done immediately followed by done, which skipped the
    re-planned step; 2026-09-30: with the first rule only, 4 gates fired on 56 human tasks while 28 stopped early).
    A reasoning step's done IS its action (the answer), so it counts as acting."""
    return plan_left > 1 or (plan_left == 1 and seg_actions == 0 and fam != REASONING)


def router_note(families):
    """What the router tells the specialist when it gates a done: the planned steps still left, described."""
    from miniwebagent.planner import FAMILY_HINTS
    left = [f"{f} ({FAMILY_HINTS.get(f, '').rstrip('.')})" if FAMILY_HINTS.get(f) else f for f in families]
    return ("The task is not finished yet. The router's plan still has " + str(len(left)) + " step(s): " + "; then ".join(left)
            + ". Re-read the task, find the part that is not done yet, and do it; call done only when every part is "
              "finished.")


def gate_done(out, remaining, gates):
    """Done-gating (2026-09-30): the router, not the specialist, decides when the task is complete. A premature `done`
    (see premature) becomes `macro_done` (the next step's specialist takes over), at most MAX_GATES times per episode.
    -> True when the reply was rewritten."""
    acts = getattr(out.completion, "action", None) or []
    if not remaining or gates >= MAX_GATES:
        return False
    for i, a in enumerate(acts):
        if "done" in a.model_dump(exclude_none=True):
            acts[i] = type(a).model_validate({"macro_done": {}})
            return True
    return False


class Routed:
    """Chooses the served model (base or an adapter) for each request; advances past a macro on macro_done. With
    `gate`, a done before the last planned macro is turned into macro_done (gate_done)."""

    def __init__(self, llm, names, gate=False, fams=None, bs=None, note=False):
        self.llm, self.names, self.k, self.trace, self.gates, self.seg = llm, names, 0, [], 0, 0
        fams = fams or [None] * len(names)
        inner = llm.ainvoke

        async def ainvoke(messages, output_format=None, **kw):
            llm.model = self.names[min(self.k, len(self.names) - 1)]
            out = await inner(messages, output_format, **kw)
            left = max(0, len(self.names) - self.k)            # planned steps not yet closed, the current one included
            fam = fams[min(self.k, len(fams) - 1)] if fams else None
            gated = gate and gate_done(out, premature(left, self.seg, fam), self.gates)
            self.gates += gated
            if gated and note and bs is not None:
                H.ROUTER_NOTES[id(bs)] = router_note(fams[self.k:] if fams and fams[0] else [])
            acts = [a.model_dump(exclude_none=True) for a in getattr(out.completion, "action", []) or []]
            self.trace.append({"model": llm.model, "action": acts, **({"gated": True} if gated else {})})
            if any("macro_done" in a for a in acts):
                self.k += 1
                self.seg = 0
            else:
                self.seg += sum(1 for a in acts if not {"done", "macro_done"} & set(a))
            return out
        llm.ainvoke = ainvoke


def action_events(trace):
    """The model's actions as human-schema action events (the verifier's action_included checks)."""
    kinds = {"input": ("type", "text"), "send_keys": ("key", "keys"), "select_dropdown": ("select", "text"),
             "click": ("click", None), "done": ("answer", "text")}
    out = []
    for t in trace:
        for a in t["action"]:
            (name, p), = a.items()
            kind, field = kinds.get(name, (name, None))
            ev = {"type": "action", "action": kind, "target": ""}
            if field:
                ev["value"] = (p or {}).get(field)
            out.append(ev)
    return out


def history_actions(hist):
    """The model's actions from browser-use histories, in the router-trace form ({name: params}). For agent_planner,
    whose trace holds one record per delegated step without per-step actions (2026-10-02: grading such arms on
    generated tasks raised KeyError 'action' before grade_inputs.json was written)."""
    acts = []
    for item in getattr(hist, "history", []) or []:
        for a in getattr(getattr(item, "model_output", None), "action", None) or []:
            acts.append(a.model_dump(exclude_none=True))
    return acts


def human_tasks(sites=None, split="test", macros=None):
    """The human-written tasks on `sites` (default: the `split` sites of data/datagen/site_split.json; a task counts
    as held out when any of its sites is) as evaluation dicts, optionally only those with one of `macros`. Start page
    and login rule as evaluation.run_agent_verify."""
    import ast
    from miniweb.annotation.storage import ANNOTATIONS_DIR
    macros_filter = macros
    from miniweb.evaluation.run_agent_verify import locate_task, recorded_start_url
    from miniwebgen import config
    sp = json.loads(config.SPLIT_PATH.read_text())
    sites = set(sites or (sp["test"] + sp["train"] if split == "all" else sp[split]))   # all: every human task (376)
    test = set(sp["test"])
    out = []
    for tj in sorted(ANNOTATIONS_DIR.glob("*/*/task.json")):
        try:
            raw = json.loads(tj.read_text())
        except (OSError, ValueError):
            continue
        if not (tj.parent / "verifier.json").exists():
            continue
        raw_sites = raw.get("sites")
        try:
            raw_sites = ast.literal_eval(raw_sites) if isinstance(raw_sites, str) else raw_sites
        except (ValueError, SyntaxError):
            raw_sites = None
        all_sites = {x["id"] if isinstance(x, dict) else x for x in (raw_sites or [raw.get("site")])}
        if split == "train" and not sites.intersection(all_sites) or split == "train" and all_sites & test:
            continue
        if split != "train" and raw.get("site") not in sites:
            continue
        key = f"{tj.parent.parent.name}/{tj.parent.name}"
        tdir, task, verifier = locate_task(key)
        lit = lambda v: ast.literal_eval(v) if isinstance(v, str) and v[:1] in "[{" else v       # noqa: E731
        macros = [m for m in (lit(task.get("macros")) or []) if m]
        if macros_filter and not set(macros_filter) & set(macros):
            continue
        task_sites = [x["id"] if isinstance(x, dict) else x for x in (lit(task.get("sites")) or [task["site"]])]
        subs = lit(task.get("macro_subtasks")) or {}
        spans = lit(task.get("macro_spans")) or {}
        order = sorted(spans, key=lambda m: spans[m][0]) if spans else macros
        gold = [(FAMILY_OF.get(m.split("#")[0], "Other"), str(subs.get(m) or "").strip()) for m in order]
        # a report is the task's deliverable: reasoning subtasks last (span order can put it first, when the annotator
        # read part of the answer early; "Report critical open count and MeridianFlow open count" needs a later page)
        gold = [g for g in gold if g[0] != REASONING] + [g for g in gold if g[0] == REASONING]
        out.append({"kind": "human", "task_id": tdir.name, "key": key, "site": task["site"], "sites": task_sites,
                    "subtasks": gold if gold and all(t for _, t in gold) else None,
                    "macro": macros[-1] if macros else "?", "macros": macros, "instruction": task.get("instruction", ""),
                    # agent_start_url: set when the recording's first page is wrong (it showed the answer) and
                    # the task is being re-recorded; it wins over the recording until the new one lands
                    "start_hint": task.get("agent_start_url") or recorded_start_url(tdir) or task.get("starting_url"),
                    "verifier": verifier,
                    "needs_login": "authenticate_by_form" in (verifier.get("macros") or {})})
    return out


async def prepare_human(bs, base, task):
    """-> (prompt, start url) after the session flags (logged out for a login task) and the start page load."""
    from urllib.parse import urlencode
    from miniweb.evaluation.run_agent_verify import app_directory, resolve_start_url
    flags = {"logout": "1"} if task["needs_login"] else {}
    await H.goto(bs, f"{base}/_admin/session-flags" + ("?" + urlencode(flags) if flags else ""))
    if len(task["sites"]) > 1:
        start, prompt = base + "/", task["instruction"] + app_directory(base, task["sites"])
    else:
        start, prompt = resolve_start_url(base, task["start_hint"], task["site"]), task["instruction"]
    await H.goto(bs, start)
    await asyncio.sleep(1.0)
    return prompt, start


async def prepare_proposed(bs, base, task):
    """miniwebgen.taskgen's proposed tasks (task-centric MiniWeb data, 2026-10-08): the default (logged-in) session on the
    task's start page. -> (prompt, start url)."""
    await H.goto(bs, f"{base}/_admin/session-flags")
    start = base + task["start_path"]
    await H.goto(bs, start)
    await asyncio.sleep(1.0)
    return task["instruction"], start


def write_proposed_rows(rows_dir, task, rec):
    """A proposed task's episode as training rows, whatever its outcome (miniwebgen.taskgen's judge picks the kept ones):
    <ep>/rows/{rows.jsonl, step_*.png, episode.json}, the layout of miniwebagent.om2w --record-rows. Only the agent's step
    calls (a JSON reply with an action) become rows; a row's family follows its first action (om2w.ROW_FAMILY)."""
    from miniwebagent.om2w import ROW_FAMILY
    from miniwebgen.replay import save_images
    rows_dir.mkdir(parents=True, exist_ok=True)
    rows = []
    for st in rec:
        try:
            act = json.loads(st["content"]).get("action") or [{}]
        except (ValueError, TypeError, AttributeError):
            continue
        name = next(iter(act[0]), "") if act and isinstance(act[0], dict) else ""
        i = len(rows)
        rows.append({"task_id": task["task_id"], "step": i, "segment": 0, "macro": None, "op": name,
                     "family": ROW_FAMILY.get(name, "Navigation"), "messages": save_images(st["messages"], rows_dir, i),
                     "target": st["content"]})
    (rows_dir / "rows.jsonl").write_text("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in rows))
    (rows_dir / "episode.json").write_text(json.dumps(
        {"task_id": task["task_id"], "instruction": task["instruction"], "site": task["site"],
         "start_path": task["start_path"], "steps": len(rows), "source": "taskgen"}, indent=1, ensure_ascii=False))
    return len(rows)


async def human_trajectory(bs, base):
    """verify_task's trajectory for this session: the recorder stream (actions + observations) with the server's
    request log merged in as network events (as evaluation.run_agent_verify.build_trajectory, per session)."""
    from miniweb.evaluation.trajectory import merge_server_log
    rec = (await H.admin_get(bs, base, "/_admin/record")).get("entries", []) or []
    log = (await H.admin_get(bs, base, "/_admin/log")).get("entries", []) or []
    return merge_server_log([e for e in rec if e.get("type") != "network"], log)


# The LLM tier of the verifier's answer checks (regex, exact and precise matchers come first) for every paper number:
# flash-lite (user, 2026-09-28). Set here, where every grading path passes (evaluate, fara_rq1 grade, grader_agreement),
# so no entry point falls back to the repo default (gemini-3.5-flash, kept for the annotation tool).
QA_MODEL = "gemini-3.5-flash-lite"


def grade_human(task, traj, answer):
    from miniweb.evaluation.verifiers import verify_task
    os.environ.setdefault("VERIFIER_JUDGE_MODEL", QA_MODEL)
    report = verify_task(task["verifier"], traj, answer or "", question=task["instruction"])
    return bool(report.get("passed")), str(report.get("summary") or report.get("by_macro") or "")[:300]


class _Hist:
    """AgentHistoryList-like view of a subtask run (one browser-use Agent per subtask, one browser session)."""

    def __init__(self, hists, answer):
        self.hists, self.answer = hists, answer
        self.history = [h for x in hists for h in x.history]

    def final_result(self):
        return self.answer

    def is_done(self):
        return bool(self.hists) and self.hists[-1].is_done()

    def errors(self):
        return [e for x in self.hists for e in x.errors()]

    def save_to_file(self, path):
        for i, x in enumerate(self.hists):
            x.save_to_file(Path(path).with_name(f"history_{i}.json"))


def sub_model(fam, args):
    """The specialist for one subtask: its family's adapter when served, else --fallback (reasoning: its own adapter,
    else the base model unless --reasoning-to-fallback)."""
    name = ADAPTER_OF_FAMILY.get(fam)
    if name in args.adapters:
        return name
    if fam == REASONING and not args.reasoning_to_fallback:
        return args.base_name
    return args.fallback or args.base_name


def arm_models(arm, task, base_name, adapters, pooled_name="pooled"):
    fams = [FAMILY_OF.get(m, "Other") for m in family_plan(task)]
    if arm == "base":
        return [base_name]
    if arm == "pooled":
        return [pooled_name]
    if arm == "oracle":
        names = [ADAPTER_OF_FAMILY.get(f, "rest") for f in fams]
        return [n if n in adapters else base_name for n in names]
    raise ValueError(arm)


def _record_client(llm, steps):
    """Wrap the ChatOpenAI client so every request (OpenAI-format messages) and reply content is kept in `steps`."""
    from types import SimpleNamespace
    real = llm.get_client()

    async def create(**kw):
        r = await real.chat.completions.create(**kw)
        steps.append({"model": kw.get("model"), "messages": kw.get("messages"), "content": r.choices[0].message.content})
        return r
    proxy = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create)))
    llm.get_client = lambda: proxy


async def run_episode(task, base, arm, args, out_dir, rep=0):
    from browser_use import Agent
    from miniwebgen.replay import _rebase
    human = task.get("kind") == "human"
    proposed = task.get("kind") == "proposed"          # miniwebgen.taskgen: no check here, judged afterwards
    if not (human or proposed):
        task = _rebase(task, base)       # values recorded with the generating server's origin -> this server's
        task["check"] = checks.relax_recorded_dates(task["check"], [task["instruction"], json.dumps(task.get("option"))])
    t0 = time.time()
    epn = task["task_id"] + (f"~r{rep}" if args.repeats > 1 else "")
    res = {"task_id": task["task_id"], "episode": epn, "macro": task["macro"], "site": task["site"], "arm": arm,
           "ok": False}
    rec = []
    bs = H.make_session()
    try:
        await bs.start()
        await H.enable_clipboard(bs)
        if human:
            prompt, start = await prepare_human(bs, base, task)
        elif proposed:
            prompt, start = await prepare_proposed(bs, base, task)
        else:
            await H.prepare(bs, base, task)
            prompt, start = task["instruction"], base + task["start"]["url"]
        llm = H.make_llm(args.base_name, base_url=args.vllm, temperature=args.temperature, max_tokens=args.max_tokens)
        if args.record_rows:
            _record_client(llm, rec)
        if arm == "agent_planner":
            router = SimpleNamespace(trace=[], plans=[])
        else:
            router = Routed(llm, arm_models(arm, task, args.base_name, args.adapters, args.pooled_name),
                            gate=(args.gate_done or args.gate_note) and arm == "oracle",
                            fams=[FAMILY_OF.get(m, "Other") for m in family_plan(task)] if arm == "oracle" else None,
                            bs=bs, note=args.gate_note)
        seen = []

        async def after_step(a):                 # viewport text of every screen (answer checks need the evidence seen)
            try:
                seen.append(await H.evaluate(bs, _VISIBLE_TEXT) or "")
            except Exception:
                pass
        seen.append(await H.evaluate(bs, _VISIBLE_TEXT) or "")
        if arm == "agent_planner":
            from miniwebagent.planner_agent import run_planner_agent
            hist, sub_trace, phist = await asyncio.wait_for(
                run_planner_agent(bs, task["instruction"], start, lambda f: sub_model(f, args), args.vllm,
                                  args.planner_name, context=prompt[len(task["instruction"]):], sub_cap=args.sub_cap,
                                  max_turns=args.planner_turns, on_end=after_step, temperature=args.temperature,
                                  version=args.agent_version),
                timeout=args.timeout)
            router.trace = sub_trace
            (out_dir / epn).mkdir(parents=True, exist_ok=True)
            phist.save_to_file(out_dir / epn / "planner_history.json")
        else:
            agent = Agent(task=H.instruction(prompt, start), llm=llm, browser_session=bs,
                          tools=H.make_tools(), max_failures=3, **H.AGENT_KWARGS)
            H.use_normalized_coordinates(bs)
            hist = await asyncio.wait_for(agent.run(max_steps=args.max_steps or H.MAX_STEPS,
                                                    on_step_start=lambda a: H.enable_clipboard(bs, a),
                                                    on_step_end=after_step), timeout=args.timeout)
        answer = hist.final_result()
        ep = out_dir / epn
        ep.mkdir(parents=True, exist_ok=True)
        try:
            if human:
                traj = await human_trajectory(bs, base)
                # + the backend evidence miniwebgen.taskgen's judge reads, so the judge can be validated on human tasks
                log, changes, clip = await H.backend_state(bs, base)
                (ep / "grade_inputs.json").write_text(json.dumps(
                    {"base": base, "traj": traj, "answer": answer, "log": log, "changes": changes, "clip": clip,
                     "seen": seen, "actions": history_actions(hist)}, default=str))
                ok, detail = grade_human(task, traj, answer)
            elif proposed:
                log, changes, clip = await H.backend_state(bs, base)
                (ep / "grade_inputs.json").write_text(json.dumps(
                    {"base": base, "answer": answer, "log": log, "changes": changes, "clip": clip, "seen": seen,
                     "actions": history_actions(hist)}, default=str))
                ok, detail = False, "pending: miniwebgen.taskgen judge"
            else:
                log, changes, clip = await H.backend_state(bs, base)
                steps = router.trace if all("action" in t for t in router.trace) else [{"action": history_actions(hist)}]
                events = action_events(steps) + [{"type": "action", "action": "clipboard_write", "target": "",
                                                         "value": v} for v in clip]
                (ep / "grade_inputs.json").write_text(json.dumps(
                    {"base": base, "log": log, "events": events, "changes": changes, "clip": clip, "answer": answer,
                     "seen": seen}, default=str))
                ok, detail, _ = checks.gate(task["check"], task["verifier"], log, events, changes, clip, answer, seen)
        except Exception as e:                   # e.g. the QA chain's LLM tier without credentials: --regrade later
            ok, detail = False, f"grade error {type(e).__name__}: {str(e)[:200]}"
            res["grade_error"] = True
        res.update(ok=bool(ok), detail=str(detail)[:300], steps=len(hist.history), answer=answer,
                   done=hist.is_done(), models=[t["model"] for t in router.trace], errors=[e for e in hist.errors() if e][:5])
        hist.save_to_file(ep / "history.json")
        (ep / "trace.json").write_text(json.dumps(router.trace, indent=1, default=str))
        if human or proposed:                        # the state after the last action, for miniwebgen.taskgen's judge
            try:
                png = await bs.take_screenshot()
                (ep / "final.png").write_bytes(base64.b64decode(png) if isinstance(png, str) else png)
            except Exception:
                pass
        if args.record_rows and proposed and rec:
            res["rows"] = write_proposed_rows(ep / "rows", task, rec)
        if args.record_rows and ok and rec and not proposed:   # a passing episode's steps, as training rows
            from miniwebgen.replay import save_images
            fam = args.record_family or FAMILY_OF.get(task["macro"], "Other")
            with open(ep / "rows.jsonl", "w") as f:
                for i, st in enumerate(rec):
                    f.write(json.dumps({"task_id": epn, "step": i, "family": fam, "source": "rejection_sampling",
                                        "model": st["model"], "messages": save_images(st["messages"], ep, i),
                                        "target": st["content"]}, ensure_ascii=False) + "\n")
            (ep / "episode.json").write_text(json.dumps({"task_id": epn, "instruction": task["instruction"],
                                                         "macros": task.get("macros"), "variant": "rejection_sampling"}))
        if arm == "agent_planner":
            (ep / "subtasks.json").write_text(json.dumps({"trace": router.trace, "plans": router.plans}, indent=1,
                                                         default=str, ensure_ascii=False))
            res["subtasks"] = [(t["family"], t["subtask"], t["steps"]) for t in router.trace]
        copy_screenshots(ep)                     # out of browser-use's /tmp dir, for miniwebagent.eval_viewer
    except asyncio.TimeoutError:
        res["error"] = "timeout"
    except Exception as e:
        res["error"] = f"{type(e).__name__}: {str(e)[:300]}"
    finally:
        res["duration_s"] = round(time.time() - t0, 1)
        try:
            await bs.kill()
        except Exception:
            pass
    return res


async def _run(tasks, bases, args):
    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    q = asyncio.Queue()
    for t in tasks:
        for rep in range(args.repeats):
            q.put_nowait((t, rep))
    tally = Counter()

    async def worker(w):
        while not q.empty():
            t, rep = q.get_nowait()
            try:
                res = await run_episode(t, bases[w % len(bases)], args.arm, args, out, rep)
            except asyncio.CancelledError as e:
                # 2026-10-08: browser-use's event bus can cancel an episode's own tasks (a DOM-watchdog screenshot
                # interrupted), and CancelledError is not an Exception: it escaped run_episode and ended the whole run
                # (taskgen shard 2 after 63 of 455 episodes). Nothing here cancels workers, so record it and go on.
                res = {"task_id": t["task_id"], "episode": t["task_id"] + (f"~r{rep}" if args.repeats > 1 else ""),
                       "macro": t.get("macro"), "site": t.get("site"), "arm": args.arm, "ok": False,
                       "error": f"CancelledError: {str(e)[:200]}", "duration_s": 0}
            res["repeat"] = rep
            tally["ok" if res["ok"] else "fail"] += 1
            with open(out / "results.jsonl", "a") as f:
                f.write(json.dumps(res, default=str) + "\n")
            print(f"[{sum(tally.values())}] {'OK  ' if res['ok'] else 'FAIL'} {res['task_id']} {res.get('steps')} steps "
                  f"{res['duration_s']}s {res.get('error') or res.get('detail', '')[:120]}", flush=True)

    await asyncio.gather(*(worker(w) for w in range(args.workers)))
    print(f"{args.arm}: {dict(tally)}", flush=True)


def main(argv=None):
    ap = argparse.ArgumentParser(prog="python -m miniwebagent.evaluate")
    ap.add_argument("--tasks", default=None, help="a datagen tasks.jsonl (e.g. from the validation root)")
    ap.add_argument("--human", action="store_true", help="the human-written tasks on the held-out sites instead")
    ap.add_argument("--human-split", default="test", choices=["test", "train", "all"], help="--human: held-out sites, train sites, or all tasks")
    ap.add_argument("--human-macros", type=lambda s: set(s.split(",")), default=None,
                    help="--human: only tasks with one of these macros")
    ap.add_argument("--record-rows", action="store_true", help="write passing episodes as training rows")
    ap.add_argument("--record-family", default=None, help="--record-rows: the rows' family (default: the task macro's)")
    ap.add_argument("--arm", required=True, choices=["base", "pooled", "oracle", "agent_planner"])
    ap.add_argument("--planner-name", default="qwen35-4b", help="agent_planner: the planner's served model name")
    ap.add_argument("--sub-cap", type=int, default=15, help="agent_planner: steps per delegated step (of the 40)")
    ap.add_argument("--planner-turns", type=int, default=12, help="agent_planner: the planner agent's own turns")
    ap.add_argument("--agent-version", default="v1", help="agent_planner: planner_agent version (v1, v2, ...)")
    ap.add_argument("--reasoning-to-fallback", action="store_true",
                    help="agent_planner: reasoning steps go to --fallback too (default: the reasoning adapter, else the base)")
    ap.add_argument("--gate-note", action="store_true",
                    help="done-gating that also tells the specialist the planned steps still left (implies --gate-done)")
    ap.add_argument("--gate-done", action="store_true",
                    help="oracle arm: the router turns a premature done into macro_done (gate_done)")
    ap.add_argument("--fallback", default=None,
                    help="agent_planner: model for a family without its adapter (default: the base)")
    ap.add_argument("--adapters", type=lambda s: set(s.split(",")), default=set(), help="adapters loaded on the server")
    ap.add_argument("--base-name", default="qwen35-4b")
    ap.add_argument("--vllm", default="http://127.0.0.1:8400/v1")
    ap.add_argument("--temperature", type=float, default=0.0)
    ap.add_argument("--max-tokens", type=int, default=1024,
                    help="reply budget per agent call; a thinking model (GLM-4.1V-Thinking) needs its thinking to fit too")
    ap.add_argument("--repeats", type=int, default=1)
    ap.add_argument("--workers", type=int, default=8)
    ap.add_argument("--timeout", type=int, default=600)
    ap.add_argument("--max-steps", type=int, default=0,
                    help="single agent: step cap (default harness.MAX_STEPS = 40). Review round 1, A3 (2026-10-05): the "
                         "untrained single agent with MiniWebAgent's model-call budget")
    ap.add_argument("--ports", type=lambda s: [int(p) for p in s.split(",")], default=[8318, 8319])
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--task-ids", type=lambda s: set(s.split(",")), default=None, help="only these task ids")
    ap.add_argument("--out", required=True)
    ap.add_argument("--pooled-name", default="pooled", help="served adapter name the pooled arm uses")
    ap.add_argument("--regrade", default=None, help="an earlier --out dir: recompute its grades from grade_inputs.json")
    ap.add_argument("--resume", action="store_true", help="skip tasks already in --out/results.jsonl (repeats=1 runs)")
    ap.add_argument("--verifier-only", action="store_true",
                    help="grade with the backend check + the task's verifier as-is (its QA tier chain included: regex, "
                         "exact, precise, then the LLM tier), never the flash macro judge. webmix runs never use the "
                         "macro judge; the flag records the intent")
    args = ap.parse_args(argv)
    os.environ.setdefault("VERIFIER_JUDGE_MODEL", QA_MODEL)       # see grade_human
    if args.human:
        tasks = human_tasks(split=args.human_split, macros=args.human_macros)
    else:
        tasks = [json.loads(line) for line in Path(args.tasks).read_text().split("\n") if line.strip()]
    if args.task_ids:
        tasks = [t for t in tasks if t["task_id"] in args.task_ids]
    if args.limit:
        tasks = tasks[:args.limit]
    if args.regrade:
        return regrade(tasks, Path(args.regrade))
    if args.resume and (Path(args.out) / "results.jsonl").exists():
        # 2026-10-03: a Slurm job ended mid-run; finish the run instead of redoing it (episodes in results.jsonl are kept)
        done = {json.loads(line)["task_id"] for line in (Path(args.out) / "results.jsonl").read_text().split("\n")
                if line.strip()}
        tasks = [t for t in tasks if t["task_id"] not in done]
        print(f"resume: {len(done)} tasks already in {args.out}, {len(tasks)} to run", flush=True)
    with B.Servers(ports=args.ports) as srv:
        asyncio.run(_run(tasks, srv.bases, args))


def regrade(tasks, out):
    """Recompute every episode's grade in `out` from its saved inputs, with this machine's full QA tier chain.
    Writes results_regraded.jsonl (the latest result per task, ok/detail replaced) and prints the tally."""
    from miniwebgen.replay import _rebase
    last = {}
    for line in (out / "results.jsonl").read_text().split("\n"):
        if line.strip():
            r = json.loads(line)
            last[r["task_id"]] = r
    by_id = {t["task_id"]: t for t in tasks}
    tally, changed = Counter(), 0
    with open(out / "results_regraded.jsonl", "w") as f:
        for tid, r in last.items():
            gi = out / tid / "grade_inputs.json"
            if tid in by_id and gi.exists() and by_id[tid].get("kind") == "human":
                g = json.loads(gi.read_text())
                ok, detail = grade_human(by_id[tid], g["traj"], g["answer"])
                changed += bool(ok) != bool(r.get("ok"))
                r = {**r, "ok": ok, "detail": detail, "regraded": True}
                r.pop("grade_error", None)
            elif tid in by_id and gi.exists():
                g = json.loads(gi.read_text())
                task = _rebase(by_id[tid], g["base"])
                task["check"] = checks.relax_recorded_dates(task["check"], [task["instruction"], json.dumps(task.get("option"))])
                ok, detail, _ = checks.gate(task["check"], task["verifier"], g["log"], g["events"], g["changes"], g["clip"],
                                            g["answer"], g["seen"])
                changed += bool(ok) != bool(r.get("ok"))
                r = {**r, "ok": bool(ok), "detail": str(detail)[:300], "regraded": True}
                r.pop("grade_error", None)
            tally["ok" if r.get("ok") else "fail"] += 1
            f.write(json.dumps(r, default=str) + "\n")
    print(f"regraded {out}: {dict(tally)} ({changed} changed)", flush=True)


if __name__ == "__main__":
    main()