From cf86997f56c9fc04e4486402c1b8523669b2c689 Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Thu, 25 Jun 2026 08:02:36 +0800 Subject: [PATCH] Add idle_keepawake: idle detection + keep machine awake Long unattended runs derail two ways: the screensaver/power policy sleeps the box mid-run, or the run should hold while a human is using the machine. The framework had neither signal. idle_seconds/is_idle read time since the last input (GetLastInputInfo on Windows) through an injectable probe; keep_awake (scoped CM) and keep_awake_on/allow_sleep (process-global on/off for JSON flows) stop the system and display sleeping through an injectable driver (SetThreadExecutionState/caffeinate/systemd-inhibit by default), restored on release. plan_keep_awake is the pure planner. All logic is unit-testable without touching the OS via the injected probe/driver. --- WHATS_NEW.md | 6 + .../doc/new_features/v204_features_doc.rst | 59 +++++ .../Zh/doc/new_features/v204_features_doc.rst | 53 +++++ je_auto_control/__init__.py | 7 + .../gui/script_builder/command_schema.py | 34 +++ .../utils/executor/action_executor.py | 36 ++++ .../utils/idle_keepawake/__init__.py | 10 + .../utils/idle_keepawake/idle_keepawake.py | 202 ++++++++++++++++++ .../utils/mcp_server/tools/_factories.py | 43 ++++ .../utils/mcp_server/tools/_handlers.py | 25 +++ .../headless/test_idle_keepawake_batch.py | 112 ++++++++++ 11 files changed, 587 insertions(+) create mode 100644 docs/source/Eng/doc/new_features/v204_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v204_features_doc.rst create mode 100644 je_auto_control/utils/idle_keepawake/__init__.py create mode 100644 je_auto_control/utils/idle_keepawake/idle_keepawake.py create mode 100644 test/unit_test/headless/test_idle_keepawake_batch.py diff --git a/WHATS_NEW.md b/WHATS_NEW.md index d9277189..11c0817d 100644 --- a/WHATS_NEW.md +++ b/WHATS_NEW.md @@ -1,5 +1,11 @@ # What's New — AutoControl +## What's new (2026-06-25) — Idle Detection + Keep the Machine Awake + +Run only when the user has stepped away, and stop an overnight run from sleeping. Full reference: [`docs/source/Eng/doc/new_features/v204_features_doc.rst`](docs/source/Eng/doc/new_features/v204_features_doc.rst). + +- **`idle_seconds` / `is_idle` / `keep_awake` / `keep_awake_on` / `allow_sleep` / `plan_keep_awake`** (`AC_idle_seconds`, `AC_is_idle`, `AC_plan_keep_awake`, `AC_keep_awake_on`, `AC_allow_sleep`): long unattended runs get derailed two ways — the screensaver / power policy sleeps the box mid-run, or the run should hold while a human is actively using the machine. The framework had neither signal. `idle_seconds` / `is_idle` report time since the last keyboard / mouse input (`GetLastInputInfo` on Windows) through an injectable `probe`; `keep_awake` (scoped context manager) and `keep_awake_on` / `allow_sleep` (process-global on/off for JSON flows) stop the system and display sleeping, applied through an injectable `driver` (`SetThreadExecutionState` / `caffeinate` / `systemd-inhibit` by default) and restored on release. `plan_keep_awake` is the pure planner. All logic is unit-testable without touching the OS via the injected probe/driver. Second feature of the ROUND-15 cross-app OS lane. No `PySide6`. + ## What's new (2026-06-25) — Open Files / URLs with the Default App Hand a file to its default app, print it, or open a URL in the browser. Full reference: [`docs/source/Eng/doc/new_features/v203_features_doc.rst`](docs/source/Eng/doc/new_features/v203_features_doc.rst). diff --git a/docs/source/Eng/doc/new_features/v204_features_doc.rst b/docs/source/Eng/doc/new_features/v204_features_doc.rst new file mode 100644 index 00000000..ba9de3f6 --- /dev/null +++ b/docs/source/Eng/doc/new_features/v204_features_doc.rst @@ -0,0 +1,59 @@ +Idle Detection + Keep the Machine Awake +======================================= + +Long unattended automation runs get derailed two ways: the screensaver / power +policy sleeps the box mid-run, or the run should hold while a human is actively +using the machine. The framework had neither signal. ``idle_keepawake`` adds +both, behind injectable seams so all logic is testable without touching the OS. + +* :func:`idle_seconds` / :func:`is_idle` — seconds since the last user keyboard / + mouse input (``GetLastInputInfo`` on Windows), through an injectable ``probe``. +* :func:`plan_keep_awake` — pure planner describing which wake flags a request + maps to. +* :func:`keep_awake` — scoped context manager that keeps the machine awake for + the duration of a ``with`` block, restoring the prior state on exit. +* :func:`keep_awake_on` / :func:`allow_sleep` — a process-global on / off pair + for JSON action flows. + +All three keep-awake entry points apply the plan through an injectable ``driver`` +(``SetThreadExecutionState`` on Windows, ``caffeinate`` on macOS, +``systemd-inhibit`` on Linux by default). Imports no ``PySide6``. + +Headless API +------------ + +.. code-block:: python + + from je_auto_control import ( + idle_seconds, is_idle, keep_awake, keep_awake_on, allow_sleep, + ) + + idle_seconds() # e.g. 3.4 — seconds since last input + is_idle(300) # True once nobody has touched the machine for 5 min + + # Scoped: keep awake only while a long step runs + with keep_awake(): + run_long_batch() + + # Flow-style: on at the start, off at the end + keep_awake_on(display=True, system=True) + try: + run_long_batch() + finally: + allow_sleep() + +:func:`is_idle` is the gate for "only run when the user has stepped away"; +:func:`keep_awake` / :func:`keep_awake_on` stop the display and system sleeping +so an overnight run is not interrupted. ``display=False`` keeps the system awake +but lets the screen blank (battery-friendly for headless boxes). + +Executor commands +----------------- + +``AC_idle_seconds`` (→ ``{idle_seconds}``), ``AC_is_idle`` (``threshold`` → +``{idle, idle_seconds}``), ``AC_plan_keep_awake`` (``display`` / ``system`` → the +plan), ``AC_keep_awake_on`` (``display`` / ``system`` → the active plan) and +``AC_allow_sleep`` (→ ``{released}``). They are exposed as the matching ``ac_*`` +MCP tools (reads read-only, keep-awake on/off side-effect-only) and as Script +Builder commands under **Shell**. The :func:`keep_awake` context manager is the +Python-API surface for scoped use. diff --git a/docs/source/Zh/doc/new_features/v204_features_doc.rst b/docs/source/Zh/doc/new_features/v204_features_doc.rst new file mode 100644 index 00000000..676c9555 --- /dev/null +++ b/docs/source/Zh/doc/new_features/v204_features_doc.rst @@ -0,0 +1,53 @@ +閒置偵測 + 保持機器清醒 +======================= + +長時間無人值守的自動化執行常因兩種情況中斷:螢幕保護 / 電源原則在執行中途讓機器睡眠,或是當有人正在 +使用機器時執行應該暫停。框架原本兩種訊號都沒有。``idle_keepawake`` 補上這兩者,並以可注入接縫實作, +所有邏輯都能在不碰作業系統的情況下測試。 + +* :func:`idle_seconds` / :func:`is_idle` ——距離使用者上次鍵盤 / 滑鼠輸入的秒數(Windows 上用 + ``GetLastInputInfo``),透過可注入的 ``probe`` 取得。 +* :func:`plan_keep_awake` ——純 planner,描述請求對應到哪些清醒旗標。 +* :func:`keep_awake` ——具範圍的 context manager,在 ``with`` 區塊期間保持機器清醒,離開時還原先前狀態。 +* :func:`keep_awake_on` / :func:`allow_sleep` ——供 JSON 動作流程使用的行程全域開 / 關配對。 + +三個 keep-awake 入口皆透過可注入的 ``driver`` 套用計畫(預設 Windows 用 +``SetThreadExecutionState``、macOS 用 ``caffeinate``、Linux 用 ``systemd-inhibit``)。不匯入 +``PySide6``。 + +無頭 API +-------- + +.. code-block:: python + + from je_auto_control import ( + idle_seconds, is_idle, keep_awake, keep_awake_on, allow_sleep, + ) + + idle_seconds() # 例如 3.4 ——距離上次輸入的秒數 + is_idle(300) # 沒人碰機器滿 5 分鐘後回傳 True + + # 具範圍:只在長步驟執行時保持清醒 + with keep_awake(): + run_long_batch() + + # 流程式:開始時開、結束時關 + keep_awake_on(display=True, system=True) + try: + run_long_batch() + finally: + allow_sleep() + +:func:`is_idle` 是「只在使用者離開時才執行」的判斷閘;:func:`keep_awake` / +:func:`keep_awake_on` 阻止螢幕與系統睡眠,讓整夜執行不被打斷。``display=False`` 會保持系統清醒但允許 +螢幕變黑(對無頭機器較省電)。 + +執行器指令 +---------- + +``AC_idle_seconds``(→ ``{idle_seconds}``)、``AC_is_idle``(``threshold`` → +``{idle, idle_seconds}``)、``AC_plan_keep_awake``(``display`` / ``system`` → 計畫)、 +``AC_keep_awake_on``(``display`` / ``system`` → 生效中的計畫)與 ``AC_allow_sleep`` +(→ ``{released}``)。皆以對應的 ``ac_*`` MCP 工具(讀取為唯讀、keep-awake 開 / 關為僅副作用) +及 Script Builder 指令(位於 **Shell** 分類下)形式提供。:func:`keep_awake` context manager +則是具範圍使用的 Python API 介面。 diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index 3cc59027..e6d113fa 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -93,6 +93,11 @@ from je_auto_control.utils.ax_events import wait_for_focus_change # Open a file with its default app / a URL in the default browser from je_auto_control.utils.shell_open import open_path, plan_open +# Detect user-idle time and keep the machine awake during unattended runs +from je_auto_control.utils.idle_keepawake import ( + allow_sleep, idle_seconds, is_idle, keep_awake, keep_awake_on, + plan_keep_awake, +) # Rich clipboard formats — RTF + CSV/TSV codecs and Windows get / set from je_auto_control.utils.clipboard_rich_formats import ( build_rtf, csv_to_rows, get_clipboard_csv, get_clipboard_rtf, rows_to_csv, @@ -1703,6 +1708,8 @@ def start_autocontrol_gui(*args, **kwargs): "get_selection", "list_views", "set_view", "wait_for_focus_change", "plan_open", "open_path", + "idle_seconds", "is_idle", "plan_keep_awake", + "keep_awake", "keep_awake_on", "allow_sleep", "build_rtf", "rtf_to_text", "rows_to_csv", "csv_to_rows", "set_clipboard_rtf", "get_clipboard_rtf", "set_clipboard_csv", "get_clipboard_csv", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 2be6e99d..ef1c0395 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -4268,6 +4268,40 @@ def _add_work_queue_specs(specs: List[CommandSpec]) -> None: ), description="Classify how a file/URL would be opened (pure, no launch).", )) + specs.append(CommandSpec( + "AC_idle_seconds", "Shell", "Idle Seconds", + fields=(), + description="Seconds since the last user keyboard / mouse input.", + )) + specs.append(CommandSpec( + "AC_is_idle", "Shell", "Is User Idle", + fields=( + FieldSpec("threshold", FieldType.FLOAT, default=300.0, + placeholder="idle seconds threshold"), + ), + description="True if the user has been idle for >= threshold seconds.", + )) + specs.append(CommandSpec( + "AC_plan_keep_awake", "Shell", "Plan Keep Awake", + fields=( + FieldSpec("display", FieldType.BOOL, optional=True, default=True), + FieldSpec("system", FieldType.BOOL, optional=True, default=True), + ), + description="Describe a keep-awake request (pure, no OS call).", + )) + specs.append(CommandSpec( + "AC_keep_awake_on", "Shell", "Keep Machine Awake", + fields=( + FieldSpec("display", FieldType.BOOL, optional=True, default=True), + FieldSpec("system", FieldType.BOOL, optional=True, default=True), + ), + description="Keep the machine awake until Allow Sleep is run.", + )) + specs.append(CommandSpec( + "AC_allow_sleep", "Shell", "Allow Machine to Sleep", + fields=(), + description="Release a previously-started keep-awake.", + )) specs.append(CommandSpec( "AC_take_golden", "Report", "Capture Golden Image", fields=(FieldSpec("path", FieldType.FILE_PATH),), diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index 4d808f57..b84a7790 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2621,6 +2621,37 @@ def _open_path(target: str, verb: str = "open") -> Dict[str, Any]: return {"opened": bool(open_path(str(target), verb=str(verb)))} +def _idle_seconds() -> Dict[str, Any]: + """Adapter: seconds since the last user input.""" + from je_auto_control.utils.idle_keepawake import idle_seconds + return {"idle_seconds": float(idle_seconds())} + + +def _is_idle(threshold: Any) -> Dict[str, Any]: + """Adapter: whether the user has been idle for >= ``threshold`` seconds.""" + from je_auto_control.utils.idle_keepawake import idle_seconds, is_idle + seconds = float(threshold) + return {"idle": bool(is_idle(seconds)), "idle_seconds": idle_seconds()} + + +def _plan_keep_awake(display: Any = True, system: Any = True) -> Dict[str, Any]: + """Adapter: describe a keep-awake request (pure, no OS call).""" + from je_auto_control.utils.idle_keepawake import plan_keep_awake + return plan_keep_awake(display=bool(display), system=bool(system)) + + +def _keep_awake_on(display: Any = True, system: Any = True) -> Dict[str, Any]: + """Adapter: keep the machine awake until ``AC_allow_sleep``.""" + from je_auto_control.utils.idle_keepawake import keep_awake_on + return keep_awake_on(display=bool(display), system=bool(system)) + + +def _allow_sleep() -> Dict[str, Any]: + """Adapter: release a previously-started keep-awake.""" + from je_auto_control.utils.idle_keepawake import allow_sleep + return {"released": bool(allow_sleep())} + + def _get_control_text(name: Optional[str] = None, role: Optional[str] = None, app_name: Optional[str] = None, automation_id: Optional[str] = None) -> Dict[str, Any]: @@ -6613,6 +6644,11 @@ def __init__(self): "AC_wait_for_focus_change": _wait_for_focus_change, "AC_plan_open": _plan_open, "AC_open_path": _open_path, + "AC_idle_seconds": _idle_seconds, + "AC_is_idle": _is_idle, + "AC_plan_keep_awake": _plan_keep_awake, + "AC_keep_awake_on": _keep_awake_on, + "AC_allow_sleep": _allow_sleep, "AC_get_control_text": _get_control_text, "AC_find_control_text": _find_control_text, "AC_select_control_text": _select_control_text, diff --git a/je_auto_control/utils/idle_keepawake/__init__.py b/je_auto_control/utils/idle_keepawake/__init__.py new file mode 100644 index 00000000..2f860ec2 --- /dev/null +++ b/je_auto_control/utils/idle_keepawake/__init__.py @@ -0,0 +1,10 @@ +"""Detect user-idle time and keep the machine awake during unattended runs.""" +from je_auto_control.utils.idle_keepawake.idle_keepawake import ( + allow_sleep, idle_seconds, is_idle, keep_awake, keep_awake_on, + plan_keep_awake, +) + +__all__ = [ + "idle_seconds", "is_idle", "plan_keep_awake", + "keep_awake", "keep_awake_on", "allow_sleep", +] diff --git a/je_auto_control/utils/idle_keepawake/idle_keepawake.py b/je_auto_control/utils/idle_keepawake/idle_keepawake.py new file mode 100644 index 00000000..9c5b0700 --- /dev/null +++ b/je_auto_control/utils/idle_keepawake/idle_keepawake.py @@ -0,0 +1,202 @@ +"""Detect how long the user has been idle, and keep the machine awake. + +Long unattended automation runs get derailed two ways: the screensaver / power +policy sleeps the box mid-run, or the run should hold while a human is actively +using the machine. The framework had neither signal. ``idle_keepawake`` adds +both, behind injectable seams so all logic is testable without touching the OS: + +* :func:`idle_seconds` / :func:`is_idle` report seconds since the last user + keyboard / mouse input (``GetLastInputInfo`` on Windows) through an injectable + ``probe``. +* :func:`plan_keep_awake` is a pure planner describing which wake flags a request + maps to. :func:`keep_awake` is a scoped context manager, and + :func:`keep_awake_on` / :func:`allow_sleep` are a process-global on / off pair + for JSON action flows — both apply the plan through an injectable ``driver`` + (``SetThreadExecutionState`` / ``caffeinate`` / ``systemd-inhibit`` by + default) and restore the prior state on release. + +Imports no ``PySide6``. +""" +import ctypes +import sys +import threading +from contextlib import contextmanager +from typing import Any, Callable, Dict, Iterator, List, Optional + +# Windows SetThreadExecutionState flags. +_ES_CONTINUOUS = 0x80000000 +_ES_SYSTEM_REQUIRED = 0x00000001 +_ES_DISPLAY_REQUIRED = 0x00000002 + +# A probe: returns seconds since the last user input. +IdleProbe = Callable[[], float] +# A driver: applies a keep-awake plan and returns a zero-arg release callable. +KeepAwakeDriver = Callable[[Dict[str, Any]], Callable[[], None]] + +# Process-global keep-awake state for the on / off serializable surface: holds +# the active release callable (0 or 1 entry), swapped atomically under the lock. +_LOCK = threading.Lock() +_ACTIVE: List[Callable[[], None]] = [] + + +class _LastInputInfo(ctypes.Structure): + """Win32 ``LASTINPUTINFO`` — tick of the last input via GetLastInputInfo.""" + + _fields_ = [("cbSize", ctypes.c_uint), ("dwTime", ctypes.c_uint)] + + +def _default_probe() -> float: + """Return seconds since last input on Windows; raise elsewhere.""" + if not sys.platform.startswith("win"): + raise RuntimeError( + "idle detection has no OS probe on this platform; pass probe=") + info = _LastInputInfo(ctypes.sizeof(_LastInputInfo), 0) + user32 = ctypes.windll.user32 + kernel32 = ctypes.windll.kernel32 + if not user32.GetLastInputInfo(ctypes.byref(info)): + raise RuntimeError("GetLastInputInfo failed") + millis = (kernel32.GetTickCount() - info.dwTime) & 0xFFFFFFFF + return max(0.0, millis / 1000.0) + + +def idle_seconds(*, probe: Optional[IdleProbe] = None) -> float: + """Return seconds since the last user keyboard / mouse input. + + Pass ``probe`` (a ``() -> float``) to supply the reading in tests; the + default queries the OS (Windows only). Never negative. + """ + source = probe if probe is not None else _default_probe + value = float(source()) + return value if value >= 0.0 else 0.0 + + +def is_idle(threshold_seconds: float, *, + probe: Optional[IdleProbe] = None) -> bool: + """Return True if the user has been idle for at least ``threshold_seconds``. + + Raises ``ValueError`` for a negative threshold. + """ + if threshold_seconds < 0: + raise ValueError("threshold_seconds must be non-negative") + return idle_seconds(probe=probe) >= float(threshold_seconds) + + +def _keepawake_backend() -> str: + if sys.platform.startswith("win"): + return "SetThreadExecutionState" + if sys.platform == "darwin": + return "caffeinate" + return "systemd-inhibit" + + +def plan_keep_awake(*, display: bool = True, + system: bool = True) -> Dict[str, Any]: + """Describe a keep-awake request (pure, no OS call). + + Returns ``{display, system, backend, flags}`` where ``flags`` is the Windows + ``SetThreadExecutionState`` bitmask the request maps to. Raises + ``ValueError`` if neither the system nor the display is to be kept awake. + """ + if not display and not system: + raise ValueError("keep_awake must keep the system or display awake") + flags = _ES_CONTINUOUS + if system: + flags |= _ES_SYSTEM_REQUIRED + if display: + flags |= _ES_DISPLAY_REQUIRED + return {"display": bool(display), "system": bool(system), + "backend": _keepawake_backend(), "flags": flags} + + +def _win_keep_awake(flags: int) -> Callable[[], None]: + kernel32 = ctypes.windll.kernel32 + kernel32.SetThreadExecutionState(ctypes.c_uint(flags)) + + def _release() -> None: + kernel32.SetThreadExecutionState(ctypes.c_uint(_ES_CONTINUOUS)) + + return _release + + +def _proc_keep_awake(argv: List[str]) -> Callable[[], None]: + import subprocess # nosec B404 # reason: fixed argv, no shell + # fixed argv list, no shell — injection-safe + proc = subprocess.Popen(argv) # nosec B603 # nosemgrep + + def _release() -> None: + proc.terminate() + + return _release + + +def _caffeinate_argv(plan: Dict[str, Any]) -> List[str]: + argv = ["caffeinate", "-i"] + if plan["system"]: + argv.append("-s") + if plan["display"]: + argv.append("-d") + return argv + + +def _systemd_argv(plan: Dict[str, Any]) -> List[str]: + what = "idle:sleep" if plan["display"] else "sleep" + return ["systemd-inhibit", f"--what={what}", + "--why=AutoControl unattended run", "sleep", "infinity"] + + +def _default_driver(plan: Dict[str, Any]) -> Callable[[], None]: + builders = { + "SetThreadExecutionState": lambda: _win_keep_awake(plan["flags"]), + "caffeinate": lambda: _proc_keep_awake(_caffeinate_argv(plan)), + "systemd-inhibit": lambda: _proc_keep_awake(_systemd_argv(plan)), + } + return builders[plan["backend"]]() + + +@contextmanager +def keep_awake(*, display: bool = True, system: bool = True, + driver: Optional[KeepAwakeDriver] = None + ) -> Iterator[Dict[str, Any]]: + """Scoped context manager keeping the machine (and display) awake. + + Inside the ``with`` block the system will not sleep / screensave; the prior + state is restored on exit. Pass ``driver`` (``plan -> release_callable``) to + intercept the OS call in tests. Yields the active plan. + """ + plan = plan_keep_awake(display=display, system=system) + acquire = driver if driver is not None else _default_driver + release = acquire(plan) + try: + yield plan + finally: + if callable(release): + release() + + +def keep_awake_on(*, display: bool = True, system: bool = True, + driver: Optional[KeepAwakeDriver] = None) -> Dict[str, Any]: + """Start keeping the machine awake until :func:`allow_sleep` is called. + + Process-global and lock-guarded; a second call replaces the prior request. + Pass ``driver`` to intercept the OS call in tests. Returns the active plan. + """ + plan = plan_keep_awake(display=display, system=system) + acquire = driver if driver is not None else _default_driver + release = acquire(plan) + with _LOCK: + previous = _ACTIVE[:] + _ACTIVE.clear() + _ACTIVE.append(release) + for old in previous: + old() + return plan + + +def allow_sleep() -> bool: + """Release a previously-started keep-awake. True if one was active.""" + with _LOCK: + pending = _ACTIVE[:] + _ACTIVE.clear() + for release in pending: + release() + return bool(pending) diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index 1b647680..67a25d1a 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -2106,6 +2106,49 @@ def process_and_shell_tools() -> List[MCPTool]: handler=h.plan_open, annotations=READ_ONLY, ), + MCPTool( + name="ac_idle_seconds", + description=("Seconds since the last user keyboard / mouse input " + "(GetLastInputInfo on Windows). Returns {idle_seconds}."), + input_schema=schema({}), + handler=h.idle_seconds, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_is_idle", + description=("Whether the user has been idle for at least " + "'threshold' seconds. Returns {idle, idle_seconds}."), + input_schema=schema({"threshold": {"type": "number"}}, + required=["threshold"]), + handler=h.is_idle, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_plan_keep_awake", + description=("Describe a keep-awake request without applying it " + "(pure): {display, system, backend, flags}."), + input_schema=schema({"display": {"type": "boolean"}, + "system": {"type": "boolean"}}), + handler=h.plan_keep_awake, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_keep_awake_on", + description=("Keep the machine (and 'display') awake until " + "ac_allow_sleep is called. Returns the active plan."), + input_schema=schema({"display": {"type": "boolean"}, + "system": {"type": "boolean"}}), + handler=h.keep_awake_on, + annotations=SIDE_EFFECT_ONLY, + ), + MCPTool( + name="ac_allow_sleep", + description=("Release a previously-started keep-awake so the machine " + "can sleep again. Returns {released}."), + input_schema=schema({}), + handler=h.allow_sleep, + annotations=SIDE_EFFECT_ONLY, + ), ] diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index b130cda5..723b7a35 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -568,6 +568,31 @@ def plan_open(target, verb="open"): return _plan_open(target, verb) +def idle_seconds(): + from je_auto_control.utils.executor.action_executor import _idle_seconds + return _idle_seconds() + + +def is_idle(threshold): + from je_auto_control.utils.executor.action_executor import _is_idle + return _is_idle(threshold) + + +def plan_keep_awake(display=True, system=True): + from je_auto_control.utils.executor.action_executor import _plan_keep_awake + return _plan_keep_awake(display, system) + + +def keep_awake_on(display=True, system=True): + from je_auto_control.utils.executor.action_executor import _keep_awake_on + return _keep_awake_on(display, system) + + +def allow_sleep(): + from je_auto_control.utils.executor.action_executor import _allow_sleep + return _allow_sleep() + + def get_clipboard() -> str: from je_auto_control.utils.clipboard.clipboard import get_clipboard as _get return _get() diff --git a/test/unit_test/headless/test_idle_keepawake_batch.py b/test/unit_test/headless/test_idle_keepawake_batch.py new file mode 100644 index 00000000..048b0aef --- /dev/null +++ b/test/unit_test/headless/test_idle_keepawake_batch.py @@ -0,0 +1,112 @@ +"""Headless tests for idle detection + keep-awake (injected probe / driver).""" +import pytest + +import je_auto_control as ac +from je_auto_control.utils.idle_keepawake import ( + allow_sleep, idle_seconds, is_idle, keep_awake, keep_awake_on, + plan_keep_awake, +) +from je_auto_control.utils.idle_keepawake.idle_keepawake import ( + _ES_CONTINUOUS, _ES_DISPLAY_REQUIRED, _ES_SYSTEM_REQUIRED, +) + + +def test_idle_seconds_uses_probe(): + assert idle_seconds(probe=lambda: 12.5) == pytest.approx(12.5) + + +def test_idle_seconds_clamps_negative(): + assert idle_seconds(probe=lambda: -3.0) == pytest.approx(0.0) + + +def test_is_idle_threshold(): + assert is_idle(5, probe=lambda: 9.0) is True + assert is_idle(15, probe=lambda: 9.0) is False + + +def test_is_idle_rejects_negative_threshold(): + with pytest.raises(ValueError): + is_idle(-1, probe=lambda: 0.0) + + +def test_plan_keep_awake_flags(): + plan = plan_keep_awake(display=True, system=True) + assert plan["flags"] == (_ES_CONTINUOUS | _ES_SYSTEM_REQUIRED + | _ES_DISPLAY_REQUIRED) + assert plan["display"] is True and plan["system"] is True + assert plan["backend"] in ("SetThreadExecutionState", "caffeinate", + "systemd-inhibit") + + +def test_plan_keep_awake_system_only_omits_display_flag(): + plan = plan_keep_awake(display=False, system=True) + assert plan["flags"] & _ES_DISPLAY_REQUIRED == 0 + assert plan["flags"] & _ES_SYSTEM_REQUIRED == _ES_SYSTEM_REQUIRED + + +def test_plan_keep_awake_rejects_all_false(): + with pytest.raises(ValueError): + plan_keep_awake(display=False, system=False) + + +def test_keep_awake_context_acquires_and_releases(): + events = [] + + def fake_driver(plan): + events.append(("acquire", plan["flags"])) + return lambda: events.append(("release", None)) + + with keep_awake(driver=fake_driver) as plan: + assert plan["backend"] is not None + assert events == [("acquire", plan["flags"])] + assert events[-1] == ("release", None) + + +def test_keep_awake_on_then_allow_sleep(): + released = [] + plan = keep_awake_on(driver=lambda p: lambda: released.append(True)) + assert plan["system"] is True + assert allow_sleep() is True + assert released == [True] + + +def test_allow_sleep_when_nothing_active(): + allow_sleep() # drain any prior state + assert allow_sleep() is False + + +def test_keep_awake_on_replaces_prior_request(): + released = [] + keep_awake_on(driver=lambda p: lambda: released.append("first")) + keep_awake_on(driver=lambda p: lambda: released.append("second")) + assert released == ["first"] # prior request released on replace + assert allow_sleep() is True + assert released == ["first", "second"] + + +# --- wiring --------------------------------------------------------------- + +def test_executor_pure_plan_path(): + from je_auto_control.utils.executor.action_executor import _plan_keep_awake + plan = _plan_keep_awake(display=True, system=False) + assert plan["flags"] & _ES_DISPLAY_REQUIRED == _ES_DISPLAY_REQUIRED + + +def test_wiring(): + known = set(ac.executor.known_commands()) + assert {"AC_idle_seconds", "AC_is_idle", "AC_plan_keep_awake", + "AC_keep_awake_on", "AC_allow_sleep"} <= known + from je_auto_control.utils.mcp_server.tools import build_default_tool_registry + names = {t.name for t in build_default_tool_registry()} + assert {"ac_idle_seconds", "ac_is_idle", "ac_plan_keep_awake", + "ac_keep_awake_on", "ac_allow_sleep"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + specs = {s.command for s in _build_specs()} + assert {"AC_idle_seconds", "AC_is_idle", "AC_plan_keep_awake", + "AC_keep_awake_on", "AC_allow_sleep"} <= specs + + +def test_facade_exports(): + for name in ("idle_seconds", "is_idle", "plan_keep_awake", + "keep_awake", "keep_awake_on", "allow_sleep"): + assert hasattr(ac, name) and name in ac.__all__