diff --git a/docs/session-restore-v1.md b/docs/session-restore-v1.md new file mode 100644 index 000000000..6be6d3c05 --- /dev/null +++ b/docs/session-restore-v1.md @@ -0,0 +1,285 @@ +# Restore Plan V1 — STEP19A + +## Statut et architecture + +**STEP19A DOES NOT RESTORE.** + +STEP19A charge un Session Snapshot V1 persistant, observe la session Sway +actuelle avec les quatre lectures STEP18, puis produit un plan JSON. Il ne +lance ni ne tue aucun processus, ne change aucun workspace, ne déplace aucune +fenêtre et n'envoie aucune commande Sway mutatrice. L'exécuteur appartient à +STEP19B ; le Spatial Canvas appartient à STEP20. + +```text +snapshot persistant validé + Session Snapshot V1 live validé + ↓ +session_restore.py (fonction pure) + ↓ +labfy.sway.restore-plan / version 1 +``` + +Le point d'entrée reste : + +```bash +session_snapshot.py plan +``` + +`session_snapshot.py` possède le chargement sûr, le verrou partagé, la +validation persistante et la collecte IPC. `session_restore.py` ne possède ni +I/O fichier, ni subprocess, ni API Sway. Il reçoit deux documents normalisés +et réutilise donc sans duplication la classification, la résolution +DesktopEntry et l'inventaire live de STEP18. + +## Enveloppe Restore Plan V1 + +```json +{ + "schema": "labfy.sway.restore-plan", + "version": 1, + "read_only": true, + "execution_supported": false, + "source_session": "dev", + "source": {}, + "live": {}, + "policy": {}, + "matches": [], + "unmatched_source_windows": [], + "actions": [], + "diagnostics": {} +} +``` + +Le planner n'ajoute pas d'heure courante. Les seuls timestamps viennent de ses +entrées. Un même couple snapshot/live produit donc le même plan. + +Chaque action contient au minimum : + +```text +action_id +action +phase +source_window +snapshot_identity +depends_on[] +``` + +`depends_on` exprime les chaînes futures, par exemple lancement, placement, +arbre puis focus. Il n'exécute pas ces opérations. + +## Inventaire live et identité runtime + +`live.windows[]` conserve : + +```text +runtime.con_id, runtime.pid, workspace, output +app_id, class, instance, desktop_entry, executable_basename +classification.category, classification.managed_by +floating, fullscreen_mode, scratchpad, focused, title_hint +``` + +Les PID et `con_id` sont exposés comme observations pour un futur exécuteur +dans la session courante, mais `policy.runtime_identity_fields` est vide. Ils +ne participent jamais au matching persistant : ils changent, et peuvent même +être réutilisés numériquement, après logout ou reboot. + +## Matching one-to-one + +Une arête snapshot→live n'existe que si une preuve d'identité applicative +existe. Les preuves, par poids décroissant, sont : + +1. même DesktopEntry résolue ; +2. même `app_id` ; +3. même classe XWayland/`StartupWMClass` ; +4. même instance ; +5. même basename exécutable compatible. + +Deux `app_id` explicites et différentes interdisent un match fondé seulement +sur un basename. La résolution DesktopEntry n'est pas réimplémentée : elle +provient de `resolve_desktop_entry()` STEP18 pour le snapshot comme pour le +live. + +Le workspace identique, le même état floating et un `title_hint` identique +servent seulement de départage. Un titre ne crée jamais un candidat et n'est +jamais une identité obligatoire. + +Le planner trie globalement toutes les arêtes par preuve, puis applique un +matching one-to-one. Une fenêtre live ne peut satisfaire qu'une fenêtre +sauvegardée. Pour deux Firefox sauvegardées et une live, une seule est +réutilisée ; l'autre reste absente. L'ordre snapshot ne sert qu'au dernier +départage. + +## Confiance + +```text +exact DesktopEntry identique + app_id ou classe identique +high DesktopEntry, app_id ou classe exacte sans preuve combinée +medium instance ou basename compatible +ambiguous plusieurs arêtes de meilleur score subsistent +unresolved aucune preuve suffisante +``` + +`exact` n'est jamais déduit du PID, du `con_id`, du titre seul ou de la seule +position dans l'arbre. + +## Classification et lancement futur + +### Applications utilisateur + +Une application absente n'est proposée au lancement que si sa DesktopEntry a +été résolue avec confiance STEP18 `exact`. Une identité heuristique ou +absente produit `manual-required` ; aucune ligne de commande n'est inventée. + +Le futur backend STEP19B est figé conceptuellement comme : + +```json +{ + "backend": "uwsm-app-desktop-entry", + "argv": ["uwsm", "app", "--", "firefox.desktop"], + "execute_in_step19a": false +} +``` + +L'audit de `uwsm 0.27.0` confirme que `uwsm app` accepte un exécutable, un +identifiant DesktopEntry ou son chemin, et lance l'application comme scope ou +service dans une slice graphique. Sa documentation recommande ce mécanisme +pour ne pas accumuler les applications dans l'unité du compositeur. STEP19B +devra donc préférer l'identifiant DesktopEntry via `uwsm app` à `gtk-launch`, +`gio launch` ou un enfant direct de Sway. STEP19A ne l'exécute jamais. + +Le lancement ne promet pas de restaurer l'état interne d'un terminal, les +onglets d'un navigateur ou le contenu propre à une application. + +### Applications gérées par autostart + +Une vue présente est réutilisée normalement. Une vue absente produit +`skip-autostart-managed`, jamais `launch-application`, car `sway-autostart` +possède son lancement. C'est notamment le contrat de `limusic-app` issu de +`exec limusic-app`. STEP19B devra attendre ou demander une intervention au +lieu de créer une double instance. + +### Infrastructure et daemons + +`session-infrastructure` et `daemon` ne produisent jamais de lancement. +QuickShell, Polkit, autotiling, wlsunset, les services systemd utilisateur et +le listener de transparence restent possédés par leur sous-système. Une +absence inattendue produit `manual-required`. + +## Workspaces et outputs + +Chaque action conserve `desired_workspace` et `desired_output`. Une fenêtre +live au mauvais endroit produit `reuse-window`, puis `move-to-workspace`, sans +lancement supplémentaire. + +Les outputs sont appariés par nom de connecteur. La politique V1 est : + +```text +connector name preferred +missing output = deferred-no-fallback +``` + +Si le connecteur sauvegardé est absent, le planner produit +`manual-required(reason=missing-output)` et n'invente ni écran ni fallback. Le +lancement d'une application absente est lui aussi différé dans ce cas. + +## Arbre Sway natif + +Le planner compare une signature sémantique par vue : layout/orientation du +workspace, chaîne des containers parents, layout/orientation/percent des +parents, branche, index et percent de la fenêtre. Les `container_id` runtime +sont exclus. + +Une différence produit `restore-tree-position` avec +`capability=partially-supported`. Sway natif sait déplacer, splitter, choisir +un layout et ajuster des proportions, mais V1 ne prétend pas reconstruire +atomiquement tout arbre arbitraire. STEP19B devra valider ses primitives et +replanifier après l'apparition des fenêtres. + +Seul `layout_engine=sway-native` est accepté. `spatial-canvas-v1` reste hors +périmètre jusqu'à STEP20. + +## Floating, fullscreen et scratchpad + +Une différence live/source produit une intention distincte : + +```text +restore-floating +restore-fullscreen +restore-scratchpad-hidden +restore-scratchpad-visible +``` + +Une fenêtre live présente dans le scratchpad alors que le snapshot ne le +demande pas produit `manual-required`, plutôt qu'une hypothèse destructrice. + +## Ordre et focus + +Le focus est toujours la dernière phase. Il n'est ajouté que si le focus live +diffère ou si une intention antérieure pourrait le perturber. + +```text +1. match-managed : réutiliser/différer/bloquer +2. launch : intentions de lancement exactes +3. wait-match : réservé au rematching STEP19B +4. workspace : placement workspace/output +5. tree-layout : arbre Sway natif +6. floating +7. fullscreen +8. scratchpad +9. focus +``` + +Le validateur refuse une dépendance future, un recul de phase ou une action +après le focus. + +## Actions V1 et refus explicites + +```text +reuse-window +launch-application +skip-autostart-managed +move-to-workspace +restore-tree-position +restore-floating +restore-fullscreen +restore-scratchpad-hidden +restore-scratchpad-visible +restore-focus +manual-required +``` + +`manual-required` porte notamment les raisons suivantes : + +```text +no-exact-launch-identity +missing-output +non-restorable-category +live-window-unexpected-scratchpad +``` + +Dire que l'automatisation ne sait pas est préférable à une commande inventée. + +## Garanties et limites STEP19A + +Le module planner n'importe pas `subprocess`, ne contient pas `swaymsg`, et ne +possède aucune fonction de lancement, d'écriture ou de suppression. Le CLI +n'appelle que : + +```text +show_session(name) # lecture et validation persistante +collect(autostart) # quatre IPC read-only STEP18 +build_restore_plan() # fonction pure +``` + +Le plan expose : + +```text +read_only = true +execution_supported = false +mutating_commands_executed = 0 +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. diff --git a/docs/session-snapshot-v1.md b/docs/session-snapshot-v1.md index eb4d1b921..cf665cbd6 100644 --- a/docs/session-snapshot-v1.md +++ b/docs/session-snapshot-v1.md @@ -317,13 +317,16 @@ session_snapshot.py save session_snapshot.py list session_snapshot.py show session_snapshot.py delete +session_snapshot.py plan ``` `--compact` produit du JSON compact. `save` collecte et publie ; `list` émet un tableau d'objets `{name, updated_at, windows, workspaces}` ; `show` retourne le document strictement validé ; `delete` supprime uniquement le fichier régulier -dérivé du nom validé. Ces commandes ne contiennent et n'appellent aucune -opération `restore`, `launch` ou `apply`. +dérivé du nom validé. `plan` charge ce document, collecte un second Snapshot +V1 live puis appelle le planner pur décrit dans `docs/session-restore-v1.md`. +Il produit uniquement des intentions : aucune commande `restore`, `launch`, +`move`, `focus` ou `apply` n'est exécutée. ### Cohérence IPC, verrouillage et publication atomique diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_restore.py b/quickshell/.config/quickshell/labfy-sway/session/session_restore.py new file mode 100644 index 000000000..e2e0c4936 --- /dev/null +++ b/quickshell/.config/quickshell/labfy-sway/session/session_restore.py @@ -0,0 +1,622 @@ +#!/usr/bin/env python3 +"""Construire, sans l'exécuter, un Restore Plan V1 déterministe.""" + +from __future__ import annotations + +from collections import Counter +import copy +from typing import Any, Mapping, Sequence + + +PLAN_SCHEMA = "labfy.sway.restore-plan" +PLAN_VERSION = 1 +SNAPSHOT_SCHEMA = "labfy.sway.session-snapshot" +SNAPSHOT_VERSION = 1 +LAYOUT_ENGINE = "sway-native" + +# CONTRACT: cet ordre est l'interface entre le planner STEP19A et le futur +# exécuteur STEP19B. Une phase tardive ne peut jamais précéder sa dépendance. +PHASES = ( + "match-managed", + "launch", + "wait-match", + "workspace", + "tree-layout", + "floating", + "fullscreen", + "scratchpad", + "focus", +) +PHASE_INDEX = {name: index for index, name in enumerate(PHASES)} +ACTION_TYPES = { + "reuse-window", + "launch-application", + "skip-autostart-managed", + "move-to-workspace", + "restore-floating", + "restore-fullscreen", + "restore-tree-position", + "restore-scratchpad-hidden", + "restore-scratchpad-visible", + "restore-focus", + "manual-required", +} +CONFIDENCE = {"exact", "high", "medium", "ambiguous", "unresolved"} +NON_RESTORABLE_CATEGORIES = {"session-infrastructure", "daemon"} + + +class RestorePlanError(RuntimeError): + """Violation du contrat du planner read-only.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise RestorePlanError(f"restore plan V1 invalide: {message}") + + +def _text(value: Any) -> str | None: + return value if isinstance(value, str) and value else None + + +def _fold(value: Any) -> str | None: + text = _text(value) + return text.casefold() if text else None + + +def _identity(window: Mapping[str, Any]) -> dict[str, str | None]: + restore = window.get("restore_identity") + restore = restore if isinstance(restore, Mapping) else {} + xwayland = window.get("xwayland") + xwayland = xwayland if isinstance(xwayland, Mapping) else {} + return { + "desktop_entry": _text(restore.get("desktop_entry")), + "app_id": _text(window.get("app_id")) or _text(restore.get("app_id")), + "class": _text(xwayland.get("class")) or _text(restore.get("class")), + "instance": _text(xwayland.get("instance")) or _text(restore.get("instance")), + "executable_basename": _text(window.get("executable_basename")), + } + + +def _stable_window_key(window: Mapping[str, Any]) -> tuple[str, ...]: + """Départager sans PID ni con_id, qui ne sont jamais des identités.""" + identity = _identity(window) + return tuple( + _fold(value) or "" + for value in ( + identity["desktop_entry"], + identity["app_id"], + identity["class"], + identity["instance"], + identity["executable_basename"], + window.get("workspace"), + window.get("output"), + window.get("title_hint"), + window.get("snapshot_identity"), + window.get("window_id"), + ) + ) + + +def _candidate(saved: Mapping[str, Any], live: Mapping[str, Any]) -> dict[str, Any] | None: + """Noter une preuve d'identité ; titre/workspace ne sont que des hints.""" + wanted = _identity(saved) + current = _identity(live) + saved_app = _fold(wanted["app_id"]) + live_app = _fold(current["app_id"]) + saved_class = _fold(wanted["class"]) + live_class = _fold(current["class"]) + desktop_equal = bool( + _fold(wanted["desktop_entry"]) + and _fold(wanted["desktop_entry"]) == _fold(current["desktop_entry"]) + ) + app_equal = bool(saved_app and saved_app == live_app) + class_equal = bool(saved_class and saved_class == live_class) + instance_equal = bool( + _fold(wanted["instance"]) + and _fold(wanted["instance"]) == _fold(current["instance"]) + ) + executable_equal = bool( + _fold(wanted["executable_basename"]) + and _fold(wanted["executable_basename"]) == _fold(current["executable_basename"]) + ) + + # Une contradiction app_id forte interdit qu'un simple basename ou titre + # fusionne deux applications différentes. + if saved_app and live_app and saved_app != live_app and not desktop_equal: + return None + if not any((desktop_equal, app_equal, class_equal, instance_equal, executable_equal)): + return None + + matched_by: list[str] = [] + score = 0 + confidence = "medium" + if desktop_equal: + score += 500 + matched_by.append("desktop-entry") + if app_equal: + score += 400 + matched_by.append("app-id") + if class_equal: + score += 350 + matched_by.append("class") + if instance_equal: + score += 250 + matched_by.append("instance") + if executable_equal: + score += 150 + matched_by.append("executable-basename") + + if desktop_equal and (app_equal or class_equal): + confidence = "exact" + elif desktop_equal or app_equal or class_equal: + confidence = "high" + + # Les hints ne peuvent jamais créer un candidat ; ils départagent + # seulement plusieurs fenêtres d'une identité application déjà prouvée. + if saved.get("workspace") == live.get("workspace"): + score += 20 + if bool(saved.get("floating")) == bool(live.get("floating")): + score += 10 + if _fold(saved.get("title_hint")) and _fold(saved.get("title_hint")) == _fold(live.get("title_hint")): + score += 5 + matched_by.append("title-hint-tiebreak") + return {"score": score, "confidence": confidence, "matched_by": matched_by} + + +def match_windows( + saved_windows: Sequence[Mapping[str, Any]], + live_windows: Sequence[Mapping[str, Any]], +) -> tuple[dict[str, dict[str, Any]], list[str]]: + """Produire un matching un-à-un stable ; une vue live n'est jamais réutilisée deux fois.""" + # WHY: une boucle source gloutonne ferait prendre Firefox-B par Firefox-A + # avant que son titre puisse départager les deux vues. + # CONTRACT: les meilleures preuves globales gagnent ; l'ordre snapshot ne + # sert qu'après identité, workspace, floating et title_hint. + edges: list[tuple[int, int, tuple[str, ...], int, dict[str, Any]]] = [] + for saved_index, saved in enumerate(saved_windows): + for live_index, live in enumerate(live_windows): + evidence = _candidate(saved, live) + if evidence is not None: + edges.append( + (-evidence["score"], saved_index, _stable_window_key(live), live_index, evidence) + ) + edges.sort() + used_saved: set[int] = set() + used_live: set[int] = set() + matches: dict[str, dict[str, Any]] = {} + for chosen in edges: + score, saved_index, _stable, live_index, evidence = chosen + if saved_index in used_saved or live_index in used_live: + continue + used_saved.add(saved_index) + used_live.add(live_index) + saved_id = str(saved_windows[saved_index].get("window_id")) + equal_competitors = sum( + 1 for item in edges + if item[0] == score and (item[1] == saved_index or item[3] == live_index) + ) + confidence = "ambiguous" if equal_competitors > 1 else evidence["confidence"] + matches[saved_id] = { + "saved_index": saved_index, + "live_index": live_index, + "confidence": confidence, + "matched_by": evidence["matched_by"], + "score": -score, + } + unmatched = [ + str(saved.get("window_id")) + for index, saved in enumerate(saved_windows) + if index not in used_saved + ] + return matches, unmatched + + +def _container_path(snapshot: Mapping[str, Any], window: Mapping[str, Any]) -> list[dict[str, Any]]: + containers = { + item.get("container_id"): item + for item in snapshot.get("containers", []) + if isinstance(item, Mapping) + } + position = window.get("tree_position") + position = position if isinstance(position, Mapping) else {} + parent = position.get("parent_container_id") + path: list[dict[str, Any]] = [] + seen: set[str] = set() + while isinstance(parent, str) and parent in containers and parent not in seen: + seen.add(parent) + item = containers[parent] + item_position = item.get("tree_position") + item_position = item_position if isinstance(item_position, Mapping) else {} + path.append({ + "layout": item.get("layout"), + "orientation": item.get("orientation"), + "branch": item_position.get("branch"), + "index": item_position.get("index"), + "percent": item.get("percent"), + }) + parent = item.get("parent_container_id") + path.reverse() + return path + + +def _placement(snapshot: Mapping[str, Any], window: Mapping[str, Any]) -> dict[str, Any]: + workspaces = { + item.get("name"): item + for item in snapshot.get("workspaces", []) + if isinstance(item, Mapping) + } + workspace = workspaces.get(window.get("workspace"), {}) + position = window.get("tree_position") + position = position if isinstance(position, Mapping) else {} + return { + "workspace_layout": workspace.get("layout"), + "workspace_orientation": workspace.get("orientation"), + "container_path": _container_path(snapshot, window), + "branch": position.get("branch"), + "index": position.get("index"), + "percent": window.get("percent"), + } + + +def _window_inventory(window: Mapping[str, Any]) -> dict[str, Any]: + runtime = window.get("runtime") + runtime = runtime if isinstance(runtime, Mapping) else {} + classification = window.get("classification") + classification = classification if isinstance(classification, Mapping) else {} + return { + "window_id": window.get("window_id"), + "snapshot_identity": window.get("snapshot_identity"), + "runtime": {"con_id": runtime.get("con_id"), "pid": runtime.get("pid")}, + "workspace": window.get("workspace"), + "output": window.get("output"), + **_identity(window), + "classification": { + "category": classification.get("category"), + "managed_by": classification.get("managed_by"), + }, + "floating": bool(window.get("floating")), + "fullscreen_mode": window.get("fullscreen_mode"), + "scratchpad": copy.deepcopy(window.get("scratchpad")), + "focused": bool(window.get("focused")), + "title_hint": window.get("title_hint"), + } + + +def _validate_input(snapshot: Mapping[str, Any], label: str, *, persistent: bool) -> None: + _require(snapshot.get("schema") == SNAPSHOT_SCHEMA, f"{label}.schema") + _require(snapshot.get("version") == SNAPSHOT_VERSION, f"{label}.version") + compositor = snapshot.get("compositor") + _require(isinstance(compositor, Mapping), f"{label}.compositor") + _require(compositor.get("layout_engine") == LAYOUT_ENGINE, f"{label}.layout_engine") + _require(isinstance(snapshot.get("windows"), list), f"{label}.windows") + _require(isinstance(snapshot.get("outputs"), list), f"{label}.outputs") + _require(isinstance(snapshot.get("workspaces"), list), f"{label}.workspaces") + metadata = snapshot.get("metadata") + _require(isinstance(metadata, Mapping), f"{label}.metadata") + if persistent: + _require(bool(_text(metadata.get("session_name"))), f"{label}.metadata.session_name") + + +def _draft( + key: str, + action: str, + phase: str, + saved: Mapping[str, Any] | None, + *, + depends: Sequence[str] = (), + **payload: Any, +) -> dict[str, Any]: + result = { + "_key": key, + "_depends": list(depends), + "action": action, + "phase": phase, + "source_window": saved.get("window_id") if saved else None, + "snapshot_identity": saved.get("snapshot_identity") if saved else None, + **payload, + } + return result + + +def _finalize_actions(drafts: list[dict[str, Any]]) -> list[dict[str, Any]]: + drafts.sort(key=lambda item: (PHASE_INDEX[item["phase"]], item.pop("_order"), item["action"], item["_key"])) + key_to_id = {item["_key"]: f"a{index:03d}" for index, item in enumerate(drafts, 1)} + actions: list[dict[str, Any]] = [] + for item in drafts: + result = {key: value for key, value in item.items() if not key.startswith("_")} + result["action_id"] = key_to_id[item["_key"]] + result["depends_on"] = [key_to_id[key] for key in item["_depends"] if key in key_to_id] + actions.append(result) + return actions + + +def validate_restore_plan(plan: Mapping[str, Any]) -> None: + """Valider l'ordre et les dépendances sans aucun effet externe.""" + _require(plan.get("schema") == PLAN_SCHEMA, "schema") + _require(plan.get("version") == PLAN_VERSION, "version") + _require(plan.get("read_only") is True, "read_only") + _require(plan.get("execution_supported") is False, "execution_supported") + actions = plan.get("actions") + _require(isinstance(actions, list), "actions") + known: set[str] = set() + last_phase = -1 + focus_seen = False + for index, action in enumerate(actions): + _require(isinstance(action, Mapping), f"actions[{index}]") + action_id = action.get("action_id") + _require(isinstance(action_id, str) and action_id not in known, "action_id") + _require(action.get("action") in ACTION_TYPES, "action type") + phase = action.get("phase") + _require(phase in PHASE_INDEX and PHASE_INDEX[phase] >= last_phase, "ordre phases") + last_phase = PHASE_INDEX[phase] + dependencies = action.get("depends_on") + _require(isinstance(dependencies, list) and all(item in known for item in dependencies), "depends_on") + if focus_seen: + _require(action.get("action") == "restore-focus", "focus doit rester dernier") + if action.get("action") == "restore-focus": + focus_seen = True + confidence = action.get("confidence") + if confidence is not None: + _require(confidence in CONFIDENCE, "confidence") + known.add(action_id) + + +def build_restore_plan(source: Mapping[str, Any], live: Mapping[str, Any]) -> dict[str, Any]: + """Comparer deux snapshots validés et retourner uniquement des intentions.""" + _validate_input(source, "source", persistent=True) + _validate_input(live, "live", persistent=False) + source_windows = [item for item in source["windows"] if isinstance(item, Mapping)] + live_windows = [item for item in live["windows"] if isinstance(item, Mapping)] + matches, unmatched = match_windows(source_windows, live_windows) + live_outputs = { + item.get("name") + for item in live["outputs"] + if isinstance(item, Mapping) and item.get("active") is True + } + live_workspaces = { + item.get("name") + for item in live["workspaces"] + if isinstance(item, Mapping) + } + live_focused = live.get("focus", {}).get("window_id") if isinstance(live.get("focus"), Mapping) else None + drafts: list[dict[str, Any]] = [] + base_keys: dict[str, str] = {} + matched_live_by_saved: dict[str, Mapping[str, Any]] = {} + + def add(item: dict[str, Any], order: int) -> str: + item["_order"] = order + drafts.append(item) + return item["_key"] + + for order, saved in enumerate(source_windows): + saved_id = str(saved.get("window_id")) + desired_output = saved.get("output") + desired_workspace = saved.get("workspace") + category = saved.get("classification", {}).get("category") if isinstance(saved.get("classification"), Mapping) else "unknown" + resolution = saved.get("restore_identity") + resolution = resolution if isinstance(resolution, Mapping) else {} + match = matches.get(saved_id) + live_window: Mapping[str, Any] | None = None + + if match is not None: + live_window = live_windows[match["live_index"]] + matched_live_by_saved[saved_id] = live_window + key = f"reuse:{saved_id}" + base_keys[saved_id] = add(_draft( + key, + "reuse-window", + "match-managed", + saved, + confidence=match["confidence"], + matched_by=match["matched_by"], + live_window=_window_inventory(live_window), + ), order) + elif desired_output is not None and desired_output not in live_outputs: + key = f"manual-output:{saved_id}" + base_keys[saved_id] = add(_draft( + key, + "manual-required", + "match-managed", + saved, + confidence="unresolved", + reason="missing-output", + desired_output=desired_output, + policy="deferred-no-fallback", + ), order) + continue + elif category == "autostart-managed-application": + key = f"managed:{saved_id}" + base_keys[saved_id] = add(_draft( + key, + "skip-autostart-managed", + "match-managed", + saved, + confidence="unresolved", + reason="launch-owned-by-sway-autostart", + managed_by=saved.get("classification", {}).get("managed_by"), + ), order) + continue + elif category in NON_RESTORABLE_CATEGORIES: + key = f"manual-category:{saved_id}" + base_keys[saved_id] = add(_draft( + key, + "manual-required", + "match-managed", + saved, + confidence="unresolved", + reason="non-restorable-category", + category=category, + ), order) + continue + elif category == "user-application" and resolution.get("confidence") == "exact" and _text(resolution.get("desktop_entry")): + key = f"launch:{saved_id}" + desktop_entry = resolution["desktop_entry"] + base_keys[saved_id] = add(_draft( + key, + "launch-application", + "launch", + saved, + confidence="exact", + desktop_entry=desktop_entry, + launch={ + "backend": "uwsm-app-desktop-entry", + "argv": ["uwsm", "app", "--", desktop_entry], + "execute_in_step19a": False, + }, + limitations=["application-internal-state-not-restored"], + ), order) + else: + key = f"manual-identity:{saved_id}" + base_keys[saved_id] = add(_draft( + key, + "manual-required", + "match-managed", + saved, + confidence="unresolved", + reason="no-exact-launch-identity", + restore_identity=copy.deepcopy(resolution), + ), order) + continue + + base = base_keys[saved_id] + current_workspace = live_window.get("workspace") if live_window else None + current_output = live_window.get("output") if live_window else None + if desired_output is not None and desired_output not in live_outputs: + add(_draft( + f"manual-output:{saved_id}", "manual-required", "workspace", saved, + depends=(base,), confidence="unresolved", reason="missing-output", + desired_output=desired_output, policy="deferred-no-fallback", + ), order) + continue + if live_window is None or current_workspace != desired_workspace or current_output != desired_output: + add(_draft( + f"workspace:{saved_id}", "move-to-workspace", "workspace", saved, + depends=(base,), desired_workspace=desired_workspace, desired_output=desired_output, + workspace_exists=desired_workspace in live_workspaces, + current_workspace=current_workspace, current_output=current_output, + ), order) + + desired_placement = _placement(source, saved) + current_placement = _placement(live, live_window) if live_window else None + if current_placement != desired_placement: + add(_draft( + f"tree:{saved_id}", "restore-tree-position", "tree-layout", saved, + depends=(base,), capability="partially-supported", + desired=desired_placement, current=current_placement, + reason="sway-native-tree-diff", + ), order) + + if live_window is None: + current_floating = False + current_fullscreen = 0 + current_scratchpad: Mapping[str, Any] = {} + else: + current_floating = bool(live_window.get("floating")) + current_fullscreen = int(live_window.get("fullscreen_mode") or 0) + raw_scratchpad = live_window.get("scratchpad") + current_scratchpad = raw_scratchpad if isinstance(raw_scratchpad, Mapping) else {} + if current_floating != bool(saved.get("floating")): + add(_draft( + f"floating:{saved_id}", "restore-floating", "floating", saved, + depends=(base,), desired_floating=bool(saved.get("floating")), + current_floating=current_floating, + ), order) + desired_fullscreen = int(saved.get("fullscreen_mode") or 0) + if current_fullscreen != desired_fullscreen: + add(_draft( + f"fullscreen:{saved_id}", "restore-fullscreen", "fullscreen", saved, + depends=(base,), desired_fullscreen_mode=desired_fullscreen, + current_fullscreen_mode=current_fullscreen, + ), order) + desired_scratchpad = saved.get("scratchpad") + desired_scratchpad = desired_scratchpad if isinstance(desired_scratchpad, Mapping) else {} + if desired_scratchpad.get("member") is True and dict(current_scratchpad) != dict(desired_scratchpad): + visibility = desired_scratchpad.get("visibility") + action = "restore-scratchpad-hidden" if visibility == "hidden" else "restore-scratchpad-visible" + add(_draft( + f"scratch:{saved_id}", action, "scratchpad", saved, + depends=(base,), desired=copy.deepcopy(desired_scratchpad), + current=copy.deepcopy(current_scratchpad), + ), order) + elif desired_scratchpad.get("member") is not True and current_scratchpad.get("member") is True: + add(_draft( + f"manual-scratch:{saved_id}", "manual-required", "scratchpad", saved, + depends=(base,), confidence="unresolved", reason="live-window-unexpected-scratchpad", + ), order) + + source_focus = source.get("focus") + source_focus = source_focus if isinstance(source_focus, Mapping) else {} + focus_saved_id = source_focus.get("window_id") + if isinstance(focus_saved_id, str) and focus_saved_id in base_keys: + focus_live = matched_live_by_saved.get(focus_saved_id) + focus_live_id = focus_live.get("window_id") if focus_live else None + mutations = [ + item["_key"] for item in drafts + if item["action"] not in {"reuse-window", "skip-autostart-managed", "manual-required"} + ] + if focus_live_id != live_focused or mutations: + dependencies = tuple( + item["_key"] for item in drafts + if item["action"] != "manual-required" + ) + add(_draft( + f"focus:{focus_saved_id}", "restore-focus", "focus", + next((item for item in source_windows if item.get("window_id") == focus_saved_id), None), + depends=dependencies, desired_focus_window=focus_saved_id, + current_live_focus=live_focused, + ), len(source_windows)) + + actions = _finalize_actions(drafts) + plan = { + "schema": PLAN_SCHEMA, + "version": PLAN_VERSION, + "read_only": True, + "execution_supported": False, + "source_session": source["metadata"]["session_name"], + "source": { + "schema": source["schema"], + "version": source["version"], + "captured_at": source["metadata"].get("captured_at"), + "layout_engine": source["compositor"]["layout_engine"], + }, + "live": { + "captured_at": live["metadata"].get("captured_at"), + "outputs": [item.get("name") for item in live["outputs"] if isinstance(item, Mapping) and item.get("active")], + "workspaces": [item.get("name") for item in live["workspaces"] if isinstance(item, Mapping)], + "windows": [_window_inventory(item) for item in live_windows], + }, + "policy": { + "runtime_identity_fields": [], + "runtime_observation_fields": ["runtime.pid", "runtime.con_id"], + "missing_output": "deferred-no-fallback", + "autostart_owner": "sway-autostart", + "launch_backend_step19b": "uwsm-app-desktop-entry", + "phase_order": list(PHASES), + "spatial_canvas": "out-of-scope-step20", + }, + "matches": [ + { + "source_window": saved_id, + "live_window": live_windows[data["live_index"]].get("window_id"), + "confidence": data["confidence"], + "matched_by": data["matched_by"], + } + for saved_id, data in sorted(matches.items(), key=lambda item: item[1]["saved_index"]) + ], + "unmatched_source_windows": unmatched, + "actions": actions, + "diagnostics": { + "action_counts": dict(sorted(Counter(item["action"] for item in actions).items())), + "match_confidence_counts": dict(sorted(Counter(item["confidence"] for item in plan_matches(matches)).items())), + "mutating_commands_executed": 0, + "applications_launched": 0, + "snapshots_written": 0, + }, + } + validate_restore_plan(plan) + return plan + + +def plan_matches(matches: Mapping[str, Mapping[str, Any]]) -> list[dict[str, Any]]: + """Projection minuscule utilisée pour les diagnostics déterministes.""" + return [{"confidence": item["confidence"]} for item in matches.values()] diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py b/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py index cd546170a..3517823e4 100755 --- a/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py +++ b/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py @@ -9,6 +9,7 @@ import configparser import datetime as dt import errno import fcntl +import importlib.util import json import os from pathlib import Path @@ -1115,8 +1116,52 @@ def delete_session( os.close(directory_fd) +def _load_restore_planner() -> Callable[[Mapping[str, Any], Mapping[str, Any]], dict[str, Any]]: + """Charger le planner voisin sans lui donner d'accès implicite au backend.""" + path = Path(__file__).with_name("session_restore.py") + spec = importlib.util.spec_from_file_location("labfy_session_restore", path) + if spec is None or spec.loader is None: + raise SnapshotError("planner de restauration indisponible") + module = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(module) + except (OSError, ImportError, SyntaxError) as exc: + raise SnapshotError(f"chargement du planner impossible: {exc}") from exc + planner = getattr(module, "build_restore_plan", None) + if not callable(planner): + raise SnapshotError("planner de restauration invalide") + return planner + + +def plan_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, + planner: Callable[[Mapping[str, Any], Mapping[str, Any]], dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Charger deux états validés et produire un plan sans jamais l'exécuter.""" + # WHY: STEP19A doit réutiliser exactement la validation persistante et la + # résolution DesktopEntry STEP18, sans ouvrir une seconde voie divergente. + # CONTRACT: loader et collector sont en lecture seule ; le planner est pur. + # INVARIANT: aucune commande applicative ou Sway mutatrice n'est accessible ici. + load = show_session if snapshot_loader is None else snapshot_loader + observe = collect if live_collector is None else live_collector + build = _load_restore_planner() if planner is None else planner + source = load(normalize_session_name(name)) + live = observe(autostart_path) + validate_snapshot(live) + try: + return build(source, live) + except Exception as exc: + if exc.__class__.__name__ != "RestorePlanError": + raise + raise SnapshotError(str(exc)) from exc + + def main(argv: Sequence[str] | None = None) -> int: - parser = argparse.ArgumentParser(description="Sauvegarde persistante read-only d'une session Sway") + parser = argparse.ArgumentParser(description="Snapshots persistants et restore planner read-only 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( @@ -1125,14 +1170,14 @@ def main(argv: Sequence[str] | None = None) -> int: 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")) + parser.add_argument("command", nargs="?", choices=("save", "list", "show", "delete", "plan")) 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") - if args.command in {"save", "show", "delete"} and args.name 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(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") @@ -1157,6 +1202,8 @@ def main(argv: Sequence[str] | None = None) -> int: print(warning, file=sys.stderr) elif args.command == "show": result = show_session(args.name) + elif args.command == "plan": + result = plan_session(args.name, args.autostart) else: delete_session(args.name) result = {"deleted": args.name} diff --git a/tests/test_session_restore.py b/tests/test_session_restore.py new file mode 100644 index 000000000..5ed221342 --- /dev/null +++ b/tests/test_session_restore.py @@ -0,0 +1,343 @@ +"""Contrat Restore Plan V1 : intentions déterministes et aucune exécution.""" + +import copy +import importlib.util +import io +import json +from pathlib import Path +import unittest +from unittest import mock + + +ROOT = Path(__file__).resolve().parents[1] +RESTORE_SOURCE = ROOT / "quickshell/.config/quickshell/labfy-sway/session/session_restore.py" +SNAPSHOT_SOURCE = ROOT / "quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py" + + +def load_module(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +restore = load_module("session_restore_test", RESTORE_SOURCE) +snapshot_backend = load_module("session_snapshot_restore_test", SNAPSHOT_SOURCE) + + +def window( + window_id, + app_id, + *, + desktop_entry=None, + identity_confidence="exact", + title=None, + workspace="1", + output="eDP-1", + category="user-application", + managed_by=None, + floating=False, + fullscreen=0, + focused=False, + con_id=100, + pid=1000, + parent=None, + index=0, + xclass=None, + instance=None, + executable=None, + scratchpad=None, +): + return { + "window_id": window_id, + "snapshot_identity": f"{app_id or xclass or executable or 'unknown'}#{window_id[1:]}", + "runtime": {"con_id": con_id, "pid": pid}, + "workspace": workspace, + "output": output, + "app_id": app_id, + "xwayland": {"class": xclass, "instance": instance}, + "executable_basename": executable or app_id, + "title_hint": title, + "restore_identity": { + "app_id": app_id, + "class": xclass, + "instance": instance, + "desktop_entry": desktop_entry, + "confidence": identity_confidence, + "matched_by": "desktop-id" if desktop_entry else None, + }, + "restore_adapter": "generic-desktop" if desktop_entry else "unknown", + "classification": {"category": category, "managed_by": managed_by}, + "focused": focused, + "urgent": False, + "floating": floating, + "fullscreen_mode": fullscreen, + "rect": {"x": 0, "y": 0, "width": 800, "height": 600}, + "percent": 1.0, + "marks": [], + "scratchpad": scratchpad or {"member": False, "visibility": "not-applicable", "state": "none"}, + "tree_position": {"parent_container_id": parent, "branch": "tiling", "index": index}, + } + + +def session( + windows, + *, + persistent=False, + name="test", + outputs=("eDP-1",), + workspace_outputs=None, + containers=None, + focused=None, + captured_at="2026-10-05T00:00:00.000Z", +): + workspace_outputs = workspace_outputs or {} + workspace_names = [] + for item in windows: + workspace = item.get("workspace") + if workspace is not None and workspace not in workspace_names: + workspace_names.append(workspace) + metadata = { + "captured_at": captured_at, + "desktop": "sway:wlroots:swayfx", + "session_type": "wayland", + } + if persistent: + metadata.update({ + "session_name": name, + "created_at": captured_at, + "updated_at": captured_at, + }) + focused_id = focused + if focused_id is None: + focused_id = next((item["window_id"] for item in windows if item.get("focused")), None) + return { + "schema": "labfy.sway.session-snapshot", + "version": 1, + "metadata": metadata, + "compositor": { + "variant": "sway", + "version": "0.6", + "sway_version": "1.12.0", + "layout_engine": "sway-native", + }, + "outputs": [ + {"name": item, "active": True, "focused": index == 0} + for index, item in enumerate(outputs) + ], + "workspaces": [ + { + "name": item, + "output": workspace_outputs.get(item, outputs[0] if outputs else None), + "layout": "splith", + "orientation": "horizontal", + } + for item in workspace_names + ], + "containers": containers or [], + "windows": copy.deepcopy(windows), + "focus": { + "output": outputs[0] if outputs else None, + "workspace": next((item.get("workspace") for item in windows if item["window_id"] == focused_id), None), + "container_id": None, + "window_id": focused_id, + }, + } + + +def action_types(plan): + return [item["action"] for item in plan["actions"]] + + +class RestorePlannerTest(unittest.TestCase): + def test_exact_reuse_identical_session_has_no_mutation(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", focused=True) + live = copy.deepcopy(saved) + live["runtime"] = {"con_id": 999, "pid": 8888} + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + self.assertEqual(action_types(plan), ["reuse-window"]) + self.assertEqual(plan["matches"][0]["confidence"], "exact") + self.assertEqual(plan["diagnostics"]["mutating_commands_executed"], 0) + + def test_missing_exact_user_application_uses_future_uwsm_backend(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", focused=True) + plan = restore.build_restore_plan(session([saved], persistent=True), session([])) + launch = next(item for item in plan["actions"] if item["action"] == "launch-application") + self.assertEqual(launch["launch"]["argv"], ["uwsm", "app", "--", "firefox.desktop"]) + self.assertFalse(launch["launch"]["execute_in_step19a"]) + + def test_synthetic_firefox_kitty_limusic_with_only_limusic_live(self): + saved = [ + window("w1", "firefox", desktop_entry="firefox.desktop"), + window("w2", "kitty", desktop_entry="kitty.desktop", executable="kitty"), + window("w3", "limusic-app", desktop_entry="limusic.desktop", workspace="10", category="autostart-managed-application", managed_by="sway-autostart"), + ] + live = [copy.deepcopy(saved[2])] + plan = restore.build_restore_plan(session(saved, persistent=True), session(live)) + launches = [item["desktop_entry"] for item in plan["actions"] if item["action"] == "launch-application"] + self.assertEqual(launches, ["firefox.desktop", "kitty.desktop"]) + self.assertEqual(sum(item["action"] == "reuse-window" for item in plan["actions"]), 1) + + def test_existing_application_on_wrong_workspace_is_reused_then_moved(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", workspace="1", focused=True) + live = window("w1", "firefox", desktop_entry="firefox.desktop", workspace="2", focused=True) + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + self.assertIn("reuse-window", action_types(plan)) + self.assertIn("move-to-workspace", action_types(plan)) + self.assertNotIn("launch-application", action_types(plan)) + move = next(item for item in plan["actions"] if item["action"] == "move-to-workspace") + self.assertEqual(move["desired_workspace"], "1") + + def test_multi_window_matches_only_best_title_and_never_reuses_twice(self): + saved_a = window("w1", "firefox", desktop_entry="firefox.desktop", title="Firefox — A") + saved_b = window("w2", "firefox", desktop_entry="firefox.desktop", title="Firefox — B") + live_b = window("w1", "firefox", desktop_entry="firefox.desktop", title="Firefox — B", pid=5000) + plan = restore.build_restore_plan(session([saved_a, saved_b], persistent=True), session([live_b])) + self.assertEqual(len(plan["matches"]), 1) + self.assertEqual(plan["matches"][0]["source_window"], "w2") + launch = next(item for item in plan["actions"] if item["action"] == "launch-application") + self.assertEqual(launch["source_window"], "w1") + + def test_title_hint_is_not_required_identity(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", title="Ancien titre") + live = window("w9", "firefox", desktop_entry="firefox.desktop", title="Nouveau titre") + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + self.assertEqual(plan["matches"][0]["source_window"], "w1") + + def test_app_id_without_desktop_entry_is_high_confidence(self): + saved = window("w1", "custom-app", identity_confidence="unresolved") + live = window("w7", "custom-app", identity_confidence="unresolved") + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + self.assertEqual(plan["matches"][0]["confidence"], "high") + self.assertIn("app-id", plan["matches"][0]["matched_by"]) + + def test_xwayland_class_without_app_id_is_high_confidence(self): + saved = window("w1", None, xclass="LegacyApp", executable=None, identity_confidence="unresolved") + live = window("w8", None, xclass="LegacyApp", executable=None, identity_confidence="unresolved") + saved["executable_basename"] = None + live["executable_basename"] = None + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + self.assertEqual(plan["matches"][0]["confidence"], "high") + self.assertEqual(plan["matches"][0]["matched_by"], ["class"]) + + def test_indistinguishable_live_candidates_are_ambiguous_but_one_to_one(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop") + first = window("w1", "firefox", desktop_entry="firefox.desktop", pid=2000) + second = window("w2", "firefox", desktop_entry="firefox.desktop", pid=3000) + plan = restore.build_restore_plan(session([saved], persistent=True), session([first, second])) + self.assertEqual(len(plan["matches"]), 1) + self.assertEqual(plan["matches"][0]["confidence"], "ambiguous") + + def test_absent_autostart_application_is_not_launched(self): + saved = window("w1", "limusic-app", desktop_entry="limusic.desktop", category="autostart-managed-application", managed_by="sway-autostart") + plan = restore.build_restore_plan(session([saved], persistent=True), session([])) + self.assertEqual(action_types(plan), ["skip-autostart-managed"]) + self.assertNotIn("launch-application", action_types(plan)) + + def test_unresolved_window_requires_manual_action(self): + saved = window("w1", None, identity_confidence="unresolved", executable=None) + saved["executable_basename"] = None + plan = restore.build_restore_plan(session([saved], persistent=True), session([])) + self.assertEqual(action_types(plan), ["manual-required"]) + self.assertEqual(plan["actions"][0]["reason"], "no-exact-launch-identity") + + def test_missing_output_is_explicit_and_has_no_fallback_launch(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", output="HDMI-A-1") + plan = restore.build_restore_plan(session([saved], persistent=True, outputs=("HDMI-A-1",)), session([], outputs=("eDP-1",))) + self.assertEqual(action_types(plan), ["manual-required"]) + self.assertEqual(plan["actions"][0]["policy"], "deferred-no-fallback") + + def test_floating_difference_produces_intention_only(self): + saved = window("w1", "kitty", desktop_entry="kitty.desktop", floating=True) + live = window("w1", "kitty", desktop_entry="kitty.desktop", floating=False) + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + action = next(item for item in plan["actions"] if item["action"] == "restore-floating") + self.assertTrue(action["desired_floating"]) + + def test_fullscreen_difference_produces_intention_only(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", fullscreen=1) + live = window("w1", "firefox", desktop_entry="firefox.desktop", fullscreen=0) + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + action = next(item for item in plan["actions"] if item["action"] == "restore-fullscreen") + self.assertEqual(action["desired_fullscreen_mode"], 1) + + def test_layout_difference_is_partially_supported(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", parent="c1") + live = window("w1", "firefox", desktop_entry="firefox.desktop") + container = { + "container_id": "c1", "parent_container_id": None, + "tree_position": {"branch": "tiling", "index": 0}, + "layout": "splitv", "orientation": "vertical", "percent": 1.0, + } + plan = restore.build_restore_plan(session([saved], persistent=True, containers=[container]), session([live])) + action = next(item for item in plan["actions"] if item["action"] == "restore-tree-position") + self.assertEqual(action["capability"], "partially-supported") + + def test_scratchpad_hidden_intention(self): + scratch = {"member": True, "visibility": "hidden", "state": "fresh"} + saved = window("w1", "kitty", desktop_entry="kitty.desktop", workspace=None, output=None, scratchpad=scratch) + live = window("w1", "kitty", desktop_entry="kitty.desktop") + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + self.assertIn("restore-scratchpad-hidden", action_types(plan)) + + def test_focus_is_last_after_mutating_intentions(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", workspace="1", focused=True) + live = window("w1", "firefox", desktop_entry="firefox.desktop", workspace="2", focused=True) + plan = restore.build_restore_plan(session([saved], persistent=True, focused="w1"), session([live], focused="w1")) + self.assertEqual(plan["actions"][-1]["action"], "restore-focus") + self.assertTrue(plan["actions"][-1]["depends_on"]) + + def test_infrastructure_is_never_launched(self): + saved = window("w1", "quickshell", desktop_entry="org.quickshell.desktop", category="session-infrastructure") + plan = restore.build_restore_plan(session([saved], persistent=True), session([])) + self.assertEqual(action_types(plan), ["manual-required"]) + self.assertNotIn("launch-application", action_types(plan)) + + def test_runtime_ids_are_observations_not_matching_identity(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop", con_id=7, pid=10) + live = window("w9", "firefox", desktop_entry="firefox.desktop", con_id=900, pid=9999) + plan = restore.build_restore_plan(session([saved], persistent=True), session([live])) + self.assertEqual(plan["matches"][0]["source_window"], "w1") + self.assertEqual(plan["policy"]["runtime_identity_fields"], []) + self.assertEqual(plan["live"]["windows"][0]["runtime"], {"con_id": 900, "pid": 9999}) + + def test_same_inputs_produce_byte_equivalent_plan(self): + saved = [ + window("w1", "firefox", desktop_entry="firefox.desktop", title="A"), + window("w2", "firefox", desktop_entry="firefox.desktop", title="B"), + ] + live = [window("w7", "firefox", desktop_entry="firefox.desktop", title="B")] + source = session(saved, persistent=True) + current = session(live) + first = restore.build_restore_plan(source, current) + second = restore.build_restore_plan(copy.deepcopy(source), copy.deepcopy(current)) + self.assertEqual(json.dumps(first, sort_keys=True), json.dumps(second, sort_keys=True)) + + def test_cli_plan_uses_only_loader_collector_and_pure_planner(self): + saved = window("w1", "firefox", desktop_entry="firefox.desktop") + source = session([saved], persistent=True, name="cli") + current = session([copy.deepcopy(saved)]) + before = copy.deepcopy(source) + output = io.StringIO() + with mock.patch.object(snapshot_backend, "show_session", return_value=source) as loader, mock.patch.object( + snapshot_backend, "collect", return_value=current + ) as collector, mock.patch.object(snapshot_backend, "validate_snapshot") as validator, mock.patch("sys.stdout", output): + self.assertEqual(snapshot_backend.main(("plan", "cli")), 0) + result = json.loads(output.getvalue()) + self.assertEqual(result["schema"], "labfy.sway.restore-plan") + loader.assert_called_once_with("cli") + collector.assert_called_once() + validator.assert_called_once_with(current) + self.assertEqual(source, before) + self.assertEqual(result["diagnostics"]["applications_launched"], 0) + + def test_planner_module_has_no_process_or_filesystem_execution_api(self): + source_text = RESTORE_SOURCE.read_text(encoding="utf-8") + self.assertNotIn("import subprocess", source_text) + self.assertNotIn("subprocess.", source_text) + self.assertNotIn("swaymsg", source_text) + self.assertNotIn("os.system", source_text) + + +if __name__ == "__main__": + unittest.main()