Blame for src/miniwebgen/taskgen.py

"""Task-centric MiniWeb data (user, 2026-10-08: "maybe we should do task-centric MiniWeb data instead", plan A): InSTA's
recipe moved onto MiniWeb's training sites, so the skill- vs task-centric comparison holds sites, harness, actor,
verification evidence and size fixed and changes only the unit of data. InSTA proposes whole-task goals per website,
an agent attempts them end to end, and a judge keeps the successes; here

  1. propose   Gemini 3.5 Flash proposes goal-level tasks for each of the 52 training sites from its pages (paths,
               titles, controls, and the visible text of its main pages, so tasks name real items); every task starts
               on the site's home page, as InSTA's start at a website's front page;
  2. attempt   the base Qwen3.5-4B, as one agent with no hint, runs each task in the MiniWeb harness and every
               episode is recorded as rows (python -m miniwebagent.evaluate --tasks <tasks> --arm base --record-rows; the
               task kind "proposed" saves the session's request log, data changes, answer, screen text, final shot);
  3. judge     Gemini 3.5 Flash judges each episode from the screenshots AND the backend evidence (which requests were
               made with which parameters, which rows changed), far more than a live-web judge sees; `validate`
               measures it against the task verifiers on human tasks of the training sites first;
  4. rows      the judged successes, sites interleaved, selected in task order up to the step budget of the
               skill-centric executor (27,644 steps), as a miniwebagent.train rows dir.

    python -m miniwebgen.taskgen propose --out data/webmix/taskgen/tasks.jsonl [--per-site 140] [--sites a,b]
    python -m miniwebgen.taskgen judge <run dir> [--workers 16]            # -> <run dir>_judge/<task_id>.json
    python -m miniwebgen.taskgen validate <human train-split run dir>      # judge vs the verifiers
    python -m miniwebgen.taskgen rows <run dir> --out data/webmix/taskgen/rows_27k [--budget 27644]
"""
from __future__ import annotations

from helpers.paths import REPO_ROOT
import argparse
import hashlib
import html
import json
import random
import re
import shutil
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

ROOT = REPO_ROOT
OUT = ROOT / "data" / "webmix" / "taskgen"
MODEL = "gemini-3.5-flash"
BUDGET = 27644                       # the skill-centric executor's training steps (pooled_v4, 3.1K trajectories)


def train_sites():
    from miniwebgen import config
    return json.loads(config.SPLIT_PATH.read_text())["train"]


# ── 1. propose ────────────────────────────────────────────────────────────────

def _text(page_html, cap):
    """The visible text of a page (scripts, styles and tags dropped), whitespace collapsed, at most `cap` chars."""
    s = re.sub(r"(?is)<(script|style|noscript|svg)[^>]*>.*?</\1>", " ", page_html)
    s = re.sub(r"(?s)<[^>]+>", "\n", s)
    lines = [re.sub(r"\s+", " ", html.unescape(x)).strip() for x in s.split("\n")]
    out, n = [], 0
    for x in lines:
        if x and (not out or out[-1] != x):
            out.append(x); n += len(x) + 1
            if n > cap:
                break
    return "\n".join(out)[:cap]


def site_context(site, client, n_pages=6, cap=1800):
    """What the proposer sees of a site: its name, every page (path, title), the controls per page (kind, label,
    a few option texts), and the visible text of its home page and the pages with the most controls."""
    from miniwebgen import config
    sm = json.loads((config.SITEMAP_DIR / f"{site}.json").read_text())
    pages = [p for p in sm["pages"] if p.get("path")]
    ctrls = sm["controls"] if isinstance(sm["controls"], list) else list(sm["controls"].values())
    by_page = {}
    for c in ctrls:
        if c.get("kind") in ("qa", "carry"):
            continue
        opts = [str(o.get("text") or o.get("value") or "")[:40] for o in (c.get("options") or [])[:4]]
        by_page.setdefault(c.get("page"), []).append(f"{c.get('kind')} '{str(c.get('label') or c.get('name') or '')[:40]}'"
                                                    + (f" [{', '.join(o for o in opts if o)}]" if any(opts) else ""))
    home = pages[0]["path"] if pages else f"/sites/{site}/"
    rich = sorted(by_page, key=lambda p: -len(by_page[p]))
    show = [home] + [p for p in rich if p and p != home][:n_pages - 1]
    texts = []
    for path in show:
        try:
            r = client.get(path, follow_redirects=True)
            texts.append(f"--- {path}\n{_text(r.get_data(as_text=True), cap)}")
        except Exception as e:                       # noqa: BLE001
            texts.append(f"--- {path}\n(could not render: {type(e).__name__})")
    page_lines = [f"{p['path']}  {p.get('title') or ''}" for p in pages[:90]]
    ctrl_lines = [f"{p}: " + "; ".join(v[:8]) for p, v in list(by_page.items())[:40] if p]
    return {"site": site, "name": sm.get("name") or site, "home": home, "paths": [p["path"] for p in pages],
            "text": "PAGES\n" + "\n".join(page_lines) + "\n\nCONTROLS BY PAGE\n" + "\n".join(ctrl_lines)
                    + "\n\nPAGE TEXT\n" + "\n\n".join(texts)}


SYSTEM_PROPOSE = """You write tasks for training a web agent, the way InSTA proposes tasks for real websites: realistic goals \
that a real user of this website would have. You see one website: its pages, its controls, and the text of its main pages."""

PROMPT_PROPOSE = """WEBSITE: {name} (all pages under {home})

{context}

Write {n} new tasks for this website. Rules:
- Each task is ONE natural instruction stating the user's goal, the way a person would ask for it ("Find ...", \
"Book ...", "Change my ...", "What is ..."), never click-by-click steps.
- Every task starts on the home page ({home}), as a visitor arriving at the site would, and is doable on this \
website alone in about 3 to 10 browser actions from there. Use real names, values and items that appear above; never \
invent items that are not on the site.
- Mix: about {n_info} information tasks (the answer is a specific fact the site shows; say what to report, e.g. \
"report the price") and {n_act} tasks that do something (search, filter, sort, create, edit, delete, book, buy, send, \
share, save, configure, ...), some of them combining two or three of these.
- {focus}
- The user is already logged in. No passwords, no other websites, no files the user would have to provide.
- Do not repeat these earlier tasks: {avoid}

Return JSON: {{"tasks": [{{"instruction": "...", "kind": "info" | "action"}}]}}"""


def _norm(s):
    return re.sub(r"[^a-z0-9]+", " ", s.lower()).strip()


def propose_site(ctx, per_site, batch, seed, model=MODEL):
    from helpers.llm import LLMClient
    llm = LLMClient(model=model, temperature=1.0)
    rng = random.Random(f"{seed}:{ctx['site']}")
    out, seen = [], set()
    focus_pages = [p for p in ctx["paths"] if p != ctx["home"]] or [ctx["home"]]
    for b in range((per_site + batch - 1) // batch + 2):            # +2: room for duplicates and invalid rows
        if len(out) >= per_site:
            break
        pick = rng.sample(focus_pages, min(4, len(focus_pages)))
        focus = ("Most of these tasks should involve these parts of the site: " + ", ".join(pick)) if b % 3 else \
            "Most of these tasks should involve the site's main sections."
        avoid = "; ".join(t["instruction"][:90] for t in out[-40:]) or "(none yet)"
        n = min(batch, per_site - len(out) + 4)
        prompt = PROMPT_PROPOSE.format(name=ctx["name"], home=ctx["home"], context=ctx["text"], n=n,
                                       n_info=round(n * 0.4), n_act=n - round(n * 0.4), focus=focus, avoid=avoid)
        try:
            raw = llm.complete(prompt, system=SYSTEM_PROPOSE, json_mode=True, max_tokens=8192)
            got = json.loads(raw or "{}")
            tasks = (got.get("tasks") if isinstance(got, dict) else got) or []   # a bare list happens too
        except Exception as e:                       # noqa: BLE001
            print(f"  {ctx['site']} batch {b}: {type(e).__name__} {str(e)[:120]}", flush=True)
            continue
        for t in tasks:
            ins = str(t.get("instruction") or "").strip()
            if len(ins) < 15 or _norm(ins) in seen:
                continue
            start = ctx["home"]                      # InSTA's rule: every task starts at the site's front page
            seen.add(_norm(ins))
            tid = f"tg-{ctx['site']}-{hashlib.md5(ins.encode()).hexdigest()[:8]}"
            out.append({"kind": "proposed", "task_id": tid, "site": ctx["site"], "sites": [ctx["site"]],
                        "macro": "proposed", "macros": [], "instruction": ins, "start_path": start,
                        "task_kind": t.get("kind") if t.get("kind") in ("info", "action") else "action",
                        "verifier": None})
    return out[:per_site]


def cmd_propose(a):
    from miniweb.app import create_app
    sites = a.sites.split(",") if a.sites else train_sites()
    out = Path(a.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    done = set()
    if out.exists():                                 # resumable: sites already in the file are kept
        done = {json.loads(l)["site"] for l in out.read_text().split("\n") if l.strip()}
    client = create_app().test_client()
    ctxs = [site_context(s, client) for s in sites if s not in done]
    print(f"proposing for {len(ctxs)} sites ({len(done)} already done), {a.per_site} each", flush=True)
    with ThreadPoolExecutor(a.workers) as ex:
        for ctx, tasks in zip(ctxs, ex.map(lambda c: propose_site(c, a.per_site, a.batch, a.seed), ctxs)):
            with open(out, "a") as f:
                f.write("".join(json.dumps(t, ensure_ascii=False) + "\n" for t in tasks))
            print(f"  {ctx['site']}: {len(tasks)} tasks ({sum(t['task_kind'] == 'info' for t in tasks)} info)", flush=True)


# ── 3. judge ──────────────────────────────────────────────────────────────────

SKIP_PATH = re.compile(r"^/(_admin|static|favicon)|\.(css|js|png|jpg|svg|ico|woff2?)$")

SYSTEM_JUDGE = """You judge whether a web agent completed a task on a website. You get the task, the agent's actions and \
final answer, the server's record of what the agent's session did, and screenshots of the episode (the last one is the \
final state). The server record is authoritative: a search, filter or sort happened only if a request carries it; an item \
was created, edited, deleted, booked, sent or saved only if the data changes show it, with the values the task asks for. \
For a question, the final answer must state the requested fact correctly and completely, as the site shows it. Judge \
the outcome the task asks for, not the route: searching instead of browsing a category is fine when the result is the \
same. A task is successful only if EVERY part of it was done; a partial, wrong or unverifiable result is NOT SUCCESS."""

PROMPT_JUDGE = """TASK: {task}

AGENT'S ACTIONS ({n_act}):
{actions}

AGENT'S FINAL ANSWER: {answer}

SERVER REQUESTS OF THE SESSION (method path, parameters):
{log}

DATA CHANGES OF THE SESSION:
{changes}

TEXT ON SCREEN AT THE END:
{seen}

Screenshots of the episode follow, in order. Split the task into its parts and check each against the evidence. \
Return JSON: {{"parts": [{{"part": "...", "done": true | false, "evidence": "..."}}], "verdict": "SUCCESS" | \
"NOT SUCCESS"}}"""


def _fmt_log(log, cap=60):
    lines = []
    for e in log or []:
        p = str(e.get("path") or "")
        if SKIP_PATH.search(p):
            continue
        q = {k: v for k, v in (e.get("query") or {}).items() if v not in ("", None)}
        b = {k: v for k, v in (e.get("body") or {}).items() if v not in ("", None) and k not in ("csrf_token",)} \
            if isinstance(e.get("body"), dict) else e.get("body")
        lines.append(f"{e.get('method')} {p}" + (f" ?{json.dumps(q, ensure_ascii=False)[:200]}" if q else "")
                     + (f" body {json.dumps(b, ensure_ascii=False)[:300]}" if b else "") + f" -> {e.get('status')}")
    return "\n".join(lines[-cap:]) or "(none)"


def _fmt_changes(changes, cap=30):
    lines = []
    for c in changes or []:
        what = c.get("collection") or c.get("table") or c.get("kind") or "?"
        op = c.get("op") or ("create" if not c.get("before") else "delete" if not c.get("after") else "update")
        ch = c.get("changed") or {}
        item = c.get("after") or c.get("before") or {}
        name = item.get("title") or item.get("name") or item.get("subject") or item.get("id")
        lines.append(f"{op} {what} ({name}): " + json.dumps({k: v for k, v in ch.items()}, ensure_ascii=False, default=str)[:400])
    return "\n".join(lines[:cap]) or "(none)"


def _fmt_actions(acts, cap=40):
    out = []
    for a in acts or []:
        if not isinstance(a, dict) or not a:
            continue
        k = next(iter(a)); v = a[k] or {}
        if k in ("done", "macro_done"):
            continue
        v = {kk: vv for kk, vv in (v.items() if isinstance(v, dict) else []) if vv not in (None, "", False)}
        out.append(f"{k} {json.dumps(v, ensure_ascii=False)[:160]}")
    return out[-cap:]


def _shots(ep, k=10):
    shots = sorted((ep / "screenshots").glob("step_*.png"), key=lambda p: int(p.stem.split("_")[1])) \
        if (ep / "screenshots").exists() else []
    if (ep / "final.png").exists():
        shots.append(ep / "final.png")
    if len(shots) <= k:
        return shots
    head = max(1, k // 4)                            # the start, then the end of the episode
    return shots[:head] + shots[-(k - head):]


def _last_seen(seen):
    """The text of the last page the agent saw; a JSON page (an API response) is kept as JSON text."""
    last = (seen or [""])[-1] or ""
    return last if isinstance(last, str) else json.dumps(last, ensure_ascii=False, default=str)


def judge_episode(ep, instruction, out_dir, model=MODEL):
    dst = out_dir / f"{ep.name}.json"
    if dst.exists():
        return json.loads(dst.read_text())
    gi = ep / "grade_inputs.json"
    if not gi.exists():
        return None
    g = json.loads(gi.read_text())
    acts = _fmt_actions(g.get("actions"))
    prompt = PROMPT_JUDGE.format(task=instruction, n_act=len(acts), actions="\n".join(acts) or "(none)",
                                 answer=(g.get("answer") or "(none)")[:1500], log=_fmt_log(g.get("log")),
                                 changes=_fmt_changes(g.get("changes")),
                                 seen=_last_seen(g.get("seen"))[:2500])
    from helpers.llm import LLMClient
    resp = None
    for _ in range(3):
        try:
            resp = LLMClient(model=model, temperature=0.0).complete(prompt, system=SYSTEM_JUDGE, max_tokens=8192,
                                                                   json_mode=True, images=[str(p) for p in _shots(ep)])
            v = str(json.loads(resp or "{}").get("verdict") or "").strip().upper()
            if v in ("SUCCESS", "NOT SUCCESS"):
                break
        except Exception as e:                       # noqa: BLE001
            resp, v = f"(judge error {type(e).__name__}: {str(e)[:200]})", ""
    ok = v in ("SUCCESS", "NOT SUCCESS")
    rec = {"task_id": ep.name, "predicted_label": int(v == "SUCCESS"), "verdict": ok, "response": resp, "model": model}
    if ok:                                           # no verdict (an error): judged again on the next pass
        dst.write_text(json.dumps(rec, indent=1, ensure_ascii=False))
    return rec


def _instructions(run):
    """task_id -> instruction for a run's episodes (proposed tasks: the run's own tasks file; human tasks: annotations)."""
    out = {}
    for f in (run / "tasks.jsonl", OUT / "tasks.jsonl"):
        if f.exists():
            out.update({t["task_id"]: t["instruction"] for t in map(json.loads, filter(str.strip, f.read_text().split("\n")))})
    if not out or not all((run / tid).exists() for tid in list(out)[:1]):
        from miniwebagent.evaluate import human_tasks
        out.update({t["task_id"]: t["instruction"] for t in human_tasks(split="all")})
    return out


def cmd_judge(a):
    run = Path(a.run)
    out = Path(str(run.rstrip("/") if isinstance(run, str) else run) + "_judge")
    out.mkdir(exist_ok=True)
    ins = _instructions(run)
    eps = [p for p in sorted(run.iterdir()) if p.is_dir() and p.name in ins]
    print(f"judging {len(eps)} episodes of {run} with {a.model} -> {out}", flush=True)
    errors = []

    def one(p):          # one episode's failure is reported, not the end of the run
        try:
            return judge_episode(p, ins[p.name], out, a.model)
        except Exception as e:  # noqa: BLE001
            errors.append(p.name)
            print(f"judge error on {p.name}: {type(e).__name__}: {e}", flush=True)
            return None
    with ThreadPoolExecutor(a.workers) as ex:
        recs = list(ex.map(one, eps))
    recs = [r for r in recs if r]
    k = sum(r["predicted_label"] for r in recs)
    print(json.dumps({"episodes": len(recs), "success": k, "rate": round(k / max(1, len(recs)), 4),
                      "no_verdict": sum(not r["verdict"] for r in recs), "errors": len(errors), "judge": a.model}),
          flush=True)
    if errors:
        sys.exit(f"{len(errors)} episodes not judged; rerun to retry them (judged ones are kept)")


def cmd_validate(a):
    """The judge against the task verifiers on a human-task run (base model, train-split sites): agreement, and the
    precision that matters for training data, P(verifier passes | judge passes)."""
    run = Path(a.run)
    a.workers = a.workers or 16
    cmd_judge(a)
    out = Path(str(run) + "_judge")
    res = {}
    for name in ("results_regraded.jsonl", "results.jsonl"):
        if (run / name).exists():
            res = {json.loads(l)["task_id"]: bool(json.loads(l)["ok"]) for l in (run / name).read_text().split("\n") if l.strip()}
            break
    pairs = [(res[t], json.loads((out / f"{t}.json").read_text())["predicted_label"] == 1) for t in res if (out / f"{t}.json").exists()]
    tp = sum(v and j for v, j in pairs); fp = sum(j and not v for v, j in pairs)
    fn = sum(v and not j for v, j in pairs); tn = sum(not v and not j for v, j in pairs)
    n = len(pairs)
    print(json.dumps({"episodes": n, "agreement": round((tp + tn) / max(1, n), 3), "verifier_pass": tp + fn,
                      "judge_pass": tp + fp, "precision": round(tp / max(1, tp + fp), 3),
                      "recall": round(tp / max(1, tp + fn), 3), "tp": tp, "fp": fp, "fn": fn, "tn": tn}), flush=True)


# ── 4. rows ───────────────────────────────────────────────────────────────────

def cmd_rows(a):
    """The judged successes in task order with sites interleaved (round robin over sites, each site's tasks in file
    order), whole episodes, until --budget steps; copied as <out>/<task_id>/{rows.jsonl, step_*.png, episode.json}."""
    run, out = Path(a.run), Path(a.out)
    judged = Path(str(run) + "_judge")
    tasks = [json.loads(l) for l in Path(a.tasks).read_text().split("\n") if l.strip()]
    by_site = {}
    for t in tasks:
        by_site.setdefault(t["site"], []).append(t)
    order, queues = [], [list(v) for _, v in sorted(by_site.items())]
    while any(queues):
        for q in queues:
            if q:
                order.append(q.pop(0))
    out.mkdir(parents=True, exist_ok=True)
    picked, steps, cand = [], 0, 0
    for t in order:
        j = judged / f"{t['task_id']}.json"
        src = run / t["task_id"] / "rows"
        if not (j.exists() and json.loads(j.read_text())["predicted_label"] == 1 and (src / "rows.jsonl").exists()):
            continue
        n = sum(1 for l in (src / "rows.jsonl").read_text().split("\n") if l.strip())
        cand += 1
        if n == 0:
            continue
        if steps >= a.budget:
            continue
        shutil.copytree(src, out / t["task_id"], dirs_exist_ok=True)
        picked.append(t["task_id"]); steps += n
    rep = {"run": str(run), "candidates": cand, "selected_trajectories": len(picked), "selected_rows": steps,
           "budget": a.budget, "short_by": max(0, a.budget - steps),
           "sites": len({t.split("-", 1)[1].rsplit("-", 1)[0] for t in picked})}
    (out / "_report.json").write_text(json.dumps(rep, indent=1))
    print(json.dumps(rep), flush=True)


def main(argv=None):
    ap = argparse.ArgumentParser(prog="python -m miniwebgen.taskgen")
    sub = ap.add_subparsers(dest="cmd", required=True)
    p = sub.add_parser("propose"); p.add_argument("--out", default=str(OUT / "tasks.jsonl"))
    p.add_argument("--per-site", type=int, default=140); p.add_argument("--batch", type=int, default=20)
    p.add_argument("--sites", default=None); p.add_argument("--workers", type=int, default=8)
    p.add_argument("--seed", type=int, default=20261008)
    for name in ("judge", "validate"):
        p = sub.add_parser(name); p.add_argument("run"); p.add_argument("--model", default=MODEL)
        p.add_argument("--workers", type=int, default=16)
    p = sub.add_parser("rows"); p.add_argument("run"); p.add_argument("--out", required=True)
    p.add_argument("--tasks", default=str(OUT / "tasks.jsonl")); p.add_argument("--budget", type=int, default=BUDGET)
    a = ap.parse_args(argv)
    {"propose": cmd_propose, "judge": cmd_judge, "validate": cmd_validate, "rows": cmd_rows}[a.cmd](a)


if __name__ == "__main__":
    sys.exit(main())