@samitouri / QOSami_Agents_Zero / commits / 07e7e1e0

Fork 0

Agent Zero AI framework

Harden desktop session reliability

Target LibreOffice saves to the requested document window and report whether the file content changed. Fail fast on dead XFCE startup, probe the live X display, replace session manifests atomically, and decode XWD screenshots through Pillow's raw decoder. Add focused regressions for each path.

Alessandro committed Jul 11, 2026 at 23:25 UTC 07e7e1e07554341d1e9815ceb815f3e691313a0a

4 files changed +269 -164

plugins/_desktop/helpers/desktop_session.py

+108 -76

@@ -5,7 +5,6 @@ import fcntl

5 import hashlib

6 import json

7 import os

8 -import re

8 import shutil

9 import socket

10 import subprocess

@@ -223,39 +222,69 @@ class DesktopSessionManager:

222 return {"ok": True, "refreshed": refreshed, "desktop": session.public(doc)}

223

224 def save(self, session_id: str, file_id: str = "") -> dict[str, Any]:

226 - session = self.require(session_id)

227 - doc = self._document_for_save(session, file_id)

228 - xdotool = shutil.which("xdotool")

229 - if not xdotool:

230 - updated = document_store.register_document(doc["path"]) if doc else None

231 - return {

232 - "ok": False,

233 - "error": "xdotool is not installed; use LibreOffice's Save control inside the canvas.",

234 - "document": _public_doc(updated) if updated else None,

235 - }

225 + with self._lock:

226 + session = self.require(session_id)

227 + doc = self._document_for_save(session, file_id)

228 + if not doc:

229 + return {

230 + "ok": True,

231 + "session_id": session.session_id,

232 + "document": None,

233 + "changed": False,

234 + }

235

237 - result = subprocess.run(

238 - [xdotool, "key", "--clearmodifiers", "ctrl+s"],

239 - check=False,

240 - capture_output=True,

241 - text=True,

242 - timeout=8,

243 - env=self._display_env(session),

244 - )

245 - time.sleep(0.8)

246 - updated = document_store.register_document(doc["path"]) if doc else None

247 - if result.returncode != 0:

248 - detail = (result.stderr or result.stdout or "").strip()

236 + xdotool = shutil.which("xdotool")

237 + if not xdotool:

238 + updated = document_store.register_document(doc["path"])

239 + return {

240 + "ok": False,

241 + "error": "xdotool is not installed; use LibreOffice's Save control inside the canvas.",

242 + "document": _public_doc(updated),

243 + }

244 +

245 + window_id = self._office_window_id_locked(

246 + session,

247 + title=str(doc.get("basename") or Path(doc["path"]).name),

248 + fallback=False,

249 + )

250 + if not window_id:

251 + return {

252 + "ok": False,

253 + "error": f"LibreOffice window not found for {doc['basename']}.",

254 + "document": _public_doc(doc),

255 + }

256 +

257 + result = subprocess.run(

258 + [

259 + xdotool,

260 + "windowactivate",

261 + "--sync",

262 + window_id,

263 + "key",

264 + "--clearmodifiers",

265 + "ctrl+s",

266 + ],

267 + check=False,

268 + capture_output=True,

269 + text=True,

270 + timeout=8,

271 + env=self._display_env(session),

272 + )

273 + time.sleep(0.8)

274 + updated = document_store.register_document(doc["path"])

275 + if result.returncode != 0:

276 + detail = (result.stderr or result.stdout or "").strip()

277 + return {

278 + "ok": False,

279 + "error": detail or "LibreOffice desktop save shortcut failed.",

280 + "document": _public_doc(updated),

281 + }

282 return {

250 - "ok": False,

251 - "error": detail or "LibreOffice desktop save shortcut failed.",

252 - "document": _public_doc(updated) if updated else None,

283 + "ok": True,

284 + "session_id": session.session_id,

285 + "document": _public_doc(updated),

286 + "changed": updated.get("sha256") != doc.get("sha256"),

287 }

254 - return {

255 - "ok": True,

256 - "session_id": session.session_id,

257 - "document": _public_doc(updated) if updated else None,

258 - }

288

289 def sync(self, session_id: str = "", file_id: str = "") -> dict[str, Any]:

290 session = self.get(session_id) if session_id else self._find_by_file_id(file_id)

@@ -820,57 +849,48 @@ class DesktopSessionManager:

849 return ""

850 env = self._display_env(session)

851 title = str(title or "").strip()

823 - searches: list[list[str]] = []

852 + try:

853 + result = subprocess.run(

854 + [xdotool, "search", "--onlyvisible", "--class", "libreoffice"],

855 + check=False,

856 + capture_output=True,

857 + text=True,

858 + timeout=2,

859 + env=env,

860 + )

861 + window_ids = [line.strip() for line in result.stdout.splitlines() if line.strip()]

862 + except (OSError, subprocess.TimeoutExpired):

863 + window_ids = []

864 if title:

825 - escaped_title = re.escape(title)

826 - searches.append([

827 - xdotool,

828 - "search",

829 - "--onlyvisible",

830 - "--name",

831 - escaped_title,

832 - ])

833 - for window_class in (

834 - "libreoffice",

835 - "libreoffice-writer",

836 - "libreoffice-calc",

837 - "libreoffice-impress",

838 - ):

839 - searches.append([

840 - xdotool,

841 - "search",

842 - "--onlyvisible",

843 - "--class",

844 - window_class,

845 - "--name",

846 - escaped_title,

847 - ])

865 + for window_id in reversed(window_ids):

866 + try:

867 + result = subprocess.run(

868 + [xdotool, "getwindowname", window_id],

869 + check=False,

870 + capture_output=True,

871 + text=True,

872 + timeout=2,

873 + env=env,

874 + )

875 + except (OSError, subprocess.TimeoutExpired):

876 + continue

877 + if title.casefold() in result.stdout.strip().casefold():

878 + return window_id

879 + if fallback and window_ids:

880 + return window_ids[-1]

881 if fallback:

849 - for window_class in (

850 - "libreoffice",

851 - "libreoffice-writer",

852 - "libreoffice-calc",

853 - "libreoffice-impress",

854 - ):

855 - searches.append([xdotool, "search", "--onlyvisible", "--class", window_class])

856 - searches.append([xdotool, "search", "--onlyvisible", "--name", "LibreOffice"])

857 - for command in searches:

882 try:

883 result = subprocess.run(

860 - command,

884 + [xdotool, "search", "--onlyvisible", "--name", "LibreOffice"],

885 check=False,

886 capture_output=True,

887 text=True,

888 timeout=2,

889 env=env,

890 )

891 + window_ids = [line.strip() for line in result.stdout.splitlines() if line.strip()]

892 except (OSError, subprocess.TimeoutExpired):

868 - continue

869 - window_ids = [

870 - line.strip()

871 - for line in result.stdout.splitlines()

872 - if line.strip()

873 - ]

893 + window_ids = []

894 if window_ids:

895 return window_ids[-1]

896 return ""

@@ -1573,14 +1593,20 @@ fi

1593 self._remove_manifest(session_id)

1594

1595 def _wait_for_display(self, session: DesktopSession) -> None:

1576 - marker = Path(f"/tmp/.X11-unix/X{session.display}")

1596 deadline = time.time() + DISPLAY_START_TIMEOUT_SECONDS

1597 while time.time() < deadline:

1598 process = session.processes.get("xvfb") or session.processes.get("xpra")

1599 if process and process.poll() is not None:

1600 raise RuntimeError("The LibreOffice X display exited before it was ready.")

1582 - if marker.exists():

1583 - return

1601 + try:

1602 + if virtual_desktop.current_display_size(

1603 + session.display,

1604 + xauthority=self._xauthority(session),

1605 + home=str(session.profile_dir),

1606 + ):

1607 + return

1608 + except (OSError, subprocess.TimeoutExpired):

1609 + pass

1610 time.sleep(0.1)

1611 raise TimeoutError("Timed out waiting for the LibreOffice X display.")

1612

@@ -1589,7 +1615,7 @@ fi

1615 while time.time() < deadline:

1616 process = session.processes.get("xfce")

1617 if process and process.poll() is not None:

1592 - return

1618 + raise RuntimeError("The XFCE desktop session exited before it was ready.")

1619 if virtual_desktop.has_window(

1620 display=session.display,

1621 name="xfce4-panel",

@@ -1619,7 +1645,13 @@ fi

1645 "owner_pid": os.getpid(),

1646 "pids": pids,

1647 }

1622 - (SESSION_DIR / f"{session.session_id}.json").write_text(json.dumps(payload), encoding="utf-8")

1648 + manifest = SESSION_DIR / f"{session.session_id}.json"

1649 + tmp = manifest.with_name(f".{manifest.name}.{uuid.uuid4().hex}.tmp")

1650 + try:

1651 + tmp.write_text(json.dumps(payload), encoding="utf-8")

1652 + os.replace(tmp, manifest)

1653 + finally:

1654 + tmp.unlink(missing_ok=True)

1655

1656 def _remove_manifest(self, session_id: str) -> None:

1657 (SESSION_DIR / f"{session_id}.json").unlink(missing_ok=True)

plugins/_desktop/helpers/desktop_state.py

+26 -55

@@ -262,50 +262,43 @@ 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, endian = parse_xwd_header(data)

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 - image_byte_order = "little" if header["byte_order"] == 0 else "big"

270 color_table_size = header["ncolors"] * 12

271 pixel_offset = header["header_size"] + color_table_size

273 - bytes_per_pixel = max((bits_per_pixel + 7) // 8, 1)

274 - if width > 0 and bytes_per_line % width == 0:

275 - bytes_per_pixel = max(bytes_per_pixel, bytes_per_line // width)

272 if width <= 0 or height <= 0 or bytes_per_line <= 0:

273 raise ValueError("invalid XWD dimensions")

278 - if pixel_offset + (height * bytes_per_line) > len(data):

274 + pixel_size = height * bytes_per_line

275 + if pixel_offset + pixel_size > len(data):

276 raise ValueError("truncated XWD pixel data")

277

281 - red_mask = header["red_mask"]

282 - green_mask = header["green_mask"]

283 - blue_mask = header["blue_mask"]

284 - red_shift, red_bits = mask_shift_and_bits(red_mask)

285 - green_shift, green_bits = mask_shift_and_bits(green_mask)

286 - blue_shift, blue_bits = mask_shift_and_bits(blue_mask)

287 - if min(red_bits, green_bits, blue_bits) <= 0:

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")

289 -

290 - pixels: list[tuple[int, int, int]] = []

291 - for row in range(height):

292 - row_start = pixel_offset + (row * bytes_per_line)

293 - for column in range(width):

294 - start = row_start + (column * bytes_per_pixel)

295 - pixel_bytes = data[start : start + bytes_per_pixel]

296 - if len(pixel_bytes) < bytes_per_pixel:

297 - raise ValueError("truncated XWD pixel")

298 - pixel = int.from_bytes(pixel_bytes, image_byte_order, signed=False)

299 - pixels.append(

300 - (

301 - scale_channel((pixel & red_mask) >> red_shift, red_bits),

302 - scale_channel((pixel & green_mask) >> green_shift, green_bits),

303 - scale_channel((pixel & blue_mask) >> blue_shift, blue_bits),

304 - ),

305 - )

306 -

307 - image = Image.new("RGB", (width, height))

308 - image.putdata(pixels)

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

@@ -348,28 +341,6 @@ def parse_xwd_header(data: bytes) -> tuple[dict[str, int], str]:

341 raise ValueError("unsupported XWD header")

342

343

351 -def mask_shift_and_bits(mask: int) -> tuple[int, int]:

352 - if mask <= 0:

353 - return 0, 0

354 - shift = 0

355 - value = mask

356 - while value and value & 1 == 0:

357 - shift += 1

358 - value >>= 1

359 - bits = 0

360 - while value & 1:

361 - bits += 1

362 - value >>= 1

363 - return shift, bits

364 -

365 -

366 -def scale_channel(value: int, bits: int) -> int:

367 - if bits >= 8:

368 - return max(0, min(255, value >> (bits - 8)))

369 - max_value = (1 << bits) - 1

370 - return 0 if max_value <= 0 else round((value / max_value) * 255)

371 -

372 -

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)

tests/test_office_desktop_state.py

+18 -33

@@ -6,6 +6,8 @@ import sys

6 import types

7 from pathlib import Path

9 +import pytest

10 +

11 PROJECT_ROOT = Path(__file__).resolve().parents[1]

12 if str(PROJECT_ROOT) not in sys.path:

13 sys.path.insert(0, str(PROJECT_ROOT))

@@ -324,7 +326,16 @@ def test_virtual_desktop_system_display_normalization_rejects_portrait_viewports

326 assert virtual_desktop.normalize_desktop_display_size(1600, 900) == (1600, 900)

327

328

327 -def test_xwd_fallback_parser_handles_truecolor_pixels(tmp_path, monkeypatch):

329 +@pytest.mark.parametrize(

330 + ("byte_order", "pixel_bytes"),

331 + (

332 + (0, bytes.fromhex("0000ff00") + bytes.fromhex("00ff0000")),

333 + (1, bytes.fromhex("00ff0000") + bytes.fromhex("0000ff00")),

334 + ),

335 +)

336 +def test_xwd_fallback_parser_handles_truecolor_pixels(tmp_path, byte_order, pixel_bytes):

337 + from PIL import Image

338 +

339 raw_path = tmp_path / "shot.xwd"

340 target = tmp_path / "shot.png"

341 header_values = [

@@ -335,7 +346,7 @@ def test_xwd_fallback_parser_handles_truecolor_pixels(tmp_path, monkeypatch):

346 2, # pixmap_width

347 1, # pixmap_height

348 0, # xoffset

338 - 1, # byte_order: MSBFirst for pixel bytes

349 + byte_order,

350 32, # bitmap_unit

351 1, # bitmap_bit_order

352 32, # bitmap_pad

@@ -354,38 +365,12 @@ def test_xwd_fallback_parser_handles_truecolor_pixels(tmp_path, monkeypatch):

365 0, # window_y

366 0, # window_bdrwidth

367 ]

357 - raw_path.write_bytes(

358 - struct.pack(">25I", *header_values)

359 - + bytes.fromhex("00ff0000")

360 - + bytes.fromhex("0000ff00")

361 - )

362 -

363 - captured: dict[str, object] = {}

364 - image_module = types.ModuleType("PIL.Image")

365 -

366 - class FakeOutputImage:

367 - def putdata(self, pixels):

368 - captured["pixels"] = list(pixels)

369 -

370 - def save(self, path):

371 - Path(path).write_bytes(b"fallback-png")

372 -

373 - def fake_new(mode, size):

374 - captured["mode"] = mode

375 - captured["size"] = size

376 - return FakeOutputImage()

377 -

378 - image_module.new = fake_new

379 - pil_module = types.ModuleType("PIL")

380 - pil_module.Image = image_module

381 -

382 - monkeypatch.setitem(sys.modules, "PIL", pil_module)

383 - monkeypatch.setitem(sys.modules, "PIL.Image", image_module)

368 + raw_path.write_bytes(struct.pack(">25I", *header_values) + pixel_bytes)

369

370 converted = desktop_state.convert_xwd_to_image(raw_path, target)

371

372 assert converted == {"width": 2, "height": 1}

388 - assert captured["mode"] == "RGB"

389 - assert captured["size"] == (2, 1)

390 - assert captured["pixels"] == [(255, 0, 0), (0, 255, 0)]

391 - assert target.read_bytes() == b"fallback-png"

373 + with Image.open(target) as image:

374 + assert image.mode == "RGB"

375 + assert image.size == (2, 1)

376 + assert list(image.getdata()) == [(255, 0, 0), (0, 255, 0)]

tests/test_office_document_store.py

+117

@@ -1037,6 +1037,123 @@ def test_official_desktop_session_manager_opens_binary_session(office_state, tmp

1037 assert manager.close(payload["session_id"], save_first=False)["persistent"] is True

1038

1039

1040 +def test_desktop_save_targets_requested_libreoffice_window(office_state, tmp_path, monkeypatch):

1041 + doc = document_store.create_document("spreadsheet", "Targeted Save", "ods", "Name,Value\nA,1")

1042 + session = desktop_session.DesktopSession(

1043 + session_id=desktop_session.SYSTEM_SESSION_ID,

1044 + file_id=doc["file_id"],

1045 + extension=doc["extension"],

1046 + path=doc["path"],

1047 + title=doc["basename"],

1048 + display=desktop_session.DISPLAY_BASE,

1049 + xpra_port=desktop_session.XPRA_PORT_BASE,

1050 + token=desktop_session.SYSTEM_SESSION_ID,

1051 + url="/desktop/session/agent-zero-desktop/index.html",

1052 + profile_dir=tmp_path / "profile",

1053 + processes={"xpra": types.SimpleNamespace(poll=lambda: None)},

1054 + )

1055 + manager = desktop_session.DesktopSessionManager()

1056 + manager._sessions[session.session_id] = session

1057 + commands = []

1058 +

1059 + monkeypatch.setattr(desktop_session.shutil, "which", lambda name: f"/usr/bin/{name}")

1060 + monkeypatch.setattr(desktop_session.time, "sleep", lambda _seconds: None)

1061 +

1062 + def fake_run(command, **_kwargs):

1063 + commands.append(command)

1064 + if command[1] == "search":

1065 + assert command == ["/usr/bin/xdotool", "search", "--onlyvisible", "--class", "libreoffice"]

1066 + return subprocess.CompletedProcess(command, 0, "222\n", "")

1067 + if command[1] == "getwindowname":

1068 + return subprocess.CompletedProcess(command, 0, f"{doc['basename']} — LibreOffice Calc\n", "")

1069 + Path(doc["path"]).write_bytes(Path(doc["path"]).read_bytes() + b"changed")

1070 + return subprocess.CompletedProcess(command, 0, "", "")

1071 +

1072 + monkeypatch.setattr(desktop_session.subprocess, "run", fake_run)

1073 +

1074 + result = manager.save(session.session_id, doc["file_id"])

1075 +

1076 + assert result["ok"] is True

1077 + assert result["changed"] is True

1078 + assert commands[-1] == [

1079 + "/usr/bin/xdotool",

1080 + "windowactivate",

1081 + "--sync",

1082 + "222",

1083 + "key",

1084 + "--clearmodifiers",

1085 + "ctrl+s",

1086 + ]

1087 +

1088 + session.file_id = desktop_session.SYSTEM_FILE_ID

1089 + command_count = len(commands)

1090 + assert manager.save(session.session_id)["changed"] is False

1091 + assert len(commands) == command_count

1092 +

1093 +

1094 +def test_desktop_startup_waiters_probe_display_and_reject_dead_xfce(tmp_path, monkeypatch):

1095 + session = desktop_session.DesktopSession(

1096 + session_id=desktop_session.SYSTEM_SESSION_ID,

1097 + file_id=desktop_session.SYSTEM_FILE_ID,

1098 + extension="desktop",

1099 + path=str(tmp_path),

1100 + title=desktop_session.SYSTEM_TITLE,

1101 + display=desktop_session.DISPLAY_BASE,

1102 + xpra_port=desktop_session.XPRA_PORT_BASE,

1103 + token=desktop_session.SYSTEM_SESSION_ID,

1104 + url="/desktop/session/agent-zero-desktop/index.html",

1105 + profile_dir=tmp_path / "profile",

1106 + processes={"xvfb": types.SimpleNamespace(poll=lambda: None)},

1107 + )

1108 + manager = desktop_session.DesktopSessionManager()

1109 + probes = iter((None, (1920, 1080)))

1110 + monkeypatch.setattr(

1111 + desktop_session.virtual_desktop,

1112 + "current_display_size",

1113 + lambda *_args, **_kwargs: next(probes),

1114 + )

1115 + monkeypatch.setattr(desktop_session.time, "sleep", lambda _seconds: None)

1116 +

1117 + manager._wait_for_display(session)

1118 +

1119 + session.processes = {"xfce": types.SimpleNamespace(poll=lambda: 1)}

1120 + with pytest.raises(RuntimeError, match="XFCE desktop session exited"):

1121 + manager._wait_for_xfce(session)

1122 +

1123 +

1124 +def test_desktop_manifest_is_replaced_atomically(tmp_path, monkeypatch):

1125 + session_dir = tmp_path / "sessions"

1126 + session_dir.mkdir()

1127 + manifest = session_dir / f"{desktop_session.SYSTEM_SESSION_ID}.json"

1128 + manifest.write_text('{"old": true}', encoding="utf-8")

1129 + session = desktop_session.DesktopSession(

1130 + session_id=desktop_session.SYSTEM_SESSION_ID,

1131 + file_id=desktop_session.SYSTEM_FILE_ID,

1132 + extension="desktop",

1133 + path=str(tmp_path),

1134 + title=desktop_session.SYSTEM_TITLE,

1135 + display=desktop_session.DISPLAY_BASE,

1136 + xpra_port=desktop_session.XPRA_PORT_BASE,

1137 + token=desktop_session.SYSTEM_SESSION_ID,

1138 + url="/desktop/session/agent-zero-desktop/index.html",

1139 + profile_dir=tmp_path / "profile",

1140 + )

1141 + real_replace = os.replace

1142 +

1143 + def assert_atomic_replace(source, destination):

1144 + assert json.loads(Path(destination).read_text(encoding="utf-8")) == {"old": True}

1145 + assert json.loads(Path(source).read_text(encoding="utf-8"))["display"] == session.display

1146 + real_replace(source, destination)

1147 +

1148 + monkeypatch.setattr(desktop_session, "SESSION_DIR", session_dir)

1149 + monkeypatch.setattr(desktop_session.os, "replace", assert_atomic_replace)

1150 +

1151 + desktop_session.DesktopSessionManager()._write_manifest(session)

1152 +

1153 + assert json.loads(manifest.read_text(encoding="utf-8"))["session_id"] == session.session_id

1154 + assert list(session_dir.glob(".*.tmp")) == []

1155 +

1156 +

1157 def test_shutdown_panel_launcher_requires_second_click(tmp_path):

1158 profile_dir = tmp_path / "desktop" / "profiles" / desktop_session.SYSTEM_SESSION_ID

1159 profile_dir.mkdir(parents=True)