Blame for src/miniwebgen/sampler.py

"""Stage 2 — parameter sampler. Programmatically picks (site, target element, argument,
start state) tuples; diversity is enforced HERE, not by the suggester.

  * train sites only, and only sites whose macro_locations entry lists the macro and whose
    site map has a usable (probed) control of the macro's kind;
  * stratified round-robin across those sites (seeded order), per-site cap, least-used
    control first within a site, options drawn without replacement;
  * dedup key (site, target element, argument) — element = the control's backend identity
    "METHOD path param", stable across re-crawls; keys kept in earlier runs are excluded;
  * start state: `fresh` (the control's page), `preset` (another control on the same listing
    already set via the start URL) or `mid_chain` (a preceding macro on the same page is
    executed first and its actions become the history). The target check then also
    requires that earlier state to survive (`keep`);
  * every tuple is dry-run with privileged calls in a throw-away session: the check must
    FAIL on the start state (not vacuous) and PASS after applying the target (solvable).
"""
from __future__ import annotations

import hashlib
import json
import random
import re
from urllib.parse import urlencode

from miniwebgen import browser as B
from miniwebgen import checks, config, kinds, sitemap

PHRASINGS = ("imperative", "goal", "conversational")


def element_key(ctrl):
    """The control's backend identity (stable across re-crawls), as its kind defines it."""
    return kinds.of(ctrl).element_key(ctrl)


def dedup_key(site, ctrl, option):
    return [site, element_key(ctrl), kinds.of(ctrl).dedup_value(option)]


def _options(ctrl, macro=None):
    return kinds.of(ctrl).arguments_for(ctrl, macro) if macro else kinds.of(ctrl).arguments(ctrl)


def _ctrl_view(ctrl):
    return kinds.of(ctrl).view(ctrl)


def candidate_sites(macro, train, sitemaps):
    from miniweb.annotation.macro_locations import MACRO_LOCATIONS
    out, dropped = {}, {}
    # a macro no train site lists (its demos are all on held-out sites) may use any train site
    # whose site map has a control serving it
    unlisted = not any(macro in (MACRO_LOCATIONS.get(s) or {}) for s in train)
    for site in train:
        if not unlisted and macro not in (MACRO_LOCATIONS.get(site) or {}):
            continue
        ctrls = [c for c in sitemap.usable_for(sitemaps.get(site), macro) if kinds.of(c).has_arguments(c)]
        if ctrls:
            out[site] = ctrls
        else:
            dropped[site] = "no usable control in site map" if sitemaps.get(site) else "no site map"
    return out, dropped


def _companions(site_ctrls, ctrl):
    """Other usable controls on the same page that hit the same endpoint (for preset / mid-chain)."""
    out = []
    if not kinds.of(ctrl).chainable:
        return out
    for c in site_ctrls:
        if c is ctrl or c["page"] != ctrl["page"] or not c.get("usable") or not kinds.of(c).chainable:
            continue
        if c["signature"]["path"] != ctrl["signature"]["path"] or c["signature"].get("param") == ctrl["signature"].get("param"):
            continue
        if _options(c):
            out.append(c)
    return out


def propose(macro, n, *, train, sitemaps, seed=0, per_site_cap=2, mid_chain_share=0.3, preset_share=0.15,
            exclude_keys=(), prior_site_counts=None, fill=False):
    """Up to `n` unvalidated tuples (the sampler over-draws; the dry run filters).
    `prior_site_counts` {site: n} = trajectories already kept for this macro by earlier runs;
    they count against the per-site cap. Start kinds follow quotas: every prefix of the list
    holds ~mid_chain_share mid-chain and ~preset_share preset tuples (where the page allows)."""
    rng = random.Random(f"{seed}:{macro}")
    cands, dropped = candidate_sites(macro, train, sitemaps)
    sites = sorted(cands)
    rng.shuffle(sites)
    used_keys = {tuple(k) for k in exclude_keys}
    per_site, ctrl_use, out = dict(prior_site_counts or {}), {}, []
    kinds = {"mid_chain": 0, "preset": 0}
    all_ctrls = {s: [c for c in sitemaps[s]["controls"] if c.get("usable")] for s in sites}
    progress, cap = True, per_site_cap
    while len(out) < n and (progress or (fill and cap == per_site_cap)):
        if not progress:                   # fill: the even cap left the macro short; sites with more take it
            cap = 10 ** 6
        progress = False
        for site in sites:
            if len(out) >= n:
                break
            if per_site.get(site, 0) >= cap:
                continue
            ctrls = sorted(cands[site], key=lambda c: (ctrl_use.get(c["control_id"], 0), rng.random()))
            pick = None
            for ctrl in ctrls:
                opts = [o for o in _options(ctrl, macro) if tuple(dedup_key(site, ctrl, o)) not in used_keys]
                if opts:
                    pick = (ctrl, rng.choice(opts))
                    break
            if not pick:
                continue
            ctrl, opt = pick
            used_keys.add(tuple(dedup_key(site, ctrl, opt)))
            per_site[site] = per_site.get(site, 0) + 1
            ctrl_use[ctrl["control_id"]] = ctrl_use.get(ctrl["control_id"], 0) + 1
            i = len(out) + 1
            want = ("mid_chain" if kinds["mid_chain"] < mid_chain_share * i else
                    "preset" if kinds["preset"] < preset_share * i else "fresh")
            t = _make(macro, site, ctrl, opt, all_ctrls[site], rng, want)
            if t["start"]["kind"] != "mid_chain" and want == "mid_chain" and kinds["preset"] < preset_share * i:
                t = _make(macro, site, ctrl, opt, all_ctrls[site], rng, "preset")
            kinds[t["start"]["kind"]] = kinds.get(t["start"]["kind"], 0) + 1
            out.append(t)
            progress = True
    return out, dropped


_LO = re.compile(r"\b(min|minimum|from|low|lower|start)|\bat least\b")
_HI = re.compile(r"\b(max|maximum|high|upper|end|until)|\b(to|at most|up to)\b")


def _bound(ctrl):
    """'lo' / 'hi' for one end of a numeric range pair (Min Price / Max Price, price_from / price_to), else None."""
    s = re.sub(r"[_\-\[\]]", " ", f"{ctrl.get('name') or ''} {ctrl.get('label') or ''}").lower()
    lo, hi = bool(_LO.search(s)), bool(_HI.search(s))
    return "lo" if lo and not hi else "hi" if hi and not lo else None


def _consistent(ctrl, opt, other, o2):
    """A preset / prefix on the other end of the same range must leave it non-empty: min < max. 2026-09-29: a
    preset Min Price above the task's Max Price gave an empty listing and a self-contradicting task."""
    a, b = _bound(ctrl), _bound(other)
    if not a or not b or a == b:
        return True
    try:
        v, w = float(opt["value"]), float(o2["value"])
    except (TypeError, ValueError, KeyError):
        return True
    return v < w if a == "lo" else w < v


def _make(macro, site, ctrl, opt, site_ctrls, rng, want="fresh"):
    start = {"kind": "fresh", "url": ctrl["page"], "keep": {}}
    comps = [c for c in _companions(site_ctrls, ctrl) if any(_consistent(ctrl, opt, c, o) for o in _options(c))]
    if comps and want == "mid_chain":
        other = rng.choice(comps)
        o2 = rng.choice([o for o in _options(other) if _consistent(ctrl, opt, other, o)])
        start = {"kind": "mid_chain", "url": ctrl["page"], "keep": {other["signature"]["param"]: o2["value"]},
                 "prefix": {"macro": kinds.macro_of(other), "control": _ctrl_view(other),
                            "option": o2}}
    elif comps and want == "preset" and ctrl["page"].rstrip("/") == ctrl["signature"]["path"].rstrip("/"):
        other = rng.choice([c for c in comps if c["signature"]["method"] == "GET"] or comps)
        o2 = rng.choice([o for o in _options(other) if _consistent(ctrl, opt, other, o)])
        if other["signature"]["method"] == "GET":
            q = {other["signature"]["param"]: o2["value"]}
            start = {"kind": "preset", "url": ctrl["page"] + "?" + urlencode(q), "keep": q,
                     "preset": {"control": _ctrl_view(other), "option": o2}}
    key = dedup_key(site, ctrl, opt)
    pid = "p-" + hashlib.sha1(json.dumps([macro, key, start["kind"], start.get("keep")]).encode()).hexdigest()[:10]
    return {"param_id": pid, "macro": macro, "site": site, "control": _ctrl_view(ctrl), "option": opt,
            **({"op": opt["op"]} if opt.get("op") else {}),
            "start": start, "phrasing": rng.choice(PHRASINGS), "dedup_key": key,
            "check": kinds.of(ctrl).check(ctrl, opt, keep=start["keep"])}


# ── feasibility dry run (privileged) ──────────────────────────────────────────

def _pin_status(chk, entries):
    """Record the status class the successful request had (see checks._status_class_ok)."""
    if chk.get("kind") != "request":
        return
    ok = [e for e in entries if checks._accepted(e) and (e.get("method") or "GET").upper() == chk["method"]
          and checks._path(e.get("path")) == checks._path(chk["path"])
          and checks._carries(e, {**chk.get("params", {}), **chk.get("keep", {})})]
    if ok and str(ok[-1].get("status", ""))[:1] == "3":
        chk["status"] = "3xx"


def _weak_check(t, before, after=(), changes=None):
    """A check that can pass without the task being done (the flash judge caught these, 2026-09-28):
    None when the check is sound, else the reason to drop the task."""
    chk = t["check"]
    if chk.get("kind") != "request":
        return None
    want = {**chk.get("params", {}), **chk.get("keep", {})}
    # a rejected login / registration re-renders the form (200) and changes nothing; other POSTs (tools,
    # joins, shares, uploads) legitimately answer 200 without a data change
    if chk["method"] == "POST" and not chk.get("response") and \
            (t.get("macro") in ("sign_by_text", "authenticate_by_form", "create_by_form", "toggle_relationship")
             or re.search(r"/(login|signin|sign-in|register|signup|sign-up)\b", chk["path"])):
        hits = [e for e in after if checks._accepted(e) and (e.get("method") or "").upper() == "POST"
                and checks._path(e.get("path")) == checks._path(chk["path"]) and checks._carries(e, want)]
        redirected = any(str(e.get("status", ""))[:1] == "3" for e in hits)
        if hits and not redirected and changes is not None and not changes:
            return ("weak check: the submission shows no success (no redirect, no data change): a rejected "
                    "form re-renders the same way (invalid login, username taken)")
    if chk["method"] == "GET" and not want and "/api/" in chk["path"]:
        return "weak check: a parameterless background API request (polling happens without the action)"
    if chk["method"] == "GET" and not want and any(checks._path(e.get("path")) == checks._path(chk["path"]) for e in before):
        return "weak check: the start state already makes this request"
    from miniwebgen import formvalues as FV
    typed = {FV.field_key(f) for f in (t["control"].get("fields") or [])
             if (f.get("type") or "text") in ("text", "textarea", "search", "email", "")}
    texts = [v for k, v in ((t.get("option") or {}).get("values") or {}).items()
             if k in typed and isinstance(v, str) and len(v.strip()) > 12 and " " in v.strip()]
    carried = json.dumps(want, ensure_ascii=False)
    missing = [v for v in texts if v.strip() not in carried]
    if missing:
        return f"weak check: typed text never reaches the checked request ({missing[0][:40]!r})"
    return None




def _oracle_once(t, base, b):
    """One privileged run of an oracle step in a fresh session: its evidence
    {entry, requests, changes, clipboard, downloads}."""
    ctx = B.new_context(b)
    page = ctx.new_page()
    downloads = []
    page.on("download", lambda dl: downloads.append({"filename": dl.suggested_filename, "url": dl.url}))
    try:
        kinds.of(t["control"]).setup(ctx, base, t["control"], t["option"])
        B.goto_start(page, base + t["start"]["url"])
        before = B.session_log(ctx, base)
        where = kinds.of(t["control"]).changes_site(t)
        ch0 = B.session_changes(ctx, base, where) or []
        ctrl = t["control"] | {"form": {"submit": t["control"].get("apply_button")}} if t["control"]["kind"] != "form" else dict(t["control"])
        e, _k, _m, _new = sitemap.apply_privileged(page, ctrl, t["option"], base, ctx, len(before))
        after = B.session_log(ctx, base)
        ch1 = B.session_changes(ctx, base, where) or []
        return {"entry": e, "requests": after[len(before):], "changes": [c for c in ch1 if c not in ch0],
                "clipboard": B.clipboard_writes(ctx, base), "downloads": downloads}
    finally:
        ctx.close()


def _stabilize(chk, changes2):
    """Keep only the expected fields a second, independent run reproduces (drops per-request
    values such as confirmation codes). None if an expected change did not recur at all."""
    out = []
    for want in chk["expect"]:
        same = [c for c in changes2 if c["collection"] == want["collection"] and c["op"] == want["op"]
                and (want["item_id"] is None or str(c["item_id"]) == str(want["item_id"]))]
        if not same:
            return None
        best = max(same, key=lambda c: sum(checks._essential(c).get(k) == v for k, v in want["fields"].items()))
        ess = checks._essential(best)
        out.append(dict(want, fields={k: v for k, v in want["fields"].items() if ess.get(k) == v}))
    return dict(chk, expect=out)


def dry_run_oracle(t, base, b):
    """(feasible, detail) for kinds whose check is recorded: run the step twice (fresh sessions),
    record what it changed (else the request it sent), keep what both runs agree on."""
    kind = kinds.of(t["control"])
    ev1 = _oracle_once(t, base, b)
    req1, ch1 = ev1["requests"], ev1["changes"]
    chk = kind.observed_check(ev1, t["site"], t["control"])
    if chk is None:
        return False, "the step changed nothing observable (no data change, request, clipboard or download)"
    if chk["kind"] != "state":
        t["check"] = chk
        ok, detail = checks.evaluate(chk, req1, ch1, ev1["clipboard"])
        return (True, detail) if ok else (False, "the recorded check does not pass its own run: " + detail)
    ch2 = _oracle_once(t, base, b)["changes"]
    stable = _stabilize(chk, ch2)
    if stable is None:
        return False, "the step's data change did not recur in a second run (unstable)"
    t["check"] = stable
    ok1, detail = checks.evaluate(stable, req1, ch1)
    ok2, detail2 = checks.evaluate(stable, [], ch2)
    if ok1 and ok2:
        return True, detail
    return False, "the recorded check does not pass its own runs: " + (detail if not ok1 else detail2)


def dry_run(t, base, b):
    """(feasible, detail). Privileged select_option/clicks in a throw-away session."""
    try:
        own = kinds.of(t["control"]).dry_run(t, base, b)
    except Exception as exc:
        return False, f"dry-run error {type(exc).__name__}: {str(exc)[:160]}"
    if own is not None:
        return own
    if getattr(kinds.of(t["control"]), "oracle", False):
        try:
            return dry_run_oracle(t, base, b)
        except Exception as exc:
            return False, f"dry-run error {type(exc).__name__}: {str(exc)[:160]}"
    ctx = B.new_context(b)
    page = ctx.new_page()
    try:
        kinds.of(t["control"]).setup(ctx, base, t["control"], t["option"])
        B.goto_start(page, base + t["start"]["url"])
        if t["start"]["kind"] == "mid_chain":
            pre = t["start"]["prefix"]
            n0 = len(B.session_log(ctx, base))
            e, _k, _m, _new = sitemap.apply_privileged(page, pre["control"] | {"form": {"submit": pre["control"].get("apply_button")}},
                                                       pre["option"], base, ctx, n0)
            if not e:
                return False, "prefix step produced no request"
        before = B.session_log(ctx, base)
        ok0, _ = checks.evaluate(t["check"], before)
        if ok0:
            return False, "vacuous: the start state already satisfies the check"
        ctrl = t["control"] | {"form": {"submit": t["control"].get("apply_button")}}
        if ctrl["kind"] in ("link", "chips"):
            ctrl["key"] = ctrl["signature"]["param"]
        e, _k, _m, _new = sitemap.apply_privileged(page, ctrl, t["option"], base, ctx, len(before))
        after = B.session_log(ctx, base)
        ok1, detail = checks.evaluate(t["check"], after)
        if not ok1:
            return False, "not solvable by applying the control: " + detail
        weak = _weak_check(t, before, after, B.session_changes(ctx, base))
        if weak:
            return False, weak
        _pin_status(t["check"], after)
        return True, detail
    except Exception as exc:
        return False, f"dry-run error {type(exc).__name__}: {str(exc)[:160]}"
    finally:
        ctx.close()


def sample(macro, n, *, train, sitemaps, bases, workers=4, seed=0, overdraw=3, log=print, **kw):
    """(feasible[:n], infeasible, dropped sites, spare feasible) — fewer than n if candidates run out."""
    import queue
    import threading
    proposed, dropped = propose(macro, n * overdraw, train=train, sitemaps=sitemaps, seed=seed, **kw)
    # keep the stratified order: validate in order, stop once n are feasible
    results = [None] * len(proposed)
    todo = queue.Queue()
    for i, t in enumerate(proposed):
        todo.put(i)
    lock = threading.Lock()
    state = {"ok": 0}

    def worker(wi):
        with B.browser() as b:
            while True:
                with lock:
                    if state["ok"] >= n:
                        return
                try:
                    i = todo.get_nowait()
                except queue.Empty:
                    return
                ok, detail = dry_run(proposed[i], bases[wi % len(bases)], b)
                results[i] = (ok, detail)
                with lock:
                    state["ok"] += ok
    threads = [threading.Thread(target=worker, args=(w,), daemon=True) for w in range(max(1, workers))]
    for th in threads:
        th.start()
    for th in threads:
        th.join()
    feasible, infeasible = [], []
    for t, r in zip(proposed, results):
        if r is None:
            continue
        t["feasibility"] = r[1]
        (feasible if r[0] else infeasible).append(t)
    # workers validate in parallel, so a few extra feasible tuples may exist: keep the first n
    # in stratified order, the rest are returned as `spare` (feasible, unused)
    feasible, spare = feasible[:n], feasible[n:]
    log(f"sample {macro}: validated {len(feasible) + len(spare) + len(infeasible)}, feasible "
        f"{len(feasible) + len(spare)} (using {len(feasible)}), infeasible {len(infeasible)}, "
        f"sites dropped {len(dropped)}")
    return feasible, infeasible, dropped, spare