code/apps/browser_suite_c.py

"""Held-out suite C: harder, multi-step tasks on live sites that appear in NO training source (train_domains.json,
finetune/sites*.txt) and not in suites A/B.  Same runner conventions and services as apps/browser_suite.py.

    python apps/browser_suite_c.py [name-filter]        REPEATS=3 SUITE_OUT=... for repeated runs

Every task has a scripted, model-free solution in apps/suite_c_scripts.py; a task is only listed here after that script
passed its check 3/3 while the check FAILED on the start page.  Checks take (url, title, text, actions): the observed
actions carry form state (checked radios/checkboxes, current <select> values), like suite A's dropdown/checkbox checks.
"""
import glob, json, os, re, sys, time
from urllib.parse import parse_qs, unquote_plus, urlparse
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import browser_suite  # noqa: E402,F401  (sets up the jev env vars)
from jev_ultrafast import Agent  # noqa: E402
from jev_ultrafast.browser import Browser, StalePage  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))


def q(u):                                   # decoded query parameters of a URL: {name: first value}
    return {k: v[0] for k, v in parse_qs(urlparse(u).query).items()}


def selected(actions, label):               # current value of the <select> whose label contains `label`
    for a in actions:
        if a.get("kind") == "select" and label.lower() in a["label"].rsplit(" → ", 1)[0].lower():
            return a.get("current_value", "")
    return None


def checked(actions, label, role=None):     # True/False for the checkbox/radio whose label contains `label`; None if absent
    for a in actions:
        if a.get("kind") == "click" and (role is None or a.get("role") == role) and label.lower() in a["label"].lower():
            return str(a.get("checked")) == "true"
    return None


def field(actions, label):                  # current text of the fillable field whose label contains `label`
    for a in actions:
        if a.get("kind") == "fill" and label.lower() in a["label"].lower():
            return a.get("value", "")
    return None


def clear_storage(url):                     # per-task setup: wipe the site's localStorage (TodoMVC keeps todos across runs)
    b = Browser(url)
    try:
        b.evaluate("(() => { localStorage.clear(); sessionStorage.clear(); return true })()")
    finally:
        b.close()


# (name, start url, goal, check(url, title, text, actions) -> bool[, setup(url)])
TASKS = [
    # ---- multi-step: search + filter + sort / open (4+ actions) ----
    ("met-sunflowers", "https://www.metmuseum.org/art/collection/search",
     "Search the collection for 'sunflowers', show only objects that have an image, and sort the results by date, oldest first.",
     lambda u, t, x, a: q(u).get("q") == "sunflowers" and q(u).get("showOnly") == "withImage" and q(u).get("sortBy") == "Date"),
    ("nuget-serilog-tool", "https://www.nuget.org/",
     "Search NuGet for 'serilog', restrict the package type to .NET tool, and sort the results by downloads.",
     lambda u, t, x, a: q(u).get("q") == "serilog" and q(u).get("packagetype") == "dotnettool" and q(u).get("sortby") == "totalDownloads-desc"),
    ("alpine-curl-filter", "https://pkgs.alpinelinux.org/packages",
     "Search for the package name 'curl' in branch v3.20, repository main, architecture aarch64.",
     lambda u, t, x, a: q(u).get("name") == "curl" and q(u).get("branch") == "v3.20" and q(u).get("repo") == "main" and q(u).get("arch") == "aarch64"),
    ("freesound-rain-cc0", "https://freesound.org/",
     "Search for sounds matching 'rain' and narrow the results to the Creative Commons 0 license and the Soundscapes category.",
     lambda u, t, x, a: q(u).get("q") == "rain" and 'license:"Creative Commons 0"' in unquote_plus(u) and 'category:"Soundscapes"' in unquote_plus(u)),
    ("ats-shampoo-haircare", "https://automationteststore.com/",
     "Search the store for 'shampoo' in the 'Hair Care' category, include product descriptions in the search, and sort the results by price from high to low.",
     lambda u, t, x, a: q(u).get("keyword") == "shampoo" and "52" in q(u).get("category_id", "").split(",") and q(u).get("description") == "1" and q(u).get("sort") == "p.price-DESC"),
    ("bnf-hugo-printed-p2", "https://catalogue.bnf.fr/",
     "Search the catalogue for 'victor hugo', keep only 'Texte imprimé et livre numérique' documents, and go to page 2 of the results.",
     lambda u, t, x, a: q(u).get("motRecherche") == "victor hugo" and "FacNatDoc_a" in q(u).get("listeAffinages", "") and q(u).get("pageEnCours") == "2"),
    ("vsm-python-installs", "https://marketplace.visualstudio.com/vscode",
     "Search for 'python' extensions and sort the results by number of installs.",
     lambda u, t, x, a: q(u).get("term") == "python" and q(u).get("sortBy") == "Installs"),
    ("wp-cache-commercial-redis", "https://wordpress.org/plugins/",
     "Search the plugin directory for 'cache', show only commercial plugins, and open the 'Redis Object Cache' plugin.",
     lambda u, t, x, a: "/plugins/redis-cache" in u),
    ("todomvc-active", "https://demo.playwright.dev/todomvc",
     "Add two todos, 'buy milk' and 'walk the dog', then show only the active todos.",
     lambda u, t, x, a: u.endswith("#/active") and "buy milk" in x and "walk the dog" in x and re.search(r"\b2\s*\|?\s*items?\s*\|?\s*left", x) is not None,
     clear_storage),
    ("setlist-radiohead-uk", "https://www.setlist.fm/",
     "Search for 'radiohead' setlists and filter the results to the artist Radiohead and the country United Kingdom.",
     lambda u, t, x, a: q(u).get("query") == "radiohead" and q(u).get("country") == "gb" and q(u).get("artist") == "bd6bd12"),
    ("jetbrains-rust-free", "https://plugins.jetbrains.com/",
     "Search for 'rust' plugins, show only free ones, and open the 'Rust' plugin published by JetBrains.",
     lambda u, t, x, a: "/plugin/22407-rust" in u),
    ("luarocks-rapidjson", "https://luarocks.org/",
     "Search LuaRocks for 'json' including non-root manifests, and open the module 'rapidjson' by xpol.",
     lambda u, t, x, a: "/modules/xpol/rapidjson" in u),
    ("letcode-dropdowns", "https://letcode.in/dropdowns",
     "In the dropdown demo: select 'Apple' as the fruit, 'Batman' as the super hero, 'Swift' as the programming language, and 'India' as the country.",
     lambda u, t, x, a: selected(a, "apple") == "Apple" and selected(a, "super hero") == "Batman" and selected(a, "programming language") == "Swift" and selected(a, "Select India") == "India"),
    ("letcode-radio", "https://letcode.in/radio",
     "On the radio-button demo: choose 'Foo', choose 'Going', tick 'I agree to the FAKE terms and conditions' and untick 'Remember me'.",
     lambda u, t, x, a: checked(a, "Foo", "radio") is True and checked(a, "Going", "radio") is True and checked(a, "I agree", "checkbox") is True and checked(a, "Remember me", "checkbox") is False),
    ("clojars-ring-page3", "https://clojars.org/",
     "Search for 'ring' and go to page 3 of the results.",
     lambda u, t, x, a: q(u).get("q") == "ring" and q(u).get("page") == "3"),
    ("fred-unemployment-pop", "https://fred.stlouisfed.org/",
     "Search FRED for 'unemployment rate' and sort the results by popularity.",
     lambda u, t, x, a: "unemployment" in q(u).get("st", "").lower() and q(u).get("ob") == "p"),
    ("modrinth-sodium", "https://modrinth.com/mods",
     "Search mods for 'sodium', filter to game version 1.21.11, and sort by downloads.",
     lambda u, t, x, a: q(u).get("q") == "sodium" and q(u).get("v") == "1.21.11" and q(u).get("s") == "downloads"),
    ("tvmaze-friends-episodes", "https://www.tvmaze.com/",
     "Find the show 'Friends' and open its episode list.",
     lambda u, t, x, a: "/shows/431/friends/episodes" in u),
    ("qaclickjet-form", "https://rahulshettyacademy.com/dropdownsPractise/",
     "Choose 'Round Trip', tick the 'Senior Citizen' discount, set the currency to USD, and enter 'India' in the country box.",
     lambda u, t, x, a: checked(a, "Round Trip", "radio") is not False and checked(a, "One Way", "radio") is not True and checked(a, "Multicity", "radio") is not True
     and checked(a, "Senior Citizen", "checkbox") is True and selected(a, "INR") == "USD" and (field(a, "Type to Select") or "").strip().lower() == "india"),
    # ---- shorter: search + open / browse (2-3 actions) ----
    ("cocktail-margarita", "https://www.thecocktaildb.com/", "Find the recipe page of the 'Margarita' cocktail.",
     lambda u, t, x, a: "/drink/11007" in u),
    ("mealdb-arrabiata", "https://www.themealdb.com/", "Open the recipe for 'Spicy Arrabiata Penne'.",
     lambda u, t, x, a: "/meal/52771" in u),
    ("gentoo-openrc-talk", "https://wiki.gentoo.org/wiki/Main_Page", "Open the discussion (talk) page of the wiki article 'OpenRC'.",
     lambda u, t, x, a: "Talk:OpenRC" in u),
    ("webkit-css-bugs", "https://bugs.webkit.org/", "Browse the WebKit product's components and open the bug list for the 'CSS' component.",
     lambda u, t, x, a: "buglist.cgi" in u and q(u).get("product") == "WebKit" and q(u).get("component") == "CSS"),
    ("govdata-wetter-energie", "https://www.govdata.de/", "Search for 'Wetter' datasets and filter them to the category 'Energie'.",
     lambda u, t, x, a: q(u).get("q") == "Wetter" and q(u).get("groups") == "ener"),
    ("fedora-vim-common", "https://packages.fedoraproject.org/", "Search for 'vim' and open the package 'vim-common'.",
     lambda u, t, x, a: "/pkgs/vim/vim-common" in u),
    ("racket-argo", "https://pkgs.racket-lang.org/", "Search packages for 'json' and open the package 'argo'.",
     lambda u, t, x, a: u.rstrip("/").endswith("/package/argo")),
    ("rdrr-ggplot", "https://rdrr.io/", "Search the R package documentation for 'ggplot'.",
     lambda u, t, x, a: "/search" in u and q(u).get("q") == "ggplot"),
]
MULTISTEP = {t[0] for t in TASKS[:19]}


def final_state(br, max_scrolls=6):
    """Whole-page view for judging: scroll to the top, then observe screen by screen and merge the actions (form state)
    and the visible text.  The observation only covers the viewport, and a check must not depend on where the agent
    left the scroll position."""
    def observe():
        for i in range(10):
            try:
                return br.observe(screenshot=False)
            except StalePage:
                time.sleep(0.5)
        return br.observe(screenshot=False)
    def to_top():                                              # instant, and wait until it took effect (smooth-scroll sites)
        br.evaluate("(() => { document.documentElement.style.scrollBehavior='auto'; window.scrollTo({top: 0, left: 0, behavior: 'instant'}); return scrollY })()")
        for _ in range(20):
            if br.evaluate("scrollY") == 0:
                break
            time.sleep(0.1)
    page = observe()
    try:
        to_top()
        page = observe()
    except Exception:
        return page
    acts, texts = {}, []
    for _ in range(max_scrolls + 1):
        for a in page["actions"]:
            if "node" in a:
                acts.setdefault((a["node"], a["kind"], a.get("value")), a)
        texts.append(page["text"])
        down = next((a for a in page["actions"] if a["id"] == "scroll_down"), None)
        if down is None:
            break
        try:
            br.act(down, page); time.sleep(0.2); page = observe()
        except Exception:
            break
    try:
        to_top()                                               # leave the page as it was found (top)
    except Exception:
        pass
    return {**page, "actions": list(acts.values()), "text": "\n".join(texts)}


def check_page(name, check, page):
    try:
        return bool(check(page["url"], page["title"], page.get("text", ""), page.get("actions", [])))
    except Exception:
        return False


def run(name, url, goal, check, setup=None, max_steps=20):
    t0 = time.time(); steps = 0; status = "error"; page = None; hist = []
    try:
        if setup:
            setup(url)
        with Agent(url, goal) as agent:
            page = agent.state["page"]
            for state in agent.run():
                steps = len(state["history"]); status = state["status"]; page = state["page"]
                hist = [{k: h.get(k) for k in ("action", "kind", "text", "url", "page_changed", "probability", "operation")} for h in state["history"]]
                if steps >= max_steps: break
            time.sleep(1.5)                                  # let a slow navigation land before judging
            try:
                page = final_state(agent.browser)
            except Exception:
                pass
    except Exception as e:
        status = f"error:{type(e).__name__}"
    wall = time.time() - t0
    ok = bool(page) and check_page(name, check, page)
    if os.environ.get("SUITE_TRACE"):   # full trajectory for case-by-case failure review
        with open(os.environ["SUITE_TRACE"], "a") as f:
            f.write(json.dumps({"name": name, "goal": goal, "start": url, "pass": ok, "status": status, "steps": steps, "history": hist,
                                "final_url": (page or {}).get("url", ""), "final_title": (page or {}).get("title", ""),
                                "final_text": (page or {}).get("text", "")[:3000]}, ensure_ascii=False) + "\n")
    return ok, steps, status, wall, (page or {}).get("url", "")


def heldout_check(tasks):
    """LEAK if a task domain (or its registrable base) appears in the training domains, finetune/sites*.txt or suites A/B."""
    dom = lambda u: urlparse(u).netloc.lower().removeprefix("www.")
    base = lambda d: ".".join(d.split(".")[-2:])
    seen = set()
    try:
        seen |= set(json.load(open(os.path.join(HERE, "..", "finetune", "out", "train_domains.json"))))
    except FileNotFoundError:
        print("held-out check: no finetune/out/train_domains.json", flush=True)
    for f in glob.glob(os.path.join(HERE, "..", "finetune", "sites*.txt")):
        for line in open(f):
            line = line.strip()
            if line and not line.startswith("#"):
                seen.add(dom(line.split()[0]))
    for f in ("browser_suite.py", "browser_suite_b.py"):
        seen |= {dom(u) for u in re.findall(r'"(https?://[^"]+)"', open(os.path.join(HERE, f)).read())}
    seen_base = {base(d) for d in seen}
    leak = sorted({dom(u) for _, u, *_ in tasks if dom(u) in seen or base(dom(u)) in seen_base})
    print("held-out check:", "OK, no suite-C domain appears in training data / sites lists / suites A-B" if not leak else f"LEAK {leak}", flush=True)
    return leak


if __name__ == "__main__":
    heldout_check(TASKS)
    flt = sys.argv[1] if len(sys.argv) > 1 else ""
    repeats = int(os.environ.get("REPEATS", "1"))
    rows = []
    for name, url, goal, check, *extra in TASKS:
        if flt and flt not in name: continue
        for rep in range(repeats):
            ok, steps, status, wall, final = run(name, url, goal, check, *extra)
            rows.append((name, ok, steps, status, wall))
            print(f"{'PASS' if ok else 'FAIL'}  {name:26s} steps={steps:2d} status={status:9s} {wall:5.1f}s  {final[:70]}", flush=True)
    n = sum(r[1] for r in rows)
    print(f"\n== {n}/{len(rows)} passed ({100*n/len(rows):.0f}%, {repeats} run(s) per task) | median wall {sorted(r[4] for r in rows)[len(rows)//2]:.1f}s")
    m = [r for r in rows if r[0] in MULTISTEP]
    if m:
        print(f"   multi-step tasks: {sum(r[1] for r in m)}/{len(m)} passed")
    per = {}
    for r in rows: per.setdefault(r[0], []).append(r[1])
    print("   per task: " + "  ".join(f"{k}={sum(v)}/{len(v)}" for k, v in per.items()))
    json.dump([dict(zip(("name", "pass", "steps", "status", "wall"), r)) for r in rows], open(os.environ.get("SUITE_OUT", "/tmp/suite_c.json"), "w"), indent=1)