From 5a14abe3231779bbd7f223f09c2b199fa16bbdb2 Mon Sep 17 00:00:00 2001 From: fy59 Date: Tue, 6 Oct 2026 06:46:36 +0200 Subject: [PATCH] feat(session): execute controlled Sway session restores --- docs/session-restore-v1.md | 178 ++++- .../labfy-sway/session/session_executor.py | 642 ++++++++++++++++++ .../labfy-sway/session/session_restore.py | 9 +- .../labfy-sway/session/session_snapshot.py | 78 ++- tests/test_session_executor.py | 479 +++++++++++++ 5 files changed, 1376 insertions(+), 10 deletions(-) create mode 100644 quickshell/.config/quickshell/labfy-sway/session/session_executor.py create mode 100644 tests/test_session_executor.py diff --git a/docs/session-restore-v1.md b/docs/session-restore-v1.md index 6be6d3c05..3443c58de 100644 --- a/docs/session-restore-v1.md +++ b/docs/session-restore-v1.md @@ -1,8 +1,9 @@ -# Restore Plan V1 — STEP19A +# Restore Plan et Executor V1 — STEP19A/STEP19B ## Statut et architecture -**STEP19A DOES NOT RESTORE.** +**Le planner STEP19A ne restaure jamais.** L'exécution contrôlée appartient +exclusivement au module séparé STEP19B décrit à la fin de ce document. STEP19A charge un Session Snapshot V1 persistant, observe la session Sway actuelle avec les quatre lectures STEP18, puis produit un plan JSON. Il ne @@ -280,6 +281,173 @@ applications_launched = 0 snapshots_written = 0 ``` -Avant STEP19B, aucune attente de fenêtre, aucune commande Sway mutatrice, -aucune restauration d'état interne, aucun fallback multi-output, aucune UI -QML et aucun comportement Spatial Canvas ne sont implémentés. +Le planner n'effectue aucune attente de fenêtre, aucune commande Sway +mutatrice, aucune restauration d'état interne, aucun fallback multi-output, +aucune UI QML et aucun comportement Spatial Canvas. + +## Executor V1 — STEP19B + +### Frontière d'effets et CLI + +`session_restore.py` reste le moteur pur. `session_executor.py` est la seule +frontière autorisée à ouvrir le lock d'exécution et à lancer des argv directs. + +```text +Session Snapshot V1 validé + ↓ +Restore Planner pur + ↓ +Restore Plan V1 frais + ↓ +Session Executor V1 + ↓ +uwsm app + Sway IPC ciblé +``` + +La mutation nécessite une intention explicite difficile à déclencher par +accident : + +```bash +session_snapshot.py apply --execute +``` + +`apply ` sans `--execute` retourne +`RESTORE_EXECUTION_REQUIRES_EXPLICIT_EXECUTE` sans charger l'exécuteur. Le CLI +n'accepte jamais un fichier Restore Plan comme autorité. Chaque apply exécute +obligatoirement : + +```text +load snapshot → collect live → build fresh plan → validate → preflight → execute +``` + +Le snapshot est hashé avant et après l'opération. Une différence produit un +échec explicite ; l'exécuteur ne possède aucune API d'écriture de snapshot. + +### Capability gate intégrale + +Avant le premier lancement ou la première commande Sway, l'exécuteur examine +toutes les actions. Le sous-ensemble STEP19B est : + +```text +reuse-window no-op revalidé +launch-application uwsm app + attente/rematching +skip-autostart-managed no-op explicite, propriétaire inchangé +move-to-workspace workspace numérique seulement +restore-floating enable/disable déterministe +restore-fullscreen modes 0 et 1 seulement +restore-focus toujours dernière mutation +``` + +Ces actions bloquent l'ensemble du plan avec +`RESTORE_BLOCKED_UNSUPPORTED_ACTION` et une liste précise, sans aucune mutation : + +```text +restore-tree-position +restore-scratchpad-hidden +restore-scratchpad-visible +manual-required +toute action inconnue +``` + +Un matching runtime `ambiguous`, un workspace non numérique et un mode +fullscreen non prouvé bloquent eux aussi le plan. Une fenêtre absente n'a pas +encore de position d'arbre observable ; le planner n'émet donc une différence +`restore-tree-position` que pour une fenêtre déjà matchée. STEP19C pourra +replanifier l'arbre après apparition. + +### Lock privé + +Un apply détient un `flock(LOCK_EX|LOCK_NB)` sur `.restore.lock` pendant toute +la séquence load/plan/execute/rehash. Le fichier est régulier, ouvert sans +suivre les symlinks et forcé à `0600` dans le répertoire sessions `0700`. Le +fichier peut persister ; la fermeture du FD libère le lock. Aucun daemon ou +scheduler supplémentaire n'est créé. + +### DesktopEntry et lancement UWSM + +`launch.argv` dans le plan reste purement descriptif et n'est jamais exécuté. +Pour chaque lancement, l'exécuteur vérifie : + +```text +action = launch-application +backend = uwsm-app-desktop-entry +classification = user-application +restore_identity.confidence = exact +DesktopEntry ID syntaxiquement bornée +résolution exacte identique dans les sources XDG courantes +``` + +Il reconstruit ensuite lui-même exactement : + +```python +["uwsm", "app", "--", desktop_entry] +``` + +Il n'utilise ni shell, ni `Sway exec`, ni `gtk-launch`, ni `gio launch`. Comme +`uwsm app` peut rester attaché pendant toute la vie de l'application, le +processus est lancé avec un argv direct, sorties vers `DEVNULL` et une nouvelle +session de processus. Son PID et son exit status ne prouvent jamais le succès. + +### Apparition et matching post-launch + +L'exécuteur conserve l'inventaire des `con_id` avant lancement, puis poll +`get_tree` via la collecte Session Snapshot V1 pendant au plus 10 secondes. +Seules les nouvelles fenêtres sont candidates. Il réutilise directement +`session_restore.match_windows()` ; il ne possède aucun second matcher. + +Zéro candidate compatible conduit à `RESTORE_LAUNCH_TIMEOUT`. Plusieurs +candidates indiscernables conduisent à `RESTORE_LAUNCH_AMBIGUOUS`. Le PID du +processus lancé n'intervient pas, ce qui couvre aussi une application qui +demanderait à une instance existante de créer une nouvelle fenêtre. + +### Revalidation runtime et commandes ciblées + +Avant chaque mutation, une nouvelle collecte doit retrouver le même `con_id`, +une identité application compatible et un workspace connu. Cela interdit de +cibler silencieusement un `con_id` disparu ou recyclé. Après la commande, une +attente bornée à une seconde relit l'arbre jusqu'à observer la postcondition, +en revalidant l'identité à chaque lecture. Elle absorbe uniquement le délai de +visibilité IPC ; son expiration reste un échec critique. + +Les workspaces exécutables sont des chaînes numériques canoniques positives. +Les noms libres restent hors périmètre afin qu'aucun quoting Sway non prouvé +ne devienne une surface d'injection. Un déplacement reconstruit uniquement : + +```text +[con_id=] move container to workspace number +``` + +Floating utilise exclusivement `floating enable` ou `floating disable`. +L'audit de `sway(5)` 1.12.0 prouve `fullscreen enable|disable [global]` : V1 +supporte `fullscreen_mode=0` avec `fullscreen disable` et le mode workspace +`fullscreen_mode=1` avec `fullscreen enable`. Le mode global et toute autre +valeur sont bloqués. Aucun état ne recourt à `toggle`. + +Le focus utilise `[con_id=] focus`, après la même revalidation, et reste la +dernière mutation. La disparition de la cible ne focalise jamais une autre +fenêtre par défaut. + +### Rapport, arrêt et absence de rollback magique + +Chaque action du rapport possède `pending`, `success`, `skipped`, `failed` ou +`blocked`, avec un `reason`. L'exécution s'arrête à la première divergence +critique et laisse les actions suivantes `pending`. + +Les lancements et mutations Sway ne sont pas transactionnels. Executor V1 ne +promet aucun rollback global : sa politique est un preflight total strict, +des commandes ciblées, une vérification après chaque commande, puis un arrêt +immédiat au premier écart. + +### Validation contrôlée STEP19B + +La validation réelle utilise exclusivement une Kitty créée pour le test sur +un workspace numérique libre. Le scénario A réutilise la fenêtre existante et +la replace par `con_id`, sans nouveau lancement. Le scénario B ferme seulement +cette fenêtre, lance `kitty.desktop` via UWSM, observe une nouvelle fenêtre, +la rematche et la replace. Le cleanup ferme uniquement le nouveau `con_id`, +supprime uniquement le snapshot de validation et restaure le focus utilisateur +initial lorsqu'il existe encore. + +Les fenêtres non marquées `TEST_ONLY` sont comparées avant/après sur +`con_id`, workspace, floating et fullscreen. L'arbre complet, le scratchpad +avancé, le Spatial Canvas et toute UI QML restent hors STEP19B. diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_executor.py b/quickshell/.config/quickshell/labfy-sway/session/session_executor.py new file mode 100644 index 000000000..c12e8f3c7 --- /dev/null +++ b/quickshell/.config/quickshell/labfy-sway/session/session_executor.py @@ -0,0 +1,642 @@ +#!/usr/bin/env python3 +"""Exécuter le sous-ensemble contrôlé STEP19B d'un Restore Plan V1 frais.""" + +from __future__ import annotations + +import contextlib +import copy +import fcntl +import hashlib +import json +import os +from pathlib import Path +import re +import stat +import subprocess +import time +from typing import Any, Callable, Iterator, Mapping, Sequence + + +REPORT_SCHEMA = "labfy.sway.restore-execution" +REPORT_VERSION = 1 +RESTORE_LOCK_FILENAME = ".restore.lock" +LAUNCH_BACKEND = "uwsm-app-desktop-entry" +DEFAULT_WINDOW_TIMEOUT_SECONDS = 10.0 +POLL_INTERVAL_SECONDS = 0.1 +POSTCONDITION_TIMEOUT_SECONDS = 1.0 +SUPPORTED_ACTIONS = { + "reuse-window", + "launch-application", + "skip-autostart-managed", + "move-to-workspace", + "restore-floating", + "restore-fullscreen", + "restore-focus", +} +UNSUPPORTED_ACTIONS = { + "restore-tree-position", + "restore-scratchpad-hidden", + "restore-scratchpad-visible", + "manual-required", +} +NUMERIC_WORKSPACE = re.compile(r"\A[1-9][0-9]*\Z") +DESKTOP_ENTRY_ID = re.compile(r"\A[A-Za-z0-9][A-Za-z0-9._-]*\.desktop\Z") + + +class RestoreExecutionError(RuntimeError): + """Échec contrôlé portant un code stable et un motif auditables.""" + + def __init__(self, code: str, reason: str): + super().__init__(f"{code}: {reason}") + self.code = code + self.reason = reason + + +def snapshot_digest(snapshot: Mapping[str, Any]) -> str: + """Hasher le document chargé, indépendamment de son formatage JSON sur disque.""" + payload = json.dumps(snapshot, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _default_runner(argv: Sequence[str], timeout: float) -> Any: + """CONTRACT: exécuter un argv construit localement, sans interpréteur shell.""" + if argv and argv[0] == "uwsm": + # WHY: uwsm app peut rester attaché pendant toute la vie de + # l'application. Attendre son exit bloquerait avant même le polling + # Sway ; son PID n'est en outre jamais l'identité de la future vue. + # CONTRACT: l'enfant direct reçoit uniquement l'argv allowlisté, sans + # shell et sans capture bornée à gérer. Le CLI STEP19B est éphémère. + # INVARIANT: seul le matching d'une nouvelle fenêtre prouve le succès. + return subprocess.Popen( # type: ignore[return-value] + list(argv), + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + close_fds=True, + start_new_session=True, + ) + return subprocess.run( + list(argv), + check=True, + capture_output=True, + text=True, + timeout=timeout, + ) + + +def _direct_uwsm_command(desktop_entry: str) -> list[str]: + """Reconstruire l'unique backend autorisé sans consulter launch.argv.""" + return ["uwsm", "app", "--", desktop_entry] + + +def _direct_sway_command(con_id: int, operation: str) -> list[str]: + """Cibler une observation runtime validée ; operation est toujours une constante locale.""" + return ["swaymsg", "-r", f"[con_id={con_id}] {operation}"] + + +def _source_windows(source: Mapping[str, Any]) -> dict[str, Mapping[str, Any]]: + return { + str(window.get("window_id")): window + for window in source.get("windows", []) + if isinstance(window, Mapping) + } + + +def _runtime_con_id(window: Mapping[str, Any]) -> int | None: + runtime = window.get("runtime") + runtime = runtime if isinstance(runtime, Mapping) else {} + value = runtime.get("con_id") + return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else None + + +def _find_con_id(live: Mapping[str, Any], con_id: int) -> Mapping[str, Any] | None: + return next( + ( + window + for window in live.get("windows", []) + if isinstance(window, Mapping) and _runtime_con_id(window) == con_id + ), + None, + ) + + +def _compatible_match( + planner_module: Any, + saved: Mapping[str, Any], + candidate: Mapping[str, Any], +) -> bool: + matches, unmatched = planner_module.match_windows([saved], [candidate]) + saved_id = str(saved.get("window_id")) + evidence = matches.get(saved_id) + return not unmatched and evidence is not None and evidence.get("confidence") != "ambiguous" + + +def _revalidate_runtime( + source_window: Mapping[str, Any], + con_id: int, + *, + collector: Callable[[], dict[str, Any]], + planner_module: Any, +) -> tuple[Mapping[str, Any], dict[str, Any]]: + """Refuser disparition, recyclage d'id, identité divergente ou workspace inconnu.""" + live = collector() + candidate = _find_con_id(live, con_id) + if candidate is None: + raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", f"con_id disparu: {con_id}") + if not _compatible_match(planner_module, source_window, candidate): + raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", f"identité modifiée pour con_id={con_id}") + if not isinstance(candidate.get("workspace"), str) or not candidate.get("workspace"): + raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", f"workspace inconnu pour con_id={con_id}") + return candidate, live + + +def _validate_desktop_entry( + saved: Mapping[str, Any], + action: Mapping[str, Any], + *, + desktop_entries: Sequence[Mapping[str, str | None]], + resolve_desktop_entry: Callable[..., dict[str, Any]], +) -> str: + """Recalculer l'identité exacte contre les sources XDG actuelles.""" + desktop_entry = action.get("desktop_entry") + launch = action.get("launch") + launch = launch if isinstance(launch, Mapping) else {} + classification = saved.get("classification") + classification = classification if isinstance(classification, Mapping) else {} + identity = saved.get("restore_identity") + identity = identity if isinstance(identity, Mapping) else {} + if ( + not isinstance(desktop_entry, str) + or DESKTOP_ENTRY_ID.fullmatch(desktop_entry) is None + or launch.get("backend") != LAUNCH_BACKEND + or classification.get("category") != "user-application" + or identity.get("confidence") != "exact" + or identity.get("desktop_entry") != desktop_entry + ): + raise RestoreExecutionError( + "RESTORE_DESKTOP_ENTRY_REJECTED", + f"identité de lancement non exacte: {action.get('source_window')}", + ) + xwayland = saved.get("xwayland") + xwayland = xwayland if isinstance(xwayland, Mapping) else {} + current = resolve_desktop_entry( + saved.get("app_id"), + xwayland, + saved.get("executable_basename"), + desktop_entries, + ) + if current.get("confidence") != "exact" or current.get("desktop_entry") != desktop_entry: + raise RestoreExecutionError( + "RESTORE_DESKTOP_ENTRY_REJECTED", + f"DesktopEntry absente, ambiguë ou devenue incompatible: {desktop_entry}", + ) + return desktop_entry + + +def _blocking_finding(action: Mapping[str, Any]) -> dict[str, Any] | None: + action_type = action.get("action") + action_id = action.get("action_id") + if action_type not in SUPPORTED_ACTIONS: + return { + "action_id": action_id, + "action": action_type, + "reason": action.get("reason") or "unsupported-action-step19b", + } + if action_type == "reuse-window" and action.get("confidence") == "ambiguous": + return { + "action_id": action_id, + "action": action_type, + "reason": "ambiguous-runtime-match", + } + if action_type == "move-to-workspace": + workspace = action.get("desired_workspace") + if not isinstance(workspace, str) or NUMERIC_WORKSPACE.fullmatch(workspace) is None: + return { + "action_id": action_id, + "action": action_type, + "reason": "unsupported-non-numeric-workspace", + "desired_workspace": workspace, + } + if action_type == "restore-fullscreen" and action.get("desired_fullscreen_mode") not in {0, 1}: + return { + "action_id": action_id, + "action": action_type, + "reason": "unsupported-fullscreen-mode", + "desired_fullscreen_mode": action.get("desired_fullscreen_mode"), + } + return None + + +def capability_gate( + source: Mapping[str, Any], + plan: Mapping[str, Any], + *, + planner_module: Any, + desktop_entries: Sequence[Mapping[str, str | None]], + resolve_desktop_entry: Callable[..., dict[str, Any]], +) -> None: + """Examiner tout le plan avant la première mutation, sans exception partielle.""" + windows = _source_windows(source) + findings: list[dict[str, Any]] = [] + for action in plan.get("actions", []): + if not isinstance(action, Mapping): + findings.append({"action_id": None, "action": None, "reason": "invalid-action"}) + continue + finding = _blocking_finding(action) + if finding is not None: + findings.append(finding) + continue + saved = windows.get(str(action.get("source_window"))) + if saved is None: + findings.append({ + "action_id": action.get("action_id"), + "action": action.get("action"), + "reason": "source-window-missing", + }) + continue + if action.get("action") == "launch-application": + try: + _validate_desktop_entry( + saved, + action, + desktop_entries=desktop_entries, + resolve_desktop_entry=resolve_desktop_entry, + ) + except RestoreExecutionError as exc: + findings.append({ + "action_id": action.get("action_id"), + "action": action.get("action"), + "reason": exc.reason, + }) + if action.get("action") == "reuse-window": + live_window = action.get("live_window") + if not isinstance(live_window, Mapping) or _runtime_con_id(live_window) is None: + findings.append({ + "action_id": action.get("action_id"), + "action": action.get("action"), + "reason": "missing-runtime-con-id", + }) + try: + planner_module.validate_restore_plan(plan) + except Exception as exc: + findings.append({"action_id": None, "action": None, "reason": f"invalid-plan: {exc}"}) + if findings: + raise RestoreExecutionError( + "RESTORE_BLOCKED_UNSUPPORTED_ACTION", + json.dumps(findings, ensure_ascii=False, sort_keys=True), + ) + + +@contextlib.contextmanager +def restore_lock(directory: Path) -> Iterator[None]: + """Détenir un flock exclusif non bloquant sur un fichier privé persistant.""" + directory_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + directory_flags |= os.O_NOFOLLOW + try: + directory_fd = os.open(directory, directory_flags) + except OSError as exc: + raise RestoreExecutionError("RESTORE_LOCK_UNSAFE", f"répertoire sessions non sûr: {exc}") from exc + if stat.S_IMODE(os.fstat(directory_fd).st_mode) != 0o700: + os.close(directory_fd) + raise RestoreExecutionError("RESTORE_LOCK_UNSAFE", "répertoire sessions non privé") + flags = os.O_RDWR | os.O_CREAT | os.O_CLOEXEC + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + lock_fd = -1 + try: + try: + lock_fd = os.open(RESTORE_LOCK_FILENAME, flags, 0o600, dir_fd=directory_fd) + except OSError as exc: + raise RestoreExecutionError("RESTORE_LOCK_UNSAFE", f"restore lock non sûr: {exc}") from exc + info = os.fstat(lock_fd) + if not stat.S_ISREG(info.st_mode): + raise RestoreExecutionError("RESTORE_LOCK_UNSAFE", "restore lock non régulier") + os.fchmod(lock_fd, 0o600) + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + raise RestoreExecutionError("RESTORE_ALREADY_RUNNING", "une restauration est déjà active") from exc + yield + finally: + if lock_fd >= 0: + try: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + finally: + os.close(lock_fd) + os.close(directory_fd) + + +def _parse_sway_success(completed: Any) -> None: + try: + payload = json.loads(completed.stdout) + except (AttributeError, TypeError, json.JSONDecodeError) as exc: + raise RestoreExecutionError("RESTORE_SWAY_COMMAND_FAILED", "réponse swaymsg JSON invalide") from exc + if not isinstance(payload, list) or not payload or not all( + isinstance(item, Mapping) and item.get("success") is True for item in payload + ): + raise RestoreExecutionError("RESTORE_SWAY_COMMAND_FAILED", f"commande refusée: {payload!r}") + + +def _wait_for_launched_window( + saved: Mapping[str, Any], + before_ids: set[int], + *, + collector: Callable[[], dict[str, Any]], + planner_module: Any, + timeout: float, + monotonic: Callable[[], float], + sleeper: Callable[[float], None], +) -> tuple[int, Mapping[str, Any]]: + deadline = monotonic() + timeout + while True: + live = collector() + candidates = [ + window + for window in live.get("windows", []) + if isinstance(window, Mapping) + and _runtime_con_id(window) is not None + and _runtime_con_id(window) not in before_ids + and _compatible_match(planner_module, saved, window) + ] + if candidates: + matches, _unmatched = planner_module.match_windows([saved], candidates) + match = matches.get(str(saved.get("window_id"))) + if match is None: + pass + elif match.get("confidence") == "ambiguous": + raise RestoreExecutionError( + "RESTORE_LAUNCH_AMBIGUOUS", + f"plusieurs nouvelles fenêtres indiscernables: {saved.get('window_id')}", + ) + else: + chosen = candidates[int(match["live_index"])] + con_id = _runtime_con_id(chosen) + if con_id is not None: + return con_id, chosen + if monotonic() >= deadline: + raise RestoreExecutionError( + "RESTORE_LAUNCH_TIMEOUT", + f"aucune nouvelle fenêtre compatible: {saved.get('window_id')}", + ) + sleeper(POLL_INTERVAL_SECONDS) + + +def _action_status(action: Mapping[str, Any]) -> dict[str, Any]: + return { + "action_id": action.get("action_id"), + "action": action.get("action"), + "source_window": action.get("source_window"), + "status": "pending", + "reason": None, + } + + +def _wait_for_postcondition( + saved: Mapping[str, Any], + con_id: int, + predicate: Callable[[Mapping[str, Any]], bool], + failure_reason: str, + *, + collector: Callable[[], dict[str, Any]], + planner_module: Any, + monotonic: Callable[[], float], + sleeper: Callable[[float], None], +) -> Mapping[str, Any]: + """Attendre brièvement la visibilité IPC d'une mutation déjà acceptée.""" + deadline = monotonic() + POSTCONDITION_TIMEOUT_SECONDS + while True: + candidate, _live = _revalidate_runtime( + saved, + con_id, + collector=collector, + planner_module=planner_module, + ) + if predicate(candidate): + return candidate + if monotonic() >= deadline: + raise RestoreExecutionError("RESTORE_POSTCONDITION_FAILED", failure_reason) + sleeper(POLL_INTERVAL_SECONDS) + + +def execute_fresh_plan( + source: Mapping[str, Any], + plan: Mapping[str, Any], + *, + collector: Callable[[], dict[str, Any]], + planner_module: Any, + desktop_entries: Sequence[Mapping[str, str | None]], + resolve_desktop_entry: Callable[..., dict[str, Any]], + runner: Callable[[Sequence[str], float], Any] = _default_runner, + timeout: float = DEFAULT_WINDOW_TIMEOUT_SECONDS, + monotonic: Callable[[], float] = time.monotonic, + sleeper: Callable[[float], None] = time.sleep, +) -> dict[str, Any]: + """Appliquer séquentiellement un plan déjà préflighté et arrêter au premier écart.""" + capability_gate( + source, + plan, + planner_module=planner_module, + desktop_entries=desktop_entries, + resolve_desktop_entry=resolve_desktop_entry, + ) + windows = _source_windows(source) + statuses = [_action_status(action) for action in plan["actions"]] + runtime_by_source: dict[str, int] = {} + initial_ids = { + con_id + for window in plan.get("live", {}).get("windows", []) + if isinstance(window, Mapping) + for con_id in [_runtime_con_id(window)] + if con_id is not None + } + applications_launched = 0 + mutating_commands = 0 + stopped_reason: str | None = None + + for action, status in zip(plan["actions"], statuses): + action_type = action["action"] + source_id = str(action["source_window"]) + saved = windows[source_id] + try: + if action_type == "reuse-window": + con_id = _runtime_con_id(action["live_window"]) + assert con_id is not None + _revalidate_runtime(saved, con_id, collector=collector, planner_module=planner_module) + runtime_by_source[source_id] = con_id + status.update(status="success", reason="fenêtre live réutilisée et revalidée") + elif action_type == "skip-autostart-managed": + status.update(status="skipped", reason="lancement possédé par sway-autostart") + elif action_type == "launch-application": + desktop_entry = _validate_desktop_entry( + saved, + action, + desktop_entries=desktop_entries, + resolve_desktop_entry=resolve_desktop_entry, + ) + # CONTRACT: l'inventaire de référence est pris immédiatement + # avant le lancement, et non réutilisé depuis le plan. Toute + # apparition/disparition imprévue depuis le plan force l'arrêt. + before_launch = collector() + before_launch_ids = { + con_id + for window in before_launch.get("windows", []) + if isinstance(window, Mapping) + for con_id in [_runtime_con_id(window)] + if con_id is not None + } + expected_ids = initial_ids | set(runtime_by_source.values()) + if before_launch_ids != expected_ids: + raise RestoreExecutionError( + "RESTORE_RUNTIME_DIVERGED", + "inventaire con_id modifié avant lancement", + ) + completed = runner(_direct_uwsm_command(desktop_entry), timeout) + if getattr(completed, "returncode", 0) not in {None, 0}: + raise RestoreExecutionError("RESTORE_LAUNCH_FAILED", f"uwsm a retourné {completed.returncode}") + applications_launched += 1 + con_id, _window = _wait_for_launched_window( + saved, + before_launch_ids, + collector=collector, + planner_module=planner_module, + timeout=timeout, + monotonic=monotonic, + sleeper=sleeper, + ) + runtime_by_source[source_id] = con_id + status.update(status="success", reason=f"nouvelle fenêtre compatible con_id={con_id}") + else: + con_id = runtime_by_source.get(source_id) + if con_id is None: + raise RestoreExecutionError( + "RESTORE_RUNTIME_DIVERGED", + f"aucun con_id validé pour {source_id}", + ) + _revalidate_runtime(saved, con_id, collector=collector, planner_module=planner_module) + if action_type == "move-to-workspace": + workspace = str(int(action["desired_workspace"])) + operation = f"move container to workspace number {workspace}" + elif action_type == "restore-floating": + operation = "floating enable" if action["desired_floating"] else "floating disable" + elif action_type == "restore-fullscreen": + operation = "fullscreen enable" if action["desired_fullscreen_mode"] == 1 else "fullscreen disable" + else: + operation = "focus" + completed = runner(_direct_sway_command(con_id, operation), 5.0) + _parse_sway_success(completed) + mutating_commands += 1 + if action_type == "move-to-workspace": + predicate = lambda item: item.get("workspace") == workspace + failure_reason = "workspace cible non atteint" + elif action_type == "restore-floating": + predicate = lambda item: bool(item.get("floating")) == bool(action["desired_floating"]) + failure_reason = "état floating non atteint" + elif action_type == "restore-fullscreen": + predicate = lambda item: int(item.get("fullscreen_mode") or 0) == action["desired_fullscreen_mode"] + failure_reason = "état fullscreen non atteint" + else: + predicate = lambda item: bool(item.get("focused")) + failure_reason = "focus cible non atteint" + _wait_for_postcondition( + saved, + con_id, + predicate, + failure_reason, + collector=collector, + planner_module=planner_module, + monotonic=monotonic, + sleeper=sleeper, + ) + status.update(status="success", reason="postcondition Sway confirmée") + except (OSError, subprocess.SubprocessError, RestoreExecutionError, AssertionError) as exc: + if isinstance(exc, RestoreExecutionError): + stopped_reason = str(exc) + else: + stopped_reason = f"RESTORE_EXTERNAL_COMMAND_FAILED: {exc}" + status.update(status="failed", reason=stopped_reason) + break + + return { + "actions": statuses, + "applications_launched": applications_launched, + "mutating_commands_executed": mutating_commands, + "status": "failed" if stopped_reason else "success", + "reason": stopped_reason, + "runtime_windows": dict(sorted(runtime_by_source.items())), + } + + +def apply_session( + name: str, + autostart_path: Path | None, + *, + sessions_path: Path, + snapshot_loader: Callable[[str], dict[str, Any]], + live_collector: Callable[[Path | None], dict[str, Any]], + planner_module: Any, + desktop_loader: Callable[[], Sequence[Mapping[str, str | None]]], + resolve_desktop_entry: Callable[..., dict[str, Any]], + runner: Callable[[Sequence[str], float], Any] = _default_runner, + timeout: float = DEFAULT_WINDOW_TIMEOUT_SECONDS, + monotonic: Callable[[], float] = time.monotonic, + sleeper: Callable[[float], None] = time.sleep, +) -> dict[str, Any]: + """Lock, snapshot frais, live frais, plan frais, gate puis exécution contrôlée.""" + report: dict[str, Any] = { + "schema": REPORT_SCHEMA, + "version": REPORT_VERSION, + "source_session": name, + "status": "blocked", + "reason": None, + "snapshot_hash_before": None, + "snapshot_hash_after": None, + "snapshot_unchanged": False, + "plan": None, + "actions": [], + "applications_launched": 0, + "mutating_commands_executed": 0, + "rollback": "not-transactional-stop-on-first-critical-divergence", + } + try: + with restore_lock(sessions_path): + source = snapshot_loader(name) + report["snapshot_hash_before"] = snapshot_digest(source) + live = live_collector(autostart_path) + plan = planner_module.build_restore_plan(source, live) + report["plan"] = copy.deepcopy(plan) + result = execute_fresh_plan( + source, + plan, + collector=lambda: live_collector(autostart_path), + planner_module=planner_module, + desktop_entries=desktop_loader(), + resolve_desktop_entry=resolve_desktop_entry, + runner=runner, + timeout=timeout, + monotonic=monotonic, + sleeper=sleeper, + ) + report.update(result) + after = snapshot_loader(name) + report["snapshot_hash_after"] = snapshot_digest(after) + report["snapshot_unchanged"] = report["snapshot_hash_after"] == report["snapshot_hash_before"] + if not report["snapshot_unchanged"]: + report["status"] = "failed" + report["reason"] = "RESTORE_SNAPSHOT_CHANGED: snapshot source modifié pendant apply" + except RestoreExecutionError as exc: + report["status"] = "blocked" + report["reason"] = str(exc) + if isinstance(report.get("plan"), Mapping): + report["actions"] = [] + for action in report["plan"].get("actions", []): + status = _action_status(action) + finding = _blocking_finding(action) + if finding is not None: + status.update(status="blocked", reason=finding["reason"]) + report["actions"].append(status) + if report["snapshot_hash_before"] is not None: + after = snapshot_loader(name) + report["snapshot_hash_after"] = snapshot_digest(after) + report["snapshot_unchanged"] = report["snapshot_hash_after"] == report["snapshot_hash_before"] + return report diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_restore.py b/quickshell/.config/quickshell/labfy-sway/session/session_restore.py index e2e0c4936..c79682ead 100644 --- a/quickshell/.config/quickshell/labfy-sway/session/session_restore.py +++ b/quickshell/.config/quickshell/labfy-sway/session/session_restore.py @@ -498,7 +498,14 @@ def build_restore_plan(source: Mapping[str, Any], live: Mapping[str, Any]) -> di desired_placement = _placement(source, saved) current_placement = _placement(live, live_window) if live_window else None - if current_placement != desired_placement: + # WHY: une fenêtre absente n'a aucune position live à comparer. Émettre + # immédiatement restore-tree-position bloquerait tout lancement STEP19B, + # même pour une simple racine de workspace. STEP19C pourra replanifier + # la topologie une fois la nouvelle fenêtre réellement observable. + # CONTRACT: le planner ne demande une reconstruction d'arbre que pour + # une fenêtre déjà matchée dont la structure diffère effectivement. + # INVARIANT: l'absence n'est jamais assimilée à une structure divergente. + if live_window is not None and current_placement != desired_placement: add(_draft( f"tree:{saved_id}", "restore-tree-position", "tree-layout", saved, depends=(base,), capability="partially-supported", diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py b/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py index 3517823e4..424952998 100755 --- a/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py +++ b/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py @@ -1133,6 +1133,62 @@ def _load_restore_planner() -> Callable[[Mapping[str, Any], Mapping[str, Any]], return planner +def _load_restore_modules() -> tuple[Any, Any]: + """Charger explicitement planner pur et frontière d'effets STEP19B.""" + planner_path = Path(__file__).with_name("session_restore.py") + executor_path = Path(__file__).with_name("session_executor.py") + + def load(name: str, path: Path) -> Any: + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise SnapshotError(f"module de restauration indisponible: {path.name}") + module = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(module) + except (OSError, ImportError, SyntaxError) as exc: + raise SnapshotError(f"chargement de {path.name} impossible: {exc}") from exc + return module + + return load("labfy_session_restore_apply", planner_path), load("labfy_session_executor", executor_path) + + +def apply_session( + name: str, + autostart_path: Path | None = None, + *, + snapshot_loader: Callable[[str], dict[str, Any]] | None = None, + live_collector: Callable[[Path | None], dict[str, Any]] | None = None, + sessions_path: Path | None = None, + planner_module: Any | None = None, + executor_module: Any | None = None, + desktop_loader: Callable[[], Sequence[Mapping[str, str | None]]] | None = None, + runner: Callable[[Sequence[str], float], Any] | None = None, + timeout: float = 10.0, +) -> dict[str, Any]: + """Construire puis exécuter un plan frais sous le verrou STEP19B.""" + # WHY: un plan persistant pourrait viser des con_id recyclés. L'API apply + # n'accepte donc qu'un nom de snapshot et recollecte toujours Sway. + # CONTRACT: l'exécuteur détient le lock pendant load→plan→execute→rehash. + # INVARIANT: aucun plan fourni par l'appelant n'est une autorité exécutable. + normalized = normalize_session_name(name) + if planner_module is None or executor_module is None: + loaded_planner, loaded_executor = _load_restore_modules() + planner_module = loaded_planner if planner_module is None else planner_module + executor_module = loaded_executor if executor_module is None else executor_module + kwargs: dict[str, Any] = { + "sessions_path": sessions_directory() if sessions_path is None else sessions_path, + "snapshot_loader": show_session if snapshot_loader is None else snapshot_loader, + "live_collector": collect if live_collector is None else live_collector, + "planner_module": planner_module, + "desktop_loader": load_desktop_entries if desktop_loader is None else desktop_loader, + "resolve_desktop_entry": resolve_desktop_entry, + "timeout": timeout, + } + if runner is not None: + kwargs["runner"] = runner + return executor_module.apply_session(normalized, autostart_path, **kwargs) + + def plan_session( name: str, autostart_path: Path | None = None, @@ -1161,26 +1217,36 @@ def plan_session( def main(argv: Sequence[str] | None = None) -> int: - parser = argparse.ArgumentParser(description="Snapshots persistants et restore planner read-only pour Sway") + parser = argparse.ArgumentParser(description="Snapshots persistants et restauration contrôlée pour Sway") parser.add_argument("--stdout", action="store_true", help="émettre le JSON validé sur stdout") parser.add_argument("--compact", action="store_true", help="émettre un JSON compact") + parser.add_argument( + "--execute", + action="store_true", + help="autoriser explicitement les mutations, uniquement avec apply", + ) parser.add_argument( "--autostart", type=Path, default=Path.home() / ".config/sway/autostart", help="fichier Sway autostart utilisé seulement pour la classification", ) - parser.add_argument("command", nargs="?", choices=("save", "list", "show", "delete", "plan")) + parser.add_argument("command", nargs="?", choices=("save", "list", "show", "delete", "plan", "apply")) parser.add_argument("name", nargs="?") args = parser.parse_args(argv) if args.stdout and args.command is not None: parser.error("--stdout ne peut pas être combiné avec une commande persistante") if not args.stdout and args.command is None: - parser.error("utiliser --stdout ou une commande save/list/show/delete/plan") - if args.command in {"save", "show", "delete", "plan"} and args.name is None: + parser.error("utiliser --stdout ou une commande save/list/show/delete/plan/apply") + if args.command in {"save", "show", "delete", "plan", "apply"} and args.name is None: parser.error(f"{args.command} requiert un nom de session") if (args.stdout or args.command == "list") and args.name is not None: parser.error("nom de session inattendu") + if args.execute and args.command != "apply": + parser.error("--execute est réservé à la commande apply") + if args.command == "apply" and not args.execute: + print("RESTORE_EXECUTION_REQUIRES_EXPLICIT_EXECUTE: ajouter --execute", file=sys.stderr) + return 1 try: if args.stdout: started = time.monotonic() @@ -1204,6 +1270,8 @@ def main(argv: Sequence[str] | None = None) -> int: result = show_session(args.name) elif args.command == "plan": result = plan_session(args.name, args.autostart) + elif args.command == "apply": + result = apply_session(args.name, args.autostart) else: delete_session(args.name) result = {"deleted": args.name} @@ -1215,6 +1283,8 @@ def main(argv: Sequence[str] | None = None) -> int: else: json.dump(result, sys.stdout, ensure_ascii=False, sort_keys=True, indent=2) sys.stdout.write("\n") + if args.command == "apply" and result.get("status") != "success": + return 1 return 0 diff --git a/tests/test_session_executor.py b/tests/test_session_executor.py new file mode 100644 index 000000000..e556b3b5b --- /dev/null +++ b/tests/test_session_executor.py @@ -0,0 +1,479 @@ +"""STEP19B : exécuteur contrôlé, tous les effets externes sont simulés.""" + +from __future__ import annotations + +import copy +import importlib.util +import io +import json +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest import mock + +try: + from tests.test_session_restore import session, window +except ModuleNotFoundError: # unittest discovery ajoute directement tests/ à sys.path. + from test_session_restore import session, window + + +ROOT = Path(__file__).resolve().parents[1] +SESSION_DIR = ROOT / "quickshell/.config/quickshell/labfy-sway/session" + + +def load_module(name, filename): + spec = importlib.util.spec_from_file_location(name, SESSION_DIR / filename) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +planner = load_module("session_restore_executor_test", "session_restore.py") +executor = load_module("session_executor_test", "session_executor.py") +snapshot_backend = load_module("session_snapshot_executor_test", "session_snapshot.py") + + +DESKTOP_ENTRIES = [{ + "desktop_entry": "kitty.desktop", + "stem": "kitty", + "startup_wm_class": "kitty", + "name": "kitty", +}] + + +def kitty_saved(**kwargs): + defaults = {"workspace": "3", "con_id": 10, "pid": 1000} + defaults.update(kwargs) + return window("w1", "kitty", desktop_entry="kitty.desktop", executable="kitty", **defaults) + + +class FakeWorld: + def __init__(self, windows=(), *, launch_window=None): + self.windows = [copy.deepcopy(item) for item in windows] + self.launch_window = copy.deepcopy(launch_window) + self.commands = [] + + def collect(self): + focused = next((item["window_id"] for item in self.windows if item.get("focused")), None) + return session(self.windows, focused=focused) + + def run(self, argv, timeout): + self.commands.append(list(argv)) + if argv[:4] == ["uwsm", "app", "--", "kitty.desktop"]: + if self.launch_window is not None: + self.windows.append(copy.deepcopy(self.launch_window)) + self.launch_window = None + return SimpleNamespace(returncode=0, stdout="", stderr="") + command = argv[-1] + prefix, operation = command.split("] ", 1) + con_id = int(prefix.removeprefix("[con_id=")) + target = next(item for item in self.windows if item["runtime"]["con_id"] == con_id) + if operation.startswith("move container to workspace number "): + target["workspace"] = operation.rsplit(" ", 1)[1] + elif operation == "floating enable": + target["floating"] = True + elif operation == "floating disable": + target["floating"] = False + elif operation == "fullscreen enable": + target["fullscreen_mode"] = 1 + elif operation == "fullscreen disable": + target["fullscreen_mode"] = 0 + elif operation == "focus": + for item in self.windows: + item["focused"] = item is target + return SimpleNamespace(returncode=0, stdout='[{"success":true}]', stderr="") + + +def execute(source, world, **kwargs): + plan = planner.build_restore_plan(source, world.collect()) + return executor.execute_fresh_plan( + source, + plan, + collector=world.collect, + planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + timeout=0.01, + sleeper=lambda _delay: None, + **kwargs, + ), plan + + +class SessionExecutorTest(unittest.TestCase): + def test_apply_requires_explicit_execute(self): + stderr = io.StringIO() + with mock.patch.object(snapshot_backend, "apply_session") as apply, mock.patch("sys.stderr", stderr): + self.assertEqual(snapshot_backend.main(("apply", "dev")), 1) + apply.assert_not_called() + self.assertIn("REQUIRES_EXPLICIT_EXECUTE", stderr.getvalue()) + + def test_apply_execute_reaches_executor(self): + report = {"schema": executor.REPORT_SCHEMA, "status": "success"} + output = io.StringIO() + with mock.patch.object(snapshot_backend, "apply_session", return_value=report) as apply, mock.patch("sys.stdout", output): + self.assertEqual(snapshot_backend.main(("apply", "dev", "--execute")), 0) + apply.assert_called_once() + self.assertEqual(json.loads(output.getvalue()), report) + + def test_plan_blocking_happens_before_mutation(self): + saved = kitty_saved(parent="c1") + live = copy.deepcopy(saved) + live["tree_position"]["parent_container_id"] = None + container = { + "container_id": "c1", "parent_container_id": None, + "tree_position": {"branch": "tiling", "index": 0}, + "layout": "splitv", "orientation": "vertical", "percent": 1.0, + } + source = session([saved], persistent=True, containers=[container]) + world = FakeWorld([live]) + plan = planner.build_restore_plan(source, world.collect()) + with self.assertRaisesRegex(executor.RestoreExecutionError, "RESTORE_BLOCKED_UNSUPPORTED_ACTION"): + executor.execute_fresh_plan( + source, plan, collector=world.collect, planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + ) + self.assertEqual(world.commands, []) + + def test_command_reconstruction_ignores_arbitrary_argv(self): + saved = kitty_saved() + source = session([saved], persistent=True) + world = FakeWorld([], launch_window=kitty_saved(con_id=22, pid=2000, workspace="3")) + plan = planner.build_restore_plan(source, world.collect()) + launch = next(item for item in plan["actions"] if item["action"] == "launch-application") + launch["launch"]["argv"] = ["sh", "-c", "touch /tmp/forbidden"] + result = executor.execute_fresh_plan( + source, plan, collector=world.collect, planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, timeout=0.01, sleeper=lambda _delay: None, + ) + self.assertEqual(result["status"], "success") + self.assertEqual(world.commands[0], ["uwsm", "app", "--", "kitty.desktop"]) + + def test_default_uwsm_launch_is_direct_and_nonblocking(self): + process = SimpleNamespace(returncode=None) + with mock.patch.object(executor.subprocess, "Popen", return_value=process) as popen, mock.patch.object( + executor.subprocess, "run" + ) as run: + self.assertIs( + executor._default_runner(["uwsm", "app", "--", "kitty.desktop"], 10.0), + process, + ) + popen.assert_called_once_with( + ["uwsm", "app", "--", "kitty.desktop"], + stdin=executor.subprocess.DEVNULL, + stdout=executor.subprocess.DEVNULL, + stderr=executor.subprocess.DEVNULL, + close_fds=True, + start_new_session=True, + ) + run.assert_not_called() + + def test_desktop_entry_is_revalidated(self): + source = session([kitty_saved()], persistent=True) + world = FakeWorld([]) + plan = planner.build_restore_plan(source, world.collect()) + with self.assertRaisesRegex(executor.RestoreExecutionError, "DesktopEntry absente"): + executor.execute_fresh_plan( + source, plan, collector=world.collect, planner_module=planner, + desktop_entries=[], resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + ) + self.assertEqual(world.commands, []) + + def test_restore_lock_is_private_and_exclusive(self): + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary) + path.chmod(0o700) + with executor.restore_lock(path): + self.assertEqual((path / ".restore.lock").stat().st_mode & 0o777, 0o600) + with self.assertRaisesRegex(executor.RestoreExecutionError, "RESTORE_ALREADY_RUNNING"): + with executor.restore_lock(path): + self.fail("second lock unexpectedly acquired") + + def test_restore_lock_rejects_non_private_directory(self): + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary) + path.chmod(0o755) + with self.assertRaisesRegex(executor.RestoreExecutionError, "répertoire sessions non privé"): + with executor.restore_lock(path): + self.fail("unsafe directory unexpectedly accepted") + + def test_ambiguous_reuse_blocks_before_mutation(self): + saved = kitty_saved() + source = session([saved], persistent=True) + world = FakeWorld([ + kitty_saved(con_id=81, pid=8001), + kitty_saved(con_id=82, pid=8002), + ]) + plan = planner.build_restore_plan(source, world.collect()) + self.assertEqual(plan["actions"][0]["confidence"], "ambiguous") + with self.assertRaisesRegex(executor.RestoreExecutionError, "ambiguous-runtime-match"): + executor.execute_fresh_plan( + source, plan, collector=world.collect, planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + ) + self.assertEqual(world.commands, []) + + def test_reuse_is_validated_noop(self): + saved = kitty_saved(con_id=10) + source = session([saved], persistent=True) + world = FakeWorld([copy.deepcopy(saved)]) + result, _plan = execute(source, world) + self.assertEqual(result["actions"][0]["status"], "success") + self.assertEqual(world.commands, []) + + def test_autostart_skip_is_validated_noop(self): + saved = kitty_saved(category="autostart-managed-application", managed_by="sway-autostart") + source = session([saved], persistent=True) + world = FakeWorld([]) + result, _plan = execute(source, world) + self.assertEqual(result["actions"][0]["status"], "skipped") + self.assertEqual(world.commands, []) + + def test_launch_waits_for_new_matching_window(self): + source = session([kitty_saved()], persistent=True) + launched = kitty_saved(con_id=55, pid=5000, workspace="3") + world = FakeWorld([], launch_window=launched) + result, _plan = execute(source, world) + self.assertEqual(result["status"], "success") + self.assertEqual(result["runtime_windows"], {"w1": 55}) + + def test_launch_timeout(self): + source = session([kitty_saved()], persistent=True) + world = FakeWorld([]) + ticks = iter((0.0, 0.0, 1.0)) + result, _plan = execute(source, world, monotonic=lambda: next(ticks)) + self.assertEqual(result["status"], "failed") + self.assertIn("RESTORE_LAUNCH_TIMEOUT", result["reason"]) + + def test_launch_aborts_if_inventory_drifted_before_uwsm(self): + source = session([kitty_saved()], persistent=True) + initial = session([]) + plan = planner.build_restore_plan(source, initial) + intruder = window("w9", "other", desktop_entry="other.desktop", con_id=99, pid=9000) + commands = [] + result = executor.execute_fresh_plan( + source, plan, collector=lambda: session([intruder]), planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=lambda argv, _timeout: commands.append(list(argv)), + ) + self.assertEqual(result["status"], "failed") + self.assertIn("inventaire con_id modifié avant lancement", result["reason"]) + self.assertEqual(commands, []) + + def test_launch_ambiguous(self): + source = session([kitty_saved()], persistent=True) + first = kitty_saved(con_id=51, pid=5001, workspace="3") + second = kitty_saved(con_id=52, pid=5002, workspace="3") + world = FakeWorld([]) + + def run(argv, timeout): + world.commands.append(list(argv)) + world.windows.extend([first, second]) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + world.run = run + result, _plan = execute(source, world) + self.assertEqual(result["status"], "failed") + self.assertIn("RESTORE_LAUNCH_AMBIGUOUS", result["reason"]) + + def test_move_is_targeted_and_confirmed(self): + saved = kitty_saved(workspace="3", con_id=10) + live = kitty_saved(workspace="2", con_id=44) + source = session([saved], persistent=True) + world = FakeWorld([live]) + result, _plan = execute(source, world) + self.assertEqual(result["status"], "success") + self.assertIn("[con_id=44] move container to workspace number 3", world.commands[-1]) + + def test_floating_enable_and_disable(self): + for desired, current, expected in ((True, False, "floating enable"), (False, True, "floating disable")): + with self.subTest(desired=desired): + source = session([kitty_saved(floating=desired)], persistent=True) + world = FakeWorld([kitty_saved(floating=current, con_id=88)]) + result, _plan = execute(source, world) + self.assertEqual(result["status"], "success") + self.assertTrue(any(expected in command[-1] for command in world.commands)) + + def test_fullscreen_workspace_mode_and_disabled(self): + for desired, current, expected in ((1, 0, "fullscreen enable"), (0, 1, "fullscreen disable")): + with self.subTest(desired=desired): + source = session([kitty_saved(fullscreen=desired)], persistent=True) + world = FakeWorld([kitty_saved(fullscreen=current, con_id=89)]) + result, _plan = execute(source, world) + self.assertEqual(result["status"], "success") + self.assertTrue(any(expected in command[-1] for command in world.commands)) + + def test_unsupported_fullscreen_blocks_before_mutation(self): + source = session([kitty_saved(fullscreen=2)], persistent=True) + world = FakeWorld([kitty_saved(fullscreen=0, con_id=90)]) + plan = planner.build_restore_plan(source, world.collect()) + with self.assertRaisesRegex(executor.RestoreExecutionError, "unsupported-fullscreen-mode"): + executor.execute_fresh_plan( + source, plan, collector=world.collect, planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + ) + self.assertEqual(world.commands, []) + + def test_focus_is_last(self): + saved = kitty_saved(workspace="3", focused=True) + source = session([saved], persistent=True, focused="w1") + world = FakeWorld([kitty_saved(workspace="2", con_id=91, focused=False)]) + result, plan = execute(source, world) + self.assertEqual(plan["actions"][-1]["action"], "restore-focus") + self.assertTrue(world.commands[-1][-1].endswith(" focus")) + self.assertEqual(result["status"], "success") + + def test_postcondition_polling_accepts_delayed_focus_visibility(self): + saved = kitty_saved(con_id=92) + states = [kitty_saved(con_id=92, focused=False), kitty_saved(con_id=92, focused=True)] + sleeps = [] + + def collect(): + return session([states.pop(0) if len(states) > 1 else states[0]]) + + result = executor._wait_for_postcondition( + saved, + 92, + lambda item: bool(item.get("focused")), + "focus cible non atteint", + collector=collect, + planner_module=planner, + monotonic=lambda: 0.0, + sleeper=sleeps.append, + ) + self.assertTrue(result["focused"]) + self.assertEqual(sleeps, [executor.POLL_INTERVAL_SECONDS]) + + def test_runtime_con_id_disappeared(self): + saved = kitty_saved() + source = session([saved], persistent=True) + initial = session([copy.deepcopy(saved)]) + plan = planner.build_restore_plan(source, initial) + result = executor.execute_fresh_plan( + source, plan, collector=lambda: session([]), planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=lambda *_args: self.fail("mutation must not run"), + ) + self.assertIn("con_id disparu", result["reason"]) + + def test_runtime_identity_changed(self): + saved = kitty_saved() + source = session([saved], persistent=True) + plan = planner.build_restore_plan(source, session([copy.deepcopy(saved)])) + changed = window("w9", "evil", desktop_entry="evil.desktop", con_id=10, pid=999) + result = executor.execute_fresh_plan( + source, plan, collector=lambda: session([changed]), planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=lambda *_args: self.fail("mutation must not run"), + ) + self.assertIn("identité modifiée", result["reason"]) + + def test_snapshot_hash_is_unchanged_and_plan_is_fresh(self): + saved = kitty_saved() + source = session([saved], persistent=True, name="fresh") + world = FakeWorld([copy.deepcopy(saved)]) + loads = [] + + def loader(name): + loads.append(name) + return copy.deepcopy(source) + + with tempfile.TemporaryDirectory() as temporary: + report = executor.apply_session( + "fresh", None, sessions_path=Path(temporary), snapshot_loader=loader, + live_collector=lambda _path: world.collect(), planner_module=planner, + desktop_loader=lambda: DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + ) + self.assertEqual(loads, ["fresh", "fresh"]) + self.assertTrue(report["snapshot_unchanged"]) + self.assertEqual(report["snapshot_hash_before"], report["snapshot_hash_after"]) + self.assertEqual(report["plan"]["live"]["windows"][0]["runtime"]["con_id"], 10) + + def test_unsupported_and_manual_actions_abort(self): + for action_type in ("restore-tree-position", "restore-scratchpad-hidden", "restore-scratchpad-visible", "manual-required", "unknown-action"): + with self.subTest(action=action_type): + source = session([kitty_saved()], persistent=True) + plan = planner.build_restore_plan(source, session([kitty_saved()])) + injected = copy.deepcopy(plan["actions"][0]) + injected.update(action_id="a999", action=action_type, phase="tree-layout", depends_on=[]) + plan["actions"] = [injected] + world = FakeWorld([kitty_saved()]) + with self.assertRaisesRegex(executor.RestoreExecutionError, "RESTORE_BLOCKED_UNSUPPORTED_ACTION"): + executor.execute_fresh_plan( + source, plan, collector=world.collect, planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + ) + self.assertEqual(world.commands, []) + + def test_malicious_workspace_is_never_a_command(self): + source = session([kitty_saved(workspace="3")], persistent=True) + world = FakeWorld([kitty_saved(workspace="2", con_id=77)]) + plan = planner.build_restore_plan(source, world.collect()) + move = next(item for item in plan["actions"] if item["action"] == "move-to-workspace") + move["desired_workspace"] = "3; exec touch /tmp/forbidden" + with self.assertRaisesRegex(executor.RestoreExecutionError, "unsupported-non-numeric-workspace"): + executor.execute_fresh_plan( + source, plan, collector=world.collect, planner_module=planner, + desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + ) + self.assertEqual(world.commands, []) + + def test_malicious_desktop_entry_is_blocked_before_command(self): + saved = kitty_saved() + saved["restore_identity"]["desktop_entry"] = "kitty.desktop;touch" + source = session([saved], persistent=True) + world = FakeWorld([]) + plan = planner.build_restore_plan(source, world.collect()) + with self.assertRaisesRegex(executor.RestoreExecutionError, "identité de lancement non exacte"): + executor.execute_fresh_plan( + source, plan, collector=world.collect, planner_module=planner, + desktop_entries=[{ + "desktop_entry": "kitty.desktop;touch", "stem": "kitty", + "startup_wm_class": "kitty", "name": "hostile", + }], + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, + runner=world.run, + ) + self.assertEqual(world.commands, []) + + def test_title_and_app_id_are_not_interpolated_into_commands(self): + malicious = kitty_saved(workspace="3") + malicious["title_hint"] = "] exec touch /tmp/forbidden" + source = session([malicious], persistent=True) + world = FakeWorld([kitty_saved(workspace="2", con_id=73)]) + result, _plan = execute(source, world) + self.assertEqual(result["status"], "success") + serialized = json.dumps(world.commands) + self.assertNotIn("touch", serialized) + self.assertNotIn("title", serialized) + + def test_source_contains_no_shell_execution_primitive(self): + source = (SESSION_DIR / "session_executor.py").read_text(encoding="utf-8") + self.assertNotIn("shell=True", source) + self.assertNotIn("os.system", source) + self.assertNotIn("eval(", source) + self.assertNotIn("exec(", source) + self.assertNotIn("shell=True", source) + + +if __name__ == "__main__": + unittest.main()