synth/generators/playwright/generate.py

"""Synthesize MCPMark *playwright* tasks.

Self-contained generation entry point for the playwright service. It deliberately
does NOT touch the shared ``synth/generate.py``; everything playwright needs lives
in this folder.

Usage (from repo root, inside the playwright-enabled conda env):

    python -m synth.generators.playwright.generate --n 4
    python -m synth.generators.playwright.generate --n 4 --types web_info_extraction
    python -m synth.generators.playwright.generate --n 4 --no-llm     # offline content

For each task it:
  1. builds a test environment (the source index.html) under test_environments/<category>/
  2. writes tasks/playwright/<suite>/<category>/<task_id>/{meta.json,description.md,verify.py}
  3. self-checks: drives the oracle solve() (a real Playwright browser) and asserts
     verify.py exits 0 on the oracle answer and non-zero on the untouched environment.
     The verifier runs with PLAYWRIGHT_TEST_DIR pointing at the environment.

Run a generated task with the normal pipeline, e.g.:

    python -m pipeline --mcp playwright --k 1 --models deepseek-v3.2-instruct \\
        --tasks synth_web_info_extraction_01 --exp-name synth-run
"""

import argparse
import datetime
import json
import os
import random
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

from synth.llm import LLMClient

from . import REGISTRY

# .../mcpmark/synth/generators/playwright/generate.py -> mcpmark
REPO_ROOT = Path(__file__).resolve().parents[3]

MCP = "playwright"
TEST_DIR_ENV = "PLAYWRIGHT_TEST_DIR"


def _state_content(env_dir: Path) -> str:
    """A short relative listing of the test environment for meta.json."""
    items = sorted(
        str(p.relative_to(env_dir)) + ("/" if p.is_dir() else "")
        for p in env_dir.rglob("*")
    )
    return "\n".join(items[:40])


def write_task(gen, spec, category, task_id, suite, tasks_root, author, env_dir):
    task_dir = tasks_root / MCP / suite / category / task_id
    task_dir.mkdir(parents=True, exist_ok=True)

    (task_dir / "description.md").write_text(gen.description(spec), encoding="utf-8")
    (task_dir / "verify.py").write_text(gen.verify_src(spec), encoding="utf-8")

    meta = {
        "task_id": task_id,
        "task_name": gen.CATEGORY_NAME,
        "category_id": category,
        "category_name": gen.CATEGORY_NAME,
        "description": gen.description(spec).split("\n\n")[1].strip()[:200],
        "author": author,
        "created_at": datetime.date.today().isoformat(),
        "difficulty": gen.DIFFICULTY,
        "tags": gen.TAGS,
        "mcp": [MCP],
        "meta_data": {
            "stateType": "synthetic",
            "stateContent": _state_content(env_dir),
            "stateUrl": None,
            "stateOriginalUrl": None,
        },
    }
    (task_dir / "meta.json").write_text(json.dumps(meta, indent=2), encoding="utf-8")
    return task_dir


def self_check(gen, spec, env_dir) -> tuple[bool, str]:
    """Solve in a temp copy and confirm verify.py accepts it (and rejects unsolved)."""
    verify_src = gen.verify_src(spec)
    tmp = Path(tempfile.mkdtemp(prefix="synth_pw_check_"))
    try:
        # Negative check: untouched environment should fail verification.
        neg = tmp / "neg"
        shutil.copytree(env_dir, neg)
        vfile = tmp / "verify.py"
        vfile.write_text(verify_src, encoding="utf-8")
        neg_rc = _run_verify(vfile, neg)
        if neg_rc == 0:
            return False, "verifier passes on the UNSOLVED environment (too weak)"

        # Positive check: oracle solution should pass verification.
        pos = tmp / "pos"
        shutil.copytree(env_dir, pos)
        httpd = None
        try:
            if getattr(gen, "NEEDS_SERVER", False):
                # Write task: bring the stateful server up so the oracle's form
                # submission is persisted to pos/state.json, then verify reads it.
                from .base import start_state_server
                httpd, port = start_state_server(tmp)
                gen.solve(pos, spec, f"http://127.0.0.1:{port}/pos/")
            else:
                gen.solve(pos, spec)  # read task: real browser over file://
        except Exception as exc:
            return False, f"oracle solve() raised {type(exc).__name__}: {exc}"
        finally:
            if httpd is not None:
                httpd.shutdown()
        pos_rc = _run_verify(vfile, pos)
        if pos_rc != 0:
            return False, "verifier REJECTS the oracle solution (rc=%d)" % pos_rc
        return True, "oracle passes, unsolved fails"
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def _run_verify(verify_file: Path, test_dir: Path) -> int:
    proc = subprocess.run(
        [sys.executable, str(verify_file)],
        env={TEST_DIR_ENV: str(test_dir), "PATH": os.environ.get("PATH", "")},
        capture_output=True,
        text=True,
    )
    return proc.returncode


def main():
    ap = argparse.ArgumentParser(description="Synthesize MCPMark playwright tasks.")
    ap.add_argument("--n", type=int, default=4, help="number of tasks to generate")
    ap.add_argument(
        "--types",
        default=",".join(REGISTRY),
        help="comma-separated task types (default: all): " + ", ".join(REGISTRY),
    )
    ap.add_argument("--suite", default="standard", choices=["standard", "easy"])
    ap.add_argument(
        "--difficulty",
        default="medium",
        choices=["easy", "medium", "hard", "none", "all"],
        help="difficulty tier for generators that support it",
    )
    ap.add_argument("--seed", type=int, default=1)
    ap.add_argument("--author", default="synth-pipeline")
    ap.add_argument("--no-llm", action="store_true", help="offline content, no API calls")
    ap.add_argument("--model", default="deepseek/deepseek-chat")
    ap.add_argument("--tasks-dir", default=str(REPO_ROOT / "tasks"))
    ap.add_argument("--env-dir", default=str(REPO_ROOT / "test_environments"))
    ap.add_argument("--prefix", default="synth", help="category-id prefix")
    args = ap.parse_args()

    types = [t.strip() for t in args.types.split(",") if t.strip()]
    for t in types:
        if t not in REGISTRY:
            ap.error(f"unknown type '{t}'. Choices: {', '.join(REGISTRY)}")

    rng = random.Random(args.seed)
    llm = LLMClient(model=args.model, enabled=not args.no_llm, seed=args.seed)
    print(f"LLM content generation: {'ON (' + args.model + ')' if llm.enabled else 'OFF (offline)'}")

    tasks_root = Path(args.tasks_dir)
    env_root = Path(args.env_dir)

    ok_count, fail_count = 0, 0
    for i in range(args.n):
        key = types[i % len(types)]
        gen = REGISTRY[key](difficulty=args.difficulty)
        category = f"{args.prefix}_{key}_{i + 1:02d}"
        env_dir = env_root / category
        if env_dir.exists():
            shutil.rmtree(env_dir)
        env_dir.mkdir(parents=True, exist_ok=True)

        spec = gen.build(env_dir, llm, rng)
        passed, detail = self_check(gen, spec, env_dir)
        if not passed:
            fail_count += 1
            print(f"  ✗ {category}: SELF-CHECK FAILED — {detail}")
            shutil.rmtree(env_dir, ignore_errors=True)
            continue

        task_dir = write_task(
            gen, spec, category, key, args.suite, tasks_root, args.author, env_dir
        )
        ok_count += 1
        try:
            rel = task_dir.relative_to(REPO_ROOT)
        except ValueError:
            rel = task_dir
        n_items = sum(1 for _ in env_dir.rglob("*") if _.is_file())
        print(f"  ✓ {category}: {n_items} files | {detail}")
        print(f"      task: {rel}")

    print(f"\nGenerated {ok_count}/{args.n} tasks ({fail_count} failed self-check).")
    if ok_count:
        print("Run one with:")
        print(
            "  python -m pipeline --mcp playwright --k 1 "
            "--models deepseek-v3.2-instruct "
            f"--tasks {args.prefix}_{types[0]}_01 --exp-name synth-run"
        )


if __name__ == "__main__":
    main()