baim/runtime.py

"""Validated executor. Permissions and completion evidence come from trusted code."""
from collections import Counter
from dataclasses import dataclass
from time import perf_counter, process_time
from urllib.parse import urlsplit
from .actions import Kind, TARGETED
from .authority import Rejected
from .state import digest


@dataclass(frozen=True)
class Outcome:
    status: str
    code: str
    wall_ms: float
    cpu_ms: float
    evidence: str | None = None


class Runtime:
    def __init__(self, browser, authority, permission, completion=None, recorder=None):
        self.browser = browser
        self.authority = authority
        # Callbacks are configured by the host, never parsed from page/model text.
        self.permission = permission
        self.completion = completion
        self.completion_task = authority.task_id
        self.recorder = recorder
        self.failures = Counter()
        self.attempts = Counter()
        self.max_steps = 100

    def execute(self, decision):
        started, cpu = perf_counter(), process_time()
        evidence = None
        status, code = "rejected", "UNINITIALIZED"
        action = decision.action
        target = None
        state = self.browser.state
        with self.authority.lock:
            try:
                if state is None:
                    raise Rejected("OBSERVATION_REQUIRED")
                self.authority.validate(decision.ticket, state)
                self.browser.validate_document()
                key = (self.authority.task_id, state.state_hash, self.action_key(action, state))
                if self.authority.sequence >= self.max_steps:
                    raise Rejected("STEP_BUDGET")
                if self.failures[key] >= 2 or self.attempts[key] >= 3:
                    raise Rejected("REPEATED_ACTION")
                if not self.permission(self.authority.goal, state, action):
                    raise Rejected("HOST_POLICY_DENIED")
                if action.kind in {Kind.NAVIGATE, Kind.OPEN_TAB}:
                    parsed = urlsplit(action.args[0])
                    if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
                        raise Rejected("UNSAFE_URL")
                if action.kind in TARGETED:
                    target, element = self.browser.resolve(action.args[0])
                    if element.sensitive:
                        raise Rejected("SENSITIVE_TARGET_REQUIRES_HOST_FLOW")
                    if action.kind == Kind.TYPE and element.role not in {"textbox", "searchbox", "combobox", "spinbutton"}:
                        raise Rejected("TYPE_ROLE_MISMATCH")
                    if action.kind == Kind.SELECT and element.role not in {"combobox", "listbox"}:
                        raise Rejected("SELECT_ROLE_MISMATCH")
                if action.kind == Kind.FINISH:
                    if not self.completion or self.completion_task != self.authority.task_id:
                        raise Rejected("NO_COMPLETION_VERIFIER")
                    evidence = self.completion(self.browser)
                    if not isinstance(evidence, str) or not evidence:
                        raise Rejected("COMPLETION_NOT_PROVEN")
                self.authority.consume()  # Attempt consumes ticket even if browser action fails.
                self.attempts[key] += 1
                try:
                    self._dispatch(action, target)
                    status, code = "ok", action.kind.name
                except Rejected:
                    self.failures[key] += 1
                    raise
                except Exception as exc:
                    self.failures[key] += 1
                    # Exceptions often contain typed values, URLs or page text. Do not log them.
                    status, code = "failed", type(exc).__name__
            except Rejected as exc:
                status, code = "rejected", str(exc)
            finally:
                if target:
                    target.dispose()
        outcome = Outcome(status, code, (perf_counter()-started)*1000, (process_time()-cpu)*1000, evidence)
        if self.recorder:
            self.recorder.record(decision, state, outcome)
        return outcome

    @staticmethod
    def action_key(action, state):
        args = list(action.args)
        if action.kind in TARGETED:
            element = next((e for e in state.elements if e.ref == args[0]), None)
            args[0] = element.signature if element else args[0]
        return digest([action.kind, args])

    def _dispatch(self, action, target):
        k, args, page = action.kind, action.args, self.browser.page
        if k == Kind.CLICK:
            target.click(timeout=2000)
        elif k == Kind.TYPE:
            target.fill(args[1], timeout=2000)
        elif k == Kind.SELECT:
            target.select_option(value=args[1], timeout=2000)
        elif k == Kind.SCROLL:
            page.mouse.wheel(*args)
        elif k == Kind.WAIT:
            page.wait_for_timeout(args[0])
        elif k == Kind.NAVIGATE:
            page.goto(args[0], wait_until="domcontentloaded", timeout=10000)
        elif k == Kind.BACK:
            page.go_back(wait_until="domcontentloaded", timeout=10000)
        elif k == Kind.FORWARD:
            page.go_forward(wait_until="domcontentloaded", timeout=10000)
        elif k == Kind.PRESS_KEY:
            if args[1] not in {"Enter", "Tab", "Escape", "ArrowUp", "ArrowDown", "ArrowLeft", "ArrowRight", "Space", "Home", "End"}:
                raise Rejected("UNSUPPORTED_KEY")
            target.press(args[1], timeout=2000)
        elif k == Kind.SUBMIT:
            target.press("Enter", timeout=2000)
        elif k == Kind.OPEN_TAB:
            self.browser.page = self.browser.context.new_page()
            self.browser.page.goto(args[0], wait_until="domcontentloaded", timeout=10000)
            self.browser.sync_tabs()
            self.browser.state = None
        elif k in {Kind.CLOSE_TAB, Kind.FOCUS_TAB}:
            self.browser.sync_tabs()
            if args[0] not in self.browser.tabs:
                raise Rejected("UNKNOWN_TAB")
            requested = self.browser.tabs[args[0]]
            if k == Kind.CLOSE_TAB:
                if len(self.browser.tabs) <= 1:
                    raise Rejected("LAST_TAB")
                requested.close()
                self.browser.sync_tabs()
                if requested == self.browser.page:
                    self.browser.page = next(iter(self.browser.tabs.values()))
            else:
                self.browser.page = requested
                requested.bring_to_front()
            self.browser.state = None
        elif k == Kind.EXTRACT:
            # Result remains in task memory, not persistent telemetry by default.
            self.extracted = target.inner_text(timeout=2000)[:16384]
        elif k == Kind.RECOVER:
            self.browser.observe()
        elif k == Kind.ASK_USER:
            self.question = args[0]
        elif k == Kind.FINISH:
            pass
        else:
            raise Rejected("UNSUPPORTED_ACTION")