Superfast action layer for computer-use agents

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

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
• 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

Three functions for the agent-facing JSON contract:

# Parse agent payload into Subtask
task = subtask_from_dict({
    "goal": "...", "verification": ["..."],
    "inputs": {...}, "constraints": [...], "max_actions": 15,
})

# 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.

from arc_cua.backends import MacOSHybridBackend
backend = MacOSHybridBackend()

MacOSAXBackend

Pure Accessibility backend. Traverses the AX tree of the frontmost application.

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

Supports CLICK, DOUBLE_CLICK, RIGHT_CLICK, TYPE_TEXT, SET_VALUE, PRESS_KEY, HOTKEY, SCROLL.

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.