| """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 |
| |
| 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") |
|
|
