Add timed input replay and declarative input-sequence DSL

JE-Chen · JE-Chen · commit b449ca39cc70 · 2026-06-19T17:11:46.000+08:00
diff --git a/README.md b/README.md
@@ -13,6 +13,7 @@
 
 ## Table of Contents
 
+- [What's new (2026-06-19) — Timed Input Macros](#whats-new-2026-06-19--timed-input-macros)
 - [What's new (2026-06-19) — Semantic Screen State](#whats-new-2026-06-19--semantic-screen-state)
 - [What's new (2026-06-19) — Set-of-Marks Overlay](#whats-new-2026-06-19--set-of-marks-overlay)
 - [What's new (2026-06-19) — Checkpoint & Resume](#whats-new-2026-06-19--checkpoint--resume)
@@ -76,6 +77,13 @@
 
 ---
 
+## What's new (2026-06-19) — Timed Input Macros
+
+Replay input with timing fidelity + a press-hold-release DSL, full stack. Full reference: [`docs/source/Eng/doc/new_features/v24_features_doc.rst`](docs/source/Eng/doc/new_features/v24_features_doc.rst).
+
+- **Timed timeline replay** — `replay_timeline(events, speed=...)` (`AC_replay_timeline`, `ac_replay_timeline`): replay events honoring each `delta_ms` gap, scaled by `speed` and clampable; ops = move/click/scroll/press/release/key.
+- **Input-sequence DSL** — `run_sequence(steps)` (`AC_input_sequence`, `ac_input_sequence`): declarative press/hold/release chords + `repeat`/`wait`. Both inject sink+sleep for deterministic tests.
+
 ## What's new (2026-06-19) — Semantic Screen State
 
 The semantic companion to the pixel diff, full stack. Full reference: [`docs/source/Eng/doc/new_features/v23_features_doc.rst`](docs/source/Eng/doc/new_features/v23_features_doc.rst).
diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md
@@ -12,6 +12,7 @@
 
 ## 目录
 
+- [本次更新 (2026-06-19) — 计时输入宏](#本次更新-2026-06-19--计时输入宏)
 - [本次更新 (2026-06-19) — 语义屏幕状态](#本次更新-2026-06-19--语义屏幕状态)
 - [本次更新 (2026-06-19) — Set-of-Marks 叠图](#本次更新-2026-06-19--set-of-marks-叠图)
 - [本次更新 (2026-06-19) — 检查点与续跑](#本次更新-2026-06-19--检查点与续跑)
@@ -75,6 +76,13 @@
 
 ---
 
+## 本次更新 (2026-06-19) — 计时输入宏
+
+以时间保真度重播输入 + 按住-放开 DSL,走完整五层。完整参考:[`docs/source/Zh/doc/new_features/v24_features_doc.rst`](../docs/source/Zh/doc/new_features/v24_features_doc.rst)。
+
+- **计时时间轴重播** — `replay_timeline(events, speed=...)`(`AC_replay_timeline`、`ac_replay_timeline`):遵守每个 `delta_ms` 间隔、按 `speed` 缩放且可夹限;op = move/click/scroll/press/release/key。
+- **输入序列 DSL** — `run_sequence(steps)`(`AC_input_sequence`、`ac_input_sequence`):声明式按住-放开组合键 + `repeat`/`wait`。两者均可注入 sink+sleep 做确定性测试。
+
 ## 本次更新 (2026-06-19) — 语义屏幕状态
 
 像素差异的语义对应物,走完整五层。完整参考:[`docs/source/Zh/doc/new_features/v23_features_doc.rst`](../docs/source/Zh/doc/new_features/v23_features_doc.rst)。
diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md
@@ -12,6 +12,7 @@
 
 ## 目錄
 
+- [本次更新 (2026-06-19) — 計時輸入巨集](#本次更新-2026-06-19--計時輸入巨集)
 - [本次更新 (2026-06-19) — 語意螢幕狀態](#本次更新-2026-06-19--語意螢幕狀態)
 - [本次更新 (2026-06-19) — Set-of-Marks 疊圖](#本次更新-2026-06-19--set-of-marks-疊圖)
 - [本次更新 (2026-06-19) — 檢查點與續跑](#本次更新-2026-06-19--檢查點與續跑)
@@ -75,6 +76,13 @@
 
 ---
 
+## 本次更新 (2026-06-19) — 計時輸入巨集
+
+以時間保真度重播輸入 + 按住-放開 DSL,走完整五層。完整參考:[`docs/source/Zh/doc/new_features/v24_features_doc.rst`](../docs/source/Zh/doc/new_features/v24_features_doc.rst)。
+
+- **計時時間軸重播** — `replay_timeline(events, speed=...)`(`AC_replay_timeline`、`ac_replay_timeline`):遵守每個 `delta_ms` 間隔、依 `speed` 縮放且可夾限;op = move/click/scroll/press/release/key。
+- **輸入序列 DSL** — `run_sequence(steps)`(`AC_input_sequence`、`ac_input_sequence`):宣告式按住-放開組合鍵 + `repeat`/`wait`。兩者皆可注入 sink+sleep 做決定性測試。
+
 ## 本次更新 (2026-06-19) — 語意螢幕狀態
 
 像素差異的語意對應物,走完整五層。完整參考:[`docs/source/Zh/doc/new_features/v23_features_doc.rst`](../docs/source/Zh/doc/new_features/v23_features_doc.rst)。
diff --git a/docs/source/Eng/doc/new_features/v24_features_doc.rst b/docs/source/Eng/doc/new_features/v24_features_doc.rst
@@ -0,0 +1,51 @@
+==================================================
+New Features (2026-06-19) — Timed Input Macros
+==================================================
+
+Replay recorded input with timing fidelity, and author press-hold-release
+combos with a small declarative DSL. Pure standard library; full stack.
+Both dispatch through an injectable sink and sleep, so they unit-test
+deterministically with a fake clock and a recording sink.
+
+.. contents::
+   :local:
+   :depth: 2
+
+
+Timed timeline replay
+====================
+
+::
+
+    from je_auto_control import replay_timeline
+
+    events = [{"op": "move", "x": 10, "y": 10, "delta_ms": 0},
+              {"op": "click", "x": 10, "y": 10, "delta_ms": 120},
+              {"op": "key", "key": "a", "delta_ms": 80}]
+    replay_timeline(events, speed=2.0)   # plays twice as fast; gaps clampable
+
+Each event's ``delta_ms`` gap is honored, divided by ``speed`` (and clamped
+to ``[min_gap, max_gap]``). Event ``op`` is one of ``move`` / ``click`` /
+``scroll`` / ``press`` / ``release`` / ``key``. Exposed as
+``AC_replay_timeline`` / ``ac_replay_timeline``.
+
+
+Input-sequence DSL
+=================
+
+::
+
+    from je_auto_control import run_sequence
+
+    run_sequence([
+        {"op": "press", "key": "ctrl"},
+        {"op": "repeat", "times": 3, "steps": [{"op": "key", "key": "a"}]},
+        {"op": "wait", "ms": 50},
+        {"op": "release", "key": "ctrl"},
+    ])
+
+A declarative mini-language for press-hold-release chords and repeated
+input: action ops (``press`` / ``release`` / ``key`` / ``click`` / ``move``
+/ ``scroll``) plus control ops ``{op: wait, ms}`` and
+``{op: repeat, times, steps:[...]}``. Returns the flattened executed log.
+Exposed as ``AC_input_sequence`` / ``ac_input_sequence``.
diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst
@@ -46,6 +46,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v21_features_doc
    doc/new_features/v22_features_doc
    doc/new_features/v23_features_doc
+   doc/new_features/v24_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc
diff --git a/docs/source/Zh/doc/new_features/v24_features_doc.rst b/docs/source/Zh/doc/new_features/v24_features_doc.rst
@@ -0,0 +1,49 @@
+==========================================
+新功能 (2026-06-19) — 計時輸入巨集
+==========================================
+
+以時間保真度重播錄製的輸入,並用一個小型宣告式 DSL 編寫按住-放開的
+組合鍵。純標準庫;走完整五層。兩者都透過可注入的 sink 與 sleep 派發,
+因此能用假時鐘與記錄式 sink 做決定性單元測試。
+
+.. contents::
+   :local:
+   :depth: 2
+
+
+計時時間軸重播
+==============
+
+::
+
+    from je_auto_control import replay_timeline
+
+    events = [{"op": "move", "x": 10, "y": 10, "delta_ms": 0},
+              {"op": "click", "x": 10, "y": 10, "delta_ms": 120},
+              {"op": "key", "key": "a", "delta_ms": 80}]
+    replay_timeline(events, speed=2.0)   # 兩倍速播放;間隔可夾限
+
+每個事件的 ``delta_ms`` 間隔都會被遵守,並除以 ``speed``(且夾限在
+``[min_gap, max_gap]``)。事件 ``op`` 為 ``move`` / ``click`` / ``scroll`` /
+``press`` / ``release`` / ``key`` 之一。對應 ``AC_replay_timeline`` /
+``ac_replay_timeline``。
+
+
+輸入序列 DSL
+============
+
+::
+
+    from je_auto_control import run_sequence
+
+    run_sequence([
+        {"op": "press", "key": "ctrl"},
+        {"op": "repeat", "times": 3, "steps": [{"op": "key", "key": "a"}]},
+        {"op": "wait", "ms": 50},
+        {"op": "release", "key": "ctrl"},
+    ])
+
+用於按住-放開組合鍵與重複輸入的宣告式迷你語言:動作 op(``press`` /
+``release`` / ``key`` / ``click`` / ``move`` / ``scroll``)加上控制 op
+``{op: wait, ms}`` 與 ``{op: repeat, times, steps:[...]}``。回傳攤平後的
+執行記錄。對應 ``AC_input_sequence`` / ``ac_input_sequence``。
diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst
@@ -46,6 +46,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v21_features_doc
    doc/new_features/v22_features_doc
    doc/new_features/v23_features_doc
+   doc/new_features/v24_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc
diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py
@@ -174,6 +174,8 @@
     describe_screen, diff_snapshots, screen_changed, snapshot,
     snapshot_screen,
 )
+# Timed input replay + declarative input-sequence DSL
+from je_auto_control.utils.input_macro import replay_timeline, run_sequence
 # Background popup/interrupt watchdog (unattended automation)
 from je_auto_control.utils.watchdog import (
     PopupWatchdog, WatchdogRule, default_popup_watchdog,
@@ -601,6 +603,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "resolve_mark",
     "describe_screen", "diff_snapshots", "screen_changed", "snapshot",
     "snapshot_screen",
+    "replay_timeline", "run_sequence",
     # MCP server
     "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt",
     "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool",
diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py
@@ -663,6 +663,21 @@ def _add_misc_specs(specs: List[CommandSpec]) -> None:
     _add_checkpoint_specs(specs)
     _add_set_of_marks_specs(specs)
     _add_screen_state_specs(specs)
+    _add_input_macro_specs(specs)
+
+
+def _add_input_macro_specs(specs: List[CommandSpec]) -> None:
+    specs.append(CommandSpec(
+        "AC_replay_timeline", "Flow", "Replay Timed Events",
+        fields=(FieldSpec("speed", FieldType.FLOAT, optional=True,
+                          default=1.0),),
+        description="Replay 'events' (JSON view) honoring delta_ms, scaled by "
+                    "speed.",
+    ))
+    specs.append(CommandSpec(
+        "AC_input_sequence", "Flow", "Run Input Sequence (DSL)",
+        description="Run 'steps' (JSON view): press/hold/release/repeat/wait.",
+    ))
 
 
 def _add_screen_state_specs(specs: List[CommandSpec]) -> None:
diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py
@@ -2781,6 +2781,19 @@ def _describe_screen(app_name: Optional[str] = None) -> Dict[str, Any]:
     return describe_screen(app_name=app_name)
 
 
+def _replay_timeline(events: List[Dict[str, Any]],
+                     speed: float = 1.0) -> Dict[str, Any]:
+    """Adapter: replay timed input events at a speed multiplier."""
+    from je_auto_control.utils.input_macro import replay_timeline
+    return {"played": replay_timeline(events, speed=float(speed))}
+
+
+def _input_sequence(steps: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Adapter: run a declarative input sequence (press/hold/repeat/...)."""
+    from je_auto_control.utils.input_macro import run_sequence
+    return {"log": run_sequence(steps)}
+
+
 class Executor:
     """
     Executor
@@ -2997,6 +3010,8 @@ def __init__(self):
             "AC_screen_diff": _screen_diff,
             "AC_screen_changed": _screen_changed,
             "AC_describe_screen": _describe_screen,
+            "AC_replay_timeline": _replay_timeline,
+            "AC_input_sequence": _input_sequence,
             "AC_a11y_record_start": _a11y_record_start,
             "AC_a11y_record_stop": _a11y_record_stop,
             "AC_a11y_record_events": _a11y_record_events,
diff --git a/je_auto_control/utils/input_macro/__init__.py b/je_auto_control/utils/input_macro/__init__.py
@@ -0,0 +1,6 @@
+"""Timed input-event replay and a declarative input-sequence DSL."""
+from je_auto_control.utils.input_macro.input_macro import (
+    replay_timeline, run_sequence,
+)
+
+__all__ = ["replay_timeline", "run_sequence"]
diff --git a/je_auto_control/utils/input_macro/input_macro.py b/je_auto_control/utils/input_macro/input_macro.py
@@ -0,0 +1,121 @@
+"""Timed input-event replay and a declarative input-sequence DSL.
+
+The recorder captures *what* happened but replays it without timing; this
+adds fidelity:
+
+* :func:`replay_timeline` plays a list of events honoring each event's
+  ``delta_ms`` gap, scaled by a global ``speed`` (2x faster / 0.5x slower).
+* :func:`run_sequence` runs a small declarative DSL — ``press`` / ``release``
+  / ``key`` / ``click`` / ``move`` / ``scroll`` / ``wait`` / ``repeat`` —
+  for press-hold-release chords and repeated input.
+
+Both dispatch each event through an injectable ``sink`` and use an
+injectable ``sleep``, so timing and sequencing are unit-tested
+deterministically with a fake clock and a recording sink — no real input.
+Imports no ``PySide6``.
+"""
+import time
+from typing import Any, Callable, Dict, List, Optional
+
+
+def _sink_move(event: Dict[str, Any]) -> None:
+    from je_auto_control.wrapper.auto_control_mouse import set_mouse_position
+    set_mouse_position(int(event.get("x", 0)), int(event.get("y", 0)))
+
+
+def _sink_click(event: Dict[str, Any]) -> None:
+    from je_auto_control.wrapper.auto_control_mouse import (
+        click_mouse, set_mouse_position)
+    x, y = int(event.get("x", 0)), int(event.get("y", 0))
+    set_mouse_position(x, y)
+    click_mouse(event.get("button", "mouse_left"), x, y)
+
+
+def _sink_scroll(event: Dict[str, Any]) -> None:
+    from je_auto_control.wrapper.auto_control_mouse import mouse_scroll
+    mouse_scroll(int(event.get("value", 1)))
+
+
+def _sink_press(event: Dict[str, Any]) -> None:
+    from je_auto_control.wrapper.auto_control_keyboard import press_keyboard_key
+    press_keyboard_key(event["key"])
+
+
+def _sink_release(event: Dict[str, Any]) -> None:
+    from je_auto_control.wrapper.auto_control_keyboard import (
+        release_keyboard_key)
+    release_keyboard_key(event["key"])
+
+
+def _sink_key(event: Dict[str, Any]) -> None:
+    from je_auto_control.wrapper.auto_control_keyboard import type_keyboard
+    type_keyboard(event["key"])
+
+
+_SINKS: Dict[str, Callable[[Dict[str, Any]], None]] = {
+    "move": _sink_move, "click": _sink_click, "scroll": _sink_scroll,
+    "press": _sink_press, "release": _sink_release, "key": _sink_key,
+}
+
+
+def _default_sink(event: Dict[str, Any]) -> None:
+    handler = _SINKS.get(event.get("op", ""))
+    if handler is not None:
+        handler(event)
+
+
+def replay_timeline(events: List[Dict[str, Any]], *, speed: float = 1.0,
+                    sink: Optional[Callable] = None,
+                    sleep: Optional[Callable] = None,
+                    min_gap: float = 0.0,
+                    max_gap: Optional[float] = None) -> int:
+    """Replay ``events`` honoring per-event ``delta_ms`` gaps; return count.
+
+    ``speed`` > 1 plays faster (gaps divided by speed). Gaps are clamped to
+    ``[min_gap, max_gap]``. Each event is dispatched via ``sink`` (default:
+    real input); ``sleep`` is injectable for tests.
+    """
+    dispatch = sink or _default_sink
+    sleeper = sleep or time.sleep
+    factor = max(float(speed), 1e-9)
+    played = 0
+    for event in events:
+        gap = float(event.get("delta_ms", 0)) / 1000.0 / factor
+        gap = max(float(min_gap), gap)
+        if max_gap is not None:
+            gap = min(gap, float(max_gap))
+        if gap > 0:
+            sleeper(gap)
+        dispatch(event)
+        played += 1
+    return played
+
+
+def _run_steps(steps: List[Dict[str, Any]], dispatch: Callable,
+               sleeper: Callable, log: List[Dict[str, Any]]) -> None:
+    for step in steps:
+        op = step.get("op")
+        if op == "repeat":
+            for _ in range(int(step.get("times", 1))):
+                _run_steps(step.get("steps", []), dispatch, sleeper, log)
+        elif op == "wait":
+            sleeper(float(step.get("ms", 0)) / 1000.0)
+            log.append({"op": "wait", "ms": step.get("ms", 0)})
+        else:
+            dispatch(step)
+            log.append(dict(step))
+
+
+def run_sequence(steps: List[Dict[str, Any]], *,
+                 sink: Optional[Callable] = None,
+                 sleep: Optional[Callable] = None) -> List[Dict[str, Any]]:
+    """Run a declarative input sequence; return the flattened executed log.
+
+    Steps are ``{op: press|release|key|click|move|scroll}`` plus control ops
+    ``{op: wait, ms}`` and ``{op: repeat, times, steps:[...]}``.
+    """
+    dispatch = sink or _default_sink
+    sleeper = sleep or time.sleep
+    log: List[Dict[str, Any]] = []
+    _run_steps(steps, dispatch, sleeper, log)
+    return log
diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py
diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py
diff --git a/test/unit_test/headless/test_input_macro_batch.py b/test/unit_test/headless/test_input_macro_batch.py
