"""Scripted expert for webgym tasks, run in the real headless Chromium through jev's Browser, so every recorded case has exactly the observation format the agent sees. Writes jev-format training cases: * goal = the full task -> gold = the expert's next operation/target * goal = the current sub-goal -> the same gold (sub-goal-conditioned policy) * goal = a sub-goal just satisfied -> gold = DONE (sub-goal completion) * on the final state, goal = full task -> DONE python finetune/webgym/expert.py [seed0=1] [kinds=flight,hotel,shop] [noise=0.1] Kinds: flight, hotel, car (search form -> results with filters/sort), shop, filter (listings), restaurant, signup (multi-page wizards with validation). Held-out seeds (seed % 10 == 0) are skipped here; eval_gym.py runs the real agent on them. """ import json, os, random, re, sys, time sys.path.insert(0, "/home/ckl/projects/S/jev-ultrafast") sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) os.environ.setdefault("BU_CDP_URL", "http://127.0.0.1:9222") from jev_ultrafast.browser import Browser, StalePage import spec as S BASE = os.environ.get("GYM_URL", "http://127.0.0.1:8811") FORM_KINDS = ("flight", "hotel", "car") # search form -> results page (+ filters / sort / open) STAGED_KINDS = ("restaurant", "signup") # multi-page wizard -> confirmation page def observe(b, tries=10): for i in range(tries): try: return b.observe(screenshot=False) except StalePage: time.sleep(0.3) raise StalePage("never settled") def js(b, expr): return b.evaluate(expr) def status(b): s = js(b, "JSON.stringify({st: window.__gym && window.__gym.status ? window.__gym.status() : null, meta: window.__gym ? window.__gym.meta : null, url: location.href})") return json.loads(s) if s else {} def node_of(b, sel): return js(b, f"window.__gym.node({json.dumps(sel)})") def where(b, sel): return js(b, f"window.__gym.where({json.dumps(sel)})") def exists(b, sel): return bool(js(b, f"!!document.querySelector({json.dumps(sel)})")) def dom_value(b, sel): return js(b, f"(document.querySelector({json.dumps(sel)})||{{}}).value") class Stop(Exception): pass def combo_ok(f, v, t): """A committed autocomplete value matches the target key (city / airport / country), or free text naming it.""" if not isinstance(v, str): return False if v == t or v == t + "-ap": return True if v.startswith("text:"): opt = next((o for o in f["options"] if o["key"] == t), None) name = (opt or {}).get("name") or (opt or {}).get("short") or t return name.lower() in v.lower() return False class Expert: def __init__(self, sp, b, rng): self.sp, self.b, self.rng = sp, b, rng self.fields = {f["key"]: f for f in sp.get("fields", [])} self.stage_of = {f["key"]: i + 1 for i, s in enumerate(sp.get("stages", [])) for f in s["fields"]} self.prefer_chip = rng.random() < 0.6 # ------------------------------------------------------------------ value comparison with the widget's semantics def same(self, key, v, t): f = self.fields.get(key) if f and f["w"] == "combo": return combo_ok(f, v, t) if isinstance(t, list): got = v if isinstance(v, list) else [x for x in str(v or "").split("|") if x] return sorted(got) == sorted(t) if isinstance(t, bool): return str(v).lower() == str(t).lower() return v == t or str(v) == str(t) # ------------------------------------------------------------------ satisfaction per sub-goal def satisfied(self, key, st, meta): sp = self.sp; meta = meta or {} if sp["kind"] in FORM_KINDS: res = meta.get("results") if key == "submit": return res is not None if key in ("filter", "sort", "ctype", "trans", "csort", "open"): if res is None: return False want = sp["post"][key] if key == "filter": return want in (res.get("filter") or "").split("|") if key == "ctype": return want in (res.get("ctype") or "").split("|") if key == "trans": return res.get("trans") == want if key == "open": return meta.get("car") == want return res.get("sort") == want t = sp["target"][key] if res is not None: # past the form: the submitted search must carry the target value return self.same(key, res.get(key), t) return self.same(key, st["state"].get(key), t) if sp["kind"] in STAGED_KINDS: stage = meta.get("stage") or 1; carried = meta.get("carried") or {} if key.startswith("go"): return stage > int(key[2:]) ks = self.stage_of[key]; t = sp["target"][key] if ks < stage or meta.get("done"): return self.same(key, carried.get(key), t) if ks == stage: return self.same(key, st["state"].get(key), t) return False if sp["kind"] == "filter": tg = sp["target"] if meta.get("product") is not None: return meta["product"] == tg["open"] if key == "open" else True lst = meta.get("listing") or {} if key == "open": return False if key.startswith("rm_"): return not lst.get(key[3:]) if key in S.FACETS: return sorted(lst.get(key) or []) == sorted(tg[key]) if key == "rating": return (lst.get("rating") or None) == tg["rating"] if key == "price": return (lst.get("pmin") or None) == tg["pmin"] and (lst.get("pmax") or None) == tg["pmax"] if key == "sort": return (lst.get("sort") or None) == tg["sort"] if key == "page": return lst.get("page") == tg["page"] return False # shop lst = meta.get("listing") or {} post = sp["post"] if key == "open": return meta.get("product") == post["open"] if meta.get("product") is not None: return True if key == "page": return lst.get("page") == post["page"] if key == "sort": return lst.get("sort") == post["sort"] if key == "brand": return lst.get("brand") == post["brand"] if key == "q": return (lst.get("q") or "").lower() == post["q"] if key == "cat": return lst.get("cat") == post["cat"] return False def wrong_results(self, meta): """Submitted with wrong or missing values: an earlier page's sub-goals are not all satisfied any more.""" meta = meta or {} if self.sp["kind"] in FORM_KINDS: if meta.get("results") is None: return False return any(not self.satisfied(k, {"state": {}}, meta) for k in self.sp["target"]) if self.sp["kind"] in STAGED_KINDS: stage = meta.get("stage") or 1 return any(not self.satisfied(k, {"state": {}}, meta) for k in self.sp["target"] if self.stage_of[k] < stage) return False def recover(self, page): """The observed action that goes back to fix wrong carried values (None when there is none).""" if self.sp["kind"] in FORM_KINDS: return next((x for x in page["actions"] if x["label"] == "Modify search"), None) if exists(self.b, "[data-g=back]"): return to_action(self.b, page, "CLICK", "[data-g=back]", None) return None def early_submit(self, page, st, meta): """DART-style detour: submit too early (wrong results / validation errors) so the expert has to recover. None if n/a.""" meta = meta or {} if self.sp["kind"] in FORM_KINDS: if meta.get("results") is not None: return None return to_action(self.b, page, "CLICK", "[data-g=submit]", None) if self.sp["kind"] in STAGED_KINDS: stage = meta.get("stage") or 1 if meta.get("done") or stage > len(self.sp["stages"]): return None req = self.sp["stages"][stage - 1]["required"] if not any(k in self.sp["target"] and not self.satisfied(k, st, meta) for k in req): return None # nothing required is missing: the wizard would accept it, so no detour here return to_action(self.b, page, "CLICK", "[data-g=submit]", None) return None def current(self, st, meta): for i, (key, text) in enumerate(self.sp["subgoals"]): if key == "ret" and self.sp["target"].get("trip") != "round": continue if not self.satisfied(key, st, meta): return i return None # ------------------------------------------------------------------ the next expert step for sub-goal key def step_for(self, key, st, page): sp, b, rng = self.sp, self.b, self.rng if sp["kind"] in FORM_KINDS: if key == "submit": other_dd = self.open_dropdown("submit") return ("CLICK", f"[data-g='{other_dd}']" if other_dd else "[data-g=submit]", None) if key == "filter": i = ["Free cancellation", "Breakfast included", "Pool", "Pet friendly"].index(sp["post"]["filter"]) return ("CLICK", f"[data-g=filter-{i}]", None) if key in ("sort", "csort"): return ("SELECT", "[data-g=sort]", sp["post"][key]) if key == "ctype": return ("CLICK", f"[data-g='ctype-{sp['post']['ctype']}']", None) if key == "trans": return ("CLICK", f"[data-g='trans-{sp['post']['trans']}']", None) if key == "open": return ("CLICK", f"[data-g='car-{sp['post']['open']}']", None) return self.field_step(key, st) if sp["kind"] in STAGED_KINDS: if key.startswith("go"): other_dd = self.open_dropdown("submit") return ("CLICK", f"[data-g='{other_dd}']" if other_dd else "[data-g=submit]", None) return self.field_step(key, st) if sp["kind"] == "filter": return self.filter_step(key, st, page) # ------------------------------------------------ shop post = sp["post"] if key == "q": val = dom_value(b, "[data-g=search]") if (val or "").strip().lower() != post["q"]: return ("TYPE_TEXT", "[data-g=search]", post["q"]) if sp["search_button"] and rng.random() < 0.4: return ("CLICK", "[data-g=search-btn]", None) return ("PRESS_ENTER", "press_enter", None) if key == "sort": return self.sort_step(post["sort"]) if key == "brand": return ("CLICK", f"[data-g=brand-{post['brand']}]", None) if key == "cat": return ("CLICK", f"[data-g='cat-{post['cat']}']", None) if key == "page": return self.page_step(post["page"]) if key == "open": return ("CLICK", f"a[href$='/p/{post['open']}']", None) raise Stop(f"no step for {key}") def sort_step(self, want): if self.sp["sort_impl"] == "select": return ("SELECT", "[data-g=sort]", want) if js(self.b, "document.getElementById('sl').hidden"): return ("CLICK", "[data-g=sort-trigger]", None) return ("CLICK_TEXT", "#sl [role=option]", want) def page_step(self, n): if exists(self.b, "[data-g=next]") and (self.sp["pager"] == "next" or self.rng.random() < 0.5): return ("CLICK", "[data-g=next]", None) return ("CLICK_TEXT", "nav.pager a", str(n)) # ------------------------------------------------ faceted listing def filter_step(self, key, st, page): sp, b, rng = self.sp, self.b, self.rng tg = sp["target"]; lst = ((status(b).get("meta") or {}).get("listing")) or {} if key.startswith("rm_"): fk = key[3:]; v = sp["pre"][fk][0] chip, box = f"[data-g='chip-{fk}-{v}']", f"[data-g='{fk}-{v}']" if self.prefer_chip or where(b, box) is None: return ("CLICK", chip, None) return ("CLICK", box, None) if key in S.FACETS: have = lst.get(key) or [] for v in tg[key]: if v in have: continue sel = f"[data-g='{key}-{v}']" if where(b, sel) is None: # hidden behind "Show more" return ("CLICK", f"[data-g='more-{key}']", None) return ("CLICK", sel, None) for v in have: if v not in tg[key]: return ("CLICK", f"[data-g='chip-{key}-{v}']", None) if key == "rating": return ("CLICK", f"[data-g='rating-{tg['rating']}']", None) if key == "price": for k in ("pmin", "pmax"): if tg[k] and (dom_value(b, f"[data-g={k}]") or "").strip() != tg[k]: return ("TYPE_TEXT", f"[data-g={k}]", tg[k]) if rng.random() < 0.3 and any(a["id"] == "press_enter" for a in page["actions"]): return ("PRESS_ENTER", "press_enter", None) return ("CLICK", "[data-g=price-apply]", None) if key == "sort": return self.sort_step(tg["sort"]) if key == "page": return self.page_step(tg["page"]) if key == "open": return ("CLICK", f"[data-g='item-{tg['open']}']", None) raise Stop(f"no step for {key}") def open_dropdown(self, key): """data-g of an expanded custom-dropdown trigger that does not belong to field `key` (it may cover other controls).""" return js(self.b, f"(([...document.querySelectorAll('[aria-haspopup=listbox][aria-expanded=true]')].find(e=>!e.getAttribute('data-g').startsWith({json.dumps(key + '-')}))||{{}}).getAttribute||(()=>null))('data-g')") # ------------------------------------------------ one form field (shared by the search forms and the wizards) def field_step(self, key, st): sp, b, rng = self.sp, self.b, self.rng open_ = set(st.get("open") or []) f = self.fields[key] t = sp["target"][key] # close a popup that belongs to another field first (it may cover what we need) for o in open_: if o and not o.startswith(key) and o.endswith("-dialog"): other = o[: -len("-dialog")] for btn in (f"[data-g={other}-close]", f"[data-g={other}-done]"): if exists(b, btn): return ("CLICK", btn, None) # another field's custom dropdown is still expanded (it can cover what we need): close it first other_dd = self.open_dropdown(key) if other_dd: return ("CLICK", f"[data-g='{other_dd}']", None) if f["w"] == "combo": dlg_open = (key + "-dialog") in open_ if f["trigger"] and not dlg_open: return ("CLICK", f"[data-g={key}-trigger]", None) opt = next((o for o in f["options"] if o["key"] == t), None) or {"name": t, "short": t, "alias": ""} name = opt.get("name") or opt["short"] if f["suggest"]: for k in ([t + "-ap", t] if rng.random() < 0.5 else [t, t + "-ap"]): sel = f"[data-g='{key}-opt-{k}']" if where(b, sel) is not None: first = js(b, f"(document.querySelector('#{key}-list [role=option]')||{{}}).getAttribute && document.querySelector('#{key}-list [role=option]').getAttribute('data-g')") if first == f"{key}-opt-{k}" and rng.random() < 0.25: return ("PRESS_ENTER", "press_enter", None) return ("CLICK", sel, None) alias = opt.get("alias") or "" typed = rng.choice([name, name, name[: max(3, len(name) // 2 + 1)], alias or name]) return ("TYPE_TEXT", f"[data-g={key}-input]", typed) return ("TYPE_TEXT", f"[data-g={key}-input]", name) if f["w"] == "choice": if f["impl"] == "select": return ("SELECT", f"[data-g={key}-select]", t) if f["impl"] == "dropdown": if js(b, f"document.querySelector('[data-g={key}-trigger]').getAttribute('aria-expanded')") != "true": return ("CLICK", f"[data-g={key}-trigger]", None) return ("CLICK", f"[data-g='{key}-{t}']", None) if f["w"] == "date": if f["impl"] == "typed": y, m, d = t.split("-") return ("TYPE_TEXT", f"[data-g={key}-input]", f"{m}/{d}/{y}") if f["impl"] == "selects": y, m, d = t.split("-") for part, want in (("month", m), ("day", d), ("year", y)): if dom_value(b, f"[data-g={key}-{part}]") != want: return ("SELECT", f"[data-g={key}-{part}]", want) raise Stop("date selects set but state differs") if (key + "-dialog") not in open_: return ("CLICK", f"[data-g={key}-trigger]", None) day = f"[data-g={key}-day-{t}]" if exists(b, day): if js(b, f"document.querySelector({json.dumps(day)}).getAttribute('aria-pressed')") != "true": return ("CLICK", day, None) return ("CLICK", f"[data-g={key}-done]", None) shown = js(b, f"document.querySelector('[data-g={key}-dialog] strong').textContent") mon, yr = shown.split() cur = int(yr) * 12 + S.MONTHS.index(mon) want = int(t[:4]) * 12 + int(t[5:7]) - 1 return ("CLICK", f"[data-g={key}-{'next' if want > cur else 'prev'}]", None) if f["w"] == "count": v = st["state"].get(key) if f["impl"] == "select": return ("SELECT", f"[data-g={key}-select]", str(t)) if f["impl"] == "popup" and (key + "-dialog") not in open_: return ("CLICK", f"[data-g={key}-trigger]", None) return ("CLICK", f"[data-g={key}-{'inc' if v < t else 'dec'}]", None) if f["w"] == "toggle": return ("CLICK", f"[data-g={key}-{'switch' if f['impl'] == 'switch' else 'check'}]", None) if f["w"] == "text": return ("TYPE_TEXT", f"[data-g={key}-input]", t) if f["w"] == "slots": return ("CLICK", f"[data-g='{key}-{t}']", None) if f["w"] == "multi": have = st["state"].get(key) or [] for v in t: if v not in have: return ("CLICK", f"[data-g='{key}-{v}']", None) for v in have: if v not in t: return ("CLICK", f"[data-g='{key}-{v}']", None) raise Stop(f"no step for field {key}") def S_name(key): return next(c[1] for c in S.CITIES if c[0] == key.replace("-ap", "")) def to_action(b, page, op, sel, val): """Map an expert step to the observed jev action (same node, same kind). None when it is not on screen.""" if op == "PRESS_ENTER": return next((a for a in page["actions"] if a["id"] == "press_enter"), None) if op == "CLICK_TEXT": node = js(b, f"window.__gym.nodeText({json.dumps(sel)}, {json.dumps(val)})") sel_for_where = None else: node = node_of(b, sel) kind = {"CLICK": "click", "CLICK_TEXT": "click", "TYPE_TEXT": "fill", "SELECT": "select"}[op] for a in page["actions"]: if a.get("node") != node or a["kind"] != kind: continue if kind == "select" and not (a.get("value") == val or a["label"].split(" → ")[-1].strip() == val): continue if kind == "click" and a["label"].startswith("Open ") and op == "CLICK" and any(x.get("node") == node and x["kind"] == "fill" for x in page["actions"]): continue # the "Open " click of an editable field is not the intended click return a return None def hist_entry(a, text, changed): return {"action": a["label"], "kind": a["kind"], "text": text, "page_changed": changed} def case(page, goal, gold_op, gold_id, a, hist, sp, tag): return {"page": -1, "url": page["url"], "title": page["title"], "goal": goal, "gold_op": gold_op, "gold_id": gold_id, "kind": a["kind"] if a else "done", "label": a["label"] if a else "", "history": list(hist)[-10:], "source": "webgym", "skill": tag, "seed": sp["seed"], "gym_kind": sp["kind"], "page_obj": {k: page[k] for k in ("url", "title", "text", "actions")}} def run_episode(kind, seed, out, rng, noise=0.1, max_steps=70): sp = S.make(kind, seed) b = Browser(f"{BASE}/t/{kind}/{seed}") ex = Expert(sp, b, rng) hist, n, prev_sub = [], 0, None # at most one premature submission per episode (wrong results / validation errors the expert must then recover from) early_done = rng.random() >= (0.35 if kind in STAGED_KINDS else 0.25) early_at = rng.randint(3, 12) try: page = observe(b) for step in range(max_steps): info = status(b); st = info.get("st") or {}; meta = info.get("meta") if st and "open" not in st: st["open"] = [] if ex.wrong_results(meta): # recovery: the search / wizard step was submitted with wrong values -> back to the form a = ex.recover(page) if a is None: raise Stop("wrong results and no way back") out.write(json.dumps(case(page, sp["goal"], "CLICK", a["id"], a, hist, sp, "recover"), ensure_ascii=False) + "\n"); n += 1 before = page["fingerprint"]; b.act(a, page); time.sleep(0.3); page = observe(b) hist.append(hist_entry(a, None, page["fingerprint"] != before)); prev_sub = None continue cur = ex.current(st, meta) # sub-goal completion: the previous sub-goal is now satisfied -> DONE under that sub-goal if prev_sub is not None and prev_sub != cur: sg = sp["subgoals"][prev_sub][1] out.write(json.dumps(case(page, sg, "DONE", "DONE", None, hist, sp, "subgoal_done"), ensure_ascii=False) + "\n"); n += 1 if cur is None: out.write(json.dumps(case(page, sp["goal"], "DONE", "DONE", None, hist, sp, "task_done"), ensure_ascii=False) + "\n") return n + 1, True key = sp["subgoals"][cur][0] op, sel, val = ex.step_for(key, st, page) # cookie banner in the way of anything below it: accept it a = to_action(b, page, op, sel, val) if a is None and op != "PRESS_ENTER": if js(b, "!!document.getElementById('cookie')"): op, sel, val = "CLICK", "[data-g=cookie-accept]", None; a = to_action(b, page, op, sel, val) elif op != "CLICK_TEXT" and where(b, sel) in (1, -1): sid = "scroll_down" if where(b, sel) == 1 else "scroll_up" a = next((x for x in page["actions"] if x["id"] == sid), None); op = sid.upper() elif op == "CLICK_TEXT": a = next((x for x in page["actions"] if x["id"] == "scroll_down"), None); op = "SCROLL_DOWN" if a is None: raise Stop(f"step {step}: {op} {sel} not actionable") gold_op = op if op in ("SCROLL_DOWN", "SCROLL_UP", "PRESS_ENTER") else {"CLICK_TEXT": "CLICK"}.get(op, op) text = val if op == "TYPE_TEXT" else None out.write(json.dumps(case(page, sp["goal"], gold_op, a["id"], a, hist, sp, key), ensure_ascii=False) + "\n") out.write(json.dumps(case(page, sp["subgoals"][cur][1], gold_op, a["id"], a, hist, sp, key + "@sub"), ensure_ascii=False) + "\n") n += 2 before = page["fingerprint"] b.act(a, page, text=text); time.sleep(0.15 if op != "CLICK" else 0.25) page = observe(b) hist.append(hist_entry(a, text, page["fingerprint"] != before)) prev_sub = cur # DART-style noise: a harmless detour the expert then has to recover from (not recorded as a gold step) if not early_done and step >= early_at: info2 = status(b); st2 = info2.get("st") or {}; st2.setdefault("open", []) z = ex.early_submit(page, st2, info2.get("meta")) if z is not None: early_done = True try: before = page["fingerprint"]; b.act(z, page); time.sleep(0.3); page = observe(b) hist.append(hist_entry(z, None, page["fingerprint"] != before)); prev_sub = None; continue except Exception: page = observe(b) if rng.random() < noise: submit_node = node_of(b, "[data-g=submit]") decoy = {"Deals", "Help", "Blog", "Careers", "Travel guides", "Gift cards", "Manage booking", "About", "Privacy", "Contact", "About us", "New arrivals"} cands = [x for x in page["actions"] if x["kind"] == "click" and x.get("node") != submit_node and ((x["label"].startswith("Open ")) or (x.get("role") == "link" and x["label"] in decoy))] if cands: z = rng.choice(cands) try: before = page["fingerprint"]; b.act(z, page); time.sleep(0.2); page = observe(b) hist.append(hist_entry(z, None, page["fingerprint"] != before)) except Exception: page = observe(b) raise Stop("step budget") finally: b.close() def main(): out_f, n_ep = sys.argv[1], int(sys.argv[2]) seed0 = int(sys.argv[3]) if len(sys.argv) > 3 else 1 kinds = (sys.argv[4] if len(sys.argv) > 4 else "flight,hotel,shop").split(",") noise = float(sys.argv[5]) if len(sys.argv) > 5 else 0.1 rng = random.Random(seed0) ok = fail = cases = 0; reasons = {} t0 = time.time() with open(out_f, "a") as out: seed = seed0 while ok + fail < n_ep: if seed % 10 == 0: seed += 1; continue kind = kinds[(ok + fail) % len(kinds)] try: c, done = run_episode(kind, seed, out, rng, noise) cases += c; ok += 1 except Exception as e: fail += 1; r = f"{type(e).__name__}: {str(e)[:60]}"; reasons[r] = reasons.get(r, 0) + 1 if os.environ.get("GYM_DEBUG"): print(f" FAIL {kind} {seed}: {r}", flush=True) seed += 1 if (ok + fail) % 25 == 0: print(f" {ok+fail} episodes ok={ok} fail={fail} cases={cases} {time.time()-t0:.0f}s", flush=True) print(f"done: ok={ok} fail={fail} cases={cases} failures={sorted(reasons.items(), key=lambda kv: -kv[1])[:6]}") if __name__ == "__main__": main()