Runnable Examples
The examples/ folder of the repository holds one short script per feature.
The six below cover the features added most recently, and each of them accepts
--validate:
pip install -e .
python examples/29_config_sync.py --validate
With --validate a script exercises the real API against an in-memory, temp
directory or loopback stand-in and exits 0 when what it observed is what
the documentation promises. It never moves the pointer, types, captures the
screen, reaches a device, or talks to anything beyond 127.0.0.1. CI runs
all six this way (test_modernization_examples.py), with every input and
process seam replaced by a tripwire.
That is evidence the calls are wired the way these pages say. It is not evidence about hardware: nothing here has touched a Wayland compositor, an Android device or an iPhone. Where a real run needs something, the script’s own docstring names it.
Script |
What |
Without the flag |
|---|---|---|
|
Probes four described desktops (GNOME before and after a refused
consent, sway with |
Probes this session. Also free of side effects. |
|
Two machines sync script files through a store in a temp directory: arrival, server restart, offline queue, a conflict and its resolution. |
Prints the recorded status; syncs this machine only with |
|
Opens an Android session over an offline transport and checks which
|
Prints a device’s setup report; taps only with |
|
Scores two template-matching versions on five labelled frames drawn in memory (HiDPI, negative origin, absent, restyled). |
|
|
Journals a run of variable and flow-control commands, rebuilds it as a candidate script and dry-runs the result. |
|
|
Drives an in-process MCP server: lists, searches, reads a schema and enables a tool. Calls no tool that touches the desktop. |
Serves MCP over stdio in progressive mode. |
Where each feature is documented
Feature |
Guide |
Entry points |
|---|---|---|
Capability and authorisation states |
|
|
Recording without a global hook |
|
|
Config sync |
|
|
Mobile device sessions |
|
|
Healing evaluation and template revisions |
|
|
Action journal and candidate scripts |
|
|
MCP tool modes |
|
|
Roles and deferred ownership |
|
|
Signed action files (Ed25519) |
|
|
Window shell, navigation, background work |
the Actions and View menus, |
|
Every environment variable |
Three things the examples show that are easy to miss
The executor’s options are keywords. The module-level
je_auto_control.execute_action(actions) forwards dry_run,
raise_on_error and step_callback to
je_auto_control.executor.execute_action; they are keyword-only there (it
used to take the action list and nothing else, so the examples written before
that call the executor object):
import je_auto_control as ac
ac.execute_action(actions, dry_run=True) # resolve, call nothing
ac.execute_action(actions, raise_on_error=True) # stop at the first failure
ac.execute_action(actions, step_callback=print) # see each action before it runs
A failed sync backs off. After a send that did not reach the server, the
next config_sync_run inside the back-off window (2 s, doubling, capped at
5 minutes) does not contact the server: it reports offline again with
error set to waiting to retry after an earlier failure. Pass
wait=True to sleep through the back-off instead — that is what a scheduled
sync wants, and a cancel event ends the wait.
Searching a tool does not enable it. In progressive mode
ac_tools_search and ac_tools_schema only describe; the session’s
tools/list changes when ac_tools_enable is called, and the client is
told with notifications/tools/list_changed.
The validation path of each script
These are included from the scripts themselves, so they are what CI runs.
def validate() -> int:
"""Probe four described desktops and check each reads the way it should."""
refused = AuthorisationLedger()
refused.transition(INPUT, AuthorisationState.DECLINED, "the user dismissed the dialog")
desktops = {
"GNOME, portal not asked yet": _context(
_WAYLAND_ENV, tools=("gnome-screenshot",), libei=True),
"GNOME, consent refused": _context(
_WAYLAND_ENV, tools=("gnome-screenshot", "ydotool"), libei=True, ledger=refused),
_SWAY_YDOTOOL_GRIM: _context(
{**_WAYLAND_ENV, "JE_AUTOCONTROL_WAYLAND_RECORD_DEVICES": "/dev/input/event3"},
tools=("ydotool", "grim"), bus=False),
"X11 backend on a Wayland session": _context(
{**_WAYLAND_ENV, "DISPLAY": ":0"}, loaded="x11"),
}
snapshots = {title: ac.probe_capabilities(context) for title, context in desktops.items()}
for title, snapshot in snapshots.items():
show(title, snapshot)
# What this program executes can be journalled with no input hook at all.
# A dry run resolves every command without calling it.
log = ac.InputStepLog()
log.run([["AC_set_var", {"name": "greeting", "value": "hello"}],
["AC_get_var", {"name": "greeting"}]], dry_run=True)
print(f"InputStepLog noted {len(log.steps)} step(s) from a dry run: "
f"{[step[0] for step in log.steps]}")
expected = {
"GNOME, portal not asked yet": ("not_requested", "libei", True),
# ydotool is installed, and is still not used: a refusal is not a fallback.
"GNOME, consent refused": ("needs_permission", "libei", True),
_SWAY_YDOTOOL_GRIM: ("available", "ydotool", True),
"X11 backend on a Wayland session": ("available", "xwayland", False),
}
problems: List[str] = []
for title, wanted in expected.items():
found = snapshots[title].input
got = (found.state.value, found.backend, found.desktop_wide)
if got != wanted:
problems.append(f"{title}: input is {got}, expected {wanted}")
if snapshots[_SWAY_YDOTOOL_GRIM].get("recording").state.value != "needs_permission":
problems.append("an unreadable recording device should read as needs_permission")
if len(log.steps) != 2:
problems.append(f"expected 2 journalled steps, got {len(log.steps)}")
for problem in problems:
print(f"FAILED: {problem}")
print("validate:", "failed" if problems else "ok")
return 1 if problems else 0
def _walkthrough(root: Path) -> Tuple[List[str], Dict[str, Any]]:
store_path = root / "server" / "config_sync.sqlite3"
server = LoopbackServer(store_path)
server.start()
laptop, desktop = Machine("laptop", root, server), Machine("desktop", root, server)
seen: Dict[str, Any] = {}
try:
print("1. a script written on the laptop reaches the desktop")
laptop.write(_SCRIPT, [["AC_set_var", {"name": "user", "value": "alice"}]])
laptop.sync()
desktop.sync()
seen["arrived"] = desktop.read(_SCRIPT)
print("2. the server restarts; the bucket is still there")
committed = ac.ConfigStore(store_path).revision(_USER)
server.stop()
server.start()
seen["restart"] = (committed, ac.ConfigStore(store_path).revision(_USER))
print("3. the laptop works offline; the change waits in its outbox")
server.stop()
laptop.write("report.json", [["AC_set_var", {"name": "mode", "value": "daily"}]])
seen["offline"] = laptop.sync()
server.start()
# A failed send backs off (2 s, doubling, capped at 5 min). A run inside
# that window does not contact the server and says "offline" again with
# error "waiting to retry after an earlier failure"; wait=True sleeps
# through the back-off instead, which is what a scheduled sync wants.
seen["early"] = laptop.sync()
seen["back_online"] = laptop.sync(wait=True)
print("4. both machines edit login.json apart: neither edit is dropped")
desktop.sync()
laptop.write(_SCRIPT, [["AC_set_var", {"name": "user", "value": "from-laptop"}]])
desktop.write(_SCRIPT, [["AC_set_var", {"name": "user", "value": "from-desktop"}]])
laptop.sync()
seen["conflict"] = desktop.sync()
details = desktop.status()["conflict_details"]
for detail in details:
origins = [choice["origin"] for choice in detail["choices"]]
print(f" {detail['section']}/{detail['key']}: choose between {origins}")
seen["details"] = details
print("5. a person picks the laptop's version on the desktop; both converge")
if details:
choices = [choice["origin"] for choice in details[0]["choices"]]
desktop.resolve(details[0]["key"], choices.index("laptop"))
seen["resolved"] = desktop.sync()
laptop.sync()
seen["final"] = (laptop.read(_SCRIPT), desktop.read(_SCRIPT))
finally:
server.stop()
return _check(seen), seen
def validate() -> int:
"""Open a session over the offline transport and check what it would send."""
adb = OfflineAdb(_SERIAL)
context = ac.DeviceContext("android", _SERIAL, timeout_s=5, label="offline pixel")
with ac.open_device(context, adb=adb) as session:
report = ac.device_setup_report(session)
show_report(report)
probing = list(adb.input_sent) # a probe must not touch the device
frame = session.capture()
print(f"captured {frame.pixel_size} px as {frame.to_dict()['point_size']} points,"
f" scale {frame.scale:g}")
session.perform(ac.Tap(540, 960))
session.perform(ac.Swipe(540, 1500, 540, 500, duration_s=0.2))
session.type_text("hello")
# A bound session is what address-less mobile commands target.
with ac.use_device(session):
ac.run_mobile_command("AC_android_tap", {"x": 10, "y": 20})
print("sent to the device:", adb.input_sent)
matrix = ac.mobile_capability_matrix()
for row in matrix["capabilities"]:
print(f" {row['capability']:<14} android={len(row['android'])} ios={len(row['ios'])} command(s)")
print(f" {len(matrix['desktop_only'])} desktop-only feature(s) documented with an alternative")
problems = []
if probing:
problems.append(f"the setup report sent input: {probing}")
if report.state != "available" or report.backend != "adb":
problems.append(f"unexpected setup report: {report.to_dict()}")
wanted = ["input tap 540 960", "input swipe 540 1500 540 500 200", "input tap 10 20"]
missing = [command for command in wanted if command not in adb.input_sent]
if missing:
problems.append(f"not sent: {missing}")
if not any(command.startswith("input text") for command in adb.input_sent):
problems.append("type_text sent no input text command")
if session.connected:
problems.append("the session was not closed on leaving the with block")
for problem in problems:
print(f"FAILED: {problem}")
print("validate:", "failed" if problems else "ok")
return 1 if problems else 0
def validate() -> int:
"""Score two template versions on the drawn frames and check the result."""
comparison = ac.evaluate_locators(build_samples(), {
"v1-single-scale": ac.template_match_strategy(threshold=0.9),
"v2-multi-scale": ac.template_match_strategy(threshold=0.9, scales=(1.0, 1.5)),
})
print(format_comparison(comparison))
gate = {"v2-multi-scale": {"min_accuracy": 1.0, "max_false_positive": 0.0}}
violations = check_thresholds(comparison, gate)
print("thresholds:", "met" if not violations else violations)
old, new = comparison.report("v1-single-scale"), comparison.report("v2-multi-scale")
problems = list(violations)
if [row.sample_id for row in comparison.failures("v1-single-scale")] != ["hidpi-150"]:
problems.append("v1 should fail exactly the HiDPI sample")
if comparison.failures("v2-multi-scale"):
problems.append("v2 should get every labelled sample right")
if (old.accuracy.value or 0) >= (new.accuracy.value or 0):
problems.append("v2 should be measurably more accurate than v1")
for problem in problems:
print(f"FAILED: {problem}")
print("validate:", "failed" if problems else "ok")
return 1 if problems else 0
def validate() -> int:
"""Record an offline run, rebuild it, and dry-run the result."""
with tempfile.TemporaryDirectory(prefix="ac_journal_") as folder:
journal = Path(folder) / "run.jsonl"
started = ac.start_action_journal(journal, run_id="example", session="validate")
try:
# The module-level ac.execute_action takes the list only; the options
# (raise_on_error, dry_run, step_callback) are on the executor object.
ac.executor.execute_action(_OFFLINE_RUN, raise_on_error=True)
finally:
stopped = ac.stop_action_journal()
print(f"journal {Path(started['path']).name}: {stopped['events']} event(s) recorded")
runs = ac.list_journal_runs(journal)
candidate = ac.generate_candidate_from_log(journal, run_id=only_run_id(journal))
describe(candidate)
written = write_candidate(candidate, Path(folder) / "test_example.py",
manifest=Path(folder) / "test_example.manifest.json")
compile(Path(written["output"]).read_text(encoding="utf-8"), written["output"], "exec")
# The candidate's action list is what the script replays. A dry run
# resolves every command name and argument without calling anything.
dry = ac.executor.execute_action(candidate.actions, dry_run=True)
problems = []
if [run["run_id"] for run in runs] != ["example"]:
problems.append(f"expected one run called 'example', got {runs}")
if candidate.actions != _OFFLINE_RUN:
problems.append("the rebuilt action list differs from what was executed")
if candidate.manifest["executed"] or candidate.manifest["outcomes_used_as_input"]:
problems.append("the manifest says something was executed or an outcome was reused")
if len(dry) != len(_OFFLINE_RUN):
problems.append(f"the dry run covered {len(dry)} of {len(_OFFLINE_RUN)} actions")
for problem in problems:
print(f"FAILED: {problem}")
print("validate:", "failed" if problems else "ok")
return 1 if problems else 0
def validate() -> int:
"""Compare a full and a progressive session, then grow the progressive one."""
full = _open("full").tool_names()
session = _open("progressive")
core = session.tool_names()
print(f"full mode offers {len(full)} tools; a progressive session starts with {len(core)}:")
print(" " + ", ".join(core))
found = session.call("ac_tools_search", {"query": _QUERY, "limit": 5})
print(f"search {_QUERY!r} -> {json.dumps(found)[:200]}...")
target = found["tools"][0]["name"]
schema = session.call("ac_tools_schema", {"name": target})
enabled = session.call("ac_tools_enable", {"names": [target]})
after = session.tool_names()
state = session.call("ac_tools_state", {})
print(f"schema of {target}: {sorted(schema)}")
print(f"enabled {target}: {json.dumps(enabled)[:160]}")
print(f"the session now lists {len(after)} tools; notifications: {session.notifications}")
print(f"state: {json.dumps(state)[:200]}")
problems = []
if len(core) >= len(full):
problems.append("progressive mode did not start smaller than full mode")
if target in core or target not in after:
problems.append(f"{target} was not added by ac_tools_enable")
if "notifications/tools/list_changed" not in session.notifications:
problems.append("enabling did not notify the session that its list changed")
for problem in problems:
print(f"FAILED: {problem}")
print("validate:", "failed" if problems else "ok")
return 1 if problems else 0