| """Single-observation, single-action text fill table. No model or desktop execution. |
| |
| All 20 slots use the shared token vocabulary; this is not a bank of action heads. |
| NA is a supervised value, never an ignored label. Source conversion is an audit, |
| not a declaration that screenshots, trajectories or task completion are verified. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import math |
| import re |
|
|
| SCHEMA = ( |
| ("Click", ("coord", "coord", "button")), |
| ("DoubleClick", ("coord", "coord")), |
| ("MoveTo", ("coord", "coord")), |
| ("TypeText", ("yes",)), |
| ("PressKeys", ("key",) * 4), |
| ("Scroll", ("direction", "amount")), |
| ("Wait", ("duration",)), |
| ("Drag", ("coord",) * 4), |
| ("Done", ("yes",)), |
| ) |
| NA = "NA" |
| MASK = "|<MASK>|" |
| MODIFIERS = ("CTRL", "ALT", "SHIFT", "WIN", "CMD") |
| |
| KEYS = frozenset( |
| ( |
| *MODIFIERS, |
| *"ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", |
| "ENTER", |
| "ESC", |
| "TAB", |
| "SPACE", |
| "BACKSPACE", |
| "DELETE", |
| "UP", |
| "DOWN", |
| "LEFT", |
| "RIGHT", |
| "HOME", |
| "END", |
| "PAGEUP", |
| "PAGEDOWN", |
| *(f"F{i}" for i in range(1, 25)), |
| ) |
| ) |
| DOMAINS = { |
| "coord": frozenset(str(i) for i in range(1001)), |
| "button": frozenset(("LEFT", "RIGHT")), |
| "yes": frozenset(("YES",)), |
| "key": KEYS, |
| "direction": frozenset(("UP", "DOWN")), |
| "amount": frozenset(("LOW", "MEDIUM", "HIGH")), |
| "duration": frozenset(("1", "2", "4")), |
| } |
| SCROLL_UNITS = {"LOW": 1, "MEDIUM": 3, "HIGH": 8} |
|
|
|
|
| def slot_token(value: str) -> str: |
| return f"<|gui_value:{value}|>" |
|
|
|
|
| def install_slot_tokens(tokenizer) -> dict[str, int]: |
| """Explicitly extend a local tokenizer, never a model, for atomic slot values. |
| |
| Preserve native single-token values. A subsequent model adapter MUST resize |
| and initialize embeddings/output vocabulary before these IDs can be used. |
| """ |
| values = sorted({NA} | set().union(*DOMAINS.values())) |
| missing = [ |
| slot_token(v) for v in values if len(tokenizer.encode(v, add_special_tokens=False).ids) != 1 |
| ] |
| tokenizer.add_tokens(missing) |
| result = {} |
| for value in values: |
| ids = tokenizer.encode(value, add_special_tokens=False).ids |
| if len(ids) != 1: |
| ids = tokenizer.encode(slot_token(value), add_special_tokens=False).ids |
| if len(ids) != 1: |
| raise ValueError("slot token installation failed") |
| result[value] = ids[0] |
| if len(set(result.values())) != len(result): |
| raise ValueError("slot token IDs must be unique") |
| return result |
|
|
|
|
| def template(mask: str = MASK) -> str: |
| return "\n".join(f"{name}({','.join([mask] * len(kinds))})" for name, kinds in SCHEMA) |
|
|
|
|
| def encode_slots(tokenizer, table: str, *, value_ids: dict[str, int] | None = None) -> dict: |
| """Compile aligned table-only IDs with a tokenizers.Tokenizer-compatible API. |
| |
| Replace masks in ID space: retokenizing filled text can merge punctuation or |
| change slot positions. Context/image tokens and attention masks are NOT built |
| here. Labels are same-position, not autoregressively shifted. |
| """ |
| selected = parse(table) |
| canonical = fill(selected["operation"], selected["values"]) |
| values = [ |
| value for row in canonical.splitlines() for value in row.split("(", 1)[1][:-1].split(",") |
| ] |
| mask_ids = tokenizer.encode(MASK, add_special_tokens=False).ids |
| if len(mask_ids) != 1: |
| raise ValueError("mask must be one token") |
| inputs = tokenizer.encode(template(), add_special_tokens=False).ids |
| positions = [i for i, value in enumerate(inputs) if value == mask_ids[0]] |
| if len(positions) != len(values): |
| raise ValueError("mask positions do not match schema") |
| targets = inputs.copy() |
| labels = [-100] * len(inputs) |
| for position, value in zip(positions, values): |
| ids = ( |
| [value_ids[value]] |
| if value_ids is not None |
| else tokenizer.encode(value, add_special_tokens=False).ids |
| ) |
| if len(ids) != 1: |
| raise ValueError(f"slot value requires an atomic token: {value}") |
| targets[position] = labels[position] = ids[0] |
| return { |
| "input_ids": inputs, |
| "target_ids": targets, |
| "labels": labels, |
| "slot_positions": positions, |
| } |
|
|
|
|
| def render_slots(slot_ids: list[int], value_ids: dict[str, int]) -> str: |
| """Render known value IDs for scoring only; does NOT authorize execution.""" |
| if len(slot_ids) != sum(len(kinds) for _, kinds in SCHEMA): |
| raise ValueError("expected twenty predictions") |
| inverse = {token_id: value for value, token_id in value_ids.items()} |
| if len(inverse) != len(value_ids): |
| raise ValueError("ambiguous value token IDs") |
| if any(type(token_id) is not int or token_id not in inverse for token_id in slot_ids): |
| raise ValueError("non-parameter token prediction") |
| values = iter(inverse[token_id] for token_id in slot_ids) |
| return "\n".join(f"{name}({','.join(next(values) for _ in kinds)})" for name, kinds in SCHEMA) |
|
|
|
|
| def decode_slots(slot_ids: list[int], value_ids: dict[str, int]) -> str: |
| """Decode twenty IDs, then fail-closed validate before execution.""" |
| table = render_slots(slot_ids, value_ids) |
| selected = parse(table) |
| return fill(selected["operation"], selected["values"]) |
|
|
|
|
| def fill(operation: str, values: list[str]) -> str: |
| if operation not in dict(SCHEMA): |
| raise ValueError("unknown operation") |
| table = "\n".join( |
| f"{name}({','.join(values if name == operation else [NA] * len(kinds))})" |
| for name, kinds in SCHEMA |
| ) |
| parse(table) |
| return table |
|
|
|
|
| def parse(table: str) -> dict: |
| """Fail closed; only harmless whitespace/key-order normalization is allowed.""" |
| if not isinstance(table, str) or len(table) > 4096: |
| raise ValueError("invalid table") |
| lines = table.strip().splitlines() |
| if len(lines) != len(SCHEMA): |
| raise ValueError("expected all nine operation rows") |
| active = [] |
| for line, (name, kinds) in zip(lines, SCHEMA): |
| match = re.fullmatch(rf"{name}\(([^()]*)\)", line.strip()) |
| if not match: |
| raise ValueError("unexpected operation order or syntax") |
| values = [v.strip() for v in match[1].split(",")] |
| if len(values) != len(kinds): |
| raise ValueError("wrong slot count") |
| if all(v == NA for v in values): |
| continue |
| for value, kind in zip(values, kinds): |
| if value not in DOMAINS[kind] and not (kind == "key" and value == NA): |
| raise ValueError("missing or invalid parameter") |
| if name == "PressKeys": |
| keys = [v for v in values if v != NA] |
| if len(keys) != len(set(keys)): |
| raise ValueError("duplicate key") |
| ordinary = [k for k in keys if k not in MODIFIERS] |
| |
| keys = [k for k in MODIFIERS if k in keys] + ordinary |
| values = keys + [NA] * (4 - len(keys)) |
| if name == "Drag" and values[:2] == values[2:]: |
| raise ValueError("zero-distance drag is not long press") |
| active.append({"operation": name, "values": values}) |
| if len(active) != 1: |
| raise ValueError("expected exactly one active operation") |
| return active[0] |
|
|
|
|
| def drag_seconds(x1: float, y1: float, x2: float, y2: float, *, width: int, height: int) -> float: |
| """Proposed distance-based movement duration, excluding press/release overhead.""" |
| if type(width) is not int or type(height) is not int or min(width, height) < 2: |
| raise ValueError("invalid screen dimensions") |
| points = (x1, y1, x2, y2) |
| if any( |
| type(p) not in (int, float) or not math.isfinite(p) or not 0 <= p <= 1000 for p in points |
| ): |
| raise ValueError("invalid coordinates") |
| if (x1, y1) == (x2, y2): |
| raise ValueError("zero-distance drag") |
| distance = math.hypot((x2 - x1) * (width - 1) / 1000, (y2 - y1) * (height - 1) / 1000) |
| return 0.2 + 0.8 * distance / math.hypot(width - 1, height - 1) |
|
|
|
|
| def normalized_coordinate(value) -> str: |
| if type(value) not in (int, float) or not math.isfinite(value) or not 0 <= value <= 1: |
| raise ValueError("source_coordinate_out_of_range") |
| return str(math.floor(value * 1000 + 0.5)) |
|
|
|
|
| def convert_source(code: str) -> dict: |
| """Audit official normalized AgentNet code without running it or splitting macros. |
| |
| Only explicit moveTo+dragTo can be represented as one complete Drag gesture. |
| Completion claims are not Done labels; wheel amounts require cursor admission. |
| """ |
| from .visual_nanojev_agentnet import _calls, adapt_sequence |
|
|
| try: |
| calls = _calls(code) |
| if any(name == "computer.terminate" for name, _, _ in calls): |
| raise ValueError("terminal_claim_requires_success_verification") |
| if len(calls) == 2 and [c[0] for c in calls] == ["pyautogui.moveTo", "pyautogui.dragTo"]: |
| result = adapt_sequence(code) |
| if result.action is None: |
| raise ValueError(result.reason) |
| actions = result.action["actions"] |
| first, last = [a["args"] for a in actions] |
| if last["button"] != "left": |
| raise ValueError("nonleft_drag") |
| values = [ |
| normalized_coordinate(p) for p in (first["x"], first["y"], last["x"], last["y"]) |
| ] |
| return { |
| "table": fill("Drag", values), |
| "reason": "candidate", |
| "notes": ["duration_replaced_by_executor_policy", "coordinates_quantized"], |
| } |
| if len(calls) != 1: |
| raise ValueError("compound_step_no_intermediate_observation") |
| result = adapt_sequence(code) |
| if result.action is None: |
| raise ValueError(result.reason) |
| if len(result.action["actions"]) != 1: |
| raise ValueError("multiple_primitives_no_intermediate_observation") |
| action = result.action["actions"][0] |
| skill, args = action["skill"], action["args"] |
| notes = [] |
| if skill in {"click", "double_click", "move_to"}: |
| name = {"click": "Click", "double_click": "DoubleClick", "move_to": "MoveTo"}[skill] |
| values = [normalized_coordinate(args[k]) for k in ("x", "y")] |
| if skill == "click": |
| values.append(args["button"].upper()) |
| elif skill == "double_click" and args["button"] != "left": |
| raise ValueError("nonleft_double_click") |
| notes.append("coordinates_quantized") |
| elif skill == "type_text": |
| name, values = "TypeText", ["YES"] |
| notes.append("source_text_not_in_gui_target_external_text_branch_required") |
| elif skill in {"press_key", "hotkey"}: |
| keys = [args["key"]] if skill == "press_key" else args["keys"] |
| aliases = { |
| "RETURN": "ENTER", |
| "ESCAPE": "ESC", |
| "CONTROL": "CTRL", |
| "COMMAND": "CMD", |
| " ": "SPACE", |
| } |
| keys = [aliases.get(k.upper(), k.upper()) for k in keys] |
| if not 1 <= len(keys) <= 4: |
| raise ValueError("key_count_out_of_range") |
| name, values = "PressKeys", keys + [NA] * (4 - len(keys)) |
| elif skill == "wheel": |
| units = args["wheel_units"] |
| amount = min(SCROLL_UNITS, key=lambda a: abs(SCROLL_UNITS[a] - abs(units))) |
| return { |
| "table": None, |
| "reason": "scroll_cursor_context_unverified", |
| "wheel_units": units, |
| "proposed_amount": amount, |
| "quantization_error": SCROLL_UNITS[amount] - abs(units), |
| } |
| else: |
| raise ValueError(f"unsupported_source_skill:{skill}") |
| table = fill(name, values) |
| selected = parse(table) |
| return { |
| "table": fill(selected["operation"], selected["values"]), |
| "reason": "candidate", |
| "notes": notes, |
| } |
| except (ValueError, TypeError, KeyError) as error: |
| return {"table": None, "reason": str(error)} |
|
|
