| 1 | from __future__ import annotations |
| 2 | |
| 3 | import argparse |
| 4 | import json |
| 5 | import os |
| 6 | import re |
| 7 | import shutil |
| 8 | import subprocess |
| 9 | import sys |
| 10 | import time |
| 11 | from pathlib import Path |
| 12 | from typing import Any |
| 13 | |
| 14 | PROJECT_ROOT = Path(__file__).resolve().parents[3] |
| 15 | if str(PROJECT_ROOT) not in sys.path: |
| 16 | sys.path.insert(0, str(PROJECT_ROOT)) |
| 17 | |
| 18 | |
| 19 | SESSION_ID = "agent-zero-desktop" |
| 20 | PLUGIN_NAME = "_desktop" |
| 21 | BASE_DIR = Path(os.environ.get("A0_BASE_DIR") or ("/a0" if Path("/a0").exists() else PROJECT_ROOT)) |
| 22 | STATE_DIR = BASE_DIR / "usr" / "plugins" / PLUGIN_NAME |
| 23 | RETIRED_STATE_DIR = BASE_DIR / "usr" / PLUGIN_NAME |
| 24 | SESSION_DIR = STATE_DIR / "sessions" |
| 25 | PROFILE_DIR = STATE_DIR / "profiles" |
| 26 | SCREENSHOT_DIR = Path(os.environ.get("A0_DESKTOP_SCREENSHOT_DIR") or BASE_DIR / "tmp" / "desktop" / "screenshots") |
| 27 | RECENT_SCREENSHOT_SECONDS = 600 |
| 28 | _SAFE_CONTEXT_RE = re.compile(r"[^a-zA-Z0-9_.-]+") |
| 29 | _SCREENSHOT_SUFFIXES = {".png", ".jpg", ".jpeg", ".xwd"} |
| 30 | |
| 31 | |
| 32 | def session_manifest_path(session_id: str = SESSION_ID) -> Path: |
| 33 | return Path(os.environ.get("A0_DESKTOP_MANIFEST") or SESSION_DIR / f"{session_id}.json") |
| 34 | |
| 35 | |
| 36 | def context_screenshot_dir(context_id: str = "") -> Path: |
| 37 | return SCREENSHOT_DIR / _safe_context_id(context_id) |
| 38 | |
| 39 | |
| 40 | def chat_screenshot_dir(context_id: str = "") -> Path: |
| 41 | return BASE_DIR / "usr" / "chats" / _safe_context_id(context_id) / "screenshots" / "desktop" |
| 42 | |
| 43 | |
| 44 | def normalize_a0_path(path: str | Path) -> str: |
| 45 | candidate = Path(path) |
| 46 | try: |
| 47 | relative = candidate.resolve(strict=False).relative_to(BASE_DIR.resolve(strict=False)) |
| 48 | except ValueError: |
| 49 | return str(candidate) |
| 50 | return "/a0/" + str(relative).replace(os.sep, "/") |
| 51 | |
| 52 | |
| 53 | def _safe_context_id(context_id: str = "") -> str: |
| 54 | raw = str(context_id or os.environ.get("A0_DESKTOP_CONTEXT_ID") or "default") |
| 55 | return _SAFE_CONTEXT_RE.sub("_", raw).strip("._") or "default" |
| 56 | |
| 57 | |
| 58 | def session_manifest_exists(session_id: str = SESSION_ID) -> bool: |
| 59 | return session_manifest_path(session_id).exists() |
| 60 | |
| 61 | |
| 62 | def collect_state( |
| 63 | *, |
| 64 | include_screenshot: bool = False, |
| 65 | screenshot_path: str | Path | None = None, |
| 66 | context_id: str = "", |
| 67 | screenshot_transport: str = "ephemeral", |
| 68 | ) -> dict[str, Any]: |
| 69 | errors: list[str] = [] |
| 70 | env_info = resolve_environment(errors=errors) |
| 71 | display = env_info["display"] |
| 72 | profile_dir = env_info["profile_dir"] |
| 73 | env = display_env(display=display, profile_dir=profile_dir) |
| 74 | |
| 75 | capabilities = collect_capabilities() |
| 76 | for name in ("xdotool", "xrandr", "xwininfo", "xprop"): |
| 77 | if not capabilities.get(name): |
| 78 | errors.append(f"{name} is not installed; install Desktop runtime dependencies through the _desktop plugin hook.") |
| 79 | |
| 80 | size = collect_display_size(env, capabilities, errors) |
| 81 | pointer = collect_pointer(env, capabilities, errors) |
| 82 | active_window = collect_active_window(env, capabilities, errors) |
| 83 | windows = collect_windows(env, capabilities, errors) |
| 84 | screenshot = latest_screenshot(context_id=context_id) |
| 85 | |
| 86 | if include_screenshot: |
| 87 | screenshot = capture_screenshot( |
| 88 | env, |
| 89 | capabilities, |
| 90 | path=screenshot_path, |
| 91 | errors=errors, |
| 92 | context_id=context_id, |
| 93 | transport=screenshot_transport, |
| 94 | ) |
| 95 | |
| 96 | return stable_state( |
| 97 | context_id=context_id, |
| 98 | display=display, |
| 99 | profile_dir=profile_dir, |
| 100 | size=size, |
| 101 | pointer=pointer, |
| 102 | active_window=active_window, |
| 103 | windows=windows, |
| 104 | screenshot=screenshot, |
| 105 | capabilities=capabilities, |
| 106 | errors=errors, |
| 107 | ) |
| 108 | |
| 109 | |
| 110 | def capture_screenshot( |
| 111 | env: dict[str, str] | None = None, |
| 112 | capabilities: dict[str, str] | None = None, |
| 113 | *, |
| 114 | path: str | Path | None = None, |
| 115 | errors: list[str] | None = None, |
| 116 | context_id: str = "", |
| 117 | transport: str = "ephemeral", |
| 118 | ) -> dict[str, Any]: |
| 119 | local_errors = errors if errors is not None else [] |
| 120 | capabilities = capabilities or collect_capabilities() |
| 121 | if not env: |
| 122 | env_errors: list[str] = [] |
| 123 | env_info = resolve_environment(errors=env_errors) |
| 124 | local_errors.extend(env_errors) |
| 125 | env = display_env(display=env_info["display"], profile_dir=env_info["profile_dir"]) |
| 126 | |
| 127 | xwd = capabilities.get("xwd") or shutil.which("xwd") or "" |
| 128 | if not xwd: |
| 129 | message = "xwd is not installed; install x11-apps through the _desktop plugin hook." |
| 130 | local_errors.append(message) |
| 131 | return {"ok": False, "path": "", "format": "", "captured_at": "", "error": message} |
| 132 | |
| 133 | explicit_path = path is not None and str(path).strip() != "" |
| 134 | transport_mode = str(transport or "").strip().lower() |
| 135 | chat_scoped = bool(not explicit_path and transport_mode == "path" and str(context_id or "").strip()) |
| 136 | ephemeral_ref = not explicit_path and transport_mode != "path" |
| 137 | screenshot_dir = chat_screenshot_dir(context_id) if chat_scoped else context_screenshot_dir(context_id) |
| 138 | if not explicit_path and not chat_scoped: |
| 139 | prune_context_screenshots(context_id=context_id) |
| 140 | screenshot_dir.mkdir(parents=True, exist_ok=True) |
| 141 | timestamp = time.strftime("%Y%m%d-%H%M%S") |
| 142 | millis = int((time.time() % 1) * 1000) |
| 143 | target = Path(path) if explicit_path else screenshot_dir / f"desktop-{timestamp}-{millis:03d}.png" |
| 144 | target.parent.mkdir(parents=True, exist_ok=True) |
| 145 | raw_path = target.with_suffix(".xwd") |
| 146 | safe_context = _safe_context_id(context_id) |
| 147 | |
| 148 | result = run([xwd, "-root", "-silent", "-out", str(raw_path)], env=env, timeout=8) |
| 149 | if result.returncode != 0: |
| 150 | detail = command_output(result) or "xwd screenshot capture failed." |
| 151 | local_errors.append(detail) |
| 152 | raw_path.unlink(missing_ok=True) |
| 153 | return {"ok": False, "path": "", "format": "", "captured_at": "", "error": detail} |
| 154 | |
| 155 | if target.suffix.lower() == ".xwd": |
| 156 | if not explicit_path and not chat_scoped: |
| 157 | prune_context_screenshots(context_id=context_id, keep_path=raw_path) |
| 158 | return { |
| 159 | "ok": True, |
| 160 | "path": str(raw_path), |
| 161 | "a0_path": normalize_a0_path(raw_path), |
| 162 | "format": "xwd", |
| 163 | "captured_at": iso_now(), |
| 164 | "recent": True, |
| 165 | "ephemeral": not explicit_path and not chat_scoped, |
| 166 | "chat_scoped": chat_scoped, |
| 167 | "context_id": safe_context, |
| 168 | "error": "", |
| 169 | } |
| 170 | |
| 171 | try: |
| 172 | from PIL import Image |
| 173 | |
| 174 | with Image.open(raw_path) as image: |
| 175 | image.save(target) |
| 176 | width = int(image.width) |
| 177 | height = int(image.height) |
| 178 | raw_path.unlink(missing_ok=True) |
| 179 | if ephemeral_ref: |
| 180 | return ephemeral_screenshot_result( |
| 181 | target, |
| 182 | context_id=context_id, |
| 183 | image_format=target.suffix.lower().lstrip(".") or "png", |
| 184 | width=width, |
| 185 | height=height, |
| 186 | ) |
| 187 | if not explicit_path and not chat_scoped: |
| 188 | prune_context_screenshots(context_id=context_id, keep_path=target) |
| 189 | return { |
| 190 | "ok": True, |
| 191 | "path": str(target), |
| 192 | "a0_path": normalize_a0_path(target), |
| 193 | "format": target.suffix.lower().lstrip(".") or "png", |
| 194 | "width": width, |
| 195 | "height": height, |
| 196 | "captured_at": iso_now(), |
| 197 | "recent": True, |
| 198 | "ephemeral": not explicit_path and not chat_scoped, |
| 199 | "chat_scoped": chat_scoped, |
| 200 | "context_id": safe_context, |
| 201 | "error": "", |
| 202 | } |
| 203 | except Exception as exc: |
| 204 | try: |
| 205 | converted = convert_xwd_to_image(raw_path, target) |
| 206 | raw_path.unlink(missing_ok=True) |
| 207 | if ephemeral_ref: |
| 208 | return ephemeral_screenshot_result( |
| 209 | target, |
| 210 | context_id=context_id, |
| 211 | image_format=target.suffix.lower().lstrip(".") or "png", |
| 212 | width=converted["width"], |
| 213 | height=converted["height"], |
| 214 | ) |
| 215 | if not explicit_path and not chat_scoped: |
| 216 | prune_context_screenshots(context_id=context_id, keep_path=target) |
| 217 | return { |
| 218 | "ok": True, |
| 219 | "path": str(target), |
| 220 | "a0_path": normalize_a0_path(target), |
| 221 | "format": target.suffix.lower().lstrip(".") or "png", |
| 222 | "width": converted["width"], |
| 223 | "height": converted["height"], |
| 224 | "captured_at": iso_now(), |
| 225 | "recent": True, |
| 226 | "ephemeral": not explicit_path and not chat_scoped, |
| 227 | "chat_scoped": chat_scoped, |
| 228 | "context_id": safe_context, |
| 229 | "error": "", |
| 230 | } |
| 231 | except Exception as fallback_exc: |
| 232 | message = f"Pillow could not convert the XWD screenshot: {exc}; fallback parser failed: {fallback_exc}" |
| 233 | local_errors.append(message) |
| 234 | if ephemeral_ref: |
| 235 | raw_path.unlink(missing_ok=True) |
| 236 | target.unlink(missing_ok=True) |
| 237 | return { |
| 238 | "ok": False, |
| 239 | "path": "", |
| 240 | "format": "", |
| 241 | "captured_at": iso_now(), |
| 242 | "recent": False, |
| 243 | "ephemeral": True, |
| 244 | "context_id": safe_context, |
| 245 | "error": message, |
| 246 | } |
| 247 | return { |
| 248 | "ok": True, |
| 249 | "path": str(raw_path), |
| 250 | "a0_path": normalize_a0_path(raw_path), |
| 251 | "format": "xwd", |
| 252 | "captured_at": iso_now(), |
| 253 | "recent": True, |
| 254 | "ephemeral": not explicit_path and not chat_scoped, |
| 255 | "chat_scoped": chat_scoped, |
| 256 | "context_id": safe_context, |
| 257 | "error": message, |
| 258 | } |
| 259 | |
| 260 | |
| 261 | def convert_xwd_to_image(raw_path: Path, target: Path) -> dict[str, int]: |
| 262 | from PIL import Image |
| 263 | |
| 264 | data = raw_path.read_bytes() |
| 265 | header, _ = parse_xwd_header(data) |
| 266 | width = header["pixmap_width"] |
| 267 | height = header["pixmap_height"] |
| 268 | bytes_per_line = header["bytes_per_line"] |
| 269 | bits_per_pixel = header["bits_per_pixel"] |
| 270 | color_table_size = header["ncolors"] * 12 |
| 271 | pixel_offset = header["header_size"] + color_table_size |
| 272 | if width <= 0 or height <= 0 or bytes_per_line <= 0: |
| 273 | raise ValueError("invalid XWD dimensions") |
| 274 | pixel_size = height * bytes_per_line |
| 275 | if pixel_offset + pixel_size > len(data): |
| 276 | raise ValueError("truncated XWD pixel data") |
| 277 | |
| 278 | if (header["red_mask"], header["green_mask"], header["blue_mask"]) != ( |
| 279 | 0x00FF0000, |
| 280 | 0x0000FF00, |
| 281 | 0x000000FF, |
| 282 | ): |
| 283 | raise ValueError("unsupported XWD visual masks") |
| 284 | raw_mode = { |
| 285 | (24, 0): "BGR", |
| 286 | (24, 1): "RGB", |
| 287 | (32, 0): "BGRX", |
| 288 | (32, 1): "XRGB", |
| 289 | }.get((bits_per_pixel, header["byte_order"])) |
| 290 | if not raw_mode: |
| 291 | raise ValueError(f"unsupported XWD pixel layout: {bits_per_pixel} bpp") |
| 292 | |
| 293 | image = Image.frombytes( |
| 294 | "RGB", |
| 295 | (width, height), |
| 296 | data[pixel_offset : pixel_offset + pixel_size], |
| 297 | "raw", |
| 298 | raw_mode, |
| 299 | bytes_per_line, |
| 300 | 1, |
| 301 | ) |
| 302 | image.save(target) |
| 303 | return {"width": width, "height": height} |
| 304 | |
| 305 | |
| 306 | def parse_xwd_header(data: bytes) -> tuple[dict[str, int], str]: |
| 307 | if len(data) < 100: |
| 308 | raise ValueError("XWD header is too short") |
| 309 | field_names = ( |
| 310 | "header_size", |
| 311 | "file_version", |
| 312 | "pixmap_format", |
| 313 | "pixmap_depth", |
| 314 | "pixmap_width", |
| 315 | "pixmap_height", |
| 316 | "xoffset", |
| 317 | "byte_order", |
| 318 | "bitmap_unit", |
| 319 | "bitmap_bit_order", |
| 320 | "bitmap_pad", |
| 321 | "bits_per_pixel", |
| 322 | "bytes_per_line", |
| 323 | "visual_class", |
| 324 | "red_mask", |
| 325 | "green_mask", |
| 326 | "blue_mask", |
| 327 | "bits_per_rgb", |
| 328 | "colormap_entries", |
| 329 | "ncolors", |
| 330 | "window_width", |
| 331 | "window_height", |
| 332 | "window_x", |
| 333 | "window_y", |
| 334 | "window_bdrwidth", |
| 335 | ) |
| 336 | for endian in ("big", "little"): |
| 337 | values = [int.from_bytes(data[index : index + 4], endian, signed=False) for index in range(0, 100, 4)] |
| 338 | header = dict(zip(field_names, values, strict=True)) |
| 339 | if 100 <= header["header_size"] <= len(data) and header["file_version"] == 7: |
| 340 | return header, endian |
| 341 | raise ValueError("unsupported XWD header") |
| 342 | |
| 343 | |
| 344 | def resolve_environment(*, errors: list[str] | None = None, session_id: str = SESSION_ID) -> dict[str, str]: |
| 345 | local_errors = errors if errors is not None else [] |
| 346 | manifest = session_manifest_path(session_id) |
| 347 | payload: dict[str, Any] = {} |
| 348 | if manifest.exists(): |
| 349 | try: |
| 350 | payload = json.loads(manifest.read_text(encoding="utf-8")) |
| 351 | except Exception as exc: |
| 352 | local_errors.append(f"Desktop session manifest is unreadable: {exc}") |
| 353 | elif not (os.environ.get("A0_DESKTOP_DISPLAY") or os.environ.get("DISPLAY")): |
| 354 | local_errors.append(f"Desktop session manifest not found at {manifest}; open the Desktop canvas before GUI control.") |
| 355 | |
| 356 | display_value = str( |
| 357 | os.environ.get("A0_DESKTOP_DISPLAY") |
| 358 | or payload.get("display") |
| 359 | or os.environ.get("DISPLAY") |
| 360 | or "" |
| 361 | ).strip() |
| 362 | if display_value.startswith(":"): |
| 363 | display = display_value |
| 364 | elif display_value: |
| 365 | display = f":{display_value}" |
| 366 | else: |
| 367 | display = "" |
| 368 | local_errors.append("Desktop DISPLAY is unavailable; the persistent Desktop session is not running.") |
| 369 | |
| 370 | profile_dir = _state_path_from_retired_root( |
| 371 | Path( |
| 372 | os.environ.get("A0_DESKTOP_PROFILE") |
| 373 | or os.environ.get("A0_DESKTOP_HOME") |
| 374 | or payload.get("profile_dir") |
| 375 | or os.environ.get("HOME") |
| 376 | or PROFILE_DIR / session_id |
| 377 | ) |
| 378 | ) |
| 379 | |
| 380 | return { |
| 381 | "display": display, |
| 382 | "profile_dir": str(profile_dir), |
| 383 | "manifest": str(manifest), |
| 384 | } |
| 385 | |
| 386 | |
| 387 | def _state_path_from_retired_root(path: Path) -> Path: |
| 388 | try: |
| 389 | relative = path.resolve(strict=False).relative_to( |
| 390 | RETIRED_STATE_DIR.resolve(strict=False) |
| 391 | ) |
| 392 | except ValueError: |
| 393 | return path |
| 394 | return STATE_DIR / relative |
| 395 | |
| 396 | |
| 397 | def display_env(*, display: str, profile_dir: str) -> dict[str, str]: |
| 398 | env = { |
| 399 | **os.environ, |
| 400 | "HOME": profile_dir, |
| 401 | "XDG_CONFIG_HOME": os.environ.get("XDG_CONFIG_HOME") or str(Path(profile_dir) / ".config"), |
| 402 | "XDG_DATA_HOME": os.environ.get("XDG_DATA_HOME") or str(Path(profile_dir) / ".local" / "share"), |
| 403 | "XDG_CACHE_HOME": os.environ.get("XDG_CACHE_HOME") or str(Path(profile_dir) / ".cache"), |
| 404 | "XDG_CURRENT_DESKTOP": os.environ.get("XDG_CURRENT_DESKTOP") or "XFCE", |
| 405 | } |
| 406 | if display: |
| 407 | env["DISPLAY"] = display |
| 408 | xauthority = os.environ.get("A0_DESKTOP_XAUTHORITY") or str(Path(profile_dir) / ".Xauthority") |
| 409 | if Path(xauthority).exists(): |
| 410 | env["XAUTHORITY"] = xauthority |
| 411 | return env |
| 412 | |
| 413 | |
| 414 | def collect_capabilities() -> dict[str, str]: |
| 415 | return { |
| 416 | name: shutil.which(name) or "" |
| 417 | for name in ( |
| 418 | "xdotool", |
| 419 | "xrandr", |
| 420 | "xwininfo", |
| 421 | "xprop", |
| 422 | "xwd", |
| 423 | "xclip", |
| 424 | ) |
| 425 | } |
| 426 | |
| 427 | |
| 428 | def collect_display_size(env: dict[str, str], capabilities: dict[str, str], errors: list[str]) -> dict[str, int]: |
| 429 | if not capabilities.get("xrandr"): |
| 430 | return {"width": 0, "height": 0} |
| 431 | result = run([capabilities["xrandr"], "-q"], env=env, timeout=4) |
| 432 | if result.returncode != 0: |
| 433 | errors.append(command_output(result) or "xrandr could not read the Desktop display.") |
| 434 | return {"width": 0, "height": 0} |
| 435 | match = re.search(r"\bcurrent\s+(\d+)\s+x\s+(\d+)", result.stdout) |
| 436 | if not match: |
| 437 | errors.append("xrandr output did not include the current Desktop size.") |
| 438 | return {"width": 0, "height": 0} |
| 439 | return {"width": int(match.group(1)), "height": int(match.group(2))} |
| 440 | |
| 441 | |
| 442 | def collect_pointer(env: dict[str, str], capabilities: dict[str, str], errors: list[str]) -> dict[str, int]: |
| 443 | if not capabilities.get("xdotool"): |
| 444 | return {"x": 0, "y": 0, "screen": 0, "window": 0} |
| 445 | result = run([capabilities["xdotool"], "getmouselocation", "--shell"], env=env, timeout=3) |
| 446 | if result.returncode != 0: |
| 447 | errors.append(command_output(result) or "xdotool could not read the pointer location.") |
| 448 | return {"x": 0, "y": 0, "screen": 0, "window": 0} |
| 449 | values = parse_shell_values(result.stdout) |
| 450 | return { |
| 451 | "x": int_value(values.get("X")), |
| 452 | "y": int_value(values.get("Y")), |
| 453 | "screen": int_value(values.get("SCREEN")), |
| 454 | "window": int_value(values.get("WINDOW")), |
| 455 | } |
| 456 | |
| 457 | |
| 458 | def collect_active_window(env: dict[str, str], capabilities: dict[str, str], errors: list[str]) -> dict[str, Any] | None: |
| 459 | if not capabilities.get("xdotool"): |
| 460 | return None |
| 461 | result = run([capabilities["xdotool"], "getactivewindow"], env=env, timeout=3) |
| 462 | if result.returncode != 0: |
| 463 | return None |
| 464 | window_id = result.stdout.strip().splitlines()[0] if result.stdout.strip() else "" |
| 465 | if not window_id: |
| 466 | return None |
| 467 | return collect_window(env, capabilities, window_id, errors) |
| 468 | |
| 469 | |
| 470 | def collect_windows(env: dict[str, str], capabilities: dict[str, str], errors: list[str]) -> list[dict[str, Any]]: |
| 471 | if not capabilities.get("xdotool"): |
| 472 | return [] |
| 473 | result = run([capabilities["xdotool"], "search", "--onlyvisible", "--name", "."], env=env, timeout=4) |
| 474 | if result.returncode != 0: |
| 475 | detail = command_output(result) |
| 476 | if detail: |
| 477 | errors.append(detail) |
| 478 | return [] |
| 479 | windows: list[dict[str, Any]] = [] |
| 480 | seen: set[str] = set() |
| 481 | for window_id in result.stdout.splitlines(): |
| 482 | window_id = window_id.strip() |
| 483 | if not window_id or window_id in seen: |
| 484 | continue |
| 485 | seen.add(window_id) |
| 486 | windows.append(collect_window(env, capabilities, window_id, errors)) |
| 487 | return windows |
| 488 | |
| 489 | |
| 490 | def collect_window( |
| 491 | env: dict[str, str], |
| 492 | capabilities: dict[str, str], |
| 493 | window_id: str, |
| 494 | errors: list[str], |
| 495 | ) -> dict[str, Any]: |
| 496 | props = collect_window_props(env, capabilities, window_id) |
| 497 | geometry = collect_window_geometry(env, capabilities, window_id) |
| 498 | return { |
| 499 | "id": str(window_id), |
| 500 | "title": props.get("title", ""), |
| 501 | "class": props.get("class", ""), |
| 502 | "name": props.get("name", ""), |
| 503 | "pid": int_value(props.get("pid")), |
| 504 | "geometry": geometry, |
| 505 | } |
| 506 | |
| 507 | |
| 508 | def collect_window_geometry(env: dict[str, str], capabilities: dict[str, str], window_id: str) -> dict[str, int]: |
| 509 | geometry = {"x": 0, "y": 0, "width": 0, "height": 0} |
| 510 | if not capabilities.get("xwininfo"): |
| 511 | return geometry |
| 512 | result = run([capabilities["xwininfo"], "-id", str(window_id)], env=env, timeout=3) |
| 513 | if result.returncode != 0: |
| 514 | return geometry |
| 515 | patterns = { |
| 516 | "x": r"Absolute upper-left X:\s*(-?\d+)", |
| 517 | "y": r"Absolute upper-left Y:\s*(-?\d+)", |
| 518 | "width": r"Width:\s*(\d+)", |
| 519 | "height": r"Height:\s*(\d+)", |
| 520 | } |
| 521 | for key, pattern in patterns.items(): |
| 522 | match = re.search(pattern, result.stdout) |
| 523 | if match: |
| 524 | geometry[key] = int(match.group(1)) |
| 525 | return geometry |
| 526 | |
| 527 | |
| 528 | def collect_window_props(env: dict[str, str], capabilities: dict[str, str], window_id: str) -> dict[str, str]: |
| 529 | props = {"title": "", "class": "", "name": "", "pid": ""} |
| 530 | xdotool = capabilities.get("xdotool") |
| 531 | if xdotool: |
| 532 | result = run([xdotool, "getwindowname", str(window_id)], env=env, timeout=3) |
| 533 | if result.returncode == 0: |
| 534 | props["title"] = result.stdout.strip() |
| 535 | xprop = capabilities.get("xprop") |
| 536 | if not xprop: |
| 537 | return props |
| 538 | result = run([xprop, "-id", str(window_id), "WM_CLASS", "WM_NAME", "_NET_WM_NAME", "_NET_WM_PID"], env=env, timeout=3) |
| 539 | if result.returncode != 0: |
| 540 | return props |
| 541 | parsed = parse_xprop(result.stdout) |
| 542 | title = parsed.get("_NET_WM_NAME") or parsed.get("WM_NAME") or props["title"] |
| 543 | props["title"] = title |
| 544 | props["class"] = parsed.get("WM_CLASS_CLASS", "") |
| 545 | props["name"] = parsed.get("WM_CLASS_NAME", "") |
| 546 | props["pid"] = parsed.get("_NET_WM_PID", "") |
| 547 | return props |
| 548 | |
| 549 | |
| 550 | def parse_xprop(output: str) -> dict[str, str]: |
| 551 | values: dict[str, str] = {} |
| 552 | for line in output.splitlines(): |
| 553 | if "=" not in line: |
| 554 | continue |
| 555 | key, raw_value = line.split("=", 1) |
| 556 | key = key.strip().split("(", 1)[0] |
| 557 | raw_value = raw_value.strip() |
| 558 | quoted = re.findall(r'"([^"]*)"', raw_value) |
| 559 | if key == "WM_CLASS" and quoted: |
| 560 | values["WM_CLASS_NAME"] = quoted[0] |
| 561 | values["WM_CLASS_CLASS"] = quoted[-1] |
| 562 | continue |
| 563 | if quoted: |
| 564 | values[key] = quoted[-1] |
| 565 | continue |
| 566 | match = re.search(r"-?\d+", raw_value) |
| 567 | values[key] = match.group(0) if match else raw_value |
| 568 | return values |
| 569 | |
| 570 | |
| 571 | def latest_screenshot(*, context_id: str = "") -> dict[str, Any]: |
| 572 | chat_dir = chat_screenshot_dir(context_id) |
| 573 | chat_latest = _latest_screenshot_from_dir( |
| 574 | chat_dir, |
| 575 | context_id=context_id, |
| 576 | ephemeral=False, |
| 577 | chat_scoped=True, |
| 578 | prune_older=False, |
| 579 | ) |
| 580 | if chat_latest.get("ok"): |
| 581 | return chat_latest |
| 582 | |
| 583 | prune_context_screenshots(context_id=context_id, max_age_seconds=RECENT_SCREENSHOT_SECONDS) |
| 584 | screenshot_dir = context_screenshot_dir(context_id) |
| 585 | return _latest_screenshot_from_dir( |
| 586 | screenshot_dir, |
| 587 | context_id=context_id, |
| 588 | ephemeral=True, |
| 589 | chat_scoped=False, |
| 590 | prune_older=True, |
| 591 | ) |
| 592 | |
| 593 | |
| 594 | def _latest_screenshot_from_dir( |
| 595 | screenshot_dir: Path, |
| 596 | *, |
| 597 | context_id: str = "", |
| 598 | ephemeral: bool, |
| 599 | chat_scoped: bool, |
| 600 | prune_older: bool, |
| 601 | ) -> dict[str, Any]: |
| 602 | if not screenshot_dir.exists(): |
| 603 | return {"ok": False, "path": "", "format": "", "captured_at": "", "recent": False} |
| 604 | candidates = [ |
| 605 | path |
| 606 | for path in screenshot_dir.iterdir() |
| 607 | if path.is_file() and path.suffix.lower() in _SCREENSHOT_SUFFIXES |
| 608 | ] |
| 609 | if not candidates: |
| 610 | return {"ok": False, "path": "", "format": "", "captured_at": "", "recent": False} |
| 611 | latest = max(candidates, key=lambda item: item.stat().st_mtime) |
| 612 | if prune_older: |
| 613 | for candidate in candidates: |
| 614 | if candidate != latest: |
| 615 | candidate.unlink(missing_ok=True) |
| 616 | age = max(0.0, time.time() - latest.stat().st_mtime) |
| 617 | return { |
| 618 | "ok": True, |
| 619 | "path": str(latest), |
| 620 | "a0_path": normalize_a0_path(latest), |
| 621 | "format": latest.suffix.lower().lstrip("."), |
| 622 | "captured_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(latest.stat().st_mtime)), |
| 623 | "recent": age <= RECENT_SCREENSHOT_SECONDS, |
| 624 | "ephemeral": ephemeral, |
| 625 | "chat_scoped": chat_scoped, |
| 626 | "context_id": _safe_context_id(context_id), |
| 627 | } |
| 628 | |
| 629 | |
| 630 | def stable_state( |
| 631 | *, |
| 632 | display: str, |
| 633 | profile_dir: str, |
| 634 | context_id: str = "", |
| 635 | size: dict[str, int] | None = None, |
| 636 | pointer: dict[str, int] | None = None, |
| 637 | active_window: dict[str, Any] | None = None, |
| 638 | windows: list[dict[str, Any]] | None = None, |
| 639 | screenshot: dict[str, Any] | None = None, |
| 640 | capabilities: dict[str, str] | None = None, |
| 641 | errors: list[str] | None = None, |
| 642 | ) -> dict[str, Any]: |
| 643 | clean_errors = [str(error) for error in errors or [] if str(error)] |
| 644 | return { |
| 645 | "ok": not clean_errors, |
| 646 | "context_id": _safe_context_id(context_id), |
| 647 | "display": display, |
| 648 | "profile_dir": profile_dir, |
| 649 | "size": size or {"width": 0, "height": 0}, |
| 650 | "pointer": pointer or {"x": 0, "y": 0, "screen": 0, "window": 0}, |
| 651 | "active_window": active_window, |
| 652 | "windows": windows or [], |
| 653 | "screenshot": screenshot or {"ok": False, "path": "", "format": "", "captured_at": "", "recent": False}, |
| 654 | "capabilities": capabilities or collect_capabilities(), |
| 655 | "errors": clean_errors, |
| 656 | } |
| 657 | |
| 658 | |
| 659 | def compact_prompt_context(state: dict[str, Any] | None = None) -> str: |
| 660 | state = state if state is not None else collect_state(include_screenshot=False) |
| 661 | if not state.get("display"): |
| 662 | return "" |
| 663 | lines = ["[DESKTOP STATE]"] |
| 664 | size = state.get("size") or {} |
| 665 | pointer = state.get("pointer") or {} |
| 666 | lines.append( |
| 667 | f"- display={state.get('display', '')} size={size.get('width', 0)}x{size.get('height', 0)} " |
| 668 | f"pointer={pointer.get('x', 0)},{pointer.get('y', 0)}" |
| 669 | ) |
| 670 | active = state.get("active_window") or {} |
| 671 | if active: |
| 672 | lines.append( |
| 673 | f"- active={active.get('title', '') or '<untitled>'} " |
| 674 | f"class={active.get('class', '') or active.get('name', '')}" |
| 675 | ) |
| 676 | visible = [] |
| 677 | for window in state.get("windows") or []: |
| 678 | title = window.get("title") or "<untitled>" |
| 679 | window_class = window.get("class") or window.get("name") or "" |
| 680 | visible.append(f"{title} ({window_class})" if window_class else title) |
| 681 | if len(visible) >= 5: |
| 682 | break |
| 683 | if visible: |
| 684 | lines.append("- visible=" + "; ".join(visible)) |
| 685 | screenshot = state.get("screenshot") or {} |
| 686 | if screenshot.get("recent") and screenshot.get("path"): |
| 687 | ephemeral = " ephemeral" if screenshot.get("ephemeral") else "" |
| 688 | screenshot_ref = screenshot.get("a0_path") or screenshot["path"] |
| 689 | lines.append(f"- recent_screenshot={screenshot_ref}{ephemeral}") |
| 690 | context_id = str(state.get("context_id") or "").strip() |
| 691 | if context_id: |
| 692 | lines.append(f"- screenshot_context={context_id}") |
| 693 | context_arg = f" --context-id {context_id}" if context_id else "" |
| 694 | lines.append( |
| 695 | "- next=plugins/_desktop/skills/linux-desktop/scripts/desktopctl.sh state --json" |
| 696 | f"{context_arg} for structured checks; use observe --json --screenshot" |
| 697 | f"{context_arg} " |
| 698 | "before coordinate or visual-OCR actions; prefer sequence/focus/key/paste/save/app-native helpers first." |
| 699 | ) |
| 700 | lines.append( |
| 701 | "- verify=for terminal/CLI-agent output, use the screenshot path from a fresh final " |
| 702 | "observe --json --screenshot captured after the response appears." |
| 703 | ) |
| 704 | if state.get("errors"): |
| 705 | lines.append("- errors=" + "; ".join(str(item) for item in state["errors"][:2])) |
| 706 | return "\n".join(lines) |
| 707 | |
| 708 | |
| 709 | def parse_shell_values(output: str) -> dict[str, str]: |
| 710 | values: dict[str, str] = {} |
| 711 | for line in output.splitlines(): |
| 712 | if "=" not in line: |
| 713 | continue |
| 714 | key, value = line.split("=", 1) |
| 715 | values[key.strip()] = value.strip().strip('"') |
| 716 | return values |
| 717 | |
| 718 | |
| 719 | def int_value(value: Any, default: int = 0) -> int: |
| 720 | try: |
| 721 | return int(str(value).strip()) |
| 722 | except (TypeError, ValueError): |
| 723 | return default |
| 724 | |
| 725 | |
| 726 | def run(command: list[str], *, env: dict[str, str], timeout: float) -> subprocess.CompletedProcess[str]: |
| 727 | try: |
| 728 | return subprocess.run( |
| 729 | command, |
| 730 | check=False, |
| 731 | capture_output=True, |
| 732 | text=True, |
| 733 | timeout=timeout, |
| 734 | env=env, |
| 735 | ) |
| 736 | except OSError as exc: |
| 737 | return subprocess.CompletedProcess(command, 127, "", str(exc)) |
| 738 | except subprocess.TimeoutExpired as exc: |
| 739 | stdout = exc.stdout.decode("utf-8", errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "") |
| 740 | stderr = exc.stderr.decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "") |
| 741 | return subprocess.CompletedProcess(command, 124, stdout, stderr or "command timed out") |
| 742 | |
| 743 | |
| 744 | def command_output(result: subprocess.CompletedProcess[str]) -> str: |
| 745 | return (result.stderr or result.stdout or "").strip() |
| 746 | |
| 747 | |
| 748 | def image_width(path: Path) -> int: |
| 749 | try: |
| 750 | from PIL import Image |
| 751 | |
| 752 | with Image.open(path) as image: |
| 753 | return int(image.width) |
| 754 | except Exception: |
| 755 | return 0 |
| 756 | |
| 757 | |
| 758 | def image_height(path: Path) -> int: |
| 759 | try: |
| 760 | from PIL import Image |
| 761 | |
| 762 | with Image.open(path) as image: |
| 763 | return int(image.height) |
| 764 | except Exception: |
| 765 | return 0 |
| 766 | |
| 767 | |
| 768 | def ephemeral_screenshot_result( |
| 769 | path: Path, |
| 770 | *, |
| 771 | context_id: str = "", |
| 772 | image_format: str = "png", |
| 773 | width: int = 0, |
| 774 | height: int = 0, |
| 775 | ) -> dict[str, Any]: |
| 776 | from helpers import ephemeral_images |
| 777 | |
| 778 | mime = "image/jpeg" if image_format.lower() in {"jpg", "jpeg"} else "image/png" |
| 779 | safe_context = _safe_context_id(context_id) |
| 780 | ref = ephemeral_images.put_image_bytes( |
| 781 | context_id=str(context_id or "").strip(), |
| 782 | mime=mime, |
| 783 | payload=path.read_bytes(), |
| 784 | name=path.name, |
| 785 | ) |
| 786 | path.unlink(missing_ok=True) |
| 787 | prune_context_screenshots(context_id=context_id) |
| 788 | return { |
| 789 | "ok": True, |
| 790 | "path": "", |
| 791 | "format": image_format, |
| 792 | "mime": mime, |
| 793 | "width": width, |
| 794 | "height": height, |
| 795 | "captured_at": iso_now(), |
| 796 | "recent": True, |
| 797 | "ephemeral": True, |
| 798 | "ephemeral_ref": ref, |
| 799 | "context_id": safe_context, |
| 800 | "vision_load": { |
| 801 | "tool_name": "vision_load", |
| 802 | "tool_args": {"paths": [ref]}, |
| 803 | }, |
| 804 | "error": "", |
| 805 | } |
| 806 | |
| 807 | |
| 808 | def prune_context_screenshots( |
| 809 | *, |
| 810 | context_id: str = "", |
| 811 | keep_path: Path | None = None, |
| 812 | max_age_seconds: float | None = None, |
| 813 | ) -> None: |
| 814 | screenshot_dir = context_screenshot_dir(context_id) |
| 815 | if not screenshot_dir.exists(): |
| 816 | return |
| 817 | keep = keep_path.resolve(strict=False) if keep_path else None |
| 818 | now = time.time() |
| 819 | for candidate in screenshot_dir.iterdir(): |
| 820 | if not candidate.is_file() or candidate.suffix.lower() not in _SCREENSHOT_SUFFIXES: |
| 821 | continue |
| 822 | if keep is not None and candidate.resolve(strict=False) == keep: |
| 823 | continue |
| 824 | if max_age_seconds is not None: |
| 825 | try: |
| 826 | if now - candidate.stat().st_mtime <= max_age_seconds: |
| 827 | continue |
| 828 | except OSError: |
| 829 | pass |
| 830 | candidate.unlink(missing_ok=True) |
| 831 | try: |
| 832 | screenshot_dir.rmdir() |
| 833 | except OSError: |
| 834 | pass |
| 835 | |
| 836 | |
| 837 | def iso_now() -> str: |
| 838 | return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) |
| 839 | |
| 840 | |
| 841 | def main(argv: list[str] | None = None) -> int: |
| 842 | parser = argparse.ArgumentParser(description="Observe the Agent Zero persistent Linux Desktop state.") |
| 843 | subparsers = parser.add_subparsers(dest="command") |
| 844 | |
| 845 | state_parser = subparsers.add_parser("state") |
| 846 | state_parser.add_argument("--json", action="store_true") |
| 847 | state_parser.add_argument("--screenshot", action="store_true") |
| 848 | state_parser.add_argument("--context-id", default="") |
| 849 | |
| 850 | observe_parser = subparsers.add_parser("observe") |
| 851 | observe_parser.add_argument("--json", action="store_true") |
| 852 | observe_parser.add_argument("--screenshot", action="store_true") |
| 853 | observe_parser.add_argument("--context-id", default="") |
| 854 | |
| 855 | screenshot_parser = subparsers.add_parser("screenshot") |
| 856 | screenshot_parser.add_argument("path", nargs="?") |
| 857 | screenshot_parser.add_argument("--json", action="store_true") |
| 858 | screenshot_parser.add_argument("--context-id", default="") |
| 859 | |
| 860 | args = parser.parse_args(argv) |
| 861 | command = args.command or "state" |
| 862 | if command in {"state", "observe"}: |
| 863 | payload = collect_state( |
| 864 | include_screenshot=bool(args.screenshot), |
| 865 | context_id=str(args.context_id or ""), |
| 866 | screenshot_transport="path", |
| 867 | ) |
| 868 | print(json.dumps(payload, sort_keys=True)) |
| 869 | return 0 if payload.get("ok") else 1 |
| 870 | |
| 871 | if command == "screenshot": |
| 872 | errors: list[str] = [] |
| 873 | env_info = resolve_environment(errors=errors) |
| 874 | payload = capture_screenshot( |
| 875 | display_env(display=env_info["display"], profile_dir=env_info["profile_dir"]), |
| 876 | collect_capabilities(), |
| 877 | path=args.path, |
| 878 | errors=errors, |
| 879 | context_id=str(args.context_id or ""), |
| 880 | transport="path", |
| 881 | ) |
| 882 | if args.json: |
| 883 | print(json.dumps(payload, sort_keys=True)) |
| 884 | else: |
| 885 | print(payload.get("path") or payload.get("error") or "") |
| 886 | return 0 if payload.get("ok") else 1 |
| 887 | |
| 888 | parser.print_help() |
| 889 | return 2 |
| 890 | |
| 891 | |
| 892 | if __name__ == "__main__": |
| 893 | raise SystemExit(main()) |