feat(session): persist Sway session snapshots safely

docs/README.md

@@ -17,7 +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 |
+| [Session Snapshot V1](session-snapshot-v1.md) | Capture Sway STEP18A et sauvegardes persistantes STEP18B, sans restauration |
 | [Shell](shell.md) | Zsh, outils interactifs, plugins et scripts utilisateur |
 | [systemd](systemd.md) | Limite système/user, unités et activations utilisateur |
 

docs/session-snapshot-v1.md

@@ -1,12 +1,12 @@
-# Session Snapshot V1 — STEP18A
+# Session Snapshot V1 — STEP18A / STEP18B
 
 ## 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
+Sway. STEP18B sauvegarde cet état de façon persistante. **Aucune restauration
-de conteneur, ne change pas le focus, ne lance ni ne tue de processus et
+n'est implémentée.** Le collecteur ne déplace pas de conteneur, ne change pas
-n'écrit aucun snapshot persistant. STEP18B possédera la persistance ; STEP19,
+le focus, ne lance ni ne tue de processus. STEP19 possédera la restauration et
-la restauration et son interface ; STEP20, le layout spatial 2D.
+son interface ; STEP20, le layout spatial 2D.
 
 Le collecteur autonome se trouve dans
 `quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py`. Il est
@@ -16,8 +16,8 @@ séparé de QML :
 python ~/.config/quickshell/labfy-sway/session/session_snapshot.py --stdout
 ```
 
-Sans `--stdout`, il refuse d'agir. Pour STEP18A, une redirection explicite vers
+`--stdout` reste la capture non persistante de STEP18A. Les commandes
-`/tmp` est réservée au diagnostic.
+persistantes explicites de STEP18B sont documentées plus bas.
 
 ## Sources et audit de disponibilité
 
@@ -270,40 +270,126 @@ 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
+## Persistance STEP18B
 
-Un titre peut révéler un document ou un site. Les snapshots futurs sont de
+### Emplacement, nom et métadonnées
-l'état utilisateur et devront aller dans :
+
+Un titre peut révéler un document ou un site. Les snapshots sont donc de
+l'état utilisateur et vont exclusivement 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
+Lorsque `XDG_STATE_HOME` est défini et non vide, la cible est
-doit être installé dans `~/.config`, le dépôt Git, `Documents` ou `/tmp` pour
+`$XDG_STATE_HOME/labfy-sway/sessions/`. Sinon elle est
-le fonctionnement normal.
+`$HOME/.local/state/labfy-sway/sessions/`. Une valeur relative de
+`XDG_STATE_HOME` est refusée. Les répertoires gérés `labfy-sway/` et
+`sessions/` sont forcés à `0700`; les snapshots et le verrou sont forcés à
+`0600`, indépendamment de l'umask. Rien n'est écrit 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
+Le nom logique est la clé persistante V1. Il contient de 1 à 64 caractères
-les permissions, écrire et valider le document complet, effectuer `fsync` du
+ASCII parmi `A-Z`, `a-z`, `0-9`, `-` et `_`. Il n'existe pas de `session_id`
-fichier si la politique de durabilité le requiert, puis publier par `rename`
+distinct en V1 : en ajouter un serait redondant tant que le renommage n'est pas
-atomique et synchroniser le répertoire si nécessaire. Une capture interrompue
+contracté. Le nom ne devient jamais un chemin libre ; l'implémentation dérive
-ne devra jamais remplacer le dernier snapshot valide.
+elle-même le seul fichier `<name>.json`. Les séparateurs, points, espaces,
+NUL, chaînes vides et traversées sont refusés.
+
+Un document persistant conserve l'enveloppe
+`labfy.sway.session-snapshot`/version `1` et ajoute à `metadata` :
+
+- `session_name`, la clé logique validée ;
+- `created_at`, fixé lors de la première sauvegarde valide ;
+- `updated_at`, avancé à chaque publication ;
+- `captured_at`, l'instant de la nouvelle capture Sway.
+
+Une mise à jour réutilise exactement le même fichier. Elle ne crée aucune
+rotation ou copie suffixée. Si la cible régulière est corrompue, un `save`
+explicitement adressé peut la remplacer par un nouveau snapshot valide ; il
+ne prétend alors pas récupérer son ancien `created_at`.
+
+### CLI
+
+```bash
+session_snapshot.py --stdout
+session_snapshot.py save <name>
+session_snapshot.py list
+session_snapshot.py show <name>
+session_snapshot.py delete <name>
+```
+
+`--compact` produit du JSON compact. `save` collecte et publie ; `list` émet un
+tableau d'objets `{name, updated_at, windows, workspaces}` ; `show` retourne le
+document strictement validé ; `delete` supprime uniquement le fichier régulier
+dérivé du nom validé. Ces commandes ne contiennent et n'appellent aucune
+opération `restore`, `launch` ou `apply`.
+
+### Cohérence IPC, verrouillage et publication atomique
+
+Les quatre requêtes IPC ne sont pas transactionnelles. Pour `save`, une
+violation relationnelle produite par cet intervalle est retentée au plus trois
+fois, avec 50 ms entre tentatives. Les erreurs IPC ordinaires ne sont pas
+masquées par ces retries. Après trois graphes incohérents, la commande retourne
+`SAVE_FAILED_INCONSISTENT_STATE` et ne touche pas au snapshot précédent.
+
+Un verrou global persistant `.sessions.lock`, fichier régulier privé ouvert
+avec `O_NOFOLLOW`, est pris par `flock`: exclusif pour `save`/`delete`, partagé
+pour `list`/`show`. Il sérialise deux mises à jour simultanées sans daemon ni
+nouvelle file de travaux.
+
+La publication suit ce contrat :
+
+```text
+snapshot validé en mémoire
+→ temporaire aléatoire dans sessions/ en 0600 et O_EXCL/O_NOFOLLOW
+→ écriture complète
+→ fsync du fichier
+→ relecture bornée, comparaison des octets, parsing et validation
+→ nouvelle vérification de la cible
+→ os.replace atomique dans le même répertoire
+→ fsync du répertoire
+```
+
+Le temporaire propre à l'opération est supprimé en cas d'échec normal. Une
+interruption avant `os.replace` laisse l'ancien fichier inchangé. Les chemins
+de cible, temporaire, verrou et suppression sont utilisés relativement au
+descripteur ouvert du répertoire. Les répertoires gérés et fichiers cibles ne
+peuvent pas être des symlinks ; une cible symlink ou non régulière est refusée,
+donc elle ne peut rediriger ni écriture ni suppression vers un autre chemin.
+
+### Chargement, corruption, version et limite
+
+Chaque lecture ouvre un fichier régulier avec `O_NOFOLLOW`, contrôle son mode
+`0600`, sa taille, son JSON, son schema, sa version, ses métadonnées
+persistantes et toutes les relations V1. La limite est de 8 MiB
+(`8388608` octets), très supérieure aux captures usuelles d'environ 10 KiB.
+Elle est appliquée avant parsing et pendant la lecture afin de borner également
+un fichier qui grandit.
+
+`show` échoue explicitement sur corruption. `list` ignore chaque fichier
+invalide indépendamment et signale un avertissement sur stderr sans perdre les
+autres sauvegardes valides. Un schema inconnu retourne
+`unsupported snapshot schema`; une version autre que `1` retourne
+`unsupported snapshot version` et n'est jamais interprétée comme V1.
+
+### Confidentialité
 
 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.
+`title_hint` demeure potentiellement privé et justifie à lui seul les modes
+`0700`/`0600`.
 
 ## Limites V1
 
-V1 ne sait pas relancer une application, retrouver le contenu d'un terminal,
+V1 ne sait pas restaurer ou 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
+Il n'existe ni sauvegarde automatique à la fermeture, ni hook logout, ni
-par Sway. Si la topologie change pendant ces quelques dizaines de
+autostart de gestionnaire de session, ni UI QuickShell dans STEP18B. Ces
-millisecondes, la validation relationnelle fait échouer la capture plutôt que
+politiques et toute mutation Sway restent hors périmètre.
-de publier un graphe incohérent. STEP18B pourra effectuer une nouvelle capture
-bornée ; il ne devra jamais persister le résultat invalide.

quickshell/.config/quickshell/labfy-sway/session/session_snapshot.py

@@ -4,17 +4,22 @@
 from __future__ import annotations
 
 import argparse
+import contextlib
 import configparser
 import datetime as dt
+import errno
+import fcntl
 import json
 import os
 from pathlib import Path
 import re
+import secrets
 import shlex
+import stat
 import subprocess
 import sys
 import time
-from typing import Any, Mapping, Sequence
+from typing import Any, Callable, Iterator, Mapping, Sequence
 
 
 SCHEMA = "labfy.sway.session-snapshot"
@@ -27,6 +32,12 @@ LAYOUT_ENGINE = "sway-native"
 LAYOUTS = {"none", "splith", "splitv", "stacked", "tabbed", "output", "dockarea"}
 CONFIDENCE = {"exact", "heuristic", "unresolved"}
 RECT_KEYS = ("x", "y", "width", "height")
+SESSION_NAME = re.compile(r"\A[A-Za-z0-9_-]{1,64}\Z")
+SNAPSHOT_SUFFIX = ".json"
+LOCK_FILENAME = ".sessions.lock"
+MAX_SNAPSHOT_BYTES = 8 * 1024 * 1024
+SAVE_ATTEMPTS = 3
+SAVE_RETRY_DELAY_SECONDS = 0.05
 DAEMON_NAMES = {
     "autotiling",
     "inactive-windows-transparency.py",
@@ -43,6 +54,118 @@ class SnapshotError(RuntimeError):
     """Erreur de collecte ou violation du contrat V1."""
 
 
+def utc_now() -> str:
+    """Retourner un instant UTC stable et sérialisable à la milliseconde."""
+    return dt.datetime.now(dt.timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
+def normalize_session_name(name: str) -> str:
+    """Valider la clé logique V1 sans jamais la transformer en chemin libre."""
+    if not isinstance(name, str) or SESSION_NAME.fullmatch(name) is None:
+        raise SnapshotError(
+            "nom de session invalide: utiliser 1 à 64 caractères parmi A-Z, a-z, 0-9, - et _"
+        )
+    return name
+
+
+def _snapshot_filename(name: str) -> str:
+    return f"{normalize_session_name(name)}{SNAPSHOT_SUFFIX}"
+
+
+def _ensure_child_directory(parent_fd: int, name: str) -> int:
+    """Créer puis ouvrir un répertoire géré sans suivre de lien symbolique."""
+    try:
+        os.mkdir(name, mode=0o700, dir_fd=parent_fd)
+    except FileExistsError:
+        pass
+    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    try:
+        fd = os.open(name, flags, dir_fd=parent_fd)
+    except OSError as exc:
+        raise SnapshotError(f"répertoire géré non sûr: {name}: {exc}") from exc
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISDIR(info.st_mode):
+            raise SnapshotError(f"répertoire géré non sûr: {name}")
+        # CONTRACT: une umask permissive ou un ancien mode ne doit jamais
+        # exposer les titres privés contenus dans les snapshots.
+        os.fchmod(fd, 0o700)
+        if stat.S_IMODE(os.fstat(fd).st_mode) != 0o700:
+            raise SnapshotError(f"permissions répertoire invalides: {name}")
+        return fd
+    except BaseException:
+        os.close(fd)
+        raise
+
+
+def sessions_directory(
+    environment: Mapping[str, str] | None = None,
+    home: Path | None = None,
+) -> Path:
+    """Résoudre et créer la cible XDG privée des sessions persistantes."""
+    env = os.environ if environment is None else environment
+    home_path = Path.home() if home is None else home
+    configured = env.get("XDG_STATE_HOME")
+    root = Path(configured) if configured else home_path / ".local/state"
+    if not root.is_absolute():
+        raise SnapshotError("XDG_STATE_HOME doit être un chemin absolu")
+    try:
+        root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        root_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
+    except OSError as exc:
+        raise SnapshotError(f"impossible d'ouvrir XDG state: {exc}") from exc
+    try:
+        app_fd = _ensure_child_directory(root_fd, "labfy-sway")
+        try:
+            sessions_fd = _ensure_child_directory(app_fd, "sessions")
+            os.close(sessions_fd)
+        finally:
+            os.close(app_fd)
+    finally:
+        os.close(root_fd)
+    return root / "labfy-sway/sessions"
+
+
+def _open_sessions_directory(path: Path) -> int:
+    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    try:
+        fd = os.open(path, flags)
+    except OSError as exc:
+        raise SnapshotError(f"répertoire sessions non sûr: {exc}") from exc
+    if stat.S_IMODE(os.fstat(fd).st_mode) != 0o700:
+        os.close(fd)
+        raise SnapshotError("permissions du répertoire sessions différentes de 0700")
+    return fd
+
+
+@contextlib.contextmanager
+def _sessions_lock(directory_fd: int, *, exclusive: bool) -> Iterator[None]:
+    """Sérialiser publications/suppressions et stabiliser lectures multi-fichiers."""
+    flags = os.O_RDWR | os.O_CREAT | os.O_CLOEXEC
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    try:
+        lock_fd = os.open(LOCK_FILENAME, flags, 0o600, dir_fd=directory_fd)
+    except OSError as exc:
+        raise SnapshotError(f"verrou sessions non sûr: {exc}") from exc
+    try:
+        info = os.fstat(lock_fd)
+        if not stat.S_ISREG(info.st_mode):
+            raise SnapshotError("verrou sessions non régulier")
+        os.fchmod(lock_fd, 0o600)
+        fcntl.flock(lock_fd, fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH)
+        yield
+    finally:
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_UN)
+        finally:
+            os.close(lock_fd)
+
+
 def _ipc(message_type: str) -> Any:
     """CONTRACT: interroger Sway sans commande mutatrice ni shell intermédiaire."""
     try:
@@ -429,7 +552,7 @@ def build_snapshot(
                 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")
+    captured = captured_at or utc_now()
     identity_counts = {key: 0 for key in CONFIDENCE}
     for window in normalizer.windows:
         identity_counts[window["restore_identity"]["confidence"]] += 1
@@ -647,7 +770,8 @@ def validate_snapshot(snapshot: Mapping[str, Any]) -> None:
 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)
+    for field in ("captured_at", "created_at", "updated_at"):
+        projected["metadata"].pop(field, None)
     projected["diagnostics"].pop("collection_duration_ms", None)
     for window in projected["windows"]:
         window.pop("title_hint", None)
@@ -672,31 +796,377 @@ def collect(autostart_path: Path | None = None) -> dict[str, Any]:
     )
 
 
+def collect_consistent(
+    autostart_path: Path | None = None,
+    *,
+    attempts: int = SAVE_ATTEMPTS,
+    retry_delay: float = SAVE_RETRY_DELAY_SECONDS,
+    collector: Callable[[Path | None], dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """Réessayer uniquement les graphes IPC relationnellement incohérents."""
+    if attempts < 1:
+        raise ValueError("attempts doit être positif")
+    operation = collect if collector is None else collector
+    last_error: SnapshotError | None = None
+    for attempt in range(attempts):
+        try:
+            snapshot = operation(autostart_path)
+            validate_snapshot(snapshot)
+            return snapshot
+        except SnapshotError as exc:
+            if not str(exc).startswith("snapshot V1 invalide:"):
+                raise
+            last_error = exc
+            if attempt + 1 < attempts and retry_delay > 0:
+                time.sleep(retry_delay)
+    raise SnapshotError(f"SAVE_FAILED_INCONSISTENT_STATE: {last_error}") from last_error
+
+
+def _validate_persistent_snapshot(snapshot: Mapping[str, Any], expected_name: str | None = None) -> None:
+    """Valider l'extension compatible de métadonnées propre à STEP18B."""
+    validate_snapshot(snapshot)
+    metadata = snapshot["metadata"]
+    for key in ("session_name", "created_at", "updated_at"):
+        _require(isinstance(metadata.get(key), str) and bool(metadata[key]), f"metadata.{key}")
+    normalized = normalize_session_name(metadata["session_name"])
+    _require(normalized == metadata["session_name"], "metadata.session_name")
+    parsed: dict[str, dt.datetime] = {}
+    for key in ("created_at", "updated_at"):
+        try:
+            parsed[key] = dt.datetime.fromisoformat(metadata[key].replace("Z", "+00:00"))
+        except ValueError as exc:
+            raise SnapshotError(f"snapshot V1 invalide: metadata.{key} UTC") from exc
+        _require(
+            parsed[key].tzinfo is not None and parsed[key].utcoffset() == dt.timedelta(0),
+            f"metadata.{key} UTC",
+        )
+    _require(parsed["updated_at"] >= parsed["created_at"], "metadata.updated_at antérieur à created_at")
+    if expected_name is not None:
+        _require(normalized == expected_name, "nom fichier/session")
+
+
+def _read_all_bounded(fd: int, size: int) -> bytes:
+    if size > MAX_SNAPSHOT_BYTES:
+        raise SnapshotError(f"snapshot trop volumineux (limite {MAX_SNAPSHOT_BYTES} octets)")
+    chunks: list[bytes] = []
+    remaining = MAX_SNAPSHOT_BYTES + 1
+    while remaining > 0:
+        chunk = os.read(fd, min(65536, remaining))
+        if not chunk:
+            break
+        chunks.append(chunk)
+        remaining -= len(chunk)
+    payload = b"".join(chunks)
+    if len(payload) > MAX_SNAPSHOT_BYTES:
+        raise SnapshotError(f"snapshot trop volumineux (limite {MAX_SNAPSHOT_BYTES} octets)")
+    return payload
+
+
+def _load_snapshot_at(directory_fd: int, filename: str, expected_name: str) -> dict[str, Any]:
+    """Lire un fichier régulier borné sans suivre de lien symbolique."""
+    flags = os.O_RDONLY | os.O_CLOEXEC
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    try:
+        fd = os.open(filename, flags, dir_fd=directory_fd)
+    except FileNotFoundError:
+        raise
+    except OSError as exc:
+        if exc.errno == errno.ELOOP:
+            raise SnapshotError(f"snapshot refusé (lien symbolique): {expected_name}") from exc
+        raise SnapshotError(f"lecture snapshot impossible: {expected_name}: {exc}") from exc
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISREG(info.st_mode):
+            raise SnapshotError(f"snapshot refusé (type non régulier): {expected_name}")
+        if stat.S_IMODE(info.st_mode) != 0o600:
+            raise SnapshotError(f"permissions snapshot différentes de 0600: {expected_name}")
+        payload = _read_all_bounded(fd, info.st_size)
+    finally:
+        os.close(fd)
+    try:
+        value = json.loads(payload)
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise SnapshotError(f"JSON snapshot invalide: {expected_name}: {exc}") from exc
+    if not isinstance(value, Mapping):
+        raise SnapshotError(f"snapshot invalide: racine JSON non objet: {expected_name}")
+    if value.get("schema") != SCHEMA:
+        raise SnapshotError(f"unsupported snapshot schema: {value.get('schema')!r}")
+    if value.get("version") != VERSION:
+        raise SnapshotError(f"unsupported snapshot version: {value.get('version')!r}")
+    snapshot = dict(value)
+    _validate_persistent_snapshot(snapshot, expected_name)
+    return snapshot
+
+
+def _serialize_snapshot(snapshot: Mapping[str, Any]) -> bytes:
+    payload = (json.dumps(snapshot, ensure_ascii=False, sort_keys=True, indent=2) + "\n").encode("utf-8")
+    if len(payload) > MAX_SNAPSHOT_BYTES:
+        raise SnapshotError(f"snapshot trop volumineux (limite {MAX_SNAPSHOT_BYTES} octets)")
+    return payload
+
+
+def _write_all(fd: int, payload: bytes) -> None:
+    view = memoryview(payload)
+    while view:
+        written = os.write(fd, view)
+        if written <= 0:
+            raise SnapshotError("écriture temporaire incomplète")
+        view = view[written:]
+
+
+def _target_is_safe(directory_fd: int, filename: str) -> None:
+    try:
+        info = os.stat(filename, dir_fd=directory_fd, follow_symlinks=False)
+    except FileNotFoundError:
+        return
+    if stat.S_ISLNK(info.st_mode):
+        raise SnapshotError("cible snapshot refusée: lien symbolique")
+    if not stat.S_ISREG(info.st_mode):
+        raise SnapshotError("cible snapshot refusée: type non régulier")
+
+
+def _atomic_publish(
+    directory_fd: int,
+    filename: str,
+    snapshot: Mapping[str, Any],
+    *,
+    before_replace: Callable[[str], None] | None = None,
+) -> None:
+    """Publier un snapshot complet sans exposer d'état intermédiaire."""
+    payload = _serialize_snapshot(snapshot)
+    temporary = f".{filename}.{os.getpid()}.{secrets.token_hex(8)}.tmp"
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    fd = -1
+    try:
+        fd = os.open(temporary, flags, 0o600, dir_fd=directory_fd)
+        os.fchmod(fd, 0o600)
+        _write_all(fd, payload)
+        os.fsync(fd)
+        os.close(fd)
+        fd = -1
+
+        # CONTRACT: relire le fichier réellement écrit avant publication ; une
+        # corruption locale ou un test d'interruption ne touche jamais l'ancien.
+        check_flags = os.O_RDONLY | os.O_CLOEXEC
+        if hasattr(os, "O_NOFOLLOW"):
+            check_flags |= os.O_NOFOLLOW
+        check_fd = os.open(temporary, check_flags, dir_fd=directory_fd)
+        try:
+            written = _read_all_bounded(check_fd, os.fstat(check_fd).st_size)
+        finally:
+            os.close(check_fd)
+        if written != payload:
+            raise SnapshotError("validation du temporaire échouée")
+        decoded = json.loads(written)
+        _validate_persistent_snapshot(decoded, snapshot["metadata"]["session_name"])
+        _target_is_safe(directory_fd, filename)
+        if before_replace is not None:
+            before_replace(temporary)
+        os.replace(temporary, filename, src_dir_fd=directory_fd, dst_dir_fd=directory_fd)
+        os.fsync(directory_fd)
+    except (OSError, json.JSONDecodeError) as exc:
+        raise SnapshotError(f"publication atomique échouée: {exc}") from exc
+    finally:
+        if fd >= 0:
+            os.close(fd)
+        try:
+            os.unlink(temporary, dir_fd=directory_fd)
+        except FileNotFoundError:
+            pass
+
+
+def _next_updated_at(previous: str | None) -> str:
+    current = utc_now()
+    if previous is None or current > previous:
+        return current
+    try:
+        parsed = dt.datetime.fromisoformat(previous.replace("Z", "+00:00"))
+    except ValueError:
+        return current
+    return (parsed + dt.timedelta(milliseconds=1)).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
+def save_session(
+    name: str,
+    autostart_path: Path | None = None,
+    *,
+    environment: Mapping[str, str] | None = None,
+    home: Path | None = None,
+    collector: Callable[[Path | None], dict[str, Any]] | None = None,
+    retry_delay: float = SAVE_RETRY_DELAY_SECONDS,
+    before_replace: Callable[[str], None] | None = None,
+) -> dict[str, Any]:
+    """Collecter, valider et remplacer atomiquement une session nommée."""
+    normalized = normalize_session_name(name)
+    started = time.monotonic()
+    snapshot = collect_consistent(
+        autostart_path,
+        retry_delay=retry_delay,
+        collector=collector,
+    )
+    snapshot["diagnostics"]["collection_duration_ms"] = round((time.monotonic() - started) * 1000, 3)
+    validate_snapshot(snapshot)
+    directory = sessions_directory(environment, home)
+    directory_fd = _open_sessions_directory(directory)
+    filename = _snapshot_filename(normalized)
+    try:
+        with _sessions_lock(directory_fd, exclusive=True):
+            _target_is_safe(directory_fd, filename)
+            existing: dict[str, Any] | None = None
+            try:
+                existing = _load_snapshot_at(directory_fd, filename, normalized)
+            except FileNotFoundError:
+                pass
+            except SnapshotError:
+                # CONTRACT: un save explicitement ciblé peut réparer un JSON
+                # corrompu, mais jamais contourner un lien ou type de fichier.
+                existing = None
+            updated_at = _next_updated_at(existing["metadata"]["updated_at"] if existing else None)
+            snapshot["metadata"].update({
+                "session_name": normalized,
+                "created_at": existing["metadata"]["created_at"] if existing else updated_at,
+                "updated_at": updated_at,
+            })
+            _validate_persistent_snapshot(snapshot, normalized)
+            _atomic_publish(
+                directory_fd,
+                filename,
+                snapshot,
+                before_replace=before_replace,
+            )
+    finally:
+        os.close(directory_fd)
+    return snapshot
+
+
+def show_session(
+    name: str,
+    *,
+    environment: Mapping[str, str] | None = None,
+    home: Path | None = None,
+) -> dict[str, Any]:
+    normalized = normalize_session_name(name)
+    directory = sessions_directory(environment, home)
+    directory_fd = _open_sessions_directory(directory)
+    try:
+        with _sessions_lock(directory_fd, exclusive=False):
+            try:
+                return _load_snapshot_at(directory_fd, _snapshot_filename(normalized), normalized)
+            except FileNotFoundError as exc:
+                raise SnapshotError(f"session introuvable: {normalized}") from exc
+    finally:
+        os.close(directory_fd)
+
+
+def list_sessions(
+    *,
+    environment: Mapping[str, str] | None = None,
+    home: Path | None = None,
+) -> tuple[list[dict[str, Any]], list[str]]:
+    directory = sessions_directory(environment, home)
+    directory_fd = _open_sessions_directory(directory)
+    sessions: list[dict[str, Any]] = []
+    warnings: list[str] = []
+    try:
+        with _sessions_lock(directory_fd, exclusive=False):
+            for filename in sorted(os.listdir(directory_fd)):
+                if not filename.endswith(SNAPSHOT_SUFFIX):
+                    continue
+                name = filename[: -len(SNAPSHOT_SUFFIX)]
+                try:
+                    normalized = normalize_session_name(name)
+                    snapshot = _load_snapshot_at(directory_fd, filename, normalized)
+                except (SnapshotError, FileNotFoundError) as exc:
+                    warnings.append(f"snapshot ignoré {filename}: {exc}")
+                    continue
+                sessions.append({
+                    "name": normalized,
+                    "updated_at": snapshot["metadata"]["updated_at"],
+                    "windows": len(snapshot["windows"]),
+                    "workspaces": len(snapshot["workspaces"]),
+                })
+    finally:
+        os.close(directory_fd)
+    return sessions, warnings
+
+
+def delete_session(
+    name: str,
+    *,
+    environment: Mapping[str, str] | None = None,
+    home: Path | None = None,
+) -> None:
+    normalized = normalize_session_name(name)
+    directory = sessions_directory(environment, home)
+    directory_fd = _open_sessions_directory(directory)
+    filename = _snapshot_filename(normalized)
+    try:
+        with _sessions_lock(directory_fd, exclusive=True):
+            _target_is_safe(directory_fd, filename)
+            try:
+                os.unlink(filename, dir_fd=directory_fd)
+            except FileNotFoundError as exc:
+                raise SnapshotError(f"session introuvable: {normalized}") from exc
+            os.fsync(directory_fd)
+    finally:
+        os.close(directory_fd)
+
+
 def main(argv: Sequence[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(description="Capture read-only d'un Session Snapshot V1 Sway")
+    parser = argparse.ArgumentParser(description="Sauvegarde persistante read-only d'une session 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("--compact", action="store_true", help="émettre un JSON compact")
     parser.add_argument(
         "--autostart",
         type=Path,
         default=Path.home() / ".config/sway/autostart",
         help="fichier Sway autostart utilisé seulement pour la classification",
     )
+    parser.add_argument("command", nargs="?", choices=("save", "list", "show", "delete"))
+    parser.add_argument("name", nargs="?")
     args = parser.parse_args(argv)
-    if not args.stdout:
+    if args.stdout and args.command is not None:
-        parser.error("STEP18A autorise uniquement --stdout")
+        parser.error("--stdout ne peut pas être combiné avec une commande persistante")
-    started = time.monotonic()
+    if not args.stdout and args.command is None:
+        parser.error("utiliser --stdout ou une commande save/list/show/delete")
+    if args.command in {"save", "show", "delete"} and args.name is None:
+        parser.error(f"{args.command} requiert un nom de session")
+    if (args.stdout or args.command == "list") and args.name is not None:
+        parser.error("nom de session inattendu")
     try:
-        snapshot = collect(args.autostart)
+        if args.stdout:
+            started = time.monotonic()
+            result: Any = collect(args.autostart)
+            result["diagnostics"]["collection_duration_ms"] = round((time.monotonic() - started) * 1000, 3)
+            validate_snapshot(result)
+        elif args.command == "save":
+            snapshot = save_session(args.name, args.autostart)
+            result = {
+                "name": snapshot["metadata"]["session_name"],
+                "created_at": snapshot["metadata"]["created_at"],
+                "updated_at": snapshot["metadata"]["updated_at"],
+                "windows": len(snapshot["windows"]),
+                "workspaces": len(snapshot["workspaces"]),
+            }
+        elif args.command == "list":
+            result, warnings = list_sessions()
+            for warning in warnings:
+                print(warning, file=sys.stderr)
+        elif args.command == "show":
+            result = show_session(args.name)
+        else:
+            delete_session(args.name)
+            result = {"deleted": args.name}
     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=(",", ":"))
+        json.dump(result, sys.stdout, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
     else:
-        json.dump(snapshot, sys.stdout, ensure_ascii=False, sort_keys=True, indent=2)
+        json.dump(result, sys.stdout, ensure_ascii=False, sort_keys=True, indent=2)
     sys.stdout.write("\n")
     return 0
 

tests/test_session_snapshot.py

@@ -1,8 +1,13 @@
 """Contrat Session Snapshot V1 sur arbres Sway synthétiques."""
 
 import copy
+import concurrent.futures
 import importlib.util
+import io
+import json
+import os
 from pathlib import Path
+import stat
 import tempfile
 import unittest
 from unittest import mock
@@ -274,5 +279,265 @@ class SessionSnapshotTest(unittest.TestCase):
         self.assertNotIn("environment", serialized_keys)
 
 
+class PersistentSessionTest(unittest.TestCase):
+    """Contrats STEP18B dans un XDG state isolé du compte utilisateur."""
+
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        self.state = self.root / "state"
+        self.environment = {"XDG_STATE_HOME": str(self.state)}
+        version, outputs, workspaces, tree, entries = fixture()
+        with mock.patch.object(snapshot_module, "_basename_from_pid", return_value=None):
+            self.snapshot = 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 collector(self, _autostart):
+        return copy.deepcopy(self.snapshot)
+
+    def save(self, name="dev", **kwargs):
+        return snapshot_module.save_session(
+            name,
+            environment=self.environment,
+            collector=self.collector,
+            retry_delay=0,
+            **kwargs,
+        )
+
+    @property
+    def sessions(self):
+        return self.state / "labfy-sway/sessions"
+
+    def write_raw(self, name, value=None, payload=None):
+        snapshot_module.sessions_directory(self.environment)
+        path = self.sessions / f"{name}.json"
+        if payload is None:
+            payload = json.dumps(value).encode("utf-8")
+        path.write_bytes(payload)
+        path.chmod(0o600)
+        return path
+
+    def test_xdg_path_fallback_and_private_directory_permissions(self):
+        path = snapshot_module.sessions_directory(self.environment)
+        self.assertEqual(path, self.sessions)
+        self.assertEqual(stat.S_IMODE((self.state / "labfy-sway").stat().st_mode), 0o700)
+        self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o700)
+        fallback = snapshot_module.sessions_directory({}, self.root / "home")
+        self.assertEqual(fallback, self.root / "home/.local/state/labfy-sway/sessions")
+        with self.assertRaises(snapshot_module.SnapshotError):
+            snapshot_module.sessions_directory({"XDG_STATE_HOME": "relative"}, self.root)
+
+    def test_safe_and_invalid_names_prevent_path_traversal(self):
+        for name in ("default", "Dev_01", "osint-prod", "A" * 64):
+            self.assertEqual(snapshot_module.normalize_session_name(name), name)
+        for name in ("", "..", ".", "a/b", "../dev", "a b", "é", "A" * 65, "nul\0name"):
+            with self.subTest(name=repr(name)):
+                with self.assertRaises(snapshot_module.SnapshotError):
+                    snapshot_module.normalize_session_name(name)
+
+    def test_atomic_first_save_permissions_and_persistent_metadata(self):
+        previous_umask = os.umask(0)
+        try:
+            saved = self.save()
+        finally:
+            os.umask(previous_umask)
+        path = self.sessions / "dev.json"
+        self.assertTrue(path.is_file())
+        self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
+        self.assertEqual(stat.S_IMODE(self.sessions.stat().st_mode), 0o700)
+        self.assertEqual(saved["schema"], snapshot_module.SCHEMA)
+        self.assertEqual(saved["version"], 1)
+        self.assertEqual(saved["metadata"]["session_name"], "dev")
+        self.assertEqual(saved["metadata"]["created_at"], saved["metadata"]["updated_at"])
+        self.assertFalse(list(self.sessions.glob("*.tmp")))
+
+    def test_atomic_update_preserves_created_at_and_advances_updated_at(self):
+        first = self.save()
+        first_bytes = (self.sessions / "dev.json").read_bytes()
+        self.snapshot["metadata"]["captured_at"] = "2026-01-01T00:00:01.000Z"
+        second = self.save()
+        self.assertEqual(second["metadata"]["created_at"], first["metadata"]["created_at"])
+        self.assertGreater(second["metadata"]["updated_at"], first["metadata"]["updated_at"])
+        self.assertNotEqual(second["metadata"]["captured_at"], first["metadata"]["captured_at"])
+        self.assertNotEqual((self.sessions / "dev.json").read_bytes(), first_bytes)
+        self.assertEqual([path.name for path in self.sessions.glob("*.json")], ["dev.json"])
+
+    def test_list_show_delete_round_trip(self):
+        saved = self.save("trainlog")
+        listed, warnings = snapshot_module.list_sessions(environment=self.environment)
+        self.assertEqual(warnings, [])
+        self.assertEqual(listed, [{
+            "name": "trainlog",
+            "updated_at": saved["metadata"]["updated_at"],
+            "windows": 7,
+            "workspaces": 1,
+        }])
+        shown = snapshot_module.show_session("trainlog", environment=self.environment)
+        self.assertEqual(shown, saved)
+        snapshot_module.delete_session("trainlog", environment=self.environment)
+        self.assertFalse((self.sessions / "trainlog.json").exists())
+        self.assertTrue(self.sessions.is_dir())
+        with self.assertRaises(snapshot_module.SnapshotError):
+            snapshot_module.show_session("trainlog", environment=self.environment)
+
+    def test_corrupt_invalid_relation_and_unknown_files_are_ignored_by_list(self):
+        self.save("valid")
+        self.write_raw("truncated", payload=b'{"schema":')
+        invalid_relation = copy.deepcopy(snapshot_module.show_session("valid", environment=self.environment))
+        invalid_relation["workspaces"][0]["roots"]["tiling"][0]["id"] = "absent"
+        self.write_raw("relation", invalid_relation)
+        unknown_schema = copy.deepcopy(invalid_relation)
+        unknown_schema["schema"] = "unknown"
+        self.write_raw("schema", unknown_schema)
+        listed, warnings = snapshot_module.list_sessions(environment=self.environment)
+        self.assertEqual([item["name"] for item in listed], ["valid"])
+        self.assertEqual(len(warnings), 3)
+        for name in ("truncated", "relation", "schema"):
+            with self.assertRaises(snapshot_module.SnapshotError):
+                snapshot_module.show_session(name, environment=self.environment)
+
+    def test_unknown_version_has_explicit_error(self):
+        snapshot = self.save("future")
+        snapshot["version"] = 99
+        self.write_raw("future", snapshot)
+        with self.assertRaisesRegex(snapshot_module.SnapshotError, "unsupported snapshot version"):
+            snapshot_module.show_session("future", environment=self.environment)
+
+    def test_oversized_snapshot_is_rejected_before_json_loading(self):
+        self.write_raw("huge", payload=b"{" + b" " * snapshot_module.MAX_SNAPSHOT_BYTES)
+        with self.assertRaisesRegex(snapshot_module.SnapshotError, "trop volumineux"):
+            snapshot_module.show_session("huge", environment=self.environment)
+
+    def test_explicit_save_repairs_corrupt_regular_target(self):
+        self.write_raw("repair", payload=b"not-json")
+        saved = self.save("repair")
+        self.assertEqual(snapshot_module.show_session("repair", environment=self.environment), saved)
+
+    def test_symlink_target_is_refused_for_save_show_and_delete(self):
+        snapshot_module.sessions_directory(self.environment)
+        external = self.root / "external"
+        external.write_text("do not touch", encoding="utf-8")
+        (self.sessions / "evil.json").symlink_to(external)
+        with self.assertRaisesRegex(snapshot_module.SnapshotError, "lien symbolique"):
+            self.save("evil")
+        with self.assertRaises(snapshot_module.SnapshotError):
+            snapshot_module.show_session("evil", environment=self.environment)
+        with self.assertRaisesRegex(snapshot_module.SnapshotError, "lien symbolique"):
+            snapshot_module.delete_session("evil", environment=self.environment)
+        self.assertEqual(external.read_text(encoding="utf-8"), "do not touch")
+
+    def test_managed_directory_symlink_is_refused(self):
+        self.state.mkdir()
+        external = self.root / "outside"
+        external.mkdir()
+        (self.state / "labfy-sway").symlink_to(external, target_is_directory=True)
+        with self.assertRaises(snapshot_module.SnapshotError):
+            snapshot_module.sessions_directory(self.environment)
+
+    def test_interrupted_update_keeps_old_snapshot_and_cleans_temporary(self):
+        self.save()
+        path = self.sessions / "dev.json"
+        before = path.read_bytes()
+
+        def interrupt(_temporary):
+            raise snapshot_module.SnapshotError("interruption simulée avant rename")
+
+        with self.assertRaisesRegex(snapshot_module.SnapshotError, "interruption simulée"):
+            self.save(before_replace=interrupt)
+        self.assertEqual(path.read_bytes(), before)
+        self.assertEqual(snapshot_module.show_session("dev", environment=self.environment)["schema"], snapshot_module.SCHEMA)
+        self.assertFalse(list(self.sessions.glob("*.tmp")))
+
+    def test_concurrent_saves_are_serialized_and_valid(self):
+        def operation(_index):
+            return self.save()["metadata"]["updated_at"]
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:
+            timestamps = list(executor.map(operation, range(12)))
+        shown = snapshot_module.show_session("dev", environment=self.environment)
+        self.assertEqual(shown["metadata"]["created_at"], min(timestamps))
+        self.assertEqual(shown["metadata"]["updated_at"], max(timestamps))
+        self.assertFalse(list(self.sessions.glob("*.tmp")))
+
+    def test_retry_inconsistent_ipc_then_success(self):
+        calls = 0
+
+        def flaky(_autostart):
+            nonlocal calls
+            calls += 1
+            if calls < 3:
+                raise snapshot_module.SnapshotError("snapshot V1 invalide: relation IPC")
+            return self.collector(None)
+
+        saved = snapshot_module.save_session(
+            "retry",
+            environment=self.environment,
+            collector=flaky,
+            retry_delay=0,
+        )
+        self.assertEqual(calls, 3)
+        self.assertEqual(saved["metadata"]["session_name"], "retry")
+
+    def test_retry_exhaustion_keeps_previous_snapshot(self):
+        self.save("stable")
+        path = self.sessions / "stable.json"
+        before = path.read_bytes()
+        calls = 0
+
+        def inconsistent(_autostart):
+            nonlocal calls
+            calls += 1
+            raise snapshot_module.SnapshotError("snapshot V1 invalide: relation IPC")
+
+        with self.assertRaisesRegex(snapshot_module.SnapshotError, "SAVE_FAILED_INCONSISTENT_STATE"):
+            snapshot_module.save_session(
+                "stable",
+                environment=self.environment,
+                collector=inconsistent,
+                retry_delay=0,
+            )
+        self.assertEqual(calls, 3)
+        self.assertEqual(path.read_bytes(), before)
+
+    def test_structural_equivalence_excludes_persistence_timestamps_and_titles(self):
+        first = self.save()
+        second = copy.deepcopy(first)
+        second["metadata"]["created_at"] = "2030-01-01T00:00:00.000Z"
+        second["metadata"]["updated_at"] = "2030-01-01T00:00:01.000Z"
+        second["metadata"]["captured_at"] = "2030-01-01T00:00:02.000Z"
+        second["windows"][0]["title_hint"] = "volatile"
+        self.assertEqual(
+            snapshot_module.structural_projection(first),
+            snapshot_module.structural_projection(second),
+        )
+
+    def test_cli_save_list_show_delete(self):
+        outputs = []
+        with mock.patch.dict(os.environ, self.environment, clear=False), mock.patch.object(
+            snapshot_module, "collect", side_effect=lambda _path: self.collector(None)
+        ):
+            for arguments in (("save", "cli"), ("list",), ("show", "cli"), ("delete", "cli")):
+                stdout = io.StringIO()
+                stderr = io.StringIO()
+                with mock.patch("sys.stdout", stdout), mock.patch("sys.stderr", stderr):
+                    self.assertEqual(snapshot_module.main(arguments), 0)
+                self.assertEqual(stderr.getvalue(), "")
+                outputs.append(json.loads(stdout.getvalue()))
+        self.assertEqual(outputs[0]["name"], "cli")
+        self.assertEqual(outputs[1][0]["name"], "cli")
+        self.assertEqual(outputs[2]["metadata"]["session_name"], "cli")
+        self.assertEqual(outputs[3], {"deleted": "cli"})
+        self.assertFalse((self.sessions / "cli.json").exists())
+
+
 if __name__ == "__main__":
     unittest.main()