Add lock_session: lock the workstation and wait for unlock

WHATS_NEW.md

@@ -2,6 +2,12 @@
 
 ## What's new (2026-06-26)
 
+### Lock the Workstation + Wait for Unlock
+
+Lock the box at the end of a run, and block until a human unlocks it before resuming. Full reference: [`docs/source/Eng/doc/new_features/v207_features_doc.rst`](docs/source/Eng/doc/new_features/v207_features_doc.rst).
+
+- **`lock_session` / `plan_lock_session` / `wait_for_unlock` / `wait_for_lock` / `classify_lock_transitions`** (`AC_lock_session`, `AC_plan_lock_session`, `AC_wait_for_unlock`, `AC_classify_lock_transitions`): `session_guard` could *detect* a locked session and raise; this adds *acting* on it. `lock_session` locks the workstation now (`LockWorkStation` on Windows, `loginctl lock-session` / `CGSession -suspend` elsewhere) through an injectable `driver`; `wait_for_unlock` / `wait_for_lock` poll `session_guard.is_session_locked` (reusing its real Windows `OpenInputDesktop` probe) until the state flips or a timeout, with injectable `clock` / `sleep` / `probe`; `plan_lock_session` is the pure per-OS planner and `classify_lock_transitions` reduces a lock-state sample stream to `{event, locked}` lock/unlock events. `wait_for_unlock` is the blocking companion to `ensure_interactive_session`. Fifth feature of the ROUND-15 cross-app OS lane. No `PySide6`.
+
 ### Read and Control the System Volume
 
 Set a known audio baseline before a run — mute, set 30%, or assert the level. Full reference: [`docs/source/Eng/doc/new_features/v206_features_doc.rst`](docs/source/Eng/doc/new_features/v206_features_doc.rst).

docs/source/Eng/doc/new_features/v207_features_doc.rst

@@ -0,0 +1,62 @@
+Lock the Workstation + Wait for Unlock
+======================================
+
+:mod:`session_guard` answers "is the session locked right now?" and raises if it
+is. The missing half is *acting* on the lock state: lock the machine at the end
+of an unattended run, block until a human unlocks it before resuming, or reduce
+a stream of lock-state samples to lock / unlock events. ``lock_session`` adds
+that, behind injectable seams so the logic is testable without touching the OS.
+
+* :func:`lock_session` — lock the workstation now (``LockWorkStation`` on
+  Windows, ``loginctl lock-session`` on Linux, ``CGSession -suspend`` on macOS)
+  through an injectable ``driver``.
+* :func:`plan_lock_session` — pure planner: how the lock would be performed on
+  this OS and whether a default is available (``{backend, argv, available}``).
+* :func:`wait_for_unlock` / :func:`wait_for_lock` — poll
+  :func:`is_session_locked` until the state flips or a timeout, with injectable
+  ``clock`` / ``sleep`` / ``probe`` for deterministic tests.
+* :func:`classify_lock_transitions` — pure: a list of lock-state samples to a
+  list of ``{event, locked}`` lock / unlock transitions.
+
+The lock probe reused by the wait helpers is :mod:`session_guard`'s — the
+Windows ``OpenInputDesktop`` check — so ``wait_for_unlock`` is the blocking
+companion to ``ensure_interactive_session`` (which only raises). Imports no
+``PySide6``.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import (
+        lock_session, wait_for_unlock, classify_lock_transitions,
+    )
+
+    # ... unattended run finishes ...
+    lock_session()                 # secure the machine
+
+    # Resume only once a human has unlocked the box
+    if wait_for_unlock(timeout_s=600):
+        run_next_stage()
+
+    # Reduce a sampled lock-state log to events
+    classify_lock_transitions([False, True, True, False])
+    # -> [{'event': 'lock', 'locked': True},
+    #     {'event': 'unlock', 'locked': False}]
+
+For tests (or any host) pass a ``driver`` / ``probe``:
+
+.. code-block:: python
+
+    locked = lock_session(driver=lambda: True)        # no real lock
+    wait_for_unlock(probe=lambda: False)              # already unlocked
+
+Executor commands
+-----------------
+
+``AC_lock_session`` (→ ``{locked}``), ``AC_plan_lock_session`` (→ the plan),
+``AC_wait_for_unlock`` (``timeout`` / ``interval`` → ``{unlocked}``) and
+``AC_classify_lock_transitions`` (``states`` JSON list → ``{events}``). They are
+exposed as the matching ``ac_*`` MCP tools (``ac_lock_session`` is destructive —
+it interrupts the session; the rest are read-only) and as Script Builder
+commands under **Shell**.

docs/source/Zh/doc/new_features/v207_features_doc.rst

@@ -0,0 +1,55 @@
+鎖定工作站 + 等待解鎖
+====================
+
+:mod:`session_guard` 回答「目前 session 是否鎖定?」並在鎖定時丟出例外。缺少的另一半是對鎖定狀態
+*採取行動*:在無人值守執行結束時鎖定機器、在恢復前阻塞直到有人解鎖,或把一連串鎖定狀態取樣化約為
+鎖定 / 解鎖事件。``lock_session`` 補上這些,並以可注入接縫實作,故邏輯能在不碰作業系統的情況下測試。
+
+* :func:`lock_session` ——立即鎖定工作站(Windows 用 ``LockWorkStation``、Linux 用
+  ``loginctl lock-session``、macOS 用 ``CGSession -suspend``),透過可注入的 ``driver``。
+* :func:`plan_lock_session` ——純 planner:此 OS 上會如何執行鎖定,以及是否有預設可用
+  (``{backend, argv, available}``)。
+* :func:`wait_for_unlock` / :func:`wait_for_lock` ——輪詢 :func:`is_session_locked`
+  直到狀態翻轉或逾時,``clock`` / ``sleep`` / ``probe`` 皆可注入以利確定性測試。
+* :func:`classify_lock_transitions` ——純函式:把一連串鎖定狀態取樣化約為
+  ``{event, locked}`` 鎖定 / 解鎖轉變的清單。
+
+wait 系列重用的鎖定 probe 即 :mod:`session_guard` 的——Windows 的 ``OpenInputDesktop`` 檢查——
+故 ``wait_for_unlock`` 是 ``ensure_interactive_session``(只會丟例外)的阻塞式搭檔。不匯入 ``PySide6``。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import (
+        lock_session, wait_for_unlock, classify_lock_transitions,
+    )
+
+    # ... 無人值守執行結束 ...
+    lock_session()                 # 鎖住機器
+
+    # 等有人解鎖後才繼續
+    if wait_for_unlock(timeout_s=600):
+        run_next_stage()
+
+    # 把取樣的鎖定狀態紀錄化約為事件
+    classify_lock_transitions([False, True, True, False])
+    # -> [{'event': 'lock', 'locked': True},
+    #     {'event': 'unlock', 'locked': False}]
+
+測試時(或任何主機)可傳入 ``driver`` / ``probe``:
+
+.. code-block:: python
+
+    locked = lock_session(driver=lambda: True)        # 不真正鎖定
+    wait_for_unlock(probe=lambda: False)              # 已解鎖
+
+執行器指令
+----------
+
+``AC_lock_session``(→ ``{locked}``)、``AC_plan_lock_session``(→ 計畫)、
+``AC_wait_for_unlock``(``timeout`` / ``interval`` → ``{unlocked}``)與
+``AC_classify_lock_transitions``(``states`` JSON 清單 → ``{events}``)。皆以對應的 ``ac_*``
+MCP 工具(``ac_lock_session`` 為破壞性——會中斷 session;其餘為唯讀)及 Script Builder 指令
+(位於 **Shell** 分類下)形式提供。

je_auto_control/__init__.py

@@ -105,6 +105,11 @@
     change_volume, get_volume, is_muted, mute, set_mute, set_volume,
     toggle_mute, unmute,
 )
+# Lock the workstation, wait for unlock, classify lock transitions
+from je_auto_control.utils.lock_session import (
+    classify_lock_transitions, lock_session, plan_lock_session,
+    wait_for_lock, wait_for_unlock,
+)
 # 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,
@@ -1720,6 +1725,8 @@ def start_autocontrol_gui(*args, **kwargs):
     "normalize_ext", "file_association",
     "get_volume", "set_volume", "change_volume",
     "is_muted", "set_mute", "mute", "unmute", "toggle_mute",
+    "lock_session", "plan_lock_session", "wait_for_unlock",
+    "wait_for_lock", "classify_lock_transitions",
     "build_rtf", "rtf_to_text", "rows_to_csv", "csv_to_rows",
     "set_clipboard_rtf", "get_clipboard_rtf",
     "set_clipboard_csv", "get_clipboard_csv",

je_auto_control/gui/script_builder/command_schema.py

@@ -4345,6 +4345,34 @@ def _add_work_queue_specs(specs: List[CommandSpec]) -> None:
         fields=(),
         description="Flip the master mute flag.",
     ))
+    specs.append(CommandSpec(
+        "AC_lock_session", "Shell", "Lock Workstation",
+        fields=(),
+        description="Lock the workstation now (interrupts the session).",
+    ))
+    specs.append(CommandSpec(
+        "AC_plan_lock_session", "Shell", "Plan Lock Workstation",
+        fields=(),
+        description="Describe how the workstation would be locked (pure).",
+    ))
+    specs.append(CommandSpec(
+        "AC_wait_for_unlock", "Shell", "Wait for Unlock",
+        fields=(
+            FieldSpec("timeout", FieldType.FLOAT, optional=True, default=30.0,
+                      placeholder="timeout seconds"),
+            FieldSpec("interval", FieldType.FLOAT, optional=True, default=0.5,
+                      placeholder="poll interval seconds"),
+        ),
+        description="Block until the session is unlocked or timeout.",
+    ))
+    specs.append(CommandSpec(
+        "AC_classify_lock_transitions", "Shell", "Classify Lock Transitions",
+        fields=(
+            FieldSpec("states", FieldType.STRING,
+                      placeholder="JSON list of booleans"),
+        ),
+        description="Reduce lock-state samples to lock / unlock events.",
+    ))
     specs.append(CommandSpec(
         "AC_normalize_ext", "Shell", "Normalize Extension",
         fields=(

je_auto_control/utils/executor/action_executor.py

@@ -2683,6 +2683,34 @@ def _toggle_mute() -> Dict[str, Any]:
     return {"muted": bool(toggle_mute())}
 
 
+def _lock_session() -> Dict[str, Any]:
+    """Adapter: lock the workstation now."""
+    from je_auto_control.utils.lock_session import lock_session
+    return {"locked": bool(lock_session())}
+
+
+def _plan_lock_session() -> Dict[str, Any]:
+    """Adapter: describe how the workstation would be locked (pure)."""
+    from je_auto_control.utils.lock_session import plan_lock_session
+    return plan_lock_session()
+
+
+def _wait_for_unlock(timeout: Any = 30.0, interval: Any = 0.5
+                     ) -> Dict[str, Any]:
+    """Adapter: block until the session is unlocked or timeout."""
+    from je_auto_control.utils.lock_session import wait_for_unlock
+    unlocked = wait_for_unlock(timeout_s=float(timeout),
+                               interval_s=float(interval))
+    return {"unlocked": bool(unlocked)}
+
+
+def _classify_lock_transitions(states: Any) -> Dict[str, Any]:
+    """Adapter: reduce lock-state samples to lock / unlock events (pure)."""
+    from je_auto_control.utils.lock_session import classify_lock_transitions
+    samples = [bool(s) for s in _coerce_list(states)] if states else []
+    return {"events": classify_lock_transitions(samples)}
+
+
 def _normalize_ext(target: str) -> Dict[str, Any]:
     """Adapter: the lowercased extension of a path / bare ext (pure)."""
     from je_auto_control.utils.file_assoc import normalize_ext
@@ -6697,6 +6725,10 @@ def __init__(self):
             "AC_change_volume": _change_volume,
             "AC_set_mute": _set_mute,
             "AC_toggle_mute": _toggle_mute,
+            "AC_lock_session": _lock_session,
+            "AC_plan_lock_session": _plan_lock_session,
+            "AC_wait_for_unlock": _wait_for_unlock,
+            "AC_classify_lock_transitions": _classify_lock_transitions,
             "AC_normalize_ext": _normalize_ext,
             "AC_file_association": _file_association,
             "AC_get_control_text": _get_control_text,

je_auto_control/utils/lock_session/__init__.py

@@ -0,0 +1,10 @@
+"""Lock the workstation, wait for unlock, and classify lock transitions."""
+from je_auto_control.utils.lock_session.lock_session import (
+    classify_lock_transitions, lock_session, plan_lock_session,
+    wait_for_lock, wait_for_unlock,
+)
+
+__all__ = [
+    "lock_session", "plan_lock_session", "wait_for_unlock", "wait_for_lock",
+    "classify_lock_transitions",
+]

je_auto_control/utils/lock_session/lock_session.py

@@ -0,0 +1,153 @@
+"""Lock the workstation, wait for it to be unlocked, and classify transitions.
+
+:mod:`session_guard` answers "is the session locked right now?" and raises if it
+is. The missing half is *acting* on the lock state: lock the machine at the end
+of an unattended run, block until a human unlocks it before resuming, or reduce
+a stream of lock-state samples to lock / unlock events. ``lock_session`` adds:
+
+* :func:`lock_session` — lock the workstation now (``LockWorkStation`` on
+  Windows, ``loginctl lock-session`` / ``CGSession -suspend`` elsewhere),
+  through an injectable ``driver`` seam.
+* :func:`plan_lock_session` — pure planner describing how the lock would be
+  performed on this OS, and whether a default is available.
+* :func:`wait_for_unlock` / :func:`wait_for_lock` — poll
+  :func:`session_guard.is_session_locked` until the state flips or a timeout,
+  with injectable ``clock`` / ``sleep`` / ``probe`` for deterministic tests.
+* :func:`classify_lock_transitions` — pure: a list of lock-state samples to a
+  list of ``{event, locked}`` lock / unlock transitions.
+
+Imports no ``PySide6``.
+"""
+import sys
+import time
+from typing import Any, Callable, Dict, List, Optional, Sequence
+
+from je_auto_control.utils.session_guard import is_session_locked
+from je_auto_control.utils.session_guard.session_guard import LockProbe
+
+# A driver performs the lock and returns whether it succeeded.
+LockDriver = Callable[[], bool]
+
+
+def _win_lock() -> bool:
+    """Lock the Windows workstation via ``LockWorkStation``."""
+    import ctypes
+    user32 = ctypes.windll.user32  # nosec B607  # reason: fixed system DLL
+    return bool(user32.LockWorkStation())
+
+
+def _lock_backend() -> str:
+    """Return the lock backend name for the current platform."""
+    if sys.platform.startswith("win"):
+        return "LockWorkStation"
+    if sys.platform == "darwin":
+        return "CGSession"
+    return "loginctl"
+
+
+_CGSESSION = ("/System/Library/CoreServices/Menu Extras/"
+              "User.menu/Contents/Resources/CGSession")
+
+
+def _lock_argv(backend: str) -> Optional[List[str]]:
+    """Return the fixed argv for a subprocess lock backend, or None."""
+    if backend == "loginctl":
+        return ["loginctl", "lock-session"]
+    if backend == "CGSession":
+        return [_CGSESSION, "-suspend"]
+    return None
+
+
+def plan_lock_session() -> Dict[str, Any]:
+    """Describe how the workstation would be locked on this OS (pure).
+
+    Returns ``{backend, argv, available}``. ``available`` is ``True`` when this
+    platform has a built-in default; otherwise :func:`lock_session` needs an
+    explicit ``driver=``.
+    """
+    backend = _lock_backend()
+    argv = _lock_argv(backend)
+    available = backend == "LockWorkStation" or argv is not None
+    return {"backend": backend, "argv": argv, "available": available}
+
+
+def _run_argv(argv: List[str]) -> bool:
+    """Run a fixed lock argv and report success (no shell)."""
+    import subprocess  # nosec B404  # reason: fixed argv, no shell
+    completed = subprocess.run(argv, check=False)  # nosec B603  # nosemgrep
+    return completed.returncode == 0
+
+
+def _default_driver() -> LockDriver:
+    """Return the OS lock driver, or raise if none is available."""
+    backend = _lock_backend()
+    if backend == "LockWorkStation":
+        return _win_lock
+    argv = _lock_argv(backend)
+    if argv is not None:
+        return lambda: _run_argv(argv)
+    raise RuntimeError(
+        "lock_session has no OS driver on this platform; pass driver=")
+
+
+def lock_session(*, driver: Optional[LockDriver] = None) -> bool:
+    """Lock the workstation now; return whether the lock was requested.
+
+    Pass ``driver`` (a ``() -> bool``) to intercept the OS call in tests; the
+    default locks via the platform backend from :func:`plan_lock_session`.
+    """
+    acquire = driver if driver is not None else _default_driver()
+    return bool(acquire())
+
+
+def _wait_lock_state(target_locked: bool, *, probe: Optional[LockProbe],
+                     timeout_s: float, interval_s: float,
+                     clock: Callable[[], float],
+                     sleep: Callable[[float], None]) -> bool:
+    """Poll until the lock state equals ``target_locked`` or timeout."""
+    deadline = clock() + float(timeout_s)
+    while True:
+        if bool(is_session_locked(probe)) == target_locked:
+            return True
+        if clock() >= deadline:
+            return False
+        sleep(float(interval_s))
+
+
+def wait_for_unlock(*, probe: Optional[LockProbe] = None,
+                    timeout_s: float = 30.0, interval_s: float = 0.5,
+                    clock: Callable[[], float] = time.monotonic,
+                    sleep: Callable[[float], None] = time.sleep) -> bool:
+    """Block until the session is unlocked; return ``True``, or ``False`` on timeout.
+
+    Reuses :func:`session_guard.is_session_locked` (Windows default probe).
+    ``clock`` / ``sleep`` / ``probe`` are injectable for deterministic tests.
+    """
+    return _wait_lock_state(False, probe=probe, timeout_s=timeout_s,
+                            interval_s=interval_s, clock=clock, sleep=sleep)
+
+
+def wait_for_lock(*, probe: Optional[LockProbe] = None,
+                  timeout_s: float = 30.0, interval_s: float = 0.5,
+                  clock: Callable[[], float] = time.monotonic,
+                  sleep: Callable[[float], None] = time.sleep) -> bool:
+    """Block until the session is locked; return ``True``, or ``False`` on timeout."""
+    return _wait_lock_state(True, probe=probe, timeout_s=timeout_s,
+                            interval_s=interval_s, clock=clock, sleep=sleep)
+
+
+def classify_lock_transitions(states: Sequence[bool]) -> List[Dict[str, Any]]:
+    """Reduce lock-state samples to lock / unlock transitions (pure).
+
+    Each adjacent ``False -> True`` yields a ``lock`` event and each
+    ``True -> False`` an ``unlock`` event; unchanged samples yield nothing.
+    """
+    events: List[Dict[str, Any]] = []
+    previous: Optional[bool] = None
+    for sample in states:
+        current = bool(sample)
+        if previous is not None and current != previous:
+            kind = "lock" if current else "unlock"
+            events.append({"event": kind, "locked": current})
+        previous = current
+    return events