From b0a754473cb415be5078cc946c2f4fff55f57a45 Mon Sep 17 00:00:00 2001 From: Sarina Li Date: Wed, 25 Feb 2026 15:28:20 +0800 Subject: [PATCH] Add trajectory recording to cua do CLI Every cua do action is now automatically recorded to a replayable trajectory at ~/.cua/trajectories/{machine}/{session}/. Viewing opens cua.ai/trajectory-viewer via a local CORS-enabled file server. New files: - trajectory_recorder.py: session management, turn writing, zip, clean - trajectory.py: cua trajectory ls/view/stop/clean commands Modified: - do.py: --no-record flag, post-action screenshot recording in all handlers, session reset on switch - main.py, __init__.py: register trajectory command - SKILL.md: document trajectory recording Co-Authored-By: Claude Opus 4.6 --- .../cua-cli/cua_cli/commands/__init__.py | 4 +- libs/python/cua-cli/cua_cli/commands/do.py | 161 ++++++-- .../cua-cli/cua_cli/commands/trajectory.py | 343 ++++++++++++++++++ libs/python/cua-cli/cua_cli/main.py | 5 +- .../cua_cli/utils/trajectory_recorder.py | 262 +++++++++++++ skills/cua-do-cli/SKILL.md | 19 + 6 files changed, 764 insertions(+), 30 deletions(-) create mode 100644 libs/python/cua-cli/cua_cli/commands/trajectory.py create mode 100644 libs/python/cua-cli/cua_cli/utils/trajectory_recorder.py diff --git a/libs/python/cua-cli/cua_cli/commands/__init__.py b/libs/python/cua-cli/cua_cli/commands/__init__.py index 8ff5abf228..f1080b896f 100644 --- a/libs/python/cua-cli/cua_cli/commands/__init__.py +++ b/libs/python/cua-cli/cua_cli/commands/__init__.py @@ -1,5 +1,5 @@ """CLI commands for CUA.""" -from . import auth, image, mcp, sandbox, skills +from . import auth, image, mcp, sandbox, skills, trajectory -__all__ = ["auth", "sandbox", "image", "skills", "mcp"] +__all__ = ["auth", "sandbox", "image", "skills", "mcp", "trajectory"] diff --git a/libs/python/cua-cli/cua_cli/commands/do.py b/libs/python/cua-cli/cua_cli/commands/do.py index 8619d27273..6a2d80111c 100644 --- a/libs/python/cua-cli/cua_cli/commands/do.py +++ b/libs/python/cua-cli/cua_cli/commands/do.py @@ -72,6 +72,7 @@ def _host_consented() -> bool: # ── output helpers ──────────────────────────────────────────────────────────── + def _ok(msg: str) -> int: print(f"✅ {msg}") return 0 @@ -84,6 +85,7 @@ def _fail(msg: str) -> int: # ── provider / connection helpers ───────────────────────────────────────────── + async def _get_api_url(provider_type: str, name: str) -> str: """Resolve the computer-server API URL for the target VM.""" from computer.providers.base import VMProviderType @@ -144,7 +146,10 @@ async def _host_dispatch(command: str, params: dict) -> dict: import cua_auto.shell as _shell import cua_auto.window as _win except ImportError as e: - return {"success": False, "error": f"cua-auto not installed: {e}. Run: pip install cua-auto"} + return { + "success": False, + "error": f"cua-auto not installed: {e}. Run: pip install cua-auto", + } try: if command == "screenshot": @@ -312,9 +317,7 @@ async def _host_dispatch(command: str, params: dict) -> dict: return {"success": bool(ok)} elif command == "set_window_position": - ok = _win.set_window_position( - params["window_id"], int(params["x"]), int(params["y"]) - ) + ok = _win.set_window_position(params["window_id"], int(params["x"]), int(params["y"])) return {"success": bool(ok)} elif command == "deactivate_window": @@ -365,16 +368,11 @@ async def _send(provider_type: str, name: str, command: str, params: dict) -> di # ── zoom / screenshot helpers ───────────────────────────────────────────────── -async def _resolve_zoom_bbox_by_id( - provider_type: str, name: str, window_id: str -) -> dict | None: + +async def _resolve_zoom_bbox_by_id(provider_type: str, name: str, window_id: str) -> dict | None: """Get the bounding box of a window by its native handle/id.""" - pos_r = await _send( - provider_type, name, "get_window_position", {"window_id": window_id} - ) - size_r = await _send( - provider_type, name, "get_window_size", {"window_id": window_id} - ) + pos_r = await _send(provider_type, name, "get_window_position", {"window_id": window_id}) + size_r = await _send(provider_type, name, "get_window_size", {"window_id": window_id}) pos = pos_r.get("position") or pos_r.get("data") size = size_r.get("size") or size_r.get("data") @@ -394,17 +392,13 @@ async def _resolve_zoom_bbox_by_id( return {"x": x, "y": y, "width": w, "height": h} -async def _resolve_zoom_bbox( - provider_type: str, name: str, window_name: str -) -> dict | None: +async def _resolve_zoom_bbox(provider_type: str, name: str, window_name: str) -> dict | None: """Get the bounding box of a window by app/window name. Filters out internal helper windows (e.g. 'Chrome Legacy Window') so the correct top-level window is always selected. """ - wins_r = await _send( - provider_type, name, "get_application_windows", {"app": window_name} - ) + wins_r = await _send(provider_type, name, "get_application_windows", {"app": window_name}) windows = wins_r.get("windows") or wins_r.get("data") or [] if not windows: return None @@ -513,12 +507,44 @@ async def _print_context(provider_type: str, name: str, state: dict | None = Non zoom_window = state.get("zoom_window") zoom_window_id = state.get("zoom_window_id") if zoom_window: - zoom_info = f"zoom: {zoom_window} ({zoom_window_id})" if zoom_window_id else f"zoom: {zoom_window}" + zoom_info = ( + f"zoom: {zoom_window} ({zoom_window_id})" if zoom_window_id else f"zoom: {zoom_window}" + ) else: zoom_info = "zoom: off" print(f"💻 {vm_label}\t🔍 {zoom_info}") +async def _take_screenshot_for_recording( + provider_type: str, name: str, state: dict +) -> bytes | None: + """Silently take a screenshot for trajectory recording. Returns None on failure.""" + try: + img_bytes, _, _ = await _take_screenshot_data(provider_type, name, state) + return img_bytes + except Exception: + return None + + +def _maybe_record_turn( + args: argparse.Namespace, + state: dict, + action_type: str, + action_params: dict, + screenshot_bytes: bytes | None = None, +) -> None: + """Record a trajectory turn if recording is enabled. Never raises.""" + if getattr(args, "no_record", False): + return + try: + from cua_cli.utils.trajectory_recorder import ensure_session, record_turn + + session_dir = ensure_session(state) + record_turn(session_dir, action_type, action_params, screenshot_bytes) + except Exception: + pass # Never interfere with the primary command + + def _require_target() -> dict | None: state = _load_state() if not state.get("provider"): @@ -529,6 +555,7 @@ def _require_target() -> dict | None: # ── subcommand handlers ─────────────────────────────────────────────────────── + def _cmd_switch(args: argparse.Namespace) -> int: provider = args.provider.lower() old_state = _load_state() @@ -543,7 +570,22 @@ def _cmd_switch(args: argparse.Namespace) -> int: file=sys.stderr, ) return 1 - _save_state({"provider": "host", "name": "", "zoom_window": None, "zoom_window_id": None, "zoom_bbox": None, "zoom_scale": 1.0}) + _save_state( + { + "provider": "host", + "name": "", + "zoom_window": None, + "zoom_window_id": None, + "zoom_bbox": None, + "zoom_scale": 1.0, + } + ) + try: + from cua_cli.utils.trajectory_recorder import reset_session + + reset_session(_load_state()) + except Exception: + pass msg = "Switched to host (local PC)" if had_zoom: msg += " — zoom reset" @@ -553,7 +595,22 @@ def _cmd_switch(args: argparse.Namespace) -> int: return _fail(f"Unknown provider '{provider}'. Choose from: {', '.join(PROVIDERS)}") name = args.name or "" - _save_state({"provider": provider, "name": name, "zoom_window": None, "zoom_window_id": None, "zoom_bbox": None, "zoom_scale": 1.0}) + _save_state( + { + "provider": provider, + "name": name, + "zoom_window": None, + "zoom_window_id": None, + "zoom_bbox": None, + "zoom_scale": 1.0, + } + ) + try: + from cua_cli.utils.trajectory_recorder import reset_session + + reset_session(_load_state()) + except Exception: + pass label = f"{provider}/{name}" if name else provider msg = f"Switched to {label}" if had_zoom: @@ -616,9 +673,9 @@ async def _list_one(ptype: str) -> int: return 0 async def _list_all() -> int: - from cua_cli.auth.store import get_api_key from computer.providers.base import VMProviderType from computer.providers.factory import VMProviderFactory + from cua_cli.auth.store import get_api_key print(" host [local]") @@ -718,6 +775,8 @@ async def _run() -> int: with open(save_path, "wb") as f: f.write(img_bytes) + _maybe_record_turn(args, state, "screenshot", {}, img_bytes) + rc = _ok(f"screenshot saved to {save_path}") await _print_context(state["provider"], state.get("name", ""), state) return rc @@ -807,7 +866,9 @@ async def _run() -> int: print() print("Interactive elements:") for el in elements: - print(f" • {el.get('name','?')} [{el.get('type','?')}] ({el.get('x','?')}, {el.get('y','?')})") + print( + f" • {el.get('name','?')} [{el.get('type','?')}] ({el.get('x','?')}, {el.get('y','?')})" + ) except json.JSONDecodeError: print(f"✅ snapshot — {save_path}") print() @@ -850,6 +911,8 @@ async def _run() -> int: if not result.get("success", True): await _print_context(p, n, state) return _fail(result.get("error", "click failed")) + _scr = await _take_screenshot_for_recording(p, n, state) + _maybe_record_turn(args, state, "click", {"x": args.x, "y": args.y}, _scr) rc = _ok(f"clicked ({args.x}, {args.y}) [{button}]") await _print_context(p, n, state) return rc @@ -877,6 +940,8 @@ async def _run() -> int: if not result.get("success", True): await _print_context(p, n, state) return _fail(result.get("error", "double-click failed")) + _scr = await _take_screenshot_for_recording(p, n, state) + _maybe_record_turn(args, state, "double_click", {"x": args.x, "y": args.y}, _scr) rc = _ok(f"double-clicked ({args.x}, {args.y})") await _print_context(p, n, state) return rc @@ -904,6 +969,7 @@ async def _run() -> int: if not result.get("success", True): await _print_context(p, n, state) return _fail(result.get("error", "move failed")) + _maybe_record_turn(args, state, "move", {"x": args.x, "y": args.y}) rc = _ok(f"cursor moved to ({args.x}, {args.y})") await _print_context(p, n, state) return rc @@ -930,6 +996,8 @@ async def _run() -> int: if not result.get("success", True): await _print_context(p, n, state) return _fail(result.get("error", "type failed")) + _scr = await _take_screenshot_for_recording(p, n, state) + _maybe_record_turn(args, state, "type", {"text": args.text}, _scr) preview = args.text[:40] + ("…" if len(args.text) > 40 else "") rc = _ok(f"typed: {preview!r}") await _print_context(p, n, state) @@ -957,6 +1025,8 @@ async def _run() -> int: if not result.get("success", True): await _print_context(p, n, state) return _fail(result.get("error", "key press failed")) + _scr = await _take_screenshot_for_recording(p, n, state) + _maybe_record_turn(args, state, "keypress", {"keys": [args.key]}, _scr) rc = _ok(f"pressed key: {args.key}") await _print_context(p, n, state) return rc @@ -985,6 +1055,8 @@ async def _run() -> int: if not result.get("success", True): await _print_context(p, n, state) return _fail(result.get("error", "hotkey failed")) + _scr = await _take_screenshot_for_recording(p, n, state) + _maybe_record_turn(args, state, "hotkey", {"keys": keys}, _scr) rc = _ok(f"hotkey: {'+'.join(keys)}") await _print_context(p, n, state) return rc @@ -1013,6 +1085,14 @@ async def _run() -> int: if not result.get("success", True): await _print_context(p, n, state) return _fail(result.get("error", "scroll failed")) + _scr = await _take_screenshot_for_recording(p, n, state) + _maybe_record_turn( + args, + state, + "scroll", + {"scroll_direction": args.direction, "scroll_amount": args.amount}, + _scr, + ) rc = _ok(f"scrolled {args.direction} {args.amount}x") await _print_context(p, n, state) return rc @@ -1035,7 +1115,9 @@ async def _run() -> int: sx2, sy2 = _coords(args.x2, args.y2, state) try: result = await _send( - p, n, "drag_to", + p, + n, + "drag_to", {"start_x": sx1, "start_y": sy1, "end_x": sx2, "end_y": sy2}, ) except Exception as e: @@ -1044,6 +1126,14 @@ async def _run() -> int: if not result.get("success", True): await _print_context(p, n, state) return _fail(result.get("error", "drag failed")) + _scr = await _take_screenshot_for_recording(p, n, state) + _maybe_record_turn( + args, + state, + "drag", + {"start_x": args.x1, "start_y": args.y1, "end_x": args.x2, "end_y": args.y2}, + _scr, + ) rc = _ok(f"dragged ({args.x1},{args.y1}) → ({args.x2},{args.y2})") await _print_context(p, n, state) return rc @@ -1080,6 +1170,7 @@ async def _run() -> int: stderr = result.get("stderr", "").strip() await _print_context(state["provider"], state.get("name", ""), state) return _fail(f"exit {rc_code}: {stderr or stdout}") + _maybe_record_turn(args, state, "shell", {"command": command}) preview = (stdout[:80] + "…") if len(stdout) > 80 else stdout rc = _ok(preview if preview else "done") await _print_context(state["provider"], state.get("name", ""), state) @@ -1107,6 +1198,7 @@ async def _run() -> int: if not result.get("success", True): await _print_context(state["provider"], state.get("name", ""), state) return _fail(result.get("error", "open failed")) + _maybe_record_turn(args, state, "open", {"path": args.path}) rc = _ok(f"opened: {args.path}") await _print_context(state["provider"], state.get("name", ""), state) return rc @@ -1116,6 +1208,7 @@ async def _run() -> int: # ── window subcommands ──────────────────────────────────────────────────────── + def _cmd_window(args: argparse.Namespace) -> int: from cua_cli.utils.async_utils import run_async @@ -1209,7 +1302,9 @@ async def _run() -> int: if action == "resize": try: result = await _send( - p, n, "set_window_size", + p, + n, + "set_window_size", {"window_id": wid, "width": args.width, "height": args.height}, ) except Exception as e: @@ -1225,7 +1320,9 @@ async def _run() -> int: if action == "move": try: result = await _send( - p, n, "set_window_position", + p, + n, + "set_window_position", {"window_id": wid, "x": args.x, "y": args.y}, ) except Exception as e: @@ -1265,6 +1362,7 @@ async def _run() -> int: # ── host consent command (registered separately as cua do-host-consent) ─────── + def register_host_consent_parser(subparsers: argparse._SubParsersAction) -> None: subparsers.add_parser( "do-host-consent", @@ -1285,6 +1383,7 @@ def execute_host_consent(args: argparse.Namespace) -> int: # ── parser registration ─────────────────────────────────────────────────────── + def register_parser(subparsers: argparse._SubParsersAction) -> None: p = subparsers.add_parser( "do", @@ -1332,6 +1431,13 @@ def register_parser(subparsers: argparse._SubParsersAction) -> None: """, ) + p.add_argument( + "--no-record", + action="store_true", + default=False, + help="Disable trajectory recording for this command", + ) + sub = p.add_subparsers(dest="do_action", metavar="action") sub.required = True @@ -1441,6 +1547,7 @@ def register_parser(subparsers: argparse._SubParsersAction) -> None: # ── dispatch ────────────────────────────────────────────────────────────────── + def execute(args: argparse.Namespace) -> int: dispatch = { "switch": _cmd_switch, diff --git a/libs/python/cua-cli/cua_cli/commands/trajectory.py b/libs/python/cua-cli/cua_cli/commands/trajectory.py new file mode 100644 index 0000000000..168f5a293e --- /dev/null +++ b/libs/python/cua-cli/cua_cli/commands/trajectory.py @@ -0,0 +1,343 @@ +"""cua trajectory — manage recorded action trajectories. + +Subcommands: + ls List trajectory sessions + view Zip + serve locally, open cua.ai/trajectory-viewer + clean Delete old sessions + stop Stop the local file server +""" + +from __future__ import annotations + +import argparse +import json +import os +import signal +import subprocess +import sys +import webbrowser +from pathlib import Path +from urllib.parse import quote + +from cua_cli.utils.trajectory_recorder import ( + clean_trajectories, + list_trajectories, + zip_trajectory, +) + +_PID_FILE = Path.home() / ".cua" / "trajectory_server.pid" + + +def register_parser(subparsers: argparse._SubParsersAction) -> None: + """Register ``cua trajectory`` (and alias ``cua traj``) commands.""" + for cmd_name in ("trajectory", "traj"): + p = subparsers.add_parser( + cmd_name, + help="Manage recorded action trajectories", + description="List, view, and clean trajectory recordings from cua do sessions.", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + cua trajectory ls List all sessions + cua trajectory ls my-container List sessions for a machine + cua trajectory view View latest session in browser + cua trajectory view my-container View latest for specific machine + cua trajectory view --port 9090 Use a custom port + cua trajectory stop Stop the file server + cua trajectory clean --older-than 7 Delete sessions older than 7 days + cua trajectory clean --machine my-ct -y Delete all sessions for a machine +""", + ) + + sub = p.add_subparsers(dest="traj_action", metavar="action") + sub.required = True + + # ls + ls_p = sub.add_parser("ls", help="List trajectory sessions") + ls_p.add_argument("machine", nargs="?", default=None, help="Filter by machine name") + ls_p.add_argument("--json", dest="as_json", action="store_true", help="Output as JSON") + + # view + view_p = sub.add_parser("view", help="Zip and open in cua.ai/trajectory-viewer") + view_p.add_argument( + "target", + nargs="?", + default=None, + help="Machine name, session timestamp, or path (default: latest session)", + ) + view_p.add_argument( + "--port", + "-p", + type=int, + default=8089, + help="Port for the local file server (default: 8089)", + ) + + # clean + clean_p = sub.add_parser("clean", help="Delete old trajectory sessions") + clean_p.add_argument( + "--older-than", + type=int, + default=None, + metavar="DAYS", + help="Only delete sessions older than DAYS days", + ) + clean_p.add_argument( + "--machine", default=None, help="Only delete sessions for this machine" + ) + clean_p.add_argument("-y", "--yes", action="store_true", help="Skip confirmation prompt") + + # stop + sub.add_parser("stop", help="Stop the trajectory file server") + + +# ── ls ─────────────────────────────────────────────────────────────────────── + + +def _cmd_ls(args: argparse.Namespace) -> int: + machine = getattr(args, "machine", None) + sessions = list_trajectories(machine=machine) + + if not sessions: + if machine: + print(f"No trajectory sessions found for '{machine}'.") + else: + print("No trajectory sessions found.") + return 0 + + if getattr(args, "as_json", False): + print(json.dumps(sessions, indent=2)) + return 0 + + print(f"{'Machine':<20} {'Session':<18} {'Turns':>5} {'Created'}") + print("-" * 70) + for s in sessions: + print(f"{s['machine']:<20} {s['session']:<18} {s['turns']:>5} {s['created']}") + return 0 + + +# ── view ───────────────────────────────────────────────────────────────────── + + +def _resolve_session(target: str | None) -> str | None: + """Resolve a target to a session path.""" + if target and Path(target).is_dir(): + return target + + sessions = list_trajectories() + if not sessions: + return None + + if target is None: + return sessions[-1]["path"] + + # Try as machine name (latest for that machine) + machine_sessions = [s for s in sessions if s["machine"] == target] + if machine_sessions: + return machine_sessions[-1]["path"] + + # Try as session timestamp + for s in sessions: + if s["session"] == target: + return s["path"] + + return None + + +# Minimal CORS-enabled HTTP server script, run as a subprocess. +_SERVER_SCRIPT = """\ +import sys, os +from http.server import HTTPServer, SimpleHTTPRequestHandler + +class CORSHandler(SimpleHTTPRequestHandler): + def end_headers(self): + self.send_header("Access-Control-Allow-Origin", "*") + self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS") + self.send_header("Access-Control-Allow-Headers", "*") + super().end_headers() + def do_OPTIONS(self): + self.send_response(204) + self.end_headers() + def log_message(self, format, *args): + pass + +os.chdir(sys.argv[1]) +port = int(sys.argv[2]) +HTTPServer(("127.0.0.1", port), CORSHandler).serve_forever() +""" + + +def _cmd_view(args: argparse.Namespace) -> int: + target = getattr(args, "target", None) + port = getattr(args, "port", 8089) + session_path = _resolve_session(target) + + if not session_path: + label = f" for '{target}'" if target else "" + print(f"No trajectory session found{label}.", file=sys.stderr) + return 1 + + session_dir = Path(session_path) + + # Zip the session + zip_path = zip_trajectory(session_dir) + zip_name = zip_path.name + + # Stop any existing server + _stop_server(quiet=True) + + # Start a CORS-enabled file server in the background + serve_dir = str(zip_path.parent) + proc = subprocess.Popen( + [sys.executable, "-c", _SERVER_SCRIPT, serve_dir, str(port)], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + + # Save PID + _PID_FILE.parent.mkdir(parents=True, exist_ok=True) + _PID_FILE.write_text(json.dumps({"pid": proc.pid, "port": port})) + + zip_url = f"http://localhost:{port}/{zip_name}" + viewer_url = f"https://cua.ai/trajectory-viewer?zip={quote(zip_url, safe='')}" + + machine = session_dir.parent.name + session_ts = session_dir.name + print(f"Serving: {machine}/{session_ts}") + print(f"Viewer: {viewer_url}") + print("Stop with: cua trajectory stop") + + try: + webbrowser.open(viewer_url) + except Exception: + pass + + return 0 + + +# ── stop ───────────────────────────────────────────────────────────────────── + + +def _is_our_server(pid: int) -> bool: + """Check if the given PID is a Python process we spawned.""" + try: + os.kill(pid, 0) # Check if process exists (doesn't actually send a signal) + except (ProcessLookupError, PermissionError): + return False + # On macOS/Linux, verify it's a Python process via /proc or ps + try: + import platform + + if platform.system() != "Windows": + result = subprocess.run( + ["ps", "-p", str(pid), "-o", "comm="], + capture_output=True, + text=True, + timeout=2, + ) + comm = result.stdout.strip().lower() + return "python" in comm + except Exception: + pass + return True # On Windows or if ps fails, trust the PID file + + +def _stop_server(quiet: bool = False) -> bool: + """Kill the background file server. Returns True if a server was stopped.""" + if not _PID_FILE.exists(): + if not quiet: + print("No file server is running.") + return False + + try: + info = json.loads(_PID_FILE.read_text()) + pid = info["pid"] + except Exception: + _PID_FILE.unlink(missing_ok=True) + if not quiet: + print("Could not read server PID file.", file=sys.stderr) + return False + + stopped = False + try: + if not _is_our_server(pid): + if not quiet: + print("Server was not running (stale PID file removed).") + _PID_FILE.unlink(missing_ok=True) + return False + os.kill(pid, signal.SIGTERM) + stopped = True + if not quiet: + print(f"Stopped file server (pid {pid}).") + except ProcessLookupError: + if not quiet: + print("Server was not running (stale PID file removed).") + except Exception as e: + if not quiet: + print(f"Failed to stop server: {e}", file=sys.stderr) + finally: + _PID_FILE.unlink(missing_ok=True) + + return stopped + + +def _cmd_stop(_args: argparse.Namespace) -> int: + _stop_server() + return 0 + + +# ── clean ──────────────────────────────────────────────────────────────────── + + +def _cmd_clean(args: argparse.Namespace) -> int: + older_than = getattr(args, "older_than", None) + machine = getattr(args, "machine", None) + yes = getattr(args, "yes", False) + + sessions = list_trajectories(machine=machine) + if not sessions: + print("No trajectory sessions to clean.") + return 0 + + if older_than is not None: + from datetime import datetime, timedelta + + cutoff = datetime.now() - timedelta(days=older_than) + sessions = [s for s in sessions if datetime.fromisoformat(s["created"]) < cutoff] + + if not sessions: + print("No sessions match the criteria.") + return 0 + + if not yes: + print(f"Will delete {len(sessions)} session(s):") + for s in sessions: + print(f" {s['machine']}/{s['session']} ({s['turns']} turns)") + try: + answer = input("Continue? [y/N] ").strip().lower() + except (EOFError, KeyboardInterrupt): + print("\nCancelled.") + return 1 + if answer != "y": + print("Cancelled.") + return 1 + + deleted = clean_trajectories(older_than_days=older_than, machine=machine) + print(f"Deleted {len(deleted)} session(s).") + return 0 + + +# ── dispatch ───────────────────────────────────────────────────────────────── + + +def execute(args: argparse.Namespace) -> int: + dispatch = { + "ls": _cmd_ls, + "view": _cmd_view, + "clean": _cmd_clean, + "stop": _cmd_stop, + } + handler = dispatch.get(args.traj_action) + if not handler: + return 1 + return handler(args) diff --git a/libs/python/cua-cli/cua_cli/main.py b/libs/python/cua-cli/cua_cli/main.py index ebf8528044..837d33b5cb 100644 --- a/libs/python/cua-cli/cua_cli/main.py +++ b/libs/python/cua-cli/cua_cli/main.py @@ -5,7 +5,7 @@ import sys from cua_cli import __version__ -from cua_cli.commands import auth, do, image, mcp, platform, sandbox, skills +from cua_cli.commands import auth, do, image, mcp, platform, sandbox, skills, trajectory from cua_cli.utils.output import print_error @@ -54,6 +54,7 @@ def create_parser() -> argparse.ArgumentParser: mcp.register_parser(subparsers) do.register_parser(subparsers) do.register_host_consent_parser(subparsers) + trajectory.register_parser(subparsers) return parser @@ -91,6 +92,8 @@ def main() -> int: return do.execute(args) elif args.command == "do-host-consent": return do.execute_host_consent(args) + elif args.command in ("trajectory", "traj"): + return trajectory.execute(args) else: print_error(f"Unknown command: {args.command}") return 1 diff --git a/libs/python/cua-cli/cua_cli/utils/trajectory_recorder.py b/libs/python/cua-cli/cua_cli/utils/trajectory_recorder.py new file mode 100644 index 0000000000..a46e2e584f --- /dev/null +++ b/libs/python/cua-cli/cua_cli/utils/trajectory_recorder.py @@ -0,0 +1,262 @@ +"""Trajectory recording utility for cua do actions. + +Records each action into a replayable trajectory compatible with the +TrajectoryViewer at cua.ai/trajectory-viewer. + +All state is file-based (each `cua do` invocation is a separate process). +The current session path is stored in ~/.cua/do_target.json under the +``trajectory_session`` key. + +Directory layout:: + + ~/.cua/trajectories/{machine_name}/{YYYYMMDD-HHMMSS}/ + turn_001/ + screenshot.png + turn_001_agent_response.json + turn_002/ + screenshot.png + turn_002_agent_response.json +""" + +from __future__ import annotations + +import json +import shutil +import time +import uuid +import zipfile +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any + +_CUA_DIR = Path.home() / ".cua" +_TRAJECTORIES_DIR = _CUA_DIR / "trajectories" +_STATE_FILE = _CUA_DIR / "do_target.json" + + +# ── state helpers ──────────────────────────────────────────────────────────── + + +def _load_state() -> dict: + if _STATE_FILE.exists(): + try: + return json.loads(_STATE_FILE.read_text()) + except Exception: + pass + return {} + + +def _save_state(state: dict) -> None: + _STATE_FILE.parent.mkdir(parents=True, exist_ok=True) + _STATE_FILE.write_text(json.dumps(state, indent=2)) + + +# ── session management ─────────────────────────────────────────────────────── + + +def ensure_session(state: dict) -> Path: + """Return the current session directory, creating one if needed. + + If ``trajectory_session`` is already set in *state* and the directory + exists, it is reused. Otherwise a new timestamped directory is created + under ``~/.cua/trajectories/{machine_name}/`` and persisted to state. + """ + existing = state.get("trajectory_session") + if existing: + p = Path(existing) + if p.is_dir(): + return p + + machine = state.get("name") or state.get("provider") or "unknown" + ts = datetime.now().strftime("%Y%m%d-%H%M%S") + session_dir = _TRAJECTORIES_DIR / machine / ts + session_dir.mkdir(parents=True, exist_ok=True) + + state["trajectory_session"] = str(session_dir) + _save_state(state) + return session_dir + + +def reset_session(state: dict) -> None: + """Clear ``trajectory_session`` from state (called on ``cua do switch``).""" + state.pop("trajectory_session", None) + _save_state(state) + + +# ── turn recording ─────────────────────────────────────────────────────────── + + +def get_next_turn_number(session_dir: Path) -> int: + """Scan for existing ``turn_NNN`` dirs and return the next number.""" + max_n = 0 + for child in session_dir.iterdir(): + if child.is_dir() and child.name.startswith("turn_"): + try: + n = int(child.name.split("_", 1)[1]) + if n > max_n: + max_n = n + except (ValueError, IndexError): + pass + return max_n + 1 + + +def _build_action_dict(action_type: str, action_params: dict[str, Any]) -> dict[str, Any]: + """Build the ``action`` dict for the agent-response JSON.""" + action: dict[str, Any] = {"type": action_type} + action.update(action_params) + return action + + +def record_turn( + session_dir: Path, + action_type: str, + action_params: dict[str, Any], + screenshot_bytes: bytes | None = None, +) -> Path: + """Write a single turn to the session directory. + + Creates ``turn_NNN/screenshot.png`` (if *screenshot_bytes* provided) + and ``turn_NNN/turn_NNN_agent_response.json``. + + Returns the turn directory. + """ + turn_num = get_next_turn_number(session_dir) + turn_name = f"turn_{turn_num:03d}" + turn_dir = session_dir / turn_name + turn_dir.mkdir(parents=True, exist_ok=True) + + # Screenshot + if screenshot_bytes: + (turn_dir / "screenshot.png").write_bytes(screenshot_bytes) + + # Agent response JSON (TrajectoryViewer-compatible) + call_id = f"call_{uuid.uuid4().hex[:12]}" + ts_ms = int(time.time()) + + response_json: dict[str, Any] = { + "model": "cua-cli", + "response": { + "id": f"resp_{int(time.time() * 1000)}", + "object": "response", + "created_at": ts_ms, + "status": "completed", + "model": "cua-cli", + "output": [ + { + "type": "computer_call", + "id": call_id, + "call_id": call_id, + "action": _build_action_dict(action_type, action_params), + "pending_safety_checks": [], + "status": "completed", + } + ], + }, + } + + json_path = turn_dir / f"{turn_name}_agent_response.json" + json_path.write_text(json.dumps(response_json, indent=2)) + + return turn_dir + + +# ── listing / inspection ───────────────────────────────────────────────────── + + +def list_trajectories(machine: str | None = None) -> list[dict[str, Any]]: + """Return a list of trajectory sessions. + + Each entry is ``{machine, session, path, turns, created}``. + If *machine* is given, only that machine's sessions are returned. + """ + results: list[dict[str, Any]] = [] + if not _TRAJECTORIES_DIR.is_dir(): + return results + + machines = [_TRAJECTORIES_DIR / machine] if machine else sorted(_TRAJECTORIES_DIR.iterdir()) + + for machine_dir in machines: + if not machine_dir.is_dir(): + continue + for session_dir in sorted(machine_dir.iterdir()): + if not session_dir.is_dir(): + continue + turns = sum( + 1 for c in session_dir.iterdir() if c.is_dir() and c.name.startswith("turn_") + ) + # Parse created timestamp from dir name + try: + created = datetime.strptime(session_dir.name, "%Y%m%d-%H%M%S") + except ValueError: + created = datetime.fromtimestamp(session_dir.stat().st_ctime) + results.append( + { + "machine": machine_dir.name, + "session": session_dir.name, + "path": str(session_dir), + "turns": turns, + "created": created.isoformat(), + } + ) + return results + + +# ── zipping ────────────────────────────────────────────────────────────────── + + +def zip_trajectory(session_path: str | Path) -> Path: + """Create a zip of a session directory in TrajectoryViewer-compatible format. + + Returns the path to the created zip file (placed next to the session dir). + """ + session_dir = Path(session_path) + zip_path = session_dir.parent / f"{session_dir.name}.zip" + with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: + for child in sorted(session_dir.rglob("*")): + if child.is_file(): + zf.write(child, child.relative_to(session_dir)) + return zip_path + + +# ── cleanup ────────────────────────────────────────────────────────────────── + + +def clean_trajectories( + older_than_days: int | None = None, + machine: str | None = None, +) -> list[str]: + """Delete trajectory sessions matching the criteria. + + Returns a list of deleted session paths. + """ + deleted: list[str] = [] + if not _TRAJECTORIES_DIR.is_dir(): + return deleted + + cutoff = datetime.now() - timedelta(days=older_than_days) if older_than_days else None + machines = [_TRAJECTORIES_DIR / machine] if machine else sorted(_TRAJECTORIES_DIR.iterdir()) + + for machine_dir in machines: + if not machine_dir.is_dir(): + continue + for session_dir in sorted(machine_dir.iterdir()): + if not session_dir.is_dir(): + continue + if cutoff: + try: + created = datetime.strptime(session_dir.name, "%Y%m%d-%H%M%S") + except ValueError: + created = datetime.fromtimestamp(session_dir.stat().st_ctime) + if created >= cutoff: + continue + shutil.rmtree(session_dir) + deleted.append(str(session_dir)) + # Also remove zip if it exists + zip_path = session_dir.parent / f"{session_dir.name}.zip" + if zip_path.exists(): + zip_path.unlink() + # Remove empty machine dirs + if machine_dir.is_dir() and not any(machine_dir.iterdir()): + machine_dir.rmdir() + + return deleted diff --git a/skills/cua-do-cli/SKILL.md b/skills/cua-do-cli/SKILL.md index a4232b0f8e..7d178edfa2 100644 --- a/skills/cua-do-cli/SKILL.md +++ b/skills/cua-do-cli/SKILL.md @@ -125,3 +125,22 @@ cua do window info - Coords are image-space: zoom + max-length scaling are applied automatically. - `do-host-consent` is permanent until the consent file is deleted (`~/.cua/host_consented`). - Set `PYTHONIOENCODING=utf-8` on Windows for correct emoji output. + +## Trajectory Recording + +Every `cua do` action is automatically recorded to a replayable trajectory. + +- Stored at `~/.cua/trajectories/{machine_name}/{session_timestamp}/` +- New session starts on `cua do switch` +- Disable with `--no-record`: `cua do --no-record click 100 200` +- After a session, mention the trajectory to the user: + "Your session was recorded. Run `cua trajectory view` to replay." + +```bash +cua trajectory ls # list all sessions +cua trajectory view # start local viewer for latest session +cua trajectory view my-container # latest for specific machine +cua trajectory view --port 9090 # use custom port +cua trajectory stop # stop the viewer server +cua trajectory clean --older-than 7 # delete old sessions +```