diff --git a/docs/README.md b/docs/README.md index 2001c2bd3..1fa134001 100644 --- a/docs/README.md +++ b/docs/README.md @@ -17,6 +17,7 @@ limites détaillées. | Document | Responsabilité | | --- | --- | | [Desktop](desktop.md) | Session greetd/UWSM/Sway et composants Wayland canoniques | +| [Session Snapshot V1](session-snapshot-v1.md) | Contrat de capture Sway STEP18A, sécurité et limites de restauration | | [Shell](shell.md) | Zsh, outils interactifs, plugins et scripts utilisateur | | [systemd](systemd.md) | Limite système/user, unités et activations utilisateur | diff --git a/docs/session-snapshot-v1.md b/docs/session-snapshot-v1.md new file mode 100644 index 000000000..b6c64c55f --- /dev/null +++ b/docs/session-snapshot-v1.md @@ -0,0 +1,309 @@ +# Session Snapshot V1 — STEP18A + +## Objet et périmètre + +STEP18A capture et valide en lecture seule l'état de session observable par +Sway. **Aucune restauration n'est implémentée.** Le collecteur ne déplace pas +de conteneur, ne change pas le focus, ne lance ni ne tue de processus et +n'écrit aucun snapshot persistant. STEP18B possédera la persistance ; STEP19, +la restauration et son interface ; STEP20, le layout spatial 2D. + +Le collecteur autonome se trouve dans +`quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py`. Il est +séparé de QML : + +```bash +python ~/.config/quickshell/labfy-sway/session/session_snapshot.py --stdout +``` + +Sans `--stdout`, il refuse d'agir. Pour STEP18A, une redirection explicite vers +`/tmp` est réservée au diagnostic. + +## Sources et audit de disponibilité + +Les seules sources IPC sont `get_version`, `get_outputs`, `get_workspaces` et +`get_tree`. Les seules variables lues sont `XDG_CURRENT_DESKTOP`, +`XDG_SESSION_TYPE` et `DESKTOP_SESSION` lorsqu'elles existent. Le collecteur +ne lit jamais `/proc/*/environ` ni `/proc/*/cmdline`. Il peut lire le lien +`/proc//exe`, puis ne conserve que son basename. + +L'audit du 5 octobre 2026 a observé SwayFX `0.6`, fondé sur Sway `1.12.0`, en +session `wayland`, avec `XDG_CURRENT_DESKTOP=sway:wlroots:swayfx` et sans +`DESKTOP_SESSION` exporté au processus de collecte. + +| Donnée IPC | Disponibilité | Volatilité | Utilité V1 | Traitement / protection | +| --- | --- | --- | --- | --- | +| output `name` | disponible | peut changer selon le connecteur | rattachement workspace et meilleur indice local | conservé ; pas supposé toujours présent au reboot | +| output `active`, `focused` | disponible | volatile | état et focus observés | conservés | +| output `rect`, `scale`, `transform`, `current_mode` | disponible | volatile | topologie multi-écran future | conservés ; refresh renommé `refresh_millihz` | +| output make/model/serial | disponible | relativement stable mais privé et parfois faux | non nécessaire au contrat actuel | exclus | +| workspace `name`, `num`, `output` | disponible | nom/output modifiables | identité locale et rattachement | conservés | +| workspace `focused`, `visible`, `urgent`, `rect` | disponible | volatile | restitution future et diagnostic | conservés | +| ordre workspace | disponible par l'ordre IPC | volatile | ordre réel observé | `order` explicite, à partir de zéro | +| container `layout`, `orientation` | disponible | volatile | reconstruction `splith`, `splitv`, `tabbed`, `stacked` | conservés | +| container ordre, parent/enfants, `floating_nodes` | disponible | volatile | structure indispensable | normalisés en références ordonnées | +| container `percent` | disponible, parfois `null` | volatile | proportions relatives | conservé | +| `rect` | disponible | volatile | géométrie effective, surtout floating | conservé | +| `deco_rect`, `window_rect`, `geometry` | disponibles | volatiles | redondants ou décoratifs pour V1 | exclus ; `rect` et `percent` suffisent au contrat actuel | +| `focus`, `focused` | disponibles | volatile | chaîne de focus observée | focus final normalisé ; listes runtime brutes exclues | +| `fullscreen_mode` | disponible | volatile | restauration future | conservé sur chaque fenêtre | +| `urgent`, `marks` | disponibles | volatiles | état utilisateur et correspondance possible | conservés | +| `app_id` | disponible pour Wayland | peut évoluer entre versions | indice d'identité principal | conservé sans le déclarer universellement stable | +| `window_properties.class/instance` | disponible pour XWayland | peut évoluer | identité XWayland | seuls `class` et `instance` sont conservés | +| `pid`, `con_id` | disponibles | strictement runtime | diagnostic de capture seulement | isolés sous `runtime`, jamais clés de restauration | +| `name` / titre | disponible | très volatile et potentiellement privé | indice diagnostique seulement | `title_hint`, exclu de l'équivalence structurelle | +| `scratchpad_state` | disponible sur les vues | volatile | membre/visible/caché | normalisé sous `scratchpad` | +| arbres output/root bruts | disponibles | contiennent beaucoup de runtime | diagnostic ponctuel seulement | jamais persistés dans V1 | + +Une propriété absente vaut `null` lorsque l'absence est sémantique. Les +rectangles sont toujours des objets entiers `{x, y, width, height}`. V1 ne +sérialise aucune structure Python ou native opaque. + +## Enveloppe et version + +Chaque document porte cette identité immuable : + +```json +{ + "schema": "labfy.sway.session-snapshot", + "version": 1 +} +``` + +La racine contient exactement les domaines contractuels suivants (des champs +compatibles pourront être ajoutés à une version mineure du producteur, mais un +lecteur doit rejeter une autre valeur de `version`) : + +```text +snapshot +├── schema, version +├── metadata +├── compositor +├── outputs[] +├── workspaces[] +├── containers[] +├── windows[] +├── focus +├── restore_hints +└── diagnostics +``` + +### `metadata` + +- `captured_at` : UTC ISO 8601 ; volatile ; +- `session_type` : valeur allowlistée, normalement `wayland` ; +- `desktop` : `XDG_CURRENT_DESKTOP`, ou repli `DESKTOP_SESSION`. + +Le hostname est volontairement omis, comme le username, le chemin HOME, +l'adresse IP, la MAC et toute identité réseau. + +### `compositor` + +- `variant`, `version`, `sway_version` viennent de `get_version` ; +- `layout_engine` vaut `sway-native` en V1. + +`layout_engine` est le point d'extension réservé à `spatial-canvas-v1` pour +STEP20. V1 n'invente ni `canvas_x`, ni `canvas_y`, ni viewport. + +### `outputs[]` + +Chaque output contient `name`, `active`, `focused`, `rect`, `scale`, +`transform` et `current_mode`. Le mode est `null` ou contient `width`, +`height`, `refresh_millihz`. + +Le nom de connecteur est l'indice préféré, non une identité universelle. +`make/model/serial` ne sont pas des clés obligatoires. Une restauration future +devra vérifier les outputs présents, appliquer une stratégie de repli encore +à définir si un écran a disparu, puis rattacher ou redistribuer les +workspaces. V1 enregistre explicitement `missing_output_policy=deferred` : il +ne prétend pas résoudre ce cas. + +### `workspaces[]` + +Chaque workspace contient : + +```text +name, num, output, order, focused, visible, urgent, rect, +layout, orientation, roots.tiling[], roots.floating[] +``` + +`num=-1` représente le comportement IPC des workspaces nommés. Un workspace +vide possède deux listes de racines vides. Chaque référence de racine vaut +`{"kind":"container|window","id":"..."}`. L'ordre des tableaux est +l'ordre Sway : il est contractuel dans le snapshot. + +### `containers[]` + +Un conteneur structurel contient : + +```text +container_id, runtime_con_id, workspace, output, +parent_container_id, tree_position, +layout, orientation, percent, rect, +focused, urgent, marks, +children[], floating_children[] +``` + +`container_id` (`c1`, `c2`, …) est local au document. Les deux listes +d'enfants conservent séparément l'ordre des branches `nodes` et +`floating_nodes`. Les layouts reconnus sont `none`, `splith`, `splitv`, +`stacked`, `tabbed`, `output` et `dockarea`. Les conteneurs ne sont pas aplatis +en une liste de fenêtres : leurs relations permettent de reconstruire +l'arbre, les groupes tabbed/stacked et les proportions. + +### `windows[]` + +Chaque vraie vue feuille contient : + +```text +window_id, snapshot_identity, runtime, +workspace, output, +app_id, xwayland.class, xwayland.instance, +executable_basename, title_hint, +restore_identity, restore_adapter, classification, +focused, urgent, floating, fullscreen_mode, +rect, percent, marks, scratchpad, tree_position +``` + +`window_id` et `snapshot_identity` sont locaux à cette capture. Les doublons +restent des fenêtres distinctes : par exemple `firefox#1` et `firefox#2`. +L'ordinal dépend de l'ordre déterministe de parcours de l'arbre ; ce n'est pas +une identité durable. `runtime.con_id` et `runtime.pid` décrivent seulement le +processus et le conteneur observés. Ils ne doivent jamais être utilisés après +un reboot. + +`floating` combine la branche de l'arbre et l'état `auto_on/user_on`. +`rect`, `workspace` et `output` donnent les informations géométriques utiles +aux fenêtres flottantes. `fullscreen_mode` conserve la valeur entière IPC +(zéro : non fullscreen). Aucune fenêtre n'est déplacée pour compléter ces +données. + +`tree_position` contient `parent_container_id`, `branch` et `index`. Une +fenêtre directement sous un workspace a un parent `null`. Une fenêtre de +scratchpad cachée n'est artificiellement rattachée à aucun workspace/output. + +## Identité restaurable et DesktopEntry + +Une fenêtre, un processus applicatif et une DesktopEntry sont trois niveaux +différents : plusieurs fenêtres peuvent appartenir au même processus ou à +plusieurs processus de la même application ; une DesktopEntry décrit un mode +de lancement, pas une fenêtre. Le collecteur ne fusionne jamais les fenêtres +ayant le même `app_id`. + +`restore_identity` contient `app_id`, `class`, `instance`, `desktop_entry`, +`confidence` et `matched_by`. La recherche déterministe parcourt les +répertoires XDG, avec priorité aux entrées utilisateur en cas de même nom : + +1. stem exact de `app_id.desktop`, `class.desktop` ou `instance.desktop` ; +2. correspondance exacte à `StartupWMClass` ; +3. stem unique égal au basename exécutable après normalisation (`heuristic`) ; +4. sinon `desktop_entry=null`, `confidence=unresolved`. + +Les niveaux sont `exact`, `heuristic`, `unresolved`. Une ambiguïté ne produit +jamais de correspondance. La valeur `Exec` des fichiers desktop n'est ni +nécessaire ni persistée. + +`restore_adapter` vaut actuellement `terminal`, `generic-desktop` ou +`unknown`. `kitty` ne permet pas de distinguer shell, Neovim, Yazi, SSH ou une +commande utilisateur : STEP18A n'essaie pas de le deviner. Les futurs +adaptateurs `browser` ou `custom` sont réservés à une politique ultérieure. + +## Autostart et infrastructure + +Le filtre primaire reste l'arbre Sway : un daemon sans fenêtre n'entre pas +dans `windows[]`. Le fichier Sway `autostart` est seulement analysé pour +classifier les programmes directs, sans exécuter leurs commandes. La +taxonomie est : + +- `user-application` ; +- `autostart-managed-application` avec `managed_by=sway-autostart` ; +- `session-infrastructure` ; +- `daemon` ; +- `unknown`. + +`limusic-app`, qui possède une vraie fenêtre, est classée application gérée +par `sway-autostart`. Une restauration future devra donc éviter une seconde +instance. QuickShell, l'agent Polkit, `autotiling`, le listener de +transparence, `wlsunset` et les services systemd user ne deviennent pas des +applications restaurables simplement parce que leur processus existe. La +classification ne lance rien et n'est pas encore une politique de relance. + +## Scratchpad et focus + +Sway expose le workspace interne `__i3_scratch`. Il est exclu de +`workspaces[]`, mais ses racines ordonnées apparaissent dans +`restore_hints.scratchpad_roots`. Chaque fenêtre porte : + +- `member` ; +- `visibility` : `hidden`, `visible` ou `not-applicable` ; +- la valeur IPC `state` (`none`, `fresh`, `changed` lorsqu'elle existe). + +Une fenêtre scratchpad cachée se trouve dans `__i3_scratch`; une vue membre +affichée dans un workspace réel garde un état scratchpad non `none`. L'audit +réel STEP18A a observé `__i3_scratch` vide : le cas caché est donc validé par +fixture synthétique, pas par manipulation de la session. + +`focus` référence l'output et le workspace par leur nom, et la feuille ou le +conteneur par son ID local au snapshot. Ces références expriment l'état +observé ; la stratégie de restitution du focus appartient à STEP19. + +## Validation et déterminisme + +Le validateur interne vérifie schema/version, collections et objets +obligatoires, unicité des IDs, ordre workspace contigu, layouts admis, +relations output/workspace, références arbre, parents et cibles du focus. Le +JSON n'est émis qu'après validation. Aucune dépendance Python externe n'est +requise. + +À état Sway inchangé, l'ordre de parcours IPC et du scan DesktopEntry est +déterministe. L'équivalence structurelle exclut seulement les chemins listés +dans `diagnostics.volatile_fields` : + +```text +metadata.captured_at +diagnostics.collection_duration_ms +windows[].title_hint +``` + +Les titres restent utiles comme indices manuels mais une animation de titre, +une page Web ou un document différent ne change pas l'identité structurelle. +Les champs runtime restent inclus dans cette comparaison : à état réellement +inchangé ils doivent rester identiques. + +## Sécurité et future persistance STEP18B + +Un titre peut révéler un document ou un site. Les snapshots futurs sont de +l'état utilisateur et devront aller dans : + +```text +${XDG_STATE_HOME:-$HOME/.local/state}/labfy-sway/sessions/ +``` + +Le répertoire devra être créé en `0700` et chaque snapshot en `0600`. Rien ne +doit être installé dans `~/.config`, le dépôt Git, `Documents` ou `/tmp` pour +le fonctionnement normal. + +STEP18B devra écrire un fichier temporaire dans le même répertoire, appliquer +les permissions, écrire et valider le document complet, effectuer `fsync` du +fichier si la politique de durabilité le requiert, puis publier par `rename` +atomique et synchroniser le répertoire si nécessaire. Une capture interrompue +ne devra jamais remplacer le dernier snapshot valide. + +Sont explicitement exclus : username, HOME absolu, hostname par défaut, +IP/MAC, make/model/serial des écrans, contenu d'environnement, lignes de +commande, arguments, tokens, mots de passe, cookies, URL privées, valeur +`Exec` des DesktopEntry, arbres IPC bruts et chemins complets d'exécutables. + +## Limites V1 + +V1 ne sait pas relancer une application, retrouver le contenu d'un terminal, +regrouper sûrement des processus en application, redistribuer un workspace si +un écran manque, restaurer le focus, interpréter un titre, faire une rotation +de snapshots ni représenter le Spatial Canvas. Ces limites sont explicites et +n'affaiblissent pas la capture normalisée de STEP18A. + +Les quatre requêtes IPC ne constituent pas une transaction atomique offerte +par Sway. Si la topologie change pendant ces quelques dizaines de +millisecondes, la validation relationnelle fait échouer la capture plutôt que +de publier un graphe incohérent. STEP18B pourra effectuer une nouvelle capture +bornée ; il ne devra jamais persister le résultat invalide. diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py b/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py new file mode 100755 index 000000000..52c46f533 --- /dev/null +++ b/quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py @@ -0,0 +1,705 @@ +#!/usr/bin/env python3 +"""Capture en lecture seule un Session Snapshot V1 normalisé de Sway.""" + +from __future__ import annotations + +import argparse +import configparser +import datetime as dt +import json +import os +from pathlib import Path +import re +import shlex +import subprocess +import sys +import time +from typing import Any, Mapping, Sequence + + +SCHEMA = "labfy.sway.session-snapshot" +VERSION = 1 +LAYOUT_ENGINE = "sway-native" +# WHY: le snapshot contient seulement une reconstruction Sway native ; STEP20 +# pourra ajouter un autre moteur sans redéfinir les coordonnées inexistantes. +# CONTRACT: schema/version/layout_engine sont les discriminants persistants V1. +# INVARIANT: aucune valeur runtime (PID/con_id) n'est une identité durable. +LAYOUTS = {"none", "splith", "splitv", "stacked", "tabbed", "output", "dockarea"} +CONFIDENCE = {"exact", "heuristic", "unresolved"} +RECT_KEYS = ("x", "y", "width", "height") +DAEMON_NAMES = { + "autotiling", + "inactive-windows-transparency.py", + "wlsunset", +} +INFRASTRUCTURE_NAMES = { + "polkit-gnome-authentication-agent-1", + "qs", + "quickshell", +} + + +class SnapshotError(RuntimeError): + """Erreur de collecte ou violation du contrat V1.""" + + +def _ipc(message_type: str) -> Any: + """CONTRACT: interroger Sway sans commande mutatrice ni shell intermédiaire.""" + try: + completed = subprocess.run( + ["swaymsg", "-t", message_type, "-r"], + check=True, + capture_output=True, + text=True, + timeout=5, + ) + return json.loads(completed.stdout) + except (OSError, subprocess.SubprocessError, json.JSONDecodeError) as exc: + raise SnapshotError(f"échec de swaymsg -t {message_type}: {exc}") from exc + + +def _rect(value: Any) -> dict[str, int]: + source = value if isinstance(value, Mapping) else {} + return {key: int(source.get(key, 0)) for key in RECT_KEYS} + + +def _nullable_number(value: Any) -> float | None: + if value is None: + return None + return float(value) + + +def _basename_from_pid(pid: Any) -> str | None: + """SECURITY: lire uniquement le lien exe et ne jamais persister son chemin.""" + if not isinstance(pid, int) or pid <= 0: + return None + try: + return Path(os.readlink(f"/proc/{pid}/exe")).name or None + except (OSError, ValueError): + return None + + +def _desktop_paths() -> list[Path]: + """Ordre déterministe ; les entrées utilisateur précèdent les entrées système.""" + data_home = Path(os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local/share"))) + data_dirs = [Path(item) for item in os.environ.get("XDG_DATA_DIRS", "/usr/local/share:/usr/share").split(":") if item] + roots = [data_home / "applications", *(item / "applications" for item in data_dirs)] + seen: set[str] = set() + result: list[Path] = [] + for root in roots: + key = str(root) + if key not in seen: + seen.add(key) + result.append(root) + return result + + +def load_desktop_entries(roots: Sequence[Path] | None = None) -> list[dict[str, str | None]]: + """Indexer les champs publics nécessaires, sans conserver Exec ni chemins locaux.""" + entries: list[dict[str, str | None]] = [] + claimed_ids: set[str] = set() + for root in roots or _desktop_paths(): + if not root.is_dir(): + continue + for path in sorted(root.glob("*.desktop"), key=lambda item: item.name.casefold()): + desktop_id = path.name + if desktop_id.casefold() in claimed_ids: + continue + parser = configparser.ConfigParser(interpolation=None, strict=False) + try: + parser.read(path, encoding="utf-8") + section = parser["Desktop Entry"] + except (OSError, UnicodeError, configparser.Error, KeyError): + continue + if section.get("Hidden", "false").casefold() == "true": + continue + claimed_ids.add(desktop_id.casefold()) + entries.append({ + "desktop_entry": desktop_id, + "stem": path.stem, + "startup_wm_class": section.get("StartupWMClass"), + "name": section.get("Name"), + }) + return entries + + +def _identity_candidates(app_id: str | None, window_properties: Mapping[str, Any]) -> list[str]: + values = [app_id, window_properties.get("class"), window_properties.get("instance")] + return [value for value in values if isinstance(value, str) and value.strip()] + + +def resolve_desktop_entry( + app_id: str | None, + window_properties: Mapping[str, Any], + executable_basename: str | None, + entries: Sequence[Mapping[str, str | None]], +) -> dict[str, Any]: + """Appliquer une chaîne de preuves explicite ; ne jamais inventer une entrée.""" + candidates = _identity_candidates(app_id, window_properties) + lowered = {candidate.casefold() for candidate in candidates} + for entry in entries: + if str(entry.get("stem") or "").casefold() in lowered: + return {"desktop_entry": entry["desktop_entry"], "confidence": "exact", "matched_by": "desktop-id"} + for entry in entries: + startup = entry.get("startup_wm_class") + if startup and startup.casefold() in lowered: + return {"desktop_entry": entry["desktop_entry"], "confidence": "exact", "matched_by": "StartupWMClass"} + + # Une heuristique n'est admise que si un basename normalisé désigne une entrée unique. + if executable_basename: + normalized = re.sub(r"[^a-z0-9]+", "", executable_basename.casefold()) + matches = [ + entry for entry in entries + if re.sub(r"[^a-z0-9]+", "", str(entry.get("stem") or "").casefold()) == normalized + ] + if len(matches) == 1: + return {"desktop_entry": matches[0]["desktop_entry"], "confidence": "heuristic", "matched_by": "executable-basename"} + return {"desktop_entry": None, "confidence": "unresolved", "matched_by": None} + + +def parse_sway_autostart(path: Path | None) -> set[str]: + """Extraire seulement le programme direct des directives exec de l'autostart.""" + managed: set[str] = set() + if path is None or not path.is_file(): + return managed + try: + lines = path.read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeError): + return managed + for raw in lines: + line = raw.strip() + if not line.startswith("exec "): + continue + try: + words = shlex.split(line[5:], comments=True) + except ValueError: + continue + while words and words[0].startswith("--"): + words.pop(0) + if not words: + continue + program = Path(words[0]).name + if program in {"env", "sh", "bash", "zsh", "python", "python3"}: + if program in {"python", "python3"} and len(words) > 1: + program = Path(words[1]).name + else: + continue + managed.add(program) + return managed + + +def _application_classification( + app_id: str | None, + executable_basename: str | None, + managed: set[str], +) -> dict[str, str | None]: + names = {name for name in (app_id, executable_basename) if name} + if names & DAEMON_NAMES: + return {"category": "daemon", "managed_by": None} + if names & INFRASTRUCTURE_NAMES: + return {"category": "session-infrastructure", "managed_by": None} + if names & managed: + return {"category": "autostart-managed-application", "managed_by": "sway-autostart"} + if names: + return {"category": "user-application", "managed_by": None} + return {"category": "unknown", "managed_by": None} + + +class Normalizer: + """Transformer l'arbre runtime en relations V1 stables dans un snapshot.""" + + def __init__(self, desktop_entries: Sequence[Mapping[str, str | None]], managed: set[str]): + self.desktop_entries = desktop_entries + self.managed = managed + self.containers: list[dict[str, Any]] = [] + self.windows: list[dict[str, Any]] = [] + self._container_seq = 0 + self._window_seq = 0 + self._app_ordinals: dict[str, int] = {} + self.focused_container: str | None = None + self.focused_window: str | None = None + + def _window( + self, + node: Mapping[str, Any], + workspace: str | None, + output: str | None, + parent_id: str | None, + branch: str, + index: int, + hidden_scratchpad: bool, + ) -> dict[str, str]: + self._window_seq += 1 + window_id = f"w{self._window_seq}" + app_id = node.get("app_id") if isinstance(node.get("app_id"), str) else None + props_source = node.get("window_properties") + props = props_source if isinstance(props_source, Mapping) else {} + xclass = props.get("class") if isinstance(props.get("class"), str) else None + instance = props.get("instance") if isinstance(props.get("instance"), str) else None + pid = node.get("pid") if isinstance(node.get("pid"), int) else None + executable = _basename_from_pid(pid) + resolution = resolve_desktop_entry(app_id, props, executable, self.desktop_entries) + identity_key = app_id or xclass or instance or executable or "unknown" + ordinal = self._app_ordinals.get(identity_key, 0) + 1 + self._app_ordinals[identity_key] = ordinal + scratch_state = node.get("scratchpad_state") + scratch_member = hidden_scratchpad or scratch_state not in (None, "none") + classification = _application_classification(app_id, executable, self.managed) + floating_raw = node.get("floating") + floating = branch == "floating" or floating_raw in {"auto_on", "user_on"} + window = { + "window_id": window_id, + "snapshot_identity": f"{identity_key}#{ordinal}", + "runtime": {"con_id": node.get("id"), "pid": pid}, + "workspace": workspace, + "output": output, + "app_id": app_id, + "xwayland": {"class": xclass, "instance": instance}, + "executable_basename": executable, + "title_hint": node.get("name") if isinstance(node.get("name"), str) else None, + "restore_identity": { + "app_id": app_id, + "class": xclass, + "instance": instance, + **resolution, + }, + "restore_adapter": "terminal" if app_id == "kitty" or xclass == "kitty" else "generic-desktop" if resolution["desktop_entry"] else "unknown", + "classification": classification, + "focused": bool(node.get("focused", False)), + "urgent": bool(node.get("urgent", False)), + "floating": floating, + "fullscreen_mode": int(node.get("fullscreen_mode") or 0), + "rect": _rect(node.get("rect")), + "percent": _nullable_number(node.get("percent")), + "marks": [mark for mark in node.get("marks", []) if isinstance(mark, str)], + "scratchpad": { + "member": scratch_member, + "visibility": "hidden" if hidden_scratchpad else "visible" if scratch_member else "not-applicable", + "state": scratch_state if isinstance(scratch_state, str) else None, + }, + "tree_position": {"parent_container_id": parent_id, "branch": branch, "index": index}, + } + self.windows.append(window) + if window["focused"]: + self.focused_window = window_id + return {"kind": "window", "id": window_id} + + def visit( + self, + node: Mapping[str, Any], + workspace: str | None, + output: str | None, + parent_id: str | None, + branch: str, + index: int, + hidden_scratchpad: bool = False, + ) -> dict[str, str]: + # WHY: les branches séparées portent une sémantique que perdrait une + # liste plate de fenêtres. + # CONTRACT: l'ordre de nodes et floating_nodes est conservé tel quel. + # INVARIANT: chaque référence produite cible exactement un objet V1. + nodes = node.get("nodes") if isinstance(node.get("nodes"), list) else [] + floating_nodes = node.get("floating_nodes") if isinstance(node.get("floating_nodes"), list) else [] + has_window_identity = any(node.get(key) is not None for key in ("app_id", "window", "window_properties", "pid")) + if not nodes and not floating_nodes and has_window_identity: + return self._window(node, workspace, output, parent_id, branch, index, hidden_scratchpad) + + self._container_seq += 1 + container_id = f"c{self._container_seq}" + container: dict[str, Any] = { + "container_id": container_id, + "runtime_con_id": node.get("id"), + "workspace": workspace, + "output": output, + "parent_container_id": parent_id, + "tree_position": {"branch": branch, "index": index}, + "layout": node.get("layout") if node.get("layout") in LAYOUTS else "none", + "orientation": node.get("orientation") if isinstance(node.get("orientation"), str) else "none", + "percent": _nullable_number(node.get("percent")), + "rect": _rect(node.get("rect")), + "focused": bool(node.get("focused", False)), + "urgent": bool(node.get("urgent", False)), + "marks": [mark for mark in node.get("marks", []) if isinstance(mark, str)], + "children": [], + "floating_children": [], + } + self.containers.append(container) + if container["focused"]: + self.focused_container = container_id + container["children"] = [ + self.visit(child, workspace, output, container_id, "tiling", child_index, hidden_scratchpad) + for child_index, child in enumerate(nodes) + if isinstance(child, Mapping) + ] + container["floating_children"] = [ + self.visit(child, workspace, output, container_id, "floating", child_index, hidden_scratchpad) + for child_index, child in enumerate(floating_nodes) + if isinstance(child, Mapping) + ] + return {"kind": "container", "id": container_id} + + +def _find_tree_output(tree: Mapping[str, Any], name: str) -> Mapping[str, Any] | None: + for node in tree.get("nodes", []): + if isinstance(node, Mapping) and node.get("type") == "output" and node.get("name") == name: + return node + return None + + +def _find_workspace(output_node: Mapping[str, Any], name: str) -> Mapping[str, Any] | None: + for node in output_node.get("nodes", []): + if isinstance(node, Mapping) and node.get("type") == "workspace" and node.get("name") == name: + return node + return None + + +def _mode(output: Mapping[str, Any]) -> dict[str, int] | None: + current = output.get("current_mode") + if not isinstance(current, Mapping): + return None + return {"width": int(current.get("width", 0)), "height": int(current.get("height", 0)), "refresh_millihz": int(current.get("refresh", 0))} + + +def build_snapshot( + version: Mapping[str, Any], + outputs_raw: Sequence[Mapping[str, Any]], + workspaces_raw: Sequence[Mapping[str, Any]], + tree: Mapping[str, Any], + *, + captured_at: str | None = None, + environment: Mapping[str, str] | None = None, + desktop_entries: Sequence[Mapping[str, str | None]] = (), + managed_autostart: set[str] | None = None, +) -> dict[str, Any]: + normalizer = Normalizer(desktop_entries, managed_autostart or set()) + outputs: list[dict[str, Any]] = [] + workspaces: list[dict[str, Any]] = [] + workspace_roots: dict[str, dict[str, list[dict[str, str]]]] = {} + + for output in outputs_raw: + outputs.append({ + "name": output.get("name"), + "active": bool(output.get("active", False)), + "focused": bool(output.get("focused", False)), + "rect": _rect(output.get("rect")), + "scale": float(output.get("scale", 1.0)), + "transform": output.get("transform") if isinstance(output.get("transform"), str) else "normal", + "current_mode": _mode(output), + }) + + for order, workspace in enumerate(workspaces_raw): + name = str(workspace.get("name")) + output_name = str(workspace.get("output")) + tree_output = _find_tree_output(tree, output_name) + tree_workspace = _find_workspace(tree_output, name) if tree_output else None + tiling: list[dict[str, str]] = [] + floating: list[dict[str, str]] = [] + if tree_workspace: + for index, node in enumerate(tree_workspace.get("nodes", [])): + if isinstance(node, Mapping): + tiling.append(normalizer.visit(node, name, output_name, None, "tiling", index)) + for index, node in enumerate(tree_workspace.get("floating_nodes", [])): + if isinstance(node, Mapping): + floating.append(normalizer.visit(node, name, output_name, None, "floating", index)) + workspace_roots[name] = {"tiling": tiling, "floating": floating} + workspaces.append({ + "name": name, + "num": int(workspace.get("num", -1)), + "output": output_name, + "order": order, + "focused": bool(workspace.get("focused", False)), + "visible": bool(workspace.get("visible", False)), + "urgent": bool(workspace.get("urgent", False)), + "rect": _rect(workspace.get("rect")), + "layout": tree_workspace.get("layout", workspace.get("layout", "none")) if tree_workspace else workspace.get("layout", "none"), + "orientation": tree_workspace.get("orientation", workspace.get("orientation", "none")) if tree_workspace else workspace.get("orientation", "none"), + "roots": workspace_roots[name], + }) + + # WHY: __i3_scratch est une racine technique, pas un workspace restaurable. + # CONTRACT: ses vues restent représentées et ne reçoivent pas un faux output. + # INVARIANT: aucune observation ne déplace une vue vers ou depuis le scratchpad. + scratch_output = _find_tree_output(tree, "__i3") + scratch_workspace = _find_workspace(scratch_output, "__i3_scratch") if scratch_output else None + scratch_roots: list[dict[str, str]] = [] + if scratch_workspace: + scratch_nodes = [*scratch_workspace.get("nodes", []), *scratch_workspace.get("floating_nodes", [])] + for index, node in enumerate(scratch_nodes): + if isinstance(node, Mapping): + scratch_roots.append(normalizer.visit(node, None, None, None, "scratchpad", index, True)) + + env = environment or {} + captured = captured_at or dt.datetime.now(dt.timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") + identity_counts = {key: 0 for key in CONFIDENCE} + for window in normalizer.windows: + identity_counts[window["restore_identity"]["confidence"]] += 1 + focused_output = next((item["name"] for item in outputs if item["focused"]), None) + focused_workspace = next((item["name"] for item in workspaces if item["focused"]), None) + snapshot = { + "schema": SCHEMA, + "version": VERSION, + "metadata": { + "captured_at": captured, + "session_type": env.get("XDG_SESSION_TYPE"), + "desktop": env.get("XDG_CURRENT_DESKTOP") or env.get("DESKTOP_SESSION"), + }, + "compositor": { + "variant": version.get("variant"), + "version": version.get("human_readable"), + "sway_version": version.get("sway_original_version"), + "layout_engine": LAYOUT_ENGINE, + }, + "outputs": outputs, + "workspaces": workspaces, + "containers": normalizer.containers, + "windows": normalizer.windows, + "focus": { + "output": focused_output, + "workspace": focused_workspace, + "container_id": normalizer.focused_container, + "window_id": normalizer.focused_window, + }, + "restore_hints": { + "output_matching": "connector-name-preferred-with-future-fallback", + "missing_output_policy": "deferred", + "scratchpad_roots": scratch_roots, + }, + "diagnostics": { + "identity_counts": identity_counts, + "volatile_fields": [ + "metadata.captured_at", + "diagnostics.collection_duration_ms", + "windows[].title_hint", + ], + "source": ["get_version", "get_outputs", "get_workspaces", "get_tree"], + }, + } + validate_snapshot(snapshot) + return snapshot + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise SnapshotError(f"snapshot V1 invalide: {message}") + + +def validate_snapshot(snapshot: Mapping[str, Any]) -> None: + """Valider types, bornes et toutes les relations entre collections V1.""" + # WHY: STEP18B ne devra jamais publier un graphe partiel ou incohérent. + # CONTRACT: toutes les références sont internes au document validé. + # INVARIANT: la validation est pure et n'interroge pas la session. + _require(snapshot.get("schema") == SCHEMA, "schema") + _require(snapshot.get("version") == VERSION, "version") + for key in ("metadata", "compositor", "focus", "restore_hints", "diagnostics"): + _require(isinstance(snapshot.get(key), Mapping), key) + for key in ("outputs", "workspaces", "containers", "windows"): + _require(isinstance(snapshot.get(key), list), key) + + metadata = snapshot["metadata"] + _require(all(key in metadata for key in ("captured_at", "session_type", "desktop")), "metadata champs") + _require(isinstance(metadata["captured_at"], str) and bool(metadata["captured_at"]), "metadata.captured_at") + _require(metadata["session_type"] is None or isinstance(metadata["session_type"], str), "metadata.session_type") + _require(metadata["desktop"] is None or isinstance(metadata["desktop"], str), "metadata.desktop") + compositor = snapshot["compositor"] + _require(all(key in compositor for key in ("variant", "version", "sway_version", "layout_engine")), "compositor champs") + _require(compositor["layout_engine"] == LAYOUT_ENGINE, "compositor.layout_engine") + _require(all(compositor[key] is None or isinstance(compositor[key], str) for key in ("variant", "version", "sway_version")), "compositor types") + + output_names = [item.get("name") for item in snapshot["outputs"]] + workspace_names = [item.get("name") for item in snapshot["workspaces"]] + container_ids = [item.get("container_id") for item in snapshot["containers"]] + window_ids = [item.get("window_id") for item in snapshot["windows"]] + _require(all(isinstance(item, str) and item for item in output_names), "output.name") + _require(len(output_names) == len(set(output_names)), "output.name dupliqué") + _require(len(workspace_names) == len(set(workspace_names)), "workspace.name dupliqué") + _require(len(container_ids) == len(set(container_ids)), "container_id dupliqué") + _require(len(window_ids) == len(set(window_ids)), "window_id dupliqué") + _require([item.get("order") for item in snapshot["workspaces"]] == list(range(len(workspace_names))), "workspace.order") + all_refs = set(container_ids) | set(window_ids) + container_by_id = {item["container_id"]: item for item in snapshot["containers"]} + window_by_id = {item["window_id"]: item for item in snapshot["windows"]} + object_by_id = {**container_by_id, **window_by_id} + + def validate_rect(value: Any, field: str) -> None: + _require(isinstance(value, Mapping), field) + _require(set(value) == set(RECT_KEYS), f"{field} clés") + _require(all(isinstance(value[key], int) and not isinstance(value[key], bool) for key in RECT_KEYS), f"{field} types") + + for output in snapshot["outputs"]: + _require(all(key in output for key in ("name", "active", "focused", "rect", "scale", "transform", "current_mode")), "output champs") + _require(isinstance(output["active"], bool) and isinstance(output["focused"], bool), "output booléens") + validate_rect(output["rect"], "output.rect") + _require(isinstance(output["scale"], (int, float)) and not isinstance(output["scale"], bool), "output.scale") + _require(isinstance(output["transform"], str), "output.transform") + mode = output["current_mode"] + _require(mode is None or ( + isinstance(mode, Mapping) + and set(mode) == {"width", "height", "refresh_millihz"} + and all(isinstance(item, int) and not isinstance(item, bool) for item in mode.values()) + ), "output.current_mode") + + def validate_ref(ref: Any) -> None: + _require(isinstance(ref, Mapping), "référence arbre") + _require(ref.get("kind") in {"container", "window"}, "référence kind") + _require(ref.get("id") in all_refs, "référence absente") + _require((ref["kind"] == "container") == (ref["id"] in set(container_ids)), "référence kind/id") + + for workspace in snapshot["workspaces"]: + _require(all(key in workspace for key in ("name", "num", "output", "order", "focused", "visible", "urgent", "rect", "layout", "orientation", "roots")), "workspace champs") + _require(workspace.get("output") in output_names, "workspace.output") + _require(workspace.get("layout") in LAYOUTS, "workspace.layout") + _require(isinstance(workspace.get("num"), int) and isinstance(workspace.get("order"), int), "workspace num/order") + _require(all(isinstance(workspace.get(key), bool) for key in ("focused", "visible", "urgent")), "workspace booléens") + validate_rect(workspace.get("rect"), "workspace.rect") + for branch in ("tiling", "floating"): + _require(isinstance(workspace.get("roots", {}).get(branch), list), f"workspace.roots.{branch}") + for ref in workspace["roots"][branch]: + validate_ref(ref) + for container in snapshot["containers"]: + _require(all(key in container for key in ("container_id", "runtime_con_id", "workspace", "output", "parent_container_id", "tree_position", "layout", "orientation", "percent", "rect", "focused", "urgent", "marks", "children", "floating_children")), "container champs") + _require(container.get("workspace") is None or container.get("workspace") in workspace_names, "container.workspace") + _require(container.get("output") is None or container.get("output") in output_names, "container.output") + _require(container.get("parent_container_id") is None or container.get("parent_container_id") in container_ids, "container.parent") + _require(container.get("layout") in LAYOUTS, "container.layout") + _require(isinstance(container.get("orientation"), str), "container.orientation") + _require(container.get("percent") is None or isinstance(container.get("percent"), (int, float)), "container.percent") + validate_rect(container.get("rect"), "container.rect") + for branch in ("children", "floating_children"): + _require(isinstance(container.get(branch), list), f"container.{branch}") + for ref in container[branch]: + validate_ref(ref) + for window in snapshot["windows"]: + _require(all(key in window for key in ("window_id", "snapshot_identity", "runtime", "workspace", "output", "app_id", "xwayland", "executable_basename", "title_hint", "restore_identity", "restore_adapter", "classification", "focused", "urgent", "floating", "fullscreen_mode", "rect", "percent", "marks", "scratchpad", "tree_position")), "window champs") + _require(window.get("workspace") is None or window.get("workspace") in workspace_names, "window.workspace") + _require(window.get("output") is None or window.get("output") in output_names, "window.output") + identity = window.get("restore_identity") + _require(isinstance(identity, Mapping) and identity.get("confidence") in CONFIDENCE, "restore_identity") + _require(all(key in identity for key in ("app_id", "class", "instance", "desktop_entry", "confidence", "matched_by")), "restore_identity champs") + _require(window.get("restore_adapter") in {"generic-desktop", "terminal", "browser", "custom", "unknown"}, "restore_adapter") + classification = window.get("classification") + _require(isinstance(classification, Mapping) and classification.get("category") in {"user-application", "autostart-managed-application", "session-infrastructure", "daemon", "unknown"}, "classification") + _require(all(isinstance(window.get(key), bool) for key in ("focused", "urgent", "floating")), "window booléens") + _require(isinstance(window.get("fullscreen_mode"), int) and not isinstance(window.get("fullscreen_mode"), bool), "window.fullscreen_mode") + validate_rect(window.get("rect"), "window.rect") + position = window.get("tree_position") + _require(isinstance(position, Mapping), "tree_position") + parent = position.get("parent_container_id") + _require(parent is None or parent in container_ids, "tree_position.parent_container_id") + _require(isinstance(window.get("runtime"), Mapping), "window.runtime") + _require(set(window["runtime"]) == {"con_id", "pid"}, "window.runtime champs") + _require(all(window["runtime"][key] is None or (isinstance(window["runtime"][key], int) and not isinstance(window["runtime"][key], bool)) for key in ("con_id", "pid")), "window.runtime types") + scratchpad = window.get("scratchpad") + _require(isinstance(scratchpad, Mapping) and isinstance(scratchpad.get("member"), bool), "window.scratchpad") + _require(scratchpad.get("visibility") in {"hidden", "visible", "not-applicable"}, "scratchpad.visibility") + focus = snapshot["focus"] + _require(set(focus) == {"output", "workspace", "container_id", "window_id"}, "focus champs") + _require(focus.get("output") is None or focus.get("output") in output_names, "focus.output") + _require(focus.get("workspace") is None or focus.get("workspace") in workspace_names, "focus.workspace") + _require(focus.get("container_id") is None or focus.get("container_id") in container_ids, "focus.container_id") + _require(focus.get("window_id") is None or focus.get("window_id") in window_ids, "focus.window_id") + + # WHY: une référence valide mais dupliquée ferait relancer/replacer deux fois + # la même vue ; un objet orphelin perdrait une partie de la session. + # CONTRACT: chaque objet a exactement une entrée depuis une racine ou un parent. + # INVARIANT: parent, branche et index concordent avec l'arête entrante. + incoming = {object_id: 0 for object_id in all_refs} + + def register(ref: Mapping[str, Any], parent: str | None, branch: str, index: int) -> None: + validate_ref(ref) + object_id = ref["id"] + incoming[object_id] += 1 + item = object_by_id[object_id] + expected_parent = item.get("parent_container_id") if ref["kind"] == "container" else item["tree_position"].get("parent_container_id") + _require(expected_parent == parent, "arête/parent") + position = item["tree_position"] + _require(position.get("branch") == branch and position.get("index") == index, "arête position") + + for workspace in snapshot["workspaces"]: + for branch in ("tiling", "floating"): + for index, ref in enumerate(workspace["roots"][branch]): + register(ref, None, branch, index) + restore_hints = snapshot["restore_hints"] + _require(all(key in restore_hints for key in ("output_matching", "missing_output_policy", "scratchpad_roots")), "restore_hints champs") + _require(restore_hints["output_matching"] == "connector-name-preferred-with-future-fallback", "restore_hints.output_matching") + _require(restore_hints["missing_output_policy"] == "deferred", "restore_hints.missing_output_policy") + scratch_roots = restore_hints.get("scratchpad_roots") + _require(isinstance(scratch_roots, list), "restore_hints.scratchpad_roots") + for index, ref in enumerate(scratch_roots): + register(ref, None, "scratchpad", index) + for container in snapshot["containers"]: + for key, branch in (("children", "tiling"), ("floating_children", "floating")): + for index, ref in enumerate(container[key]): + register(ref, container["container_id"], branch, index) + _require(all(count == 1 for count in incoming.values()), "objet orphelin ou référencé plusieurs fois") + + # Les comptes d'identité servent au gate réel et doivent être dérivables. + diagnostics = snapshot["diagnostics"] + _require(diagnostics.get("source") == ["get_version", "get_outputs", "get_workspaces", "get_tree"], "diagnostics.source") + _require(isinstance(diagnostics.get("volatile_fields"), list) and all(isinstance(item, str) for item in diagnostics["volatile_fields"]), "diagnostics.volatile_fields") + counts = diagnostics.get("identity_counts") + _require(isinstance(counts, Mapping) and set(counts) == CONFIDENCE, "diagnostics.identity_counts") + expected_counts = {key: 0 for key in CONFIDENCE} + for window in snapshot["windows"]: + expected_counts[window["restore_identity"]["confidence"]] += 1 + _require(dict(counts) == expected_counts, "diagnostics.identity_counts valeurs") + + +def structural_projection(snapshot: Mapping[str, Any]) -> dict[str, Any]: + """Projection déterministe : les indices purement volatils n'y participent pas.""" + projected = json.loads(json.dumps(snapshot)) + projected["metadata"].pop("captured_at", None) + projected["diagnostics"].pop("collection_duration_ms", None) + for window in projected["windows"]: + window.pop("title_hint", None) + return projected + + +def collect(autostart_path: Path | None = None) -> dict[str, Any]: + """Collecter exactement les quatre sources IPC autorisées et l'allowlist env.""" + environment = { + key: os.environ[key] + for key in ("XDG_CURRENT_DESKTOP", "XDG_SESSION_TYPE", "DESKTOP_SESSION") + if key in os.environ + } + return build_snapshot( + _ipc("get_version"), + _ipc("get_outputs"), + _ipc("get_workspaces"), + _ipc("get_tree"), + environment=environment, + desktop_entries=load_desktop_entries(), + managed_autostart=parse_sway_autostart(autostart_path), + ) + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Capture read-only d'un Session Snapshot V1 Sway") + parser.add_argument("--stdout", action="store_true", help="émettre le JSON validé sur stdout") + parser.add_argument("--compact", action="store_true", help="JSON compact (avec --stdout)") + parser.add_argument( + "--autostart", + type=Path, + default=Path.home() / ".config/sway/autostart", + help="fichier Sway autostart utilisé seulement pour la classification", + ) + args = parser.parse_args(argv) + if not args.stdout: + parser.error("STEP18A autorise uniquement --stdout") + started = time.monotonic() + try: + snapshot = collect(args.autostart) + except SnapshotError as exc: + print(str(exc), file=sys.stderr) + return 1 + snapshot["diagnostics"]["collection_duration_ms"] = round((time.monotonic() - started) * 1000, 3) + validate_snapshot(snapshot) + if args.compact: + json.dump(snapshot, sys.stdout, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + else: + json.dump(snapshot, sys.stdout, ensure_ascii=False, sort_keys=True, indent=2) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_session_snapshot.py b/tests/test_session_snapshot.py new file mode 100644 index 000000000..3318d3f62 --- /dev/null +++ b/tests/test_session_snapshot.py @@ -0,0 +1,278 @@ +"""Contrat Session Snapshot V1 sur arbres Sway synthétiques.""" + +import copy +import importlib.util +from pathlib import Path +import tempfile +import unittest +from unittest import mock + + +SOURCE = ( + Path(__file__).resolve().parents[1] + / "quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py" +) +SPEC = importlib.util.spec_from_file_location("session_snapshot", SOURCE) +snapshot_module = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(snapshot_module) + + +def leaf( + con_id, + app_id, + *, + focused=False, + floating="auto_off", + fullscreen=0, + properties=None, + scratchpad_state="none", +): + return { + "id": con_id, + "type": "con", + "name": f"titre {con_id}", + "app_id": app_id, + "window": con_id if properties else None, + "window_properties": properties, + "pid": 90000 + con_id, + "nodes": [], + "floating_nodes": [], + "layout": "none", + "orientation": "none", + "percent": 0.5, + "rect": {"x": con_id, "y": 2, "width": 300, "height": 200}, + "focused": focused, + "urgent": False, + "marks": [f"mark-{con_id}"], + "floating": floating, + "fullscreen_mode": fullscreen, + "scratchpad_state": scratchpad_state, + } + + +def container(con_id, layout, children, floating_nodes=None): + return { + "id": con_id, + "type": "con", + "name": None, + "nodes": children, + "floating_nodes": floating_nodes or [], + "layout": layout, + "orientation": "horizontal" if layout in {"splith", "tabbed"} else "vertical", + "percent": 1.0, + "rect": {"x": 0, "y": 0, "width": 1200, "height": 800}, + "focused": False, + "urgent": False, + "marks": [], + } + + +def fixture(): + firefox_one = leaf(11, "firefox") + firefox_two = leaf(12, "firefox", fullscreen=1) + kitty = leaf(13, "kitty", focused=True) + xwayland = leaf(14, None, properties={"class": "LegacyApp", "instance": "legacy"}) + unknown = leaf(15, "not-catalogued") + floating = leaf(16, "float-app", floating="user_on") + tabbed = container(21, "tabbed", [firefox_one, firefox_two]) + stacked = container(22, "stacked", [kitty, xwayland]) + splitv = container(23, "splitv", [tabbed, stacked]) + splith = container(24, "splith", [splitv, unknown], [floating]) + workspace = { + "id": 4, + "type": "workspace", + "name": "dev", + "nodes": [splith], + "floating_nodes": [], + "layout": "splith", + "orientation": "horizontal", + } + scratch = leaf(30, "scratch-app", floating="user_on", scratchpad_state="changed") + tree = { + "id": 1, + "type": "root", + "nodes": [ + { + "id": 2, + "type": "output", + "name": "__i3", + "nodes": [{"id": 3, "type": "workspace", "name": "__i3_scratch", "nodes": [], "floating_nodes": [scratch]}], + "floating_nodes": [], + }, + {"id": 5, "type": "output", "name": "DP-1", "nodes": [workspace], "floating_nodes": []}, + ], + } + outputs = [{ + "name": "DP-1", + "active": True, + "focused": True, + "rect": {"x": 0, "y": 0, "width": 1200, "height": 800}, + "scale": 1.25, + "transform": "normal", + "current_mode": {"width": 1920, "height": 1080, "refresh": 60000}, + }] + workspaces = [{ + "name": "dev", + "num": -1, + "output": "DP-1", + "focused": True, + "visible": True, + "urgent": False, + "rect": {"x": 0, "y": 0, "width": 1200, "height": 800}, + "layout": "splith", + "orientation": "horizontal", + }] + version = {"variant": "sway", "human_readable": "0.6", "sway_original_version": "1.12.0"} + entries = [ + {"desktop_entry": "firefox.desktop", "stem": "firefox", "startup_wm_class": "firefox", "name": "Firefox"}, + {"desktop_entry": "kitty.desktop", "stem": "kitty", "startup_wm_class": "kitty", "name": "Kitty"}, + {"desktop_entry": "legacy.desktop", "stem": "legacy-desktop", "startup_wm_class": "LegacyApp", "name": "Legacy"}, + ] + return version, outputs, workspaces, tree, entries + + +class SessionSnapshotTest(unittest.TestCase): + def build(self): + version, outputs, workspaces, tree, entries = fixture() + with mock.patch.object(snapshot_module, "_basename_from_pid", return_value=None): + return snapshot_module.build_snapshot( + version, + outputs, + workspaces, + tree, + captured_at="2026-01-01T00:00:00.000Z", + environment={"XDG_SESSION_TYPE": "wayland", "XDG_CURRENT_DESKTOP": "sway"}, + desktop_entries=entries, + managed_autostart={"float-app"}, + ) + + def test_layout_tree_order_and_simple_workspace(self): + snapshot = self.build() + self.assertEqual(snapshot["workspaces"][0]["name"], "dev") + self.assertEqual(snapshot["workspaces"][0]["roots"]["tiling"], [{"kind": "container", "id": "c1"}]) + layouts = {item["layout"] for item in snapshot["containers"]} + self.assertTrue({"splith", "splitv", "tabbed", "stacked"}.issubset(layouts)) + top = snapshot["containers"][0] + self.assertEqual([ref["kind"] for ref in top["children"]], ["container", "window"]) + + def test_simple_direct_window_and_empty_workspace(self): + version, outputs, workspaces, tree, entries = fixture() + direct = leaf(40, "firefox") + real_output = tree["nodes"][1] + real_output["nodes"][0]["nodes"] = [direct] + empty = { + "id": 41, "type": "workspace", "name": "2", "nodes": [], + "floating_nodes": [], "layout": "splith", "orientation": "horizontal", + } + real_output["nodes"].append(empty) + workspaces.append({ + "name": "2", "num": 2, "output": "DP-1", "focused": False, + "visible": False, "urgent": False, + "rect": {"x": 0, "y": 0, "width": 1200, "height": 800}, + "layout": "splith", "orientation": "horizontal", + }) + with mock.patch.object(snapshot_module, "_basename_from_pid", return_value=None): + snapshot = snapshot_module.build_snapshot(version, outputs, workspaces, tree, desktop_entries=entries) + self.assertEqual(snapshot["workspaces"][0]["roots"]["tiling"][0]["kind"], "window") + self.assertEqual(snapshot["workspaces"][1]["roots"], {"tiling": [], "floating": []}) + + def test_floating_fullscreen_focus_and_marks(self): + snapshot = self.build() + floating = next(window for window in snapshot["windows"] if window["app_id"] == "float-app") + fullscreen = next(window for window in snapshot["windows"] if window["runtime"]["con_id"] == 12) + kitty = next(window for window in snapshot["windows"] if window["app_id"] == "kitty") + self.assertTrue(floating["floating"]) + self.assertEqual(floating["classification"]["managed_by"], "sway-autostart") + self.assertEqual(fullscreen["fullscreen_mode"], 1) + self.assertEqual(snapshot["focus"]["window_id"], kitty["window_id"]) + self.assertEqual(kitty["marks"], ["mark-13"]) + + def test_duplicate_app_ids_keep_distinct_snapshot_ordinals(self): + windows = [window for window in self.build()["windows"] if window["app_id"] == "firefox"] + self.assertEqual([window["snapshot_identity"] for window in windows], ["firefox#1", "firefox#2"]) + self.assertNotEqual(windows[0]["window_id"], windows[1]["window_id"]) + + def test_desktop_identity_exact_xwayland_and_unresolved(self): + snapshot = self.build() + firefox = next(window for window in snapshot["windows"] if window["app_id"] == "firefox") + legacy = next(window for window in snapshot["windows"] if window["xwayland"]["class"] == "LegacyApp") + unknown = next(window for window in snapshot["windows"] if window["app_id"] == "not-catalogued") + self.assertEqual(firefox["restore_identity"]["desktop_entry"], "firefox.desktop") + self.assertEqual(legacy["restore_identity"]["matched_by"], "StartupWMClass") + self.assertEqual(legacy["restore_identity"]["confidence"], "exact") + self.assertEqual(unknown["restore_identity"]["confidence"], "unresolved") + + def test_hidden_scratchpad_is_not_a_user_workspace(self): + snapshot = self.build() + scratch = next(window for window in snapshot["windows"] if window["app_id"] == "scratch-app") + self.assertIsNone(scratch["workspace"]) + self.assertTrue(scratch["scratchpad"]["member"]) + self.assertEqual(scratch["scratchpad"]["visibility"], "hidden") + self.assertEqual(snapshot["restore_hints"]["scratchpad_roots"], [{"kind": "window", "id": scratch["window_id"]}]) + + def test_visible_scratchpad_member_stays_on_real_workspace(self): + version, outputs, workspaces, tree, entries = fixture() + visible = leaf(42, "scratch-visible", scratchpad_state="fresh") + tree["nodes"][1]["nodes"][0]["floating_nodes"] = [visible] + with mock.patch.object(snapshot_module, "_basename_from_pid", return_value=None): + snapshot = snapshot_module.build_snapshot(version, outputs, workspaces, tree, desktop_entries=entries) + window = next(item for item in snapshot["windows"] if item["app_id"] == "scratch-visible") + self.assertEqual(window["workspace"], "dev") + self.assertEqual(window["scratchpad"]["visibility"], "visible") + + def test_unique_executable_basename_is_only_heuristic(self): + entries = [{"desktop_entry": "odd-name.desktop", "stem": "tool", "startup_wm_class": None, "name": "Tool"}] + identity = snapshot_module.resolve_desktop_entry("unrelated", {}, "tool", entries) + self.assertEqual(identity, { + "desktop_entry": "odd-name.desktop", + "confidence": "heuristic", + "matched_by": "executable-basename", + }) + + def test_title_and_timestamp_do_not_affect_structural_projection(self): + first = self.build() + second = copy.deepcopy(first) + second["metadata"]["captured_at"] = "2027-02-02T00:00:00Z" + second["windows"][0]["title_hint"] = "titre dynamique différent" + self.assertEqual( + snapshot_module.structural_projection(first), + snapshot_module.structural_projection(second), + ) + + def test_validator_rejects_broken_tree_relation(self): + snapshot = self.build() + snapshot["workspaces"][0]["roots"]["tiling"][0]["id"] = "absent" + with self.assertRaises(snapshot_module.SnapshotError): + snapshot_module.validate_snapshot(snapshot) + + def test_validator_rejects_duplicate_reference(self): + snapshot = self.build() + root = snapshot["workspaces"][0]["roots"]["tiling"][0] + snapshot["workspaces"][0]["roots"]["tiling"].append(copy.deepcopy(root)) + with self.assertRaises(snapshot_module.SnapshotError): + snapshot_module.validate_snapshot(snapshot) + + def test_autostart_parser_excludes_shell_wrappers_and_keeps_direct_apps(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "autostart" + path.write_text( + "exec limusic-app\n" + "exec python3 ~/.config/sway/scripts/inactive-windows-transparency.py\n" + "exec sh -c 'private command'\n", + encoding="utf-8", + ) + self.assertEqual( + snapshot_module.parse_sway_autostart(path), + {"limusic-app", "inactive-windows-transparency.py"}, + ) + + def test_no_command_line_or_environment_field_exists(self): + serialized_keys = repr(self.build()).casefold() + self.assertNotIn("cmdline", serialized_keys) + self.assertNotIn("command_line", serialized_keys) + self.assertNotIn("/proc/", serialized_keys) + self.assertNotIn("environment", serialized_keys) + + +if __name__ == "__main__": + unittest.main()