Add portable agent-trajectory trace (record / replay)

README/WHATS_NEW_zh-CN.md

@@ -1,5 +1,11 @@
 # 本次更新 — AutoControl
 
+## 本次更新 (2026-06-23) — 可携式 Agent 轨迹记录(录制与重播)
+
+记录 agent 的观测→动作步骤并重播。完整参考:[`docs/source/Zh/doc/new_features/v154_features_doc.rst`](../docs/source/Zh/doc/new_features/v154_features_doc.rst)。
+
+- **`record_step` / `to_jsonl` / `from_jsonl` / `replay_trace`**(`AC_replay_trace`):`agent_trace` 记录 OTel span(观测性)、`trajectory_eval` 只评分、`semantic_recording` 重播人类宏——都不是可重播的观测→动作转录。本功能是 OmniTool 风格的 `{step, observation, action, result}` JSONL,加确定性重播驱动器(可注入 `runner`、无需即时模型)。执行器命令透过执行器重播每一步的 AC 动作。纯标准库、可无头测试;可从 agent 执行建立回归 / 训练数据集。
+
 ## 本次更新 (2026-06-23) — 动作前接地防护
 
 拒绝越界点击;把接近偏离者吸附到真正的元素。完整参考:[`docs/source/Zh/doc/new_features/v153_features_doc.rst`](../docs/source/Zh/doc/new_features/v153_features_doc.rst)。

README/WHATS_NEW_zh-TW.md

@@ -1,5 +1,11 @@
 # 本次更新 — AutoControl
 
+## 本次更新 (2026-06-23) — 可攜式 Agent 軌跡記錄(錄製與重播)
+
+記錄 agent 的觀測→動作步驟並重播。完整參考:[`docs/source/Zh/doc/new_features/v154_features_doc.rst`](../docs/source/Zh/doc/new_features/v154_features_doc.rst)。
+
+- **`record_step` / `to_jsonl` / `from_jsonl` / `replay_trace`**(`AC_replay_trace`):`agent_trace` 記錄 OTel span(觀測性)、`trajectory_eval` 只評分、`semantic_recording` 重播人類巨集——都不是可重播的觀測→動作轉錄。本功能是 OmniTool 風格的 `{step, observation, action, result}` JSONL,加決定性重播驅動器(可注入 `runner`、無需即時模型)。執行器命令透過執行器重播每一步的 AC 動作。純標準函式庫、可無頭測試;可從 agent 執行建立回歸 / 訓練資料集。
+
 ## 本次更新 (2026-06-23) — 動作前接地防護
 
 拒絕越界點擊;把接近偏離者吸附到真正的元素。完整參考:[`docs/source/Zh/doc/new_features/v153_features_doc.rst`](../docs/source/Zh/doc/new_features/v153_features_doc.rst)。

WHATS_NEW.md

@@ -1,5 +1,11 @@
 # What's New — AutoControl
 
+## What's new (2026-06-23) — Portable Agent-Trajectory Trace (Record & Replay)
+
+Log an agent's observation→action steps and replay them. Full reference: [`docs/source/Eng/doc/new_features/v154_features_doc.rst`](docs/source/Eng/doc/new_features/v154_features_doc.rst).
+
+- **`record_step` / `to_jsonl` / `from_jsonl` / `replay_trace`** (`AC_replay_trace`): `agent_trace` records OTel spans (observability), `trajectory_eval` only scores, `semantic_recording` replays human macros — none is a replayable obs→action transcript. This is the OmniTool-style `{step, observation, action, result}` JSONL with a deterministic replay driver (injectable `runner`, no live model). The executor command replays each step's AC action through the executor. Pure-stdlib, headless-testable; build regression / training datasets from agent runs.
+
 ## What's new (2026-06-23) — Pre-Action Grounding Guard
 
 Reject out-of-bounds clicks; snap near-misses onto the real element. Full reference: [`docs/source/Eng/doc/new_features/v153_features_doc.rst`](docs/source/Eng/doc/new_features/v153_features_doc.rst).

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

@@ -0,0 +1,45 @@
+Portable Agent-Trajectory Trace (Record & Replay)
+=================================================
+
+``agent_trace`` records OpenTelemetry GenAI *spans* (tokens / latency / cost) — that is
+observability, not a replayable observation→action transcript; ``trajectory_eval``
+*scores* a trajectory but defines no persisted format and cannot replay it; and
+``semantic_recording`` replays recorded *human input macros*, not *agent* decisions.
+This adds the OmniTool-style "log the trajectory to build a replay / training dataset"
+format: ``{step, observation, action, result}`` JSONL with a deterministic replay
+driver.
+
+Pure-stdlib JSONL; the replay driver takes an injectable ``runner`` (no live model), so
+it is fully unit-testable. Imports no ``PySide6``.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import record_step, to_jsonl, from_jsonl, replay_trace
+
+    trace = []
+    record_step(trace, observation="login screen",
+                action=["AC_click_mouse", {"x": 120, "y": 80}])
+    record_step(trace, observation="typed user", action=["AC_write",
+                {"write_string": "alice"}], result={"ok": True})
+
+    open("run.jsonl", "w").write(to_jsonl(trace))     # persist a dataset
+
+    # Later — replay every step through any runner (here a fake for tests).
+    results = replay_trace(from_jsonl(open("run.jsonl").read()),
+                           runner=lambda action: do(action))
+
+``record_step`` appends an indexed ``{step, observation, action[, result]}`` entry;
+``to_jsonl`` / ``from_jsonl`` round-trip the trace as newline-delimited JSON;
+``replay_trace`` runs each step's ``action`` through ``runner(action)`` and returns the
+``{step, action, result}`` outcomes in order.
+
+Executor command
+----------------
+
+``AC_replay_trace`` replays a ``trace`` (JSON array or JSONL) by running each step's
+``action`` (an AC action list) through the executor, returning ``{count, results}``. It
+is exposed as the MCP tool ``ac_replay_trace`` (side-effecting) and as a Script Builder
+command under **Flow**.

docs/source/Eng/eng_index.rst

@@ -176,6 +176,7 @@ Comprehensive guides for all AutoControl features.
    doc/new_features/v151_features_doc
    doc/new_features/v152_features_doc
    doc/new_features/v153_features_doc
+   doc/new_features/v154_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

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

@@ -0,0 +1,38 @@
+可攜式 Agent 軌跡記錄(錄製與重播)
+====================================
+
+``agent_trace`` 記錄 OpenTelemetry GenAI *span*(符記 / 延遲 / 成本)——那是觀測性,不是可重播的觀測→動作轉錄;
+``trajectory_eval`` *評分*軌跡但未定義持久格式也無法重播;``semantic_recording`` 重播錄製的*人類輸入巨集*,而非
+*agent* 決策。本功能加入 OmniTool 風格的「記錄軌跡以建立重播 / 訓練資料集」格式:``{step, observation, action,
+result}`` JSONL,加上決定性的重播驅動器。
+
+純標準函式庫 JSONL;重播驅動器接受可注入的 ``runner``(無需即時模型),因此完全可單元測試。不匯入 ``PySide6``。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import record_step, to_jsonl, from_jsonl, replay_trace
+
+    trace = []
+    record_step(trace, observation="login screen",
+                action=["AC_click_mouse", {"x": 120, "y": 80}])
+    record_step(trace, observation="typed user", action=["AC_write",
+                {"write_string": "alice"}], result={"ok": True})
+
+    open("run.jsonl", "w").write(to_jsonl(trace))     # 持久化資料集
+
+    # 之後——透過任意 runner 重播每一步(此處為測試用 fake)。
+    results = replay_trace(from_jsonl(open("run.jsonl").read()),
+                           runner=lambda action: do(action))
+
+``record_step`` 附加一個有索引的 ``{step, observation, action[, result]}`` 條目;``to_jsonl`` / ``from_jsonl`` 以
+換行分隔 JSON 往返;``replay_trace`` 透過 ``runner(action)`` 執行每一步的 ``action``,並依序回傳
+``{step, action, result}`` 結果。
+
+執行器命令
+----------
+
+``AC_replay_trace`` 透過執行器執行每一步的 ``action``(AC 動作清單)來重播 ``trace``(JSON 陣列或 JSONL),回傳
+``{count, results}``。它以 MCP 工具 ``ac_replay_trace``(有副作用)以及 Script Builder 中 **Flow** 分類下的命令提供。

docs/source/Zh/zh_index.rst

@@ -176,6 +176,7 @@ AutoControl 所有功能的完整使用指南。
    doc/new_features/v151_features_doc
    doc/new_features/v152_features_doc
    doc/new_features/v153_features_doc
+   doc/new_features/v154_features_doc
    doc/ocr_backends/ocr_backends_doc
    doc/observability/observability_doc
    doc/operations_layer/operations_layer_doc

je_auto_control/__init__.py

@@ -373,6 +373,10 @@
 from je_auto_control.utils.action_grounding import (
     in_bounds, snap_to_element, validate_action,
 )
+# Portable agent-trajectory trace (record observation->action steps, replay)
+from je_auto_control.utils.agent_replay import (
+    from_jsonl, record_step, replay_trace, to_jsonl,
+)
 # CI workflow annotations (GitHub Actions)
 from je_auto_control.utils.ci_annotations import (
     emit_annotations, format_annotation,
@@ -1257,6 +1261,10 @@ def start_autocontrol_gui(*args, **kwargs):
     "in_bounds",
     "snap_to_element",
     "validate_action",
+    "record_step",
+    "to_jsonl",
+    "from_jsonl",
+    "replay_trace",
     "emit_annotations", "format_annotation",
     "ClipboardHistory", "default_clipboard_history",
     "analyze_heal_log", "heal_stats", "scan_secrets",

je_auto_control/gui/script_builder/command_schema.py

@@ -899,6 +899,14 @@ def _add_flow_specs(specs: List[CommandSpec]) -> None:
         ),
         description="Aggregate many checks and report all failures (not just first).",
     ))
+    specs.append(CommandSpec(
+        "AC_replay_trace", "Flow", "Replay Agent Trace",
+        fields=(
+            FieldSpec("trace", FieldType.STRING,
+                      placeholder='[{"action":["AC_click_mouse",{...}]}]'),
+        ),
+        description="Replay a recorded trajectory's actions through the executor.",
+    ))
     specs.append(CommandSpec(
         "AC_wait_pixel", "Flow", "Wait for Pixel",
         fields=(

je_auto_control/utils/agent_replay/__init__.py

@@ -0,0 +1,6 @@
+"""Portable agent-trajectory trace (record observation->action steps, replay)."""
+from je_auto_control.utils.agent_replay.agent_replay import (
+    from_jsonl, record_step, replay_trace, to_jsonl,
+)
+
+__all__ = ["from_jsonl", "record_step", "replay_trace", "to_jsonl"]

je_auto_control/utils/agent_replay/agent_replay.py

@@ -0,0 +1,57 @@
+"""Portable agent-trajectory trace — record observation→action steps, replay them.
+
+``agent_trace`` records OpenTelemetry GenAI *spans* (tokens / latency / cost) — that is
+observability, not a replayable observation→action transcript; ``trajectory_eval``
+*scores* a trajectory but defines no persisted on-disk format and cannot replay it; and
+``semantic_recording`` replays recorded *human input macros*, not *agent* decisions.
+This is the OmniTool-style "log the trajectory to build a replay / training dataset"
+format: ``{step, observation, action, result}`` JSONL with a deterministic replay
+driver.
+
+Pure-stdlib JSONL; the replay driver takes an injectable ``runner`` (no live model), so
+it is fully unit-testable. Imports no ``PySide6``.
+"""
+import json
+from typing import Any, Callable, Dict, List, Mapping, Sequence
+
+Step = Dict[str, Any]
+
+
+def record_step(trace: List[Step], observation: Any, action: Any,
+                result: Any = None) -> Step:
+    """Append an ``{step, observation, action[, result]}`` entry to ``trace``.
+
+    Mutates and returns the new step; ``step`` is the running index.
+    """
+    step: Step = {"step": len(trace), "observation": observation,
+                  "action": action}
+    if result is not None:
+        step["result"] = result
+    trace.append(step)
+    return step
+
+
+def to_jsonl(trace: Sequence[Mapping[str, Any]]) -> str:
+    """Serialize a trace to newline-delimited JSON (one step per line)."""
+    return "\n".join(json.dumps(step, ensure_ascii=False, sort_keys=True)
+                     for step in trace)
+
+
+def from_jsonl(text: str) -> List[Step]:
+    """Parse a JSONL trace back into a list of step dicts."""
+    return [json.loads(line) for line in text.splitlines() if line.strip()]
+
+
+def replay_trace(trace: Sequence[Mapping[str, Any]],
+                 runner: Callable[[Any], Any]) -> List[Step]:
+    """Replay each step's ``action`` through ``runner``; return the replay results.
+
+    ``runner(action)`` performs the action and returns its result. The output is a list
+    of ``{step, action, result}`` in order — the basis for agent regression testing.
+    """
+    results: List[Step] = []
+    for index, step in enumerate(trace):
+        action = step.get("action")
+        results.append({"step": step.get("step", index), "action": action,
+                        "result": runner(action)})
+    return results

je_auto_control/utils/executor/action_executor.py

@@ -3878,6 +3878,22 @@ def _validate_action(action: Any, screen: Any = None,
                            targets=list(targets) if targets else None)
 
 
+def _replay_trace(trace: Any) -> Dict[str, Any]:
+    """Adapter: replay a trajectory by running each step's action via the executor."""
+    import json
+    from je_auto_control.utils.agent_replay import from_jsonl, replay_trace
+    if isinstance(trace, str):
+        trace = (json.loads(trace) if trace.strip().startswith("[")
+                 else from_jsonl(trace))
+
+    def runner(action):
+        record = executor.execute_action([list(action)])
+        return next(iter(record.values()), None)
+
+    results = replay_trace(list(trace), runner)
+    return {"count": len(results), "results": results}
+
+
 def _with_modifiers(modifiers: Any, actions: Any) -> Dict[str, Any]:
     """Adapter: run nested actions while modifier keys are held down."""
     import json
@@ -5636,6 +5652,7 @@ def __init__(self):
             "AC_serialize_observation": _serialize_observation,
             "AC_observation_index": _observation_index,
             "AC_validate_action": _validate_action,
+            "AC_replay_trace": _replay_trace,
             "AC_tile_rect": _tile_rect,
             "AC_grid_rects": _grid_rects,
             "AC_cascade_rects": _cascade_rects,

je_auto_control/utils/mcp_server/tools/_factories.py

@@ -3329,6 +3329,22 @@ def action_grounding_tools() -> List[MCPTool]:
     ]
 
 
+def agent_replay_tools() -> List[MCPTool]:
+    return [
+        MCPTool(
+            name="ac_replay_trace",
+            description=("Replay a recorded agent trajectory: run each step's "
+                         "'action' (an AC action list) through the executor, in "
+                         "order. 'trace' is a JSON array or JSONL of {step, "
+                         "observation, action, result} steps. Returns {count, "
+                         "results}. Side-effecting (runs the actions)."),
+            input_schema=schema({"trace": {"type": "array"}}, required=["trace"]),
+            handler=h.replay_trace,
+            annotations=SIDE_EFFECT_ONLY,
+        ),
+    ]
+
+
 def ssim_tools() -> List[MCPTool]:
     return [
         MCPTool(
@@ -6837,7 +6853,8 @@ def media_assert_tools() -> List[MCPTool]:
     locator_chain_tools, rich_clipboard_tools, img_histogram_tools,
     motion_regions_tools, window_zorder_tools, soft_assert_tools,
     perceptual_diff_tools, window_geometry_tools, cua_action_tools,
-    observation_tools, action_grounding_tools, plugin_sdk_tools, governance_tools,
+    observation_tools, action_grounding_tools, agent_replay_tools,
+    plugin_sdk_tools, governance_tools,
     credential_lease_tools, egress_tools, approval_testing_tools,
     trajectory_eval_tools, compliance_tools, agent_trace_tools,
     video_report_tools, fuzzy_tools, artifact_store_tools, image_dedup_tools,

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -2325,6 +2325,11 @@ def validate_action(action, screen=None, targets=None):
     return _validate_action(action, screen, targets)
 
 
+def replay_trace(trace):
+    from je_auto_control.utils.executor.action_executor import _replay_trace
+    return _replay_trace(trace)
+
+
 def detect_drift(reference, current, threshold=0.25, bins=10):
     from je_auto_control.utils.executor.action_executor import _detect_drift
     return _detect_drift(reference, current, threshold, bins)

test/unit_test/headless/test_agent_replay_batch.py

@@ -0,0 +1,58 @@
+"""Headless tests for the agent-trajectory trace. No Qt; runner is injected."""
+import je_auto_control as ac
+from je_auto_control.utils.agent_replay import (
+    from_jsonl, record_step, replay_trace, to_jsonl,
+)
+
+
+def _trace():
+    trace = []
+    record_step(trace, "obs0", ["AC_click_mouse", {"x": 1, "y": 2}])
+    record_step(trace, "obs1", ["AC_write", {"write_string": "hi"}],
+                result={"ok": True})
+    return trace
+
+
+def test_record_step_indexes_and_keeps_result():
+    trace = _trace()
+    assert [s["step"] for s in trace] == [0, 1]
+    assert trace[0]["observation"] == "obs0"
+    assert "result" not in trace[0] and trace[1]["result"] == {"ok": True}
+
+
+def test_jsonl_round_trip():
+    trace = _trace()
+    text = to_jsonl(trace)
+    assert len(text.splitlines()) == 2
+    assert from_jsonl(text) == trace
+
+
+def test_from_jsonl_skips_blank_lines():
+    assert from_jsonl('{"step": 0}\n\n  \n{"step": 1}\n') == [{"step": 0},
+                                                             {"step": 1}]
+
+
+def test_replay_runs_each_action_in_order():
+    calls = []
+    results = replay_trace(_trace(), lambda action: calls.append(action[0])
+                           or f"ran:{action[0]}")
+    assert calls == ["AC_click_mouse", "AC_write"]
+    assert [(r["step"], r["result"]) for r in results] == [
+        (0, "ran:AC_click_mouse"), (1, "ran:AC_write")]
+
+
+# --- wiring ---------------------------------------------------------------
+
+def test_wiring():
+    assert "AC_replay_trace" in set(ac.executor.known_commands())
+    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_replay_trace" in names
+    from je_auto_control.gui.script_builder.command_schema import _build_specs
+    specs = {s.command for s in _build_specs()}
+    assert "AC_replay_trace" in specs
+
+
+def test_facade_exports():
+    for attr in ("record_step", "to_jsonl", "from_jsonl", "replay_trace"):
+        assert hasattr(ac, attr) and attr in ac.__all__