feat(session): define Sway session snapshot contract
This commit is contained in:
parent
71a8ee189f
commit
40fa33eef3
4 changed files with 1293 additions and 0 deletions
|
|
@ -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 |
|
||||
|
||||
|
|
|
|||
309
docs/session-snapshot-v1.md
Normal file
309
docs/session-snapshot-v1.md
Normal file
|
|
@ -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/<pid>/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.
|
||||
705
quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py
Executable file
705
quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py
Executable file
|
|
@ -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())
|
||||
278
tests/test_session_snapshot.py
Normal file
278
tests/test_session_snapshot.py
Normal file
|
|
@ -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()
|
||||
Loading…
Reference in a new issue