Blame for src/miniweb/browsergym/task.py

"""BrowserGym task wrapper for MiniWeb — PoC (browser-gym branch).

Bridges MiniWeb onto BrowserGym's AbstractBrowserTask:
  setup(page)                 -> reset session, navigate to start URL, return goal
  validate(page, chat)        -> fetch THIS session's trajectory, run verify_task
  teardown()                  -> noop (server is shared)

The grade reuses MiniWeb's existing verifier verbatim; the only new bit is fetching
the trajectory session-scoped THROUGH the browser (page.request carries the
_data_overlay_sid cookie), so many BrowserGym envs can share one MiniWeb server.
"""
from helpers.paths import REPO_ROOT
from helpers.paths import SITES_DIR
import json
import os
import pathlib

from browsergym.core.task import AbstractBrowserTask

from miniweb.evaluation.verifiers import verify_task
from miniweb.evaluation.trajectory import merge_server_log

ROOT = REPO_ROOT
ANN = ROOT / "data" / "annotations"


def _start_url(base, tdir, task):
    """Where the agent starts, mapped onto the local server.

    Multi-site tasks start on the PORTAL directory (`/`): crossing between sites
    is part of the task, so neither site's deep page is a fair start. Single-site
    tasks start where the reference recording began — the first OBSERVATION url
    (the page the human was actually on), then the first ACTION url. NEVER a
    network/XHR event — those are API calls (e.g. /api/prices/live?symbols=SPY)
    that render as raw JSON/CSV, not the page.
    """
    if len(task.get("sites") or []) > 1:
        return base + "/"
    tj = tdir / "trajectory.json"
    if tj.exists():
        try:
            traj = json.loads(tj.read_text())
            for kind in ("observation", "action"):
                for e in traj:
                    u = e.get("url")
                    if e.get("type") == kind and u and "/sites/" in u:
                        return base + u[u.find("/sites/"):]
        except (json.JSONDecodeError, OSError):
            pass
    sites = task.get("sites") or []
    sid = (sites[0].get("id") if isinstance(sites[0], dict) else sites[0]) if sites else ""
    return f"{base}/sites/{sid}/"


def _assemble(recorded, log, beacons):
    """Same assembly as run_agent_verify.build_trajectory, but from session-scoped feeds.

    Network events come from the server request log (union with any client-recorded
    ones) — queries intact, nothing synthesized from actions."""
    traj = [e for e in recorded if e.get("type") != "network"]
    if not any(e.get("type") == "action" for e in traj):
        for b in beacons:
            traj.append({"type": "action", **b})
    return merge_server_log(traj, log)


def fetch_session_trajectory(page, base):
    """Pull THIS browser session's trajectory (the cookie selects the session)."""
    def entries(path):
        try:
            return page.request.get(base + path).json().get("entries", []) or []
        except Exception:
            return []
    return _assemble(entries("/_admin/record"), entries("/_admin/log"),
                     entries("/_admin/beacon"))


def _site_brand(sid):
    """The site's brand/display name from site.json (falls back to the id)."""
    try:
        return json.loads((SITES_DIR / sid / "site.json").read_text()).get("name", sid)
    except Exception:
        return sid


def _nav_preamble(task):
    """Natural, assistant-style orientation (kept free of evaluation cues so the
    agent behaves as it would for a real user): describes the browser home page and
    how to move between the user's sites, and how to wrap up. No goto tool — the
    agent uses the page's own controls (search bar, site shortcuts, tab bar)."""
    names = [_site_brand(s.get("id") if isinstance(s, dict) else s)
             for s in (task.get("sites") or []) if (s.get("id") if isinstance(s, dict) else s)]
    sites_block = (" You may need: " + ", ".join(names) + ".\n\n") if names else " "
    return ("You are a helpful assistant operating the user's web browser. You are on "
            "the user's home page, which lists their websites — open and move between "
            "them using the page's search box, the site shortcuts, and the tab bar." +
            sites_block +
            "When you have done what the user asked, call finish_task() to wrap up. If "
            "they asked a question, reply with report_answer(<answer>) instead. Use "
            "send_msg_to_user only for brief progress updates.\n\n"
            "The user says: ")


from miniweb.browsergym.actions import FINAL_ANSWER_PREFIX, TASK_DONE_PREFIX


def _agent_texts(chat_messages):
    """The messages the AGENT produced this episode. BrowserGym seeds a stock
    assistant greeting ("Hi! I am your UI assistant...") BEFORE the goal, so we take
    only assistant messages that come AFTER the goal (the first user-role message)."""
    msgs = list(chat_messages or [])
    start = 0
    for i, m in enumerate(msgs):
        if isinstance(m, dict) and m.get("role") == "user":
            start = i + 1
            break
    out = []
    for m in msgs[start:]:
        if isinstance(m, dict):
            if m.get("role") in (None, "assistant", "infeasible") and m.get("message"):
                out.append(str(m["message"]))
        elif isinstance(m, str) and m:
            out.append(m)
    return out


def _final_answer(chat_messages):
    """The agent's explicitly-reported final answer, via report_answer (last marked
    message, marker stripped). '' if the agent never reported one."""
    for t in reversed(_agent_texts(chat_messages)):
        if FINAL_ANSWER_PREFIX in t:
            return t.split(FINAL_ANSWER_PREFIX, 1)[1].strip()
    return ""


def _task_finished(chat_messages):
    """True once the agent has declared it is done — via finish_task (no answer) or
    report_answer (with an answer). Both end the task; grading runs at that point."""
    return any(FINAL_ANSWER_PREFIX in t or TASK_DONE_PREFIX in t
               for t in _agent_texts(chat_messages))


def _collect_answer(chat_messages):
    """Fallback for agents that don't report an explicit answer: everything the agent
    said to the human, joined (so a multi-part answer split across messages is whole)."""
    return "\n".join(_agent_texts(chat_messages)).strip()


class MiniWebTask(AbstractBrowserTask):
    def __init__(self, seed: int = 0, task_id: str = None, base_url: str = None):
        super().__init__(seed)
        self.task_id = task_id
        self.base = (base_url or os.environ.get("MINIWEB_URL", "http://localhost:8099")).rstrip("/")
        self.tdir = ANN / task_id
        self.task = json.loads((self.tdir / "task.json").read_text())
        self.verifier = json.loads((self.tdir / "verifier.json").read_text())
        self.expected = (self.task.get("expected_answer") or "").strip()

    def setup(self, page):
        # session-scoped reset (safe on a shared server), then land on the start page.
        # authenticate_by_form tasks must start LOGGED OUT: ?no_autologin=1 sets the
        # session flag the auto-login guard honours (per-session equivalent of the
        # native runner's MINIWEB_NO_AUTOLOGIN server env).
        reset = f"{self.base}/_reset_data"
        if "authenticate_by_form" in (self.verifier.get("macros") or {}):
            reset += "?no_autologin=1"
        try:
            page.goto(reset)
        except Exception:
            pass
        page.goto(_start_url(self.base, self.tdir, self.task))
        goal = _nav_preamble(self.task) + self.task["instruction"]
        return goal, {"task_id": self.task_id}

    def validate(self, page, chat_messages):
        # The agent's answer: its explicit report_answer/finish_task value if given,
        # else its messages to the human (BrowserGym's seed greeting is excluded).
        answer = _final_answer(chat_messages) or _collect_answer(chat_messages)
        traj = fetch_session_trajectory(page, self.base)
        report = verify_task(self.verifier, traj, answer,
                             question=self.task.get("instruction", ""))
        passed = report["passed"]
        # The agent ends the task by calling finish_task (or report_answer); grading
        # runs at that point, pass or fail. A solved task also ends immediately.
        # Otherwise the agent keeps going until the step limit (BrowserGym truncates).
        finished = _task_finished(chat_messages)
        done = passed or finished
        reward = 1.0 if passed else 0.0
        return reward, done, "", {"task_id": self.task_id,
                                  "by_macro": report["by_macro"],
                                  "ended_by": "verifier" if passed else
                                              ("agent" if finished else "")}

    def teardown(self):
        pass