Spaces:
Running
Running
| """Stage 4 — executor. A coding LLM writes a script against the restricted wrapper | |
| (src/miniwebgen/actions.py API_DOC) using the macro guide and privileged DOM facts; the harness | |
| runs it in a fresh session and records only screenshots + pixel actions. | |
| Per attempt (fresh browser context = fresh Flask session): | |
| goto start URL (the only navigation) -> [prefix step script] -> target step script | |
| -> final screenshot -> backend gate (checks.gate over the session request log) | |
| A failed attempt (policy violation, script error, gate failure) is kept with its reason — | |
| it is a DPO negative — and the next attempt's prompt carries the error and the page state. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import random | |
| import re | |
| import time | |
| import traceback | |
| from pathlib import Path | |
| from miniwebgen import browser as B | |
| from miniwebgen import checks, config, elements, kinds | |
| from miniwebgen.actions import (API_DOC, Act, ActionError, Dom, Expect, PolicyViolation, Recorder, Session, | |
| run_script) | |
| from miniwebgen.driver import PlaywrightDriver | |
| SYSTEM = ("You are a browser-automation engineer. You write short, robust Python scripts that complete one " | |
| "UI step using ONLY the API below. The script's actions are recorded as a demonstration, so act " | |
| "like a careful human: scroll to things before using them, one purposeful action at a time, no " | |
| "unnecessary clicks.\n\n" + API_DOC) | |
| def _clip(s, n): | |
| s = str(s or "") | |
| return s if len(s) <= n else s[:n] + "…" | |
| def step_spec(step): | |
| lines = [f"MACRO: {step['macro']}", f"STEP: {step['subtask']}", f"(full task: {step['instruction']})"] | |
| op = step["control"].get("opener") | |
| if op: | |
| lines.append(f"OPEN FIRST: the control is revealed by clicking css={op['css']!r} text={op.get('text')!r} " | |
| f"(a dialog, tab or panel) — click it, then re-query the control (task['opener_css'])") | |
| lines += kinds.of(step["control"]).step_spec(step) | |
| lines.append(f"BACKEND PROOF checked by expect.backend(): {checks.describe(step['check'])}") | |
| return "\n".join(lines) | |
| def script_task(step): | |
| """The `task` dict a script sees: the exact targets, so selectors are never re-typed.""" | |
| t = kinds.of(step["control"]).script_task(step) | |
| if step["control"].get("opener"): | |
| t["opener_css"] = step["control"]["opener"]["css"] | |
| return t | |
| def page_state(driver, step): | |
| """Privileged facts about the live page for the prompt.""" | |
| facts = [f"URL: {driver.url()}", f"VIEWPORT: {driver.viewport()[0]}x{driver.viewport()[1]}"] | |
| try: | |
| hits = kinds.of(step["control"]).locate(driver, step) | |
| for _k, info in hits[:2]: | |
| vw, vh = driver.viewport() | |
| b = info.get("box") or [0, 0, 0, 0] | |
| on = info.get("visible") and b[1] >= 0 and b[1] + b[3] <= vh | |
| facts.append(f"TARGET NOW: <{info['tag']}> visible={info.get('visible')} on_screen={bool(on)} " | |
| f"box=({b[0]:.0f},{b[1]:.0f},{b[2]:.0f},{b[3]:.0f})" | |
| + (f" selected={info.get('selected_text')!r}" if info["tag"] == "select" else "")) | |
| if not hits: | |
| facts.append("TARGET NOW: not found with that selector/text — inspect the outline") | |
| except Exception as exc: | |
| facts.append(f"TARGET NOW: lookup failed ({exc})") | |
| facts.append("PAGE OUTLINE (first elements):\n" + _clip(driver.outline(60), 3500)) | |
| return "\n".join(facts) | |
| def _code(raw): | |
| m = re.search(r"```(?:python)?\s*\n(.*?)```", raw or "", re.S) | |
| return (m.group(1) if m else (raw or "")).strip() | |
| def write_script(step, guide_txt, state, feedback=None, model=config.MODEL_EXECUTOR): | |
| from helpers.llm import call_llm | |
| example = kinds.of(step["control"]).example | |
| prompt = (f"GUIDE for {step['macro']} (distilled from human demonstrations):\n{guide_txt}\n\n" | |
| f"{step_spec(step)}\n\nLIVE PAGE:\n{state}\n\nEXAMPLE (different page):\n```python\n{example}\n```\n") | |
| if feedback: | |
| prompt += f"\nA PREVIOUS ATTEMPT FAILED:\n{feedback}\nFix the cause.\n" | |
| prompt += (f"\nThe script receives task = {json.dumps(script_task(step), ensure_ascii=False)}\n" | |
| "Use task[...] for selectors/texts verbatim — never rewrite a selector. After any action that may " | |
| "reload the page (committing a choice, clicking Apply/a link), re-query elements with dom.one(...) " | |
| "before using them again.\nWrite the script now, in one ```python block. End it with expect.backend().") | |
| raw = call_llm(prompt, system=SYSTEM, model=model, max_tokens=8000, temperature=0.3) | |
| return _code(raw) | |
| def _step_of(task, which): | |
| if which == "prefix": | |
| pre = task["start"]["prefix"] | |
| return {"macro": pre["macro"], "subtask": task.get("prefix_subtask") or "", "instruction": task["instruction"], | |
| "control": pre["control"], "option": pre["option"], "check": pre["check"]} | |
| return {"macro": task["macro"], "subtask": task["subtask"], "instruction": task["instruction"], | |
| "control": task["control"], "option": task["option"], "check": task["check"]} | |
| def run_attempt(task, base, b, out_dir, n, guides, feedback=None, model=config.MODEL_EXECUTOR, log=print): | |
| """One attempt in a fresh session. Returns the attempt record (also written to disk).""" | |
| out_dir = Path(out_dir) | |
| out_dir.mkdir(parents=True, exist_ok=True) | |
| rng = random.Random(f"{task['task_id']}:{n}") | |
| ctx = B.new_context(b) | |
| page = ctx.new_page() | |
| vp = config.VIEWPORT | |
| driver = PlaywrightDriver(page, vp) | |
| rec = Recorder(out_dir, vp) | |
| logs, entries = [], [] | |
| record = {"attempt": n, "task_id": task["task_id"], "ok": False, "executed": False, "stage": None, | |
| "error": None, "scripts": {}, "t0": time.time()} | |
| try: | |
| kinds.of(task["control"]).setup(ctx, base, task["control"], task["option"]) | |
| B.goto_start(page, base + task["start"]["url"]) | |
| record["start_url"] = page.url | |
| which_steps = (["prefix"] if task["start"]["kind"] == "mid_chain" else []) + ["target"] | |
| for which in which_steps: | |
| step = _step_of(task, which) | |
| rec.span = which | |
| record["stage"] = which | |
| state = page_state(driver, step) | |
| fb = (feedback or {}).get(which) | |
| code = write_script(step, guides[step["macro"]], state, feedback=fb, model=model) | |
| record["scripts"][which] = code | |
| (out_dir / f"{which}.py").write_text(code) | |
| if not code: | |
| raise RuntimeError("the LLM returned no script") | |
| session = Session(driver, rec, rng, max_actions=30, time_budget=180) | |
| chk = step["check"] | |
| session.files = kinds.of(step["control"]).files(step) | |
| session.upload_input = step["control"]["css"] if step["control"]["kind"] == "upload" else None | |
| session.open_urls = kinds.of(step["control"]).open_urls(step) | |
| session.forbidden_keys = kinds.of(step["control"]).forbidden_keys(step) | |
| expect = Expect(session, lambda chk=chk: checks.evaluate( | |
| chk, B.session_log(ctx, base), B.session_changes(ctx, base) if _needs_changes(chk) else None, | |
| B.clipboard_writes(ctx, base) if chk.get("kind") == "clipboard" else None, rec.answer, rec.seen)) | |
| run_script(code, act=Act(session), dom=Dom(session), expect=expect, task=script_task(step), | |
| log=lambda m: logs.append(f"[{which}] {m}")) | |
| record["executed"] = True | |
| record["stage"] = "backend" | |
| except PolicyViolation as exc: | |
| record["error"] = f"policy violation: {exc}" | |
| except AssertionError as exc: | |
| record["error"] = f"assertion failed: {str(exc)[:400]}" | |
| except ActionError as exc: | |
| record["error"] = f"action refused: {exc}" | |
| except Exception as exc: | |
| tb = traceback.extract_tb(exc.__traceback__) | |
| where = next((f"script line {f.lineno}" for f in reversed(tb) if f.filename == "<executor-script>"), "") | |
| record["error"] = f"{type(exc).__name__}: {str(exc)[:300]} {where}".strip() | |
| finally: | |
| try: | |
| record["final_screenshot"] = rec.final(driver.screenshot()) | |
| record["final_url"] = driver.page.url | |
| html = driver.page.content() | |
| (out_dir / "final.html").write_text(html[:400000]) | |
| except Exception: | |
| pass | |
| entries = B.session_log(ctx, base) | |
| changes = B.session_changes(ctx, base) | |
| records = B.session_record(ctx, base) | |
| record["dialogs"], record["downloads"] = driver.dialogs, driver.downloads | |
| try: | |
| ctx.close() | |
| except Exception: | |
| pass | |
| (out_dir / "server_log.json").write_text(json.dumps(entries, indent=1, default=str)) | |
| (out_dir / "data_changes.json").write_text(json.dumps(changes, indent=1, default=str)) | |
| clip = [e.get("value") for e in records if e.get("action") == "clipboard_write"] | |
| copies = [{"type": "action", "action": "clipboard_write", "target": "", "value": v} for v in clip] | |
| ok, detail, report = checks.gate(task["check"], task["verifier"], entries, action_events(rec.steps) + copies, | |
| changes, clip, rec.answer, rec.seen) | |
| record["answer"] = rec.answer | |
| record["backend"] = {"ok": ok, "detail": detail, "verify_task": report.get("by_macro")} | |
| record["ok"] = bool(record["executed"] and ok) | |
| if record["executed"] and not ok: | |
| record["error"] = f"backend check failed: {detail}" | |
| record["steps"] = rec.steps | |
| elements.write_attempt(out_dir, rec.steps) # privileged element side file (every attempt) | |
| record["logs"] = logs[-50:] | |
| record["duration_s"] = round(time.time() - record.pop("t0"), 1) | |
| (out_dir / "attempt.json").write_text(json.dumps(record, indent=1, default=str)) | |
| return record | |
| def _needs_changes(chk): | |
| return chk.get("kind") == "state" or (chk.get("kind") == "answer" and (chk.get("requires") or {}).get("kind") == "state") | |
| def action_events(steps): | |
| """Recorded pixel actions as human-schema action events (for verify_task / the judge digest).""" | |
| out = [] | |
| for s in steps: | |
| tgt = s.get("target") or {} | |
| desc = f"{tgt.get('tag', '')} '{tgt.get('label') or tgt.get('text') or tgt.get('name') or ''}'".strip() | |
| ev = {"type": "action", "action": s["type"], "target": desc if tgt else "", "url": s.get("url"), | |
| "timestamp": s.get("t")} | |
| if s["type"] == "type": | |
| ev["value"] = s["text"] | |
| elif s["type"] == "key": | |
| ev["value"] = s["key"] | |
| elif s["type"] == "scroll": | |
| ev["value"] = f"dy={s.get('dy')}" | |
| elif s["type"] == "answer": | |
| ev["value"] = s["text"] | |
| out.append(ev) | |
| return out | |
| def execute_task(task, base, b, episodes_dir, guides, attempts=3, model=config.MODEL_EXECUTOR, log=print): | |
| """Up to `attempts` fresh attempts; stops at the first that passes the backend gate.""" | |
| tdir = Path(episodes_dir) / task["task_id"] | |
| records, feedback = [], None | |
| for n in range(1, attempts + 1): | |
| rec = run_attempt(task, base, b, tdir / f"a{n}", n, guides, feedback=feedback, model=model, log=log) | |
| rec["dir"] = str((tdir / f"a{n}").relative_to(config.ROOT)) | |
| records.append(rec) | |
| log(f" {task['task_id']} a{n}: {'OK' if rec['ok'] else 'FAIL'} " | |
| f"({len(rec['steps'])} actions, {rec['duration_s']}s) {rec.get('error') or ''}"[:220]) | |
| if rec["ok"]: | |
| break | |
| stage = rec.get("stage") if rec.get("stage") in ("prefix", "target") else "target" | |
| prev = rec["scripts"].get(stage, "") | |
| feedback = {stage: f"error: {rec.get('error')}\nscript was:\n```python\n{prev}\n```\n" | |
| f"actions it recorded: {[s['type'] for s in rec['steps'] if s.get('span') == stage]}"} | |
| return records | |