feat: stop record when --until-cmd exits 0

openadapt_flow/__main__.py

@@ -997,6 +997,18 @@ def _replay_desktop(
     )
 
 
+def _record_until_stop(args: argparse.Namespace):
+    """Poll ``--until-cmd`` until it exits 0. Ctrl-C remains the fallback."""
+
+    raw = getattr(args, "until_cmd", None)
+    if not raw:
+        return None
+    from openadapt_flow.desktop_record import UntilCommand, parse_until_cmd
+
+    print("Recording until the check succeeds. Ctrl-C still stops.")
+    return UntilCommand(parse_until_cmd(raw))
+
+
 def _cmd_record(args: argparse.Namespace) -> int:
     # The interactive (web) recorder installs in-page DOM listeners against a
     # headed Playwright page. The DESKTOP recorder (--backend windows) captures
@@ -1095,6 +1107,7 @@ def _cmd_record(args: argparse.Namespace) -> int:
             headless=args.headless,
             cdp_endpoint=getattr(args, "browser_cdp_endpoint", None),
             browser_page_url=getattr(args, "browser_page_url", None),
+            stop_when=_record_until_stop(args),
             surface="web",
         )
     except BrowserAttachError as exc:
@@ -1277,6 +1290,7 @@ def _cmd_record_desktop(args: argparse.Namespace, backend: str) -> int:
         replay_window=getattr(args, "rdp_window", None),
         replay_window_title=getattr(args, "rdp_window_title", None),
         readiness_text=getattr(args, "rdp_readiness_text", None),
+        stop=_record_until_stop(args),
     )
     _stamp_recording_surface(out, backend)
     print(f"Recording written to {out}")
@@ -5366,6 +5380,16 @@ def build_parser() -> argparse.ArgumentParser:
             "(stored in the recording metadata)."
         ),
     )
+    p.add_argument(
+        "--until-cmd",
+        default=None,
+        metavar="CMD",
+        help=(
+            "Run CMD on an interval while recording. Exit 0 stops the record. "
+            "The command is not captured as clicks. Ctrl-C still stops. "
+            "A non-zero exit keeps recording (probe failure is not done)."
+        ),
+    )
     p.add_argument(
         "--window",
         default=None,

openadapt_flow/desktop_record.py

@@ -44,6 +44,8 @@
 
 import functools
 import json
+import shlex
+import subprocess
 import sys
 import time
 from pathlib import Path
@@ -93,31 +95,112 @@ def _default_recorder_factory(
     capture session and surfaced by the capture adapter into ``meta.json``.
     """
     try:
+        from loguru import logger
         from openadapt_capture import Recorder as CaptureRecorder
+        from openadapt_capture import recorder as capture_recorder
     except ImportError as exc:  # pragma: no cover - exercised via install state
         raise ImportError(
             "openadapt-capture is required to record a desktop workflow but is "
             "not installed. Install the optional extra:\n\n"
             "    pip install 'openadapt-flow[capture]'\n"
         ) from exc
-    return CaptureRecorder(
-        task_description=task_description, capture_dir=capture_dir, window=window
-    )
+    import inspect
+
+    params = inspect.signature(CaptureRecorder.__init__).parameters
+    if "window" not in params:
+        raise SystemExit(
+            "record: this openadapt-capture does not accept window=. "
+            "Need openadapt-capture>=1.2.2. Run: pip install -U "
+            "'openadapt-capture==1.2.2'"
+        )
+    if hasattr(capture_recorder, "LOG_LEVEL"):
+        capture_recorder.LOG_LEVEL = "WARNING"
+    logger.remove()
+    logger.add(sys.stderr, level="WARNING")
+    kwargs: dict[str, Any] = {
+        "task_description": task_description,
+        "capture_dir": capture_dir,
+        "window": window,
+    }
+    if "plot_performance" in params:
+        kwargs["plot_performance"] = False
+    return CaptureRecorder(**kwargs)
+
+
+def parse_until_cmd(raw: str) -> list[str]:
+    """Split ``--until-cmd`` into argv. Empty input is a usage error."""
+
+    argv = shlex.split(raw)
+    if not argv:
+        raise SystemExit("record: --until-cmd is empty. Nothing was recorded.")
+    return argv
+
+
+class UntilCommand:
+    """Poll a command. Exit 0 stops the record. Other exits keep recording.
+
+    The command is not part of the demonstration. Ctrl-C still stops.
+    Probe failures must not stop the record (fail closed on the check, not
+    on the capture).
+    """
+
+    def __init__(
+        self,
+        argv: list[str],
+        *,
+        interval_s: float = 2.0,
+        timeout_s: float = 30.0,
+        run: Optional[Callable[..., Any]] = None,
+        announce: bool = True,
+    ) -> None:
+        if not argv:
+            raise ValueError("until-cmd argv must not be empty")
+        self.argv = list(argv)
+        self.interval_s = interval_s
+        self.timeout_s = timeout_s
+        self._run = run or subprocess.run
+        self._announce = announce
+        self._next = 0.0
+        self._fired = False
+
+    def __call__(self) -> bool:
+        if self._fired:
+            return True
+        now = time.monotonic()
+        if now < self._next:
+            return False
+        self._next = now + self.interval_s
+        try:
+            result = self._run(
+                self.argv,
+                capture_output=True,
+                timeout=self.timeout_s,
+            )
+        except Exception:
+            return False
+        if int(getattr(result, "returncode", 1)) != 0:
+            return False
+        self._fired = True
+        if self._announce:
+            print("\n[record] check succeeded; stopping.")
+        return True
 
 
 def _wait_for_stop(stop: Optional[Callable[[], bool]]) -> None:
     """Block until the operator interrupts (Ctrl-C) or ``stop()`` returns True.
 
-    ``stop`` is a test/programmatic hook; in the interactive CLI it is None and
-    the loop runs until KeyboardInterrupt.
+    ``stop`` is a test/programmatic hook or ``--until-cmd``. Ctrl-C remains
+    the fallback when the check has not succeeded yet.
     """
     try:
         while True:
             if stop is not None and stop():
                 return
             time.sleep(0.2)
     except KeyboardInterrupt:
-        print("\n[record] stopping…")
+        print("\n[record] stopping (Ctrl-C).")
+        if stop is not None:
+            print("[record] If you set --until-cmd, run that check before compile.")
 
 
 def record_desktop_capture(
@@ -236,9 +319,16 @@ def record_desktop_capture(
                 "  Window-scoped capture is active (recording that window's "
                 "own pixels; target selectors are not printed).\n"
             )
+        until_line = ""
+        if stop is not None:
+            until_line = (
+                "  Recording stops when the check succeeds.\n"
+                "  Ctrl-C still stops (fallback).\n"
+            )
         print(
             f"Recording desktop workflow (task: {task_description!r}).\n"
             f"{scope_line}"
+            f"{until_line}"
             "  Perform your workflow on the target desktop now.\n"
             "  Press Ctrl-C here to finish."
         )

tests/test_desktop_record.py

@@ -18,7 +18,11 @@
 import pytest
 from PIL import Image, ImageDraw
 
-from openadapt_flow.desktop_record import record_desktop_capture
+from openadapt_flow.desktop_record import (
+    UntilCommand,
+    parse_until_cmd,
+    record_desktop_capture,
+)
 
 VIEWPORT = (800, 600)
 SEARCH = (120, 90)
@@ -956,3 +960,79 @@ def test_record_desktop_no_window_ok_on_any_platform(
         announce=False,
     )
     assert out == tmp_path / "rec"
+
+
+def test_parse_until_cmd_rejects_empty() -> None:
+    with pytest.raises(SystemExit, match="--until-cmd is empty"):
+        parse_until_cmd("   ")
+
+
+def test_until_command_stops_only_on_exit_zero() -> None:
+    calls: list[list[str]] = []
+
+    class _Result:
+        def __init__(self, code: int) -> None:
+            self.returncode = code
+
+    def run(argv: list[str], **kwargs: Any) -> _Result:
+        calls.append(list(argv))
+        return _Result(1 if len(calls) < 2 else 0)
+
+    stop = UntilCommand(
+        ["python3", "check_awake.py"],
+        interval_s=0.0,
+        run=run,
+        announce=False,
+    )
+    assert stop() is False
+    assert stop() is True
+    assert stop() is True
+    assert calls == [
+        ["python3", "check_awake.py"],
+        ["python3", "check_awake.py"],
+    ]
+
+
+def test_until_command_probe_error_keeps_recording() -> None:
+    def run(*args: Any, **kwargs: Any) -> Any:
+        raise OSError("probe down")
+
+    stop = UntilCommand(
+        ["python3", "check_awake.py"],
+        interval_s=0.0,
+        run=run,
+        announce=False,
+    )
+    assert stop() is False
+
+
+def test_cli_record_macos_until_cmd_stops_capture(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    captured: dict[str, Any] = {}
+
+    def fake_record(out_dir, **kwargs):
+        captured["stop"] = kwargs.get("stop")
+        Path(out_dir).mkdir(parents=True, exist_ok=True)
+        return Path(out_dir)
+
+    monkeypatch.setattr(
+        "openadapt_flow.desktop_record.record_desktop_capture", fake_record
+    )
+    rc = _run_cli(
+        [
+            "record",
+            "--backend",
+            "macos",
+            "--macos-app",
+            "Google Chrome",
+            "--out",
+            str(tmp_path / "rec"),
+            "--until-cmd",
+            "python3 check_awake.py",
+        ]
+    )
+    assert rc == 0
+    stop = captured["stop"]
+    assert isinstance(stop, UntilCommand)
+    assert stop.argv == ["python3", "check_awake.py"]