File size: 3,176 Bytes
795f737 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """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")
|