executor.py

"""
ActionExecutor — the only module permitted to mutate the browser.

Every state change flows through here; the LLM never touches Playwright. That
boundary is what makes SafetyGuard enforceable — there is exactly one door.

What changed from the first implementation
------------------------------------------
* **`select()` targets the element it was asked to.** The JS path ran
  `document.querySelectorAll('select')` and used the first dropdown whose
  options matched, ignoring `target_id`. On a page with several dropdowns
  (dates, passengers, cabin class) it would silently set the wrong one.
* **No site-specific knowledge.** A `sort_value_map` of
  `{"price (low to high)": "price-asc"}` encoded one site's query parameters
  into what is supposed to be a general-purpose executor.
* **Locator resolution reports ambiguity.** `get_by_role()` returns lazily and
  never raises, so the old `try/except` around it was dead code; a label
  matching three elements surfaced later as an opaque strict-mode violation.
* **Waits are settle-based.** Fixed `wait_for_timeout(800)` calls are
  simultaneously too slow for static pages and too short for slow ones.
"""

from __future__ import annotations

import asyncio
import random
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime

from playwright.async_api import Locator, Page

from logger import get_logger

log = get_logger(__name__)

SettleFn = Callable[[Page], Awaitable[bool]]


class ActionExecutor:
    def __init__(
        self,
        *,
        settle: SettleFn | None = None,
        humanize: bool = True,
        max_attempts: int = 3,
        action_timeout_ms: int = 5000,
    ) -> None:
        """
        settle
            Async callable awaited after each mutating action to let the page
            reach a stable state. Injected rather than imported so the executor
            keeps no dependency on perception. Falls back to a short fixed wait.
        humanize
            Insert small randomised delays before input. Real users do not click
            with zero latency, and uniform timing is a bot signal on sites that
            look for it. Tests disable it.
        """
        self.element_map: dict[int, dict] = {}
        self._settle = settle
        self.humanize = humanize
        self.max_attempts = max_attempts
        self.action_timeout_ms = action_timeout_ms

    def build_element_map(self, elements) -> None:
        """Refresh the id -> element index after each perception pass."""
        items = getattr(elements, "elements", elements)
        self.element_map = {el["id"]: el for el in items if el.get("id") is not None}

    # ------------------------------------------------------------------
    # Dispatch
    # ------------------------------------------------------------------

    async def execute(self, action: dict, page: Page) -> dict:
        kind = str(action.get("action") or "").strip().lower()
        target_id = action.get("target_id")
        value = action.get("value")

        handlers = {
            "click": lambda: self.click(target_id, page),
            "type": lambda: self.type(target_id, value, page,
                                      submit=bool(action.get("submit"))),
            "select": lambda: self.select(target_id, value, page),
            "scroll": lambda: self.scroll(value or "down", page),
            "wait": lambda: self.wait(value),
            "navigate": lambda: self.navigate(value, page),
            "back": lambda: self.back(page),
        }
        handler = handlers.get(kind)
        if handler is None:
            return self._result(kind, target_id, "", success=False,
                                error=f"unknown action type {kind!r}")
        try:
            return await handler()
        except Exception as exc:
            # A handler raising would abort the whole run; the loop needs a
            # result it can reason about instead.
            return self._result(kind, target_id, self._label_for(target_id),
                                success=False, error=f"{type(exc).__name__}: {exc}"[:200])

    # ------------------------------------------------------------------
    # Actions
    # ------------------------------------------------------------------

    async def click(self, target_id: int | None, page: Page) -> dict:
        label = self._label_for(target_id)
        url_before = page.url

        for attempt in range(self.max_attempts):
            try:
                locator = await self._resolve(target_id, page)
                await self._human_pause()
                await locator.click(timeout=self.action_timeout_ms)
                changed = await self._after_mutation(page, url_before)
                return self._result("click", target_id, label, success=True,
                                    attempts=attempt + 1, page_changed=changed,
                                    url_before=url_before, url_after=page.url)
            except Exception as exc:
                if attempt < self.max_attempts - 1:
                    await asyncio.sleep(0.4 * (attempt + 1))
                    continue
                return self._failure("click", target_id, label, exc, url_before)

    async def type(
        self, target_id: int | None, value, page: Page, *, submit: bool = False
    ) -> dict:
        label = self._label_for(target_id)
        url_before = page.url
        text = "" if value is None else str(value)

        for attempt in range(self.max_attempts):
            try:
                locator = await self._resolve(target_id, page)
                await self._human_pause()
                await locator.fill("", timeout=self.action_timeout_ms)
                if self.humanize:
                    for char in text:
                        await locator.press_sequentially(
                            char, delay=random.randint(40, 110)
                        )
                else:
                    await locator.fill(text, timeout=self.action_timeout_ms)

                if submit:
                    # Many search fields have no visible submit control.
                    await locator.press("Enter")

                changed = await self._after_mutation(page, url_before)
                return self._result("type", target_id, label, success=True,
                                    attempts=attempt + 1, page_changed=changed,
                                    url_before=url_before, url_after=page.url)
            except Exception as exc:
                if attempt < self.max_attempts - 1:
                    await asyncio.sleep(0.4 * (attempt + 1))
                    continue
                return self._failure("type", target_id, label, exc, url_before)

    async def select(self, target_id: int | None, value, page: Page) -> dict:
        """
        Set a dropdown, trying progressively less native strategies.

        Every strategy operates on the *resolved* element. The previous version
        fell back to a global `querySelectorAll('select')` scan, so on a page
        with more than one dropdown it could set a different control than the
        one the agent asked for.
        """
        label = self._label_for(target_id)
        url_before = page.url
        wanted = "" if value is None else str(value)

        try:
            locator = await self._resolve(target_id, page)
        except Exception as exc:
            return self._failure("select", target_id, label, exc, url_before)

        # Read the options up front rather than guessing. Blindly trying
        # select_option(label=...) then select_option(value=...) costs a full
        # Playwright timeout per miss — several seconds of dead waiting on every
        # failed select, paid by the agent as much as by the test suite.
        options, is_select = await self._read_options(locator)

        if is_select:
            match = self._match_option(options, wanted)
            if match is None:
                return self._result(
                    "select", target_id, label, success=False,
                    reason=f"no option matches {wanted!r}",
                    suggestion=f"available options: {[o['text'] for o in options][:10]}",
                    url_before=url_before, url_after=page.url,
                )

            # Strategy 1 — native, using the exact value we just read, so this
            # cannot miss on a label/value mismatch.
            try:
                await locator.select_option(value=match["value"], timeout=2500)
                changed = await self._after_mutation(page, url_before)
                if changed:
                    return self._result("select", target_id, label, success=True,
                                        method="native", selected=match["text"],
                                        page_changed=True, url_before=url_before,
                                        url_after=page.url)
            except Exception:
                changed = False

            # Strategy 2 — set the value directly and dispatch the events a real
            # selection produces, for widgets that listen rather than navigate.
            try:
                in_form = await locator.evaluate(
                    """(el, value) => {
                        el.value = value;
                        el.dispatchEvent(new Event('input',  { bubbles: true }));
                        el.dispatchEvent(new Event('change', { bubbles: true }));
                        return Boolean(el.closest('form'));
                    }""",
                    match["value"],
                )
                changed = await self._after_mutation(page, url_before)
                if not changed and in_form:
                    # A change handler meant to submit may not have run; submit
                    # the owning form directly.
                    try:
                        await locator.evaluate("(el) => el.closest('form').submit()")
                        changed = await self._after_mutation(page, url_before)
                    except Exception:
                        pass
                return self._result("select", target_id, label, success=True,
                                    method="dispatch_change", selected=match["text"],
                                    page_changed=changed, url_before=url_before,
                                    url_after=page.url)
            except Exception:
                pass

        available = [o["text"] for o in options]

        # Strategy 3 — custom widget: open it, then click the option.
        try:
            await locator.click(timeout=2500)
            await self._settle_page(page)
            option = page.get_by_role("option", name=wanted, exact=False).first
            if await option.count() == 0:
                option = page.get_by_text(wanted, exact=False).first
            await option.click(timeout=2500)
            changed = await self._after_mutation(page, url_before)
            return self._result("select", target_id, label, success=True,
                                method="click_option", page_changed=changed,
                                url_before=url_before, url_after=page.url)
        except Exception:
            pass

        return self._result(
            "select", target_id, label, success=False,
            reason=f"no strategy could set {wanted!r}",
            suggestion=(
                f"available options: {available[:10]}" if available else
                "click the control first, then click the option as a separate action"
            ),
            url_before=url_before, url_after=page.url,
        )

    async def scroll(self, direction, page: Page) -> dict:
        direction = str(direction or "down").strip().lower()
        js = {
            "down": "window.scrollBy(0, window.innerHeight * 0.8)",
            "up": "window.scrollBy(0, -window.innerHeight * 0.8)",
            "top": "window.scrollTo(0, 0)",
            "bottom": "window.scrollTo(0, document.body.scrollHeight)",
        }.get(direction, "window.scrollBy(0, window.innerHeight * 0.8)")

        before = await page.evaluate("() => window.scrollY")
        await page.evaluate(js)
        await self._settle_page(page)
        after = await page.evaluate("() => window.scrollY")

        return self._result("scroll", None, direction, success=True,
                            page_changed=before != after,
                            scrolled_from=before, scrolled_to=after)

    async def wait(self, value) -> dict:
        try:
            ms = max(0, min(int(value), 10_000))
        except (TypeError, ValueError):
            ms = 1000
        await asyncio.sleep(ms / 1000)
        return self._result("wait", None, "", success=True, duration_ms=ms)

    async def navigate(self, url, page: Page) -> dict:
        """
        Direct navigation. The agent needs this to recover when a required link
        is not present on the page it landed on.
        """
        target = str(url or "").strip()
        url_before = page.url
        if not target.startswith(("http://", "https://")):
            return self._result("navigate", None, target, success=False,
                                error="only http(s) URLs may be navigated to")
        try:
            await page.goto(target, wait_until="domcontentloaded",
                            timeout=self.action_timeout_ms * 3)
            await self._settle_page(page)
            return self._result("navigate", None, target, success=True,
                                page_changed=page.url != url_before,
                                url_before=url_before, url_after=page.url)
        except Exception as exc:
            return self._failure("navigate", None, target, exc, url_before)

    async def back(self, page: Page) -> dict:
        url_before = page.url
        try:
            await page.go_back(wait_until="domcontentloaded",
                               timeout=self.action_timeout_ms)
            await self._settle_page(page)
            return self._result("back", None, "", success=True,
                                page_changed=page.url != url_before,
                                url_before=url_before, url_after=page.url)
        except Exception as exc:
            return self._failure("back", None, "", exc, url_before)

    @staticmethod
    async def _read_options(locator: Locator) -> tuple[list[dict], bool]:
        """Return this element's options and whether it is a native <select>."""
        try:
            return await locator.evaluate(
                """(el) => el.options
                    ? [Array.from(el.options).map(
                        (o) => ({ text: o.text.trim(), value: o.value })), true]
                    : [[], false]"""
            )
        except Exception:
            return [], False

    @staticmethod
    def _match_option(options: list[dict], wanted: str) -> dict | None:
        """
        Exact matches win over substring ones, and label over value, so
        "Price (low to high)" cannot be beaten by an option merely containing
        the word "price".
        """
        want = wanted.strip().lower()
        if not want:
            return None
        for key in ("text", "value"):
            for option in options:
                if option.get(key, "").strip().lower() == want:
                    return option
        for key in ("text", "value"):
            for option in options:
                if want in option.get(key, "").strip().lower():
                    return option
        return None

    # ------------------------------------------------------------------
    # Locator resolution
    # ------------------------------------------------------------------

    async def _resolve(self, target_id: int | None, page: Page) -> Locator:
        """
        Turn an element id into a Locator that matches exactly one node.

        Candidates are tried most-specific first. A candidate matching several
        nodes is narrowed to `.first` rather than discarded — a strict-mode
        violation deep inside `.click()` is far harder to diagnose than a
        logged ambiguity here.
        """
        element = self.element_map.get(target_id)
        if element is None:
            raise ValueError(
                f"target_id {target_id!r} is not in the current element map "
                f"(known ids: {sorted(self.element_map)[:20]})"
            )

        label = (element.get("label") or "").strip()
        role = (element.get("type") or "").strip()
        name = (element.get("name") or "").strip()
        placeholder = (element.get("placeholder") or "").strip()

        candidates: list[tuple[str, Locator]] = []
        if name:
            candidates.append((f"[name={name}]", page.locator(f"[name={name!r}]")))
        if placeholder:
            candidates.append((f"placeholder={placeholder}",
                               page.get_by_placeholder(placeholder, exact=True)))
        if role and label:
            candidates.append((f"role={role} name={label} exact",
                               page.get_by_role(role, name=label, exact=True)))
            candidates.append((f"role={role} name={label}",
                               page.get_by_role(role, name=label)))
        if label:
            candidates.append((f"label={label}", page.get_by_label(label, exact=True)))
            candidates.append((f"text={label}", page.get_by_text(label, exact=True)))

        first_match: Locator | None = None
        for description, locator in candidates:
            try:
                count = await locator.count()
            except Exception:
                continue
            if count == 1:
                return locator
            if count > 1 and first_match is None:
                log.info(
                    "executor_ambiguous_locator",
                    extra={
                        "timestamp": datetime.now(UTC).isoformat(),
                        "src_module": "executor",
                        "target_id": target_id,
                        "strategy": description,
                        "match_count": count,
                    },
                )
                first_match = locator.first

        if first_match is not None:
            return first_match
        raise ValueError(
            f"could not locate element {target_id} "
            f"(role={role!r}, label={label!r}) on the current page"
        )

    # ------------------------------------------------------------------
    # Waiting
    # ------------------------------------------------------------------

    async def _settle_page(self, page: Page) -> None:
        if self._settle is not None:
            try:
                await self._settle(page)
                return
            except Exception:
                pass
        await page.wait_for_timeout(250)

    async def _after_mutation(self, page: Page, url_before: str) -> bool:
        """Wait for the page to stabilise; report whether it actually changed."""
        try:
            await page.wait_for_load_state("domcontentloaded", timeout=3000)
        except Exception:
            pass
        await self._settle_page(page)
        return page.url != url_before

    async def _human_pause(self) -> None:
        if self.humanize:
            await asyncio.sleep(random.uniform(0.12, 0.35))

    # ------------------------------------------------------------------
    # Results
    # ------------------------------------------------------------------

    def _label_for(self, target_id: int | None) -> str:
        element = self.element_map.get(target_id)
        return (element or {}).get("label", "") if element else ""

    def _failure(self, kind, target_id, label, exc: Exception, url_before: str) -> dict:
        message = str(exc).splitlines()[0][:180] if str(exc) else type(exc).__name__
        ambiguous = "strict mode violation" in str(exc).lower()
        return self._result(
            kind, target_id, label, success=False,
            attempts=self.max_attempts,
            error=f"{type(exc).__name__}: {message}",
            reason=(
                "the label matches several elements on the page" if ambiguous else
                "the element could not be interacted with — it may be covered, "
                "off-screen, or gone since the page was read"
            ),
            suggestion=(
                "pick a more specific element" if ambiguous else
                "scroll to reveal it, or choose a different element"
            ),
            url_before=url_before,
        )

    def _result(
        self, action: str, target_id, label: str, *, success: bool,
        attempts: int = 1, page_changed: bool = False, **extra,
    ) -> dict:
        result = {
            "action": action,
            "target_id": target_id,
            # `label` is what SiteMemory stores as the route step; without it a
            # remembered route reads as `click(), select()` and helps nobody.
            "label": label,
            "success": success,
            "attempts": attempts,
            "page_changed": page_changed,
            **extra,
        }
        log.info(
            "action_executed",
            extra={
                "timestamp": datetime.now(UTC).isoformat(),
                "src_module": "executor",
                "action": action,
                "target_id": target_id,
                "label": label[:60],
                "success": success,
                "attempts": attempts,
                "page_changed": page_changed,
            },
        )
        return result