Superfast action layer for computer-use agents, powered by decision models

from arc_cua import (
    ActionKind, Bounds, Decision, DesktopElement, DesktopExecutor,
    DesktopSnapshot, ExecutableAction, ExecutionResult, RuntimeConfig,
    StepEvent, Subtask, TerminalKind, VerifyFn,
    execute_payload, result_to_dict, subtask_from_dict,
)

Subtask

Contract supplied by the external agent/planner. The executor never invents verification criteria or free-form text.

• goal str Natural-language description of what to accomplish
• verification tuple[str, ...] Observable criteria the policy checks for completion
• inputs Mapping[str, str|int|float|bool] Literal values the executor may use (default: empty)
• constraints tuple[str, ...] Things the executor must not do (default: empty)
• max_actions int Action budget (default: 30)
• metadata Mapping[str, Any] Opaque pass-through for the caller
• shortcuts Mapping[str, str] Keyboard chords and descriptions added to this subtask's HOTKEY choices (default: empty)

Caller-supplied shortcuts

Supply a keyboard chord and its meaning to JEV through the Python or JSON API:

task = Subtask(
    goal="Save the current document",
    verification=("The document has no unsaved changes",),
    shortcuts={"MOD+S": "Save the current document in this editor"},
)

JEV chooses from the defaults plus the supplied shortcuts when it selects HOTKEY. Defaults are unchanged; extra choices apply only to this subtask. Supplying an existing default chord gives it a contextual description.

Chords use uppercase keys and one or more MOD, CTRL, ALT, or SHIFT modifiers. MOD means Command on macOS. Keys include A-Z, 0-9, F1-F20, navigation keys, and named punctuation keys; see the full reference for the vocabulary. The macOS backend uses US/ANSI physical key positions. Each shortcut is one chord.

Invalid declarations are rejected when the subtask is created. Runtime validation also rejects undeclared non-default hotkeys, including those emitted by custom policies.

DesktopElement

A normalized, currently observable UI element. IDs are stable for the session lifetime and are never model-invented.

• id str Stable identifier for this element
• role str Semantic role (button, text_field, visible_text, ...)
• name str Human-readable label
• value str|int|float|bool|None Current value (text field contents, slider position, ...)
• actions tuple[ActionKind, ...] Legal operations on this element
• enabled bool Whether the element is interactive (default: True)
• visible bool Whether the element is visible (default: True)
• source str Perception source: macos_ax, macos_ocr, etc.
• bounds Bounds|None Screen coordinates (x, y, width, height)
• accepts_drop bool Valid DRAG_TO destination

DesktopSnapshot

Immutable capture of the current desktop state.

• application str Frontmost application name
• window str Active window title
• revision str SHA-256 fingerprint of element state
• elements tuple[DesktopElement, ...] All observable elements
• context Mapping[str, Any] Backend metadata (pid, window_id, ...)

element(element_id) -> DesktopElement — O(1) lookup by ID. Raises KeyError.

compact() -> dict — Planner-friendly summary (visible elements only).

ActionKind

MOD means Cmd on macOS, Ctrl elsewhere.

DesktopExecutor

The core executor. Takes a backend and a policy, runs subtasks.

executor = DesktopExecutor(backend, policy, config=RuntimeConfig())

result = executor.run(task)           # Returns ExecutionResult
for event in executor.run_iter(task): # Yields StepEvent per cycle
    print(event.step, event.action)
executor.cancel()                     # Thread-safe stop signal

run() executes the subtask and returns the result. run_iter() is a generator yielding a StepEvent after each decision cycle for live observability. cancel() can be called from another thread to stop execution after the current action.

RuntimeConfig

• stale_retries int = 8 Max freshness re-observations before giving up
• no_change_limit int = 3 Consecutive no-change actions before BLOCKED
• post_action_settle_s float = 0.03 Minimum settling delay
• settle_reaction_s float = 0.6 Max wait for a visible reaction
• settle_quiet_s float = 0.15 Unchanged time required after a reaction
• settle_timeout_s float = 2.0 Upper bound on probe-based settling
• timeout_s float|None = None Wall-clock timeout (None = no limit)
• verify VerifyFn|None = None Verification callback for SUBTASK_COMPLETE

VerifyFn

VerifyFn = Callable[[DesktopSnapshot, Subtask], bool]

When set on RuntimeConfig, called before accepting SUBTASK_COMPLETE. If it returns False, the result becomes NEEDS_AGENT.

StepEvent

Emitted by run_iter() after each decision cycle.

• step int Step counter
• snapshot DesktopSnapshot Desktop state at decision time
• decision Decision The policy's decision
• action ExecutableAction|None Materialized action (None for terminal)
• record ActionRecord|None Execution record (None for terminal)
• result ExecutionResult|None Set only on the final event

terminal property returns True when result is set.

ExecutionResult

• status TerminalKind How execution ended
• subtask Subtask The original subtask
• final_snapshot DesktopSnapshot Desktop state at termination
• history tuple[ActionRecord, ...] Complete action history
• observations tuple[str, ...] Satisfaction notes for verification criteria
• reason str|None Explanation for non-complete termination

actions_taken property returns len(history).

JSON boundary

goal must be a non-empty string. verification must be a non-empty array of non-empty strings, and constraints must be an array of non-empty strings. Even one criterion needs an array: "verification": ["The folder exists"]. Bare strings are rejected before a provider request.

inputs maps names to literal strings, finite numbers, or booleans. Supply every value that needs to be typed here. max_actions must be a positive integer; metadata and shortcuts must be objects. Invalid fields raise descriptive errors without silent type conversion. Use uppercase chords such as MOD+SHIFT+N; unmodified Return is the built-in key ENTER.

JEV requests share one element table and subtask across questions. Target choices reference observed IDs, preserving the element facts and available choices while reducing repeated data. Provider token-limit failures report max_tokens_exceeded without executing an action.

Three functions for the agent-facing JSON contract:

# Parse agent payload into Subtask
task = subtask_from_dict({
    "goal": "...", "verification": ["..."],
    "inputs": {...}, "constraints": [...], "max_actions": 15,
    "shortcuts": {"MOD+S": "Save the current document in this editor"},
})

# Convert result to planner-friendly dict
d = result_to_dict(result)
# {"status", "actions_taken", "reason", "observations", "history", "final_snapshot"}

# One-shot convenience
d = execute_payload(executor, payload)

Errors

All inherit from JevDesktopError(RuntimeError):

MacOSHybridBackend

The default production backend. Combines Accessibility and Apple Vision OCR into a single element stream. Handles modal detection, OCR deduplication, and coordinate-based execution for OCR targets.

Acts on one app, given by process ID, in the background: the user's pointer does not move, the window is not raised and the front app does not change. As a context manager it also reaches a minimized window or a hidden app's windows, out of sight. Raises TargetUnavailable once the app quits or has no usable window.

from arc_cua.backends import MacOSApp, MacOSHybridBackend
with MacOSHybridBackend(MacOSApp.from_bundle_id("com.apple.TextEdit").pid) as backend:
    ...

From any language: arc-cua run reads one JSON request (app, subtask, provider) on standard input, prints one JSON line per action and a final result line, and exits. See the full reference.

ChromeBackend

One Chrome tab through the DevTools protocol, on macOS, Linux and Windows. Reads visible controls and text from the DOM in one call (about 9 ms on a large article), including shadow DOM and same-origin iframes. Input does not move the real pointer or need window focus.

from arc_cua.backends import ChromeBackend  # pip install 'arc-cua[browser]'
backend = ChromeBackend.launch("https://example.com")

MacOSAXBackend

Pure Accessibility backend. Traverses the AX tree of one app, given by process ID, with background input.

from arc_cua.backends import MacOSAXBackend
backend = MacOSAXBackend(pid, max_elements=1200, max_depth=18, cache=False)

Supports CLICK, DOUBLE_CLICK, RIGHT_CLICK, TYPE_TEXT, SET_VALUE, PRESS_KEY, HOTKEY, SCROLL. Reads only what is on screen. cache=True re-reads only elements the app reports as changed, so repeat observations take a few milliseconds.

StateMachineBackend

Deterministic backend for tests and demos. No OS dependencies.

from arc_cua.backends import StateMachineBackend

backend = StateMachineBackend(
    initial_state={"key": "value"},
    snapshot_factory=my_snapshot_fn,  # (state) -> DesktopSnapshot
    transition=my_transition_fn,      # (state, action) -> None
)

TypeSafeJevPolicy

Production policy. Builds a dynamic action space from the current DesktopSnapshot and sends it to the JEV API.

from arc_cua.policies import TypeSafeJevPolicy

policy = TypeSafeJevPolicy(
    api_key="...",          # or TYPESAFE_API_KEY env var
    model="jev-latest",     # or TYPESAFE_MODEL env var
    timeout_s=12.0,
)

ScriptedPolicy

Deterministic policy for tests. Consumes decisions in order.

from arc_cua.policies import ScriptedPolicy

policy = ScriptedPolicy([
    Decision(kind=ActionKind.CLICK, target_id="btn"),
    Decision(terminal=TerminalKind.SUBTASK_COMPLETE),
])

DesktopBackend

class DesktopBackend(Protocol):
    def observe(self) -> DesktopSnapshot: ...
    def is_fresh(self, snapshot: DesktopSnapshot, action: ExecutableAction) -> bool: ...
    def execute(self, snapshot: DesktopSnapshot, action: ExecutableAction) -> None: ...

observe() captures current desktop state. is_fresh() checks the target guard still matches reality. execute() performs the action — raise StaleDesktopState if the target changed, InvalidDecision if the action is illegal.

DecisionPolicy

class DecisionPolicy(Protocol):
    def decide(
        self, *,
        subtask: Subtask,
        snapshot: DesktopSnapshot,
        history: Sequence[ActionRecord],
    ) -> Decision: ...

Return Decision(kind=..., target_id=...) for an action, or Decision(terminal=...) to end execution.