"""Strict action wire format: opcode followed by a compact JSON argument list.""" from dataclasses import dataclass from enum import StrEnum import json import math class Kind(StrEnum): NAVIGATE = "N" CLICK = "C" TYPE = "T" SELECT = "O" SCROLL = "S" WAIT = "W" BACK = "B" FORWARD = "G" OPEN_TAB = "U" CLOSE_TAB = "X" FOCUS_TAB = "H" PRESS_KEY = "K" SUBMIT = "J" EXTRACT = "E" ASK_USER = "A" RECOVER = "R" FINISH = "F" SCHEMA = { Kind.NAVIGATE: (str,), Kind.CLICK: (str,), Kind.TYPE: (str, str), Kind.SELECT: (str, str), Kind.SCROLL: (int, int), Kind.WAIT: (int,), Kind.BACK: (), Kind.FORWARD: (), Kind.OPEN_TAB: (str,), Kind.CLOSE_TAB: (str,), Kind.FOCUS_TAB: (str,), Kind.PRESS_KEY: (str, str), Kind.SUBMIT: (str,), Kind.EXTRACT: (str,), Kind.ASK_USER: (str,), Kind.RECOVER: (), Kind.FINISH: (), } TARGETED = {Kind.CLICK, Kind.TYPE, Kind.SELECT, Kind.PRESS_KEY, Kind.SUBMIT, Kind.EXTRACT} @dataclass(frozen=True) class Action: kind: Kind args: tuple = () def __post_init__(self): if not isinstance(self.kind, Kind) or not isinstance(self.args, tuple): raise ValueError("invalid action representation") expected = SCHEMA[self.kind] if len(self.args) != len(expected) or any(type(v) is not t for v, t in zip(self.args, expected)): raise ValueError("invalid action arguments") if any(isinstance(v, str) and (len(v) > 8192 or "\x00" in v) for v in self.args): raise ValueError("oversized or NUL-containing argument") if self.kind == Kind.WAIT and not 0 <= self.args[0] <= 2000: raise ValueError("wait must be 0..2000 ms") if self.kind == Kind.SCROLL and any(abs(v) > 2000 for v in self.args): raise ValueError("scroll exceeds viewport step limit") def encode(self): return self.kind.value + json.dumps(self.args, ensure_ascii=False, separators=(",", ":")) @classmethod def parse(cls, wire): if not isinstance(wire, str) or not 2 <= len(wire) <= 20000: raise ValueError("invalid action wire length") try: args = json.loads(wire[1:]) if type(args) is not list: raise ValueError("arguments must be array") return cls(Kind(wire[0]), tuple(args)) except (json.JSONDecodeError, KeyError, TypeError) as exc: raise ValueError("malformed action") from exc @dataclass(frozen=True) class Ticket: session_id: str task_id: str epoch: int sequence: int document_id: str revision: int state_hash: str @dataclass(frozen=True) class Decision: ticket: Ticket action: Action # None means uncalibrated. Never invent confidence for untrained policies. action_confidence: float | None = None target_confidence: float | None = None def __post_init__(self): for value in (self.action_confidence, self.target_confidence): if value is not None and (not math.isfinite(value) or not 0 <= value <= 1): raise ValueError("confidence must be a calibrated probability or None")