diff --git a/docs/session-restore-v1.md b/docs/session-restore-v1.md index 3443c58de..f58f64e2b 100644 --- a/docs/session-restore-v1.md +++ b/docs/session-restore-v1.md @@ -1,4 +1,4 @@ -# Restore Plan et Executor V1 — STEP19A/STEP19B +# Restore Plan et Executor V1 — STEP19A/STEP19B/STEP19C ## Statut et architecture @@ -451,3 +451,214 @@ initial lorsqu'il existe encore. Les fenêtres non marquées `TEST_ONLY` sont comparées avant/après sur `con_id`, workspace, floating et fullscreen. L'arbre complet, le scratchpad avancé, le Spatial Canvas et toute UI QML restent hors STEP19B. + +## Tree Restore Compiler et scratchpad — STEP19C + +### Audit local Sway 1.12 / SwayFX 0.6 + +L'implémentation STEP19C est fondée sur la documentation réellement installée +et sur des expériences `get_tree` avec des Kitty `TEST_ONLY`, pas sur une +supposition i3. La version validée est : + +```text +swayfx version 0.6 (based on sway 1.12.0) +swaymsg version 0.6 (based on sway 1.12.0) +``` + +`sway(5)` confirme localement : + +```text +layout default|splith|splitv|stacking|tabbed +split vertical|horizontal|none|toggle +focus parent|child +move left|right|up|down +move container to mark +mark --add|--replace [--toggle] +unmark [] +resize ... px|ppt +move container to scratchpad +scratchpad show +``` + +L'IPC expose le layout `stacked`, tandis que la commande correspondante est +`layout stacking`. Cette conversion est une table locale fermée. `scratchpad +show` sans critère est cyclique et n'est jamais utilisé ; STEP19C reconstruit +toujours `[con_id=] scratchpad show`. + +### Matrice de capacités + +| Élément | Niveau STEP19C | Contrat | +|---|---|---| +| Racine `splith` | `supported-subset` | Container racine homogène, fenêtres toutes matchées | +| Racine `splitv` | `supported-subset` | Même contrat | +| Racine `tabbed` | `supported-subset` | Même contrat, layout final relu | +| Racine `stacked` | `supported-subset` | Commande locale `stacking`, layout IPC `stacked` | +| Ordre des enfants | `supported-subset` | Insertion ciblée après une fenêtre de référence | +| `H[A,V[B,C]]` et symétrique | `supported-subset` | Un seul container imbriqué, deux feuilles, splits orthogonaux | +| Arbre imbriqué arbitraire | `unsupported` | Bloqué avant mutation, jamais aplati silencieusement | +| Proportions `percent` | `unsupported` | Observées mais aucune promesse de restitution V1 | +| Branche floating | `supported-subset` | Préservée, jamais compilée comme enfant tiling | +| Scratchpad caché | `supported-subset` | `move scratchpad` ciblé + relecture V1 | +| Scratchpad visible | `supported-subset` | Workspace cible numérique déjà focalisé + show ciblé | +| Spatial Canvas | `unsupported/out-of-scope` | Aucun comportement `spatial-canvas-v1` | + +Cette matrice ne signifie pas « arbitrary Sway tree restore ». Un changement +de style entre des fenêtres directement sous le workspace et un container +racine explicite, plusieurs containers imbriqués, un nesting en tête, des +splits imbriqués redondants ou un arbre live déjà imbriqué hors forme cible +sont refusés. + +### Architecture et représentation normalisée + +```text +Session Snapshot V1 target + + +Session Snapshot V1 live frais + + +matching source_window → runtime con_id + ↓ +session_tree_restore.py (pur) + ↓ +Tree Restore compilation + ↓ +operations[] typées + ↓ +session_executor.py (effets externes) +``` + +`TreeNode` ne conserve que `kind`, identité de fenêtre sauvegardée, layout, +percent observé et enfants. Il ne prend jamais `container_id`, `con_id` ou PID +sauvegardé comme identité persistante. Le lien live est reconstruit uniquement +depuis le matching frais du planner et le `con_id` observé dans cette session. + +Le compilateur valide d'abord toutes les références V1, l'unicité des feuilles, +la branche tiling/floating, les layouts et la forme. Il compare ensuite les +arbres normalisés. Un arbre identique produit zéro opération. Pour le +sous-ensemble supporté, il produit seulement : + +```text +set-container-layout +set-split-orientation +move-relative(relation=after) +``` + +Les opérations scratchpad sont portées par les actions du plan : + +```text +move-to-scratchpad +show-scratchpad-window +``` + +Elles ne sont jamais des chaînes Sway provenant du snapshot. Les enums sont +validés puis traduits par l'exécuteur en argv directs, sans shell. + +### Ordre et marks temporaires + +`move left/right` n'est pas une primitive d'ordre universelle : sur la version +locale, elle peut extraire une vue d'un container `tabbed` ou `stacked`. +STEP19C utilise donc un mark temporaire uniquement pour `move-relative` : + +```text +[reference] mark --add __labfy_restore_ +[source] move container to mark __labfy_restore_ +[reference] unmark __labfy_restore_ +``` + +Le préfixe est réservé, le suffixe contient 64 bits aléatoires encodés en hex, +et jusqu'à huit candidats sont comparés à tous les marks live avant création. +Le mark est ajouté uniquement à une fenêtre cible déjà revalidée, n'utilise +jamais `--replace`, et est supprimé dans un `finally`. Son apparition et sa +disparition sont relues. Une fuite produit `RESTORE_TEMPORARY_MARK_LEAK`. + +L'ordre cible est construit en gardant le premier enfant comme ancre, puis en +insérant chaque enfant suivant après son prédécesseur cible. Cette règle est +déterministe et fonctionne indépendamment de la direction géométrique du +layout. + +### Nesting supporté + +Le nesting V1 supporté contient exactement un container interne de deux +fenêtres, précédé d'au moins une feuille, avec orientations orthogonales : + +```text +H[A,V[B,C]] +V[A,H[B,C]] +``` + +Le compilateur remet d'abord l'ordre plat, applique le layout racine, crée le +split interne sur B, puis insère C après B via le mark temporaire. `get_tree` +doit ensuite être exactement égal à la topologie cible normalisée. Un arbre +plus profond ou une transition depuis un arbre live imbriqué non reconnu est +`unsupported-live-nested-tree` et bloque le plan entier. + +### Proportions + +Sway 1.12 expose `resize ... ppt`, mais les ajustements sont relatifs aux +frères, arrondis par la géométrie disponible et dépendants de l'ordre des +redimensionnements. Les expériences STEP19C ne justifient donc pas une +restitution générale déterministe des `percent`. Le compilateur conserve les +valeurs pour le diagnostic mais publie : + +```text +percent_capability = unsupported +``` + +Aucune commande `resize` n'est émise et le rapport ne prétend pas restaurer +les proportions exactes ou approchées. + +### Preflight global et dérive + +Avant la première mutation, le capability gate vérifie toutes les actions, +les identités de lancement, les workspaces/outputs, les modes fullscreen, les +cibles scratchpad et la compilation complète d'arbre. Pour chaque workspace +compilé : + +```text +toutes les fenêtres target sont matchées +aucune fenêtre tiling live inattendue n'existe +aucune fenêtre floating ne devient tiling +layout et forme sont dans le sous-ensemble +toutes les opérations sont typées et validées +aucune action manual-required ne subsiste +``` + +Une seule erreur rend le plan entier `RESTORE_BLOCKED_UNSUPPORTED_ACTION` +avant lancement ou mutation. + +Entre deux mutations d'arbre, l'exécuteur recalcule une empreinte contenant le +layout du workspace, les parents, les layouts et l'ordre des fenêtres ciblées. +Elle doit être égale à la postcondition de l'opération précédente. Une action +utilisateur concurrente produit `RESTORE_TREE_DRIFTED` et la commande suivante +n'est pas lancée. Après chaque opération, une nouvelle empreinte doit devenir +visible ; enfin le compilateur est relancé sur le live final et doit produire +zéro opération avec une topologie target/live égale. + +### Scratchpad ciblé et isolation utilisateur + +Le snapshot V1 distingue : + +```text +member=true, visibility=hidden, workspace=null, branch=scratchpad +member=true, visibility=visible, workspace=, branch=floating +``` + +Pour une cible cachée, STEP19C exécute `move scratchpad` si la fenêtre est +hors scratchpad ou visible, puis exige la représentation cachée ci-dessus. +Pour une cible visible, une fenêtre hors scratchpad est d'abord admise, puis +`[con_id=] scratchpad show` est envoyé uniquement à cette fenêtre. Le +workspace numérique cible doit déjà être focalisé au preflight ; sinon le +placement de la vue visible serait ambigu et tout le plan est bloqué. + +Les autres fenêtres scratchpad, y compris celles de l'utilisateur, ne sont ni +cyclées ni marquées. Une cible disparue, une identité changée ou un matching +multiple ambigu arrête l'exécution. Cette stratégie reste sûre même avec +plusieurs membres scratchpad, car aucune commande `scratchpad show` globale +n'existe dans l'exécuteur. + +### Limite non transactionnelle + +Les commandes Sway restent non transactionnelles. STEP19C maximise la sûreté +par preflight intégral, revalidation avant chaque commande, postcondition après +chaque mutation, marks nettoyés et arrêt au premier écart. Il ne promet ni +rollback global, ni restauration d'un arbre arbitraire, ni reconstruction des +proportions, ni UI. `spatial-canvas-v1` demeure entièrement hors périmètre. diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_executor.py b/quickshell/.config/quickshell/labfy-sway/session/session_executor.py index c12e8f3c7..e9f7b1d9b 100644 --- a/quickshell/.config/quickshell/labfy-sway/session/session_executor.py +++ b/quickshell/.config/quickshell/labfy-sway/session/session_executor.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Exécuter le sous-ensemble contrôlé STEP19B d'un Restore Plan V1 frais.""" +"""Exécuter le sous-ensemble contrôlé STEP19B/STEP19C d'un plan V1 frais.""" from __future__ import annotations @@ -7,10 +7,12 @@ import contextlib import copy import fcntl import hashlib +import importlib.util import json import os from pathlib import Path import re +import secrets import stat import subprocess import time @@ -31,12 +33,12 @@ SUPPORTED_ACTIONS = { "move-to-workspace", "restore-floating", "restore-fullscreen", - "restore-focus", -} -UNSUPPORTED_ACTIONS = { "restore-tree-position", "restore-scratchpad-hidden", "restore-scratchpad-visible", + "restore-focus", +} +UNSUPPORTED_ACTIONS = { "manual-required", } NUMERIC_WORKSPACE = re.compile(r"\A[1-9][0-9]*\Z") @@ -94,6 +96,20 @@ def _direct_sway_command(con_id: int, operation: str) -> list[str]: return ["swaymsg", "-r", f"[con_id={con_id}] {operation}"] +def _load_tree_compiler() -> Any: + """Charger le compilateur pur voisin sans lui donner d'effet externe.""" + path = Path(__file__).with_name("session_tree_restore.py") + spec = importlib.util.spec_from_file_location("labfy_session_tree_restore", path) + if spec is None or spec.loader is None: + raise RestoreExecutionError("RESTORE_TREE_COMPILER_UNAVAILABLE", path.name) + module = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(module) + except (OSError, ImportError, SyntaxError) as exc: + raise RestoreExecutionError("RESTORE_TREE_COMPILER_UNAVAILABLE", str(exc)) from exc + return module + + def _source_windows(source: Mapping[str, Any]) -> dict[str, Mapping[str, Any]]: return { str(window.get("window_id")): window @@ -137,6 +153,7 @@ def _revalidate_runtime( *, collector: Callable[[], dict[str, Any]], planner_module: Any, + allow_scratchpad: bool = False, ) -> tuple[Mapping[str, Any], dict[str, Any]]: """Refuser disparition, recyclage d'id, identité divergente ou workspace inconnu.""" live = collector() @@ -145,7 +162,12 @@ def _revalidate_runtime( raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", f"con_id disparu: {con_id}") if not _compatible_match(planner_module, source_window, candidate): raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", f"identité modifiée pour con_id={con_id}") - if not isinstance(candidate.get("workspace"), str) or not candidate.get("workspace"): + scratchpad = candidate.get("scratchpad") + scratch_member = isinstance(scratchpad, Mapping) and scratchpad.get("member") is True + if ( + (not isinstance(candidate.get("workspace"), str) or not candidate.get("workspace")) + and not (allow_scratchpad and scratch_member) + ): raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", f"workspace inconnu pour con_id={con_id}") return candidate, live @@ -224,17 +246,29 @@ def _blocking_finding(action: Mapping[str, Any]) -> dict[str, Any] | None: "reason": "unsupported-fullscreen-mode", "desired_fullscreen_mode": action.get("desired_fullscreen_mode"), } + if action_type in {"restore-scratchpad-hidden", "restore-scratchpad-visible"}: + desired = action.get("desired") + visibility = desired.get("visibility") if isinstance(desired, Mapping) else None + expected = "hidden" if action_type.endswith("hidden") else "visible" + if not isinstance(desired, Mapping) or desired.get("member") is not True or visibility != expected: + return { + "action_id": action_id, + "action": action_type, + "reason": "invalid-scratchpad-target", + } return None def capability_gate( source: Mapping[str, Any], plan: Mapping[str, Any], + live: Mapping[str, Any], *, planner_module: Any, + tree_module: Any, desktop_entries: Sequence[Mapping[str, str | None]], resolve_desktop_entry: Callable[..., dict[str, Any]], -) -> None: +) -> Mapping[str, Any]: """Examiner tout le plan avant la première mutation, sans exception partielle.""" windows = _source_windows(source) findings: list[dict[str, Any]] = [] @@ -276,15 +310,51 @@ def capability_gate( "action": action.get("action"), "reason": "missing-runtime-con-id", }) + if action.get("action") == "restore-scratchpad-visible": + desired_workspace = saved.get("workspace") + live_focus = live.get("focus") + focused_workspace = live_focus.get("workspace") if isinstance(live_focus, Mapping) else None + if ( + not isinstance(desired_workspace, str) + or NUMERIC_WORKSPACE.fullmatch(desired_workspace) is None + or focused_workspace != desired_workspace + ): + findings.append({ + "action_id": action.get("action_id"), + "action": action.get("action"), + "reason": "scratchpad-visible-requires-focused-target-workspace", + }) try: planner_module.validate_restore_plan(plan) except Exception as exc: findings.append({"action_id": None, "action": None, "reason": f"invalid-plan: {exc}"}) + if any( + isinstance(action, Mapping) and action.get("action") == "restore-tree-position" + for action in plan.get("actions", []) + ): + try: + tree_compilation = tree_module.compile_tree_restore(source, live, plan) + tree_module.validate_compilation(tree_compilation) + except Exception as exc: + findings.append({ + "action_id": None, + "action": "restore-tree-position", + "reason": f"tree-preflight: {exc}", + }) + tree_compilation = {"capability": "unsupported", "operations": [], "workspaces": []} + else: + tree_compilation = { + "capability": "supported-subset", + "percent_capability": "unsupported", + "operations": [], + "workspaces": [], + } if findings: raise RestoreExecutionError( "RESTORE_BLOCKED_UNSUPPORTED_ACTION", json.dumps(findings, ensure_ascii=False, sort_keys=True), ) + return tree_compilation @contextlib.contextmanager @@ -402,6 +472,7 @@ def _wait_for_postcondition( planner_module: Any, monotonic: Callable[[], float], sleeper: Callable[[float], None], + allow_scratchpad: bool = False, ) -> Mapping[str, Any]: """Attendre brièvement la visibilité IPC d'une mutation déjà acceptée.""" deadline = monotonic() + POSTCONDITION_TIMEOUT_SECONDS @@ -411,6 +482,7 @@ def _wait_for_postcondition( con_id, collector=collector, planner_module=planner_module, + allow_scratchpad=allow_scratchpad, ) if predicate(candidate): return candidate @@ -419,6 +491,172 @@ def _wait_for_postcondition( sleeper(POLL_INTERVAL_SECONDS) +def _tree_argv( + operation: Mapping[str, Any], + runtime_by_source: Mapping[str, int], + tree_module: Any, +) -> list[str]: + """Construire une commande fermée depuis une opération prévalidée.""" + tree_module.validate_operation(operation) + source_id = str(operation["source_window"]) + con_id = runtime_by_source.get(source_id) + if con_id is None: + raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", f"fenêtre arbre absente: {source_id}") + kind = operation["operation"] + if kind == "set-container-layout": + command_layout = "stacking" if operation["layout"] == "stacked" else operation["layout"] + sway_operation = f"layout {command_layout}" + elif kind == "set-split-orientation": + sway_operation = "split horizontal" if operation["layout"] == "splith" else "split vertical" + else: + raise RestoreExecutionError("RESTORE_TREE_OPERATION_INVALID", "move-relative requiert un mark temporaire") + return _direct_sway_command(con_id, sway_operation) + + +def _all_live_marks(live: Mapping[str, Any]) -> set[str]: + result: set[str] = set() + for collection in (live.get("containers", []), live.get("windows", [])): + for item in collection if isinstance(collection, list) else []: + if isinstance(item, Mapping): + result.update(mark for mark in item.get("marks", []) if isinstance(mark, str)) + return result + + +def _move_relative_with_mark( + operation: Mapping[str, Any], + runtime_by_source: Mapping[str, int], + *, + collector: Callable[[], dict[str, Any]], + runner: Callable[[Sequence[str], float], Any], + monotonic: Callable[[], float], + sleeper: Callable[[float], None], +) -> int: + """Insérer une vue après une référence exacte sans direction géométrique.""" + source_con = runtime_by_source.get(str(operation["source_window"])) + reference_con = runtime_by_source.get(str(operation["reference_window"])) + if source_con is None or reference_con is None or source_con == reference_con: + raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", "référence move-relative invalide") + before = collector() + existing = _all_live_marks(before) + mark = "" + for _attempt in range(8): + candidate = f"__labfy_restore_{secrets.token_hex(8)}" + if candidate not in existing: + mark = candidate + break + if not mark: + raise RestoreExecutionError("RESTORE_TEMPORARY_MARK_COLLISION", "aucun mark unique disponible") + + mark_created = False + commands = 0 + try: + completed = runner(_direct_sway_command(reference_con, f"mark --add {mark}"), 5.0) + _parse_sway_success(completed) + commands += 1 + mark_created = True + marked = collector() + reference = _find_con_id(marked, reference_con) + if reference is None or mark not in reference.get("marks", []): + raise RestoreExecutionError("RESTORE_POSTCONDITION_FAILED", "mark temporaire absent") + completed = runner(_direct_sway_command(source_con, f"move container to mark {mark}"), 5.0) + _parse_sway_success(completed) + commands += 1 + finally: + if mark_created: + try: + completed = runner(_direct_sway_command(reference_con, f"unmark {mark}"), 5.0) + _parse_sway_success(completed) + commands += 1 + except (OSError, subprocess.SubprocessError, RestoreExecutionError): + # Le caller vérifiera l'absence du mark ; l'erreur originale + # reste prioritaire si le move lui-même a échoué. + pass + deadline = monotonic() + POSTCONDITION_TIMEOUT_SECONDS + while True: + after = collector() + if mark not in _all_live_marks(after): + return commands + if monotonic() >= deadline: + raise RestoreExecutionError("RESTORE_TEMPORARY_MARK_LEAK", mark) + sleeper(POLL_INTERVAL_SECONDS) + + +def _execute_tree_compilation( + source: Mapping[str, Any], + plan: Mapping[str, Any], + compilation: Mapping[str, Any], + runtime_by_source: Mapping[str, int], + *, + collector: Callable[[], dict[str, Any]], + planner_module: Any, + tree_module: Any, + runner: Callable[[Sequence[str], float], Any], + monotonic: Callable[[], float], + sleeper: Callable[[float], None], +) -> int: + """Exécuter le bytecode typé avec détection de dérive entre mutations.""" + source_windows = _source_windows(source) + last_fingerprint: dict[str, Any] = {} + commands = 0 + for operation in compilation.get("operations", []): + workspace = str(operation["workspace"]) + before = collector() + for source_id, con_id in runtime_by_source.items(): + saved = source_windows.get(source_id) + if saved is not None and saved.get("workspace") == workspace: + candidate = _find_con_id(before, con_id) + if candidate is None or not _compatible_match(planner_module, saved, candidate): + raise RestoreExecutionError( + "RESTORE_RUNTIME_DIVERGED", + f"inventaire arbre modifié: {source_id}", + ) + fingerprint = tree_module.tree_fingerprint(before, workspace, runtime_by_source) + if workspace in last_fingerprint and fingerprint != last_fingerprint[workspace]: + raise RestoreExecutionError( + "RESTORE_TREE_DRIFTED", + f"arbre modifié entre deux opérations: workspace {workspace}", + ) + if operation["operation"] == "move-relative": + commands += _move_relative_with_mark( + operation, + runtime_by_source, + collector=collector, + runner=runner, + monotonic=monotonic, + sleeper=sleeper, + ) + else: + completed = runner(_tree_argv(operation, runtime_by_source, tree_module), 5.0) + _parse_sway_success(completed) + commands += 1 + deadline = monotonic() + POSTCONDITION_TIMEOUT_SECONDS + while True: + after = collector() + changed = tree_module.tree_fingerprint(after, workspace, runtime_by_source) + if changed != fingerprint: + last_fingerprint[workspace] = changed + break + if monotonic() >= deadline: + raise RestoreExecutionError( + "RESTORE_POSTCONDITION_FAILED", + f"opération arbre sans effet observable: {operation['operation']}", + ) + sleeper(POLL_INTERVAL_SECONDS) + + final_live = collector() + try: + remaining = tree_module.compile_tree_restore(source, final_live, plan) + tree_module.validate_compilation(remaining) + except Exception as exc: + raise RestoreExecutionError("RESTORE_POSTCONDITION_FAILED", f"arbre final invalide: {exc}") from exc + if remaining.get("operations"): + raise RestoreExecutionError("RESTORE_POSTCONDITION_FAILED", "arbre cible non atteint") + for workspace in remaining.get("workspaces", []): + if workspace.get("target_tree") != workspace.get("live_tree"): + raise RestoreExecutionError("RESTORE_POSTCONDITION_FAILED", "topologie finale différente") + return commands + + def execute_fresh_plan( source: Mapping[str, Any], plan: Mapping[str, Any], @@ -431,12 +669,17 @@ def execute_fresh_plan( timeout: float = DEFAULT_WINDOW_TIMEOUT_SECONDS, monotonic: Callable[[], float] = time.monotonic, sleeper: Callable[[float], None] = time.sleep, + tree_module: Any | None = None, ) -> dict[str, Any]: """Appliquer séquentiellement un plan déjà préflighté et arrêter au premier écart.""" - capability_gate( + tree_module = _load_tree_compiler() if tree_module is None else tree_module + preflight_live = collector() + tree_compilation = capability_gate( source, plan, + preflight_live, planner_module=planner_module, + tree_module=tree_module, desktop_entries=desktop_entries, resolve_desktop_entry=resolve_desktop_entry, ) @@ -453,6 +696,7 @@ def execute_fresh_plan( applications_launched = 0 mutating_commands = 0 stopped_reason: str | None = None + tree_executed = False for action, status in zip(plan["actions"], statuses): action_type = action["action"] @@ -462,7 +706,17 @@ def execute_fresh_plan( if action_type == "reuse-window": con_id = _runtime_con_id(action["live_window"]) assert con_id is not None - _revalidate_runtime(saved, con_id, collector=collector, planner_module=planner_module) + saved_scratchpad = saved.get("scratchpad") + _revalidate_runtime( + saved, + con_id, + collector=collector, + planner_module=planner_module, + allow_scratchpad=( + isinstance(saved_scratchpad, Mapping) + and saved_scratchpad.get("member") is True + ), + ) runtime_by_source[source_id] = con_id status.update(status="success", reason="fenêtre live réutilisée et revalidée") elif action_type == "skip-autostart-managed": @@ -513,7 +767,34 @@ def execute_fresh_plan( "RESTORE_RUNTIME_DIVERGED", f"aucun con_id validé pour {source_id}", ) - _revalidate_runtime(saved, con_id, collector=collector, planner_module=planner_module) + saved_scratchpad = saved.get("scratchpad") + _revalidate_runtime( + saved, + con_id, + collector=collector, + planner_module=planner_module, + allow_scratchpad=( + isinstance(saved_scratchpad, Mapping) + and saved_scratchpad.get("member") is True + ), + ) + if action_type == "restore-tree-position": + if not tree_executed: + mutating_commands += _execute_tree_compilation( + source, + plan, + tree_compilation, + runtime_by_source, + collector=collector, + planner_module=planner_module, + tree_module=tree_module, + runner=runner, + monotonic=monotonic, + sleeper=sleeper, + ) + tree_executed = True + status.update(status="success", reason="topologie Sway cible confirmée") + continue if action_type == "move-to-workspace": workspace = str(int(action["desired_workspace"])) operation = f"move container to workspace number {workspace}" @@ -521,6 +802,61 @@ def execute_fresh_plan( operation = "floating enable" if action["desired_floating"] else "floating disable" elif action_type == "restore-fullscreen": operation = "fullscreen enable" if action["desired_fullscreen_mode"] == 1 else "fullscreen disable" + elif action_type in {"restore-scratchpad-hidden", "restore-scratchpad-visible"}: + current = collector() + candidate = _find_con_id(current, con_id) + if candidate is None or not _compatible_match(planner_module, saved, candidate): + raise RestoreExecutionError("RESTORE_RUNTIME_DIVERGED", f"scratchpad cible absente: {con_id}") + scratch = candidate.get("scratchpad") + scratch = scratch if isinstance(scratch, Mapping) else {} + must_hide = ( + action_type == "restore-scratchpad-hidden" + and scratch.get("visibility") != "hidden" + ) + must_admit = ( + action_type == "restore-scratchpad-visible" + and scratch.get("member") is not True + ) + if must_hide or must_admit: + completed = runner(_direct_sway_command(con_id, "move scratchpad"), 5.0) + _parse_sway_success(completed) + mutating_commands += 1 + _wait_for_postcondition( + saved, + con_id, + lambda item: isinstance(item.get("scratchpad"), Mapping) + and item["scratchpad"].get("member") is True + and item["scratchpad"].get("visibility") == "hidden", + "fenêtre non déplacée vers le scratchpad caché", + collector=collector, + planner_module=planner_module, + monotonic=monotonic, + sleeper=sleeper, + allow_scratchpad=True, + ) + scratch = {"member": True, "visibility": "hidden"} + if ( + action_type == "restore-scratchpad-visible" + and scratch.get("visibility") != "visible" + ): + completed = runner(_direct_sway_command(con_id, "scratchpad show"), 5.0) + _parse_sway_success(completed) + mutating_commands += 1 + _wait_for_postcondition( + saved, + con_id, + lambda item: isinstance(item.get("scratchpad"), Mapping) + and item["scratchpad"].get("member") is True + and item["scratchpad"].get("visibility") == "visible", + "fenêtre scratchpad ciblée non visible", + collector=collector, + planner_module=planner_module, + monotonic=monotonic, + sleeper=sleeper, + allow_scratchpad=True, + ) + status.update(status="success", reason="scratchpad ciblé confirmé") + continue else: operation = "focus" completed = runner(_direct_sway_command(con_id, operation), 5.0) diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_restore.py b/quickshell/.config/quickshell/labfy-sway/session/session_restore.py index c79682ead..f7395154f 100644 --- a/quickshell/.config/quickshell/labfy-sway/session/session_restore.py +++ b/quickshell/.config/quickshell/labfy-sway/session/session_restore.py @@ -481,6 +481,9 @@ def build_restore_plan(source: Mapping[str, Any], live: Mapping[str, Any]) -> di base = base_keys[saved_id] current_workspace = live_window.get("workspace") if live_window else None current_output = live_window.get("output") if live_window else None + saved_scratchpad = saved.get("scratchpad") + saved_scratchpad = saved_scratchpad if isinstance(saved_scratchpad, Mapping) else {} + scratchpad_target = saved_scratchpad.get("member") is True if desired_output is not None and desired_output not in live_outputs: add(_draft( f"manual-output:{saved_id}", "manual-required", "workspace", saved, @@ -488,7 +491,10 @@ def build_restore_plan(source: Mapping[str, Any], live: Mapping[str, Any]) -> di desired_output=desired_output, policy="deferred-no-fallback", ), order) continue - if live_window is None or current_workspace != desired_workspace or current_output != desired_output: + if ( + not scratchpad_target + and (live_window is None or current_workspace != desired_workspace or current_output != desired_output) + ): add(_draft( f"workspace:{saved_id}", "move-to-workspace", "workspace", saved, depends=(base,), desired_workspace=desired_workspace, desired_output=desired_output, @@ -505,7 +511,11 @@ def build_restore_plan(source: Mapping[str, Any], live: Mapping[str, Any]) -> di # CONTRACT: le planner ne demande une reconstruction d'arbre que pour # une fenêtre déjà matchée dont la structure diffère effectivement. # INVARIANT: l'absence n'est jamais assimilée à une structure divergente. - if live_window is not None and current_placement != desired_placement: + if ( + live_window is not None + and saved_scratchpad.get("member") is not True + and current_placement != desired_placement + ): add(_draft( f"tree:{saved_id}", "restore-tree-position", "tree-layout", saved, depends=(base,), capability="partially-supported", diff --git a/quickshell/.config/quickshell/labfy-sway/session/session_tree_restore.py b/quickshell/.config/quickshell/labfy-sway/session/session_tree_restore.py new file mode 100644 index 000000000..84bd6b34c --- /dev/null +++ b/quickshell/.config/quickshell/labfy-sway/session/session_tree_restore.py @@ -0,0 +1,423 @@ +#!/usr/bin/env python3 +"""Compiler un sous-ensemble prouvé d'arbres Sway en opérations typées.""" + +from __future__ import annotations + +import copy +import re +from typing import Any, Mapping, NamedTuple, Sequence + + +SUPPORTED_LAYOUTS = {"splith", "splitv", "tabbed", "stacked"} +SPLIT_LAYOUTS = {"splith", "splitv"} +SAFE_SOURCE_ID = re.compile(r"\Aw[1-9][0-9]*\Z") +SAFE_WORKSPACE = re.compile(r"\A[1-9][0-9]*\Z") + + +class TreeRestoreError(RuntimeError): + """Topologie invalide ou extérieure au sous-ensemble STEP19C.""" + + def __init__(self, reason: str): + super().__init__(reason) + self.reason = reason + + +class TreeNode(NamedTuple): + """Nœud indépendant des identifiants de containers Sway runtime.""" + + kind: str + key: str | None + layout: str | None + percent: float | None + children: tuple["TreeNode", ...] = () + + +def _require(condition: bool, reason: str) -> None: + if not condition: + raise TreeRestoreError(reason) + + +def _number(value: Any) -> float | None: + if value is None: + return None + _require(isinstance(value, (int, float)) and not isinstance(value, bool), "invalid-percent") + result = float(value) + _require(0.0 <= result <= 1.0, "invalid-percent") + return result + + +def _collections(snapshot: Mapping[str, Any]) -> tuple[dict[str, Mapping[str, Any]], dict[str, Mapping[str, Any]]]: + containers = { + str(item.get("container_id")): item + for item in snapshot.get("containers", []) + if isinstance(item, Mapping) and isinstance(item.get("container_id"), str) + } + windows = { + str(item.get("window_id")): item + for item in snapshot.get("windows", []) + if isinstance(item, Mapping) and isinstance(item.get("window_id"), str) + } + return containers, windows + + +def _node_from_ref( + ref: Mapping[str, Any], + containers: Mapping[str, Mapping[str, Any]], + windows: Mapping[str, Mapping[str, Any]], + window_keys: Mapping[str, str], + seen: set[str], +) -> TreeNode: + kind = ref.get("kind") + object_id = ref.get("id") + _require(kind in {"container", "window"} and isinstance(object_id, str), "invalid-tree-reference") + _require(object_id not in seen, "duplicate-tree-reference") + seen.add(object_id) + if kind == "window": + _require(object_id in windows and object_id in window_keys, "missing-target-window") + window = windows[object_id] + _require(window.get("floating") is False, "floating-window-in-tiling-branch") + return TreeNode("window", window_keys[object_id], None, _number(window.get("percent"))) + _require(object_id in containers, "missing-target-container") + container = containers[object_id] + layout = container.get("layout") + _require(layout in SUPPORTED_LAYOUTS, "unsupported-layout") + floating_children = container.get("floating_children") + _require(isinstance(floating_children, list) and not floating_children, "floating-child-in-tiling-container") + children = container.get("children") + _require(isinstance(children, list) and children, "empty-or-invalid-container") + return TreeNode( + "container", + None, + str(layout), + _number(container.get("percent")), + tuple( + _node_from_ref(child, containers, windows, window_keys, seen) + for child in children + if isinstance(child, Mapping) + ), + ) + + +def normalized_workspace_tree( + snapshot: Mapping[str, Any], + workspace_name: str, + window_keys: Mapping[str, str], +) -> tuple[TreeNode, ...]: + """Projeter uniquement la branche tiling en identités persistantes/matchées.""" + _require(SAFE_WORKSPACE.fullmatch(workspace_name) is not None, "unsupported-non-numeric-workspace") + workspace = next( + ( + item for item in snapshot.get("workspaces", []) + if isinstance(item, Mapping) and item.get("name") == workspace_name + ), + None, + ) + _require(workspace is not None, "missing-workspace") + roots = workspace.get("roots") + roots = roots if isinstance(roots, Mapping) else {} + tiling = roots.get("tiling") + _require(isinstance(tiling, list), "invalid-workspace-roots") + containers, windows = _collections(snapshot) + seen: set[str] = set() + return tuple( + _node_from_ref(ref, containers, windows, window_keys, seen) + for ref in tiling + if isinstance(ref, Mapping) + ) + + +def _leaves(nodes: Sequence[TreeNode]) -> list[str]: + result: list[str] = [] + for node in nodes: + if node.kind == "window": + assert node.key is not None + result.append(node.key) + else: + result.extend(_leaves(node.children)) + return result + + +def _shape(nodes: Sequence[TreeNode]) -> Any: + return [ + ("window", node.key) + if node.kind == "window" + else ("container", node.layout, _shape(node.children)) + for node in nodes + ] + + +def _workspace_layout(snapshot: Mapping[str, Any], workspace_name: str) -> str: + workspace = next( + ( + item for item in snapshot.get("workspaces", []) + if isinstance(item, Mapping) and item.get("name") == workspace_name + ), + None, + ) + _require(workspace is not None and workspace.get("layout") in SUPPORTED_LAYOUTS, "unsupported-workspace-layout") + return str(workspace["layout"]) + + +def _flat_parent(nodes: Sequence[TreeNode], workspace_layout: str) -> tuple[str, list[str], str]: + """Retourner layout, ordre et style de racine pour une fratrie sans nesting.""" + if nodes and all(node.kind == "window" for node in nodes): + return workspace_layout, _leaves(nodes), "workspace-direct" + if len(nodes) == 1 and nodes[0].kind == "container" and all( + child.kind == "window" for child in nodes[0].children + ): + container = nodes[0] + assert container.layout is not None + return container.layout, _leaves(container.children), "container-root" + raise TreeRestoreError("unsupported-nested-tree") + + +def _target_shape(nodes: Sequence[TreeNode], workspace_layout: str) -> dict[str, Any]: + try: + layout, order, style = _flat_parent(nodes, workspace_layout) + return {"kind": "flat", "layout": layout, "order": order, "style": style} + except TreeRestoreError: + pass + + outer: TreeNode | None = None + style = "workspace-direct" + children: Sequence[TreeNode] = nodes + if len(nodes) == 1 and nodes[0].kind == "container": + outer = nodes[0] + children = outer.children + style = "container-root" + nested_indexes = [index for index, child in enumerate(children) if child.kind == "container"] + _require(len(nested_indexes) == 1, "unsupported-nested-tree") + nested_index = nested_indexes[0] + nested = children[nested_index] + _require(all(child.kind == "window" for child in children if child is not nested), "unsupported-nested-tree") + _require(len(nested.children) == 2 and all(child.kind == "window" for child in nested.children), "unsupported-nested-tree") + outer_layout = outer.layout if outer is not None else workspace_layout + _require(outer_layout in SPLIT_LAYOUTS and nested.layout in SPLIT_LAYOUTS, "unsupported-nested-layout") + _require(outer_layout != nested.layout, "redundant-nested-split") + _require(nested_index > 0, "unsupported-leading-nested-container") + return { + "kind": "two-level", + "layout": outer_layout, + "nested_layout": nested.layout, + "nested_index": nested_index, + "order": _leaves(children), + "nested_order": _leaves(nested.children), + "style": style, + } + + +def _operation(kind: str, workspace: str, **values: Any) -> dict[str, Any]: + # CONTRACT: ces dictionnaires sont une somme fermée ; aucune chaîne Sway + # provenant d'un snapshot n'est transportée jusqu'à l'exécuteur. + return {"operation": kind, "workspace": workspace, **values} + + +def _bubble_order(workspace: str, current: list[str], target: Sequence[str]) -> list[dict[str, Any]]: + operations: list[dict[str, Any]] = [] + # WHY: move left/right peut extraire une vue d'un container tabbed/stacked. + # CONTRACT: move-to-mark insère la source après une référence exacte ; le + # premier enfant cible sert donc d'ancre et chaque suivant est réinséré. + for wanted_index in range(1, len(target)): + wanted = target[wanted_index] + reference = target[wanted_index - 1] + current_index = current.index(wanted) + reference_index = current.index(reference) + if current_index == reference_index + 1: + continue + operations.append(_operation( + "move-relative", + workspace, + source_window=wanted, + relation="after", + reference_window=reference, + )) + current.pop(current_index) + reference_index = current.index(reference) + current.insert(reference_index + 1, wanted) + return operations + + +def _workspace_names_for_tree_actions( + source: Mapping[str, Any], + plan: Mapping[str, Any], +) -> list[str]: + source_windows = { + str(item.get("window_id")): item + for item in source.get("windows", []) + if isinstance(item, Mapping) + } + result: set[str] = set() + for action in plan.get("actions", []): + if not isinstance(action, Mapping) or action.get("action") != "restore-tree-position": + continue + window = source_windows.get(str(action.get("source_window"))) + workspace = window.get("workspace") if window else None + _require(isinstance(workspace, str), "tree-action-without-workspace") + result.add(workspace) + return sorted(result, key=lambda value: int(value)) + + +def compile_tree_restore( + source: Mapping[str, Any], + live: Mapping[str, Any], + plan: Mapping[str, Any], +) -> dict[str, Any]: + """Compiler sans effet externe le sous-ensemble localement démontré. + + WHY: les containers et PID sauvegardés sont volatils ; seules les fenêtres + rematchées relient les deux arbres. CONTRACT: toute fenêtre tiling du + workspace doit appartenir au target, sinon le preflight bloque. INVARIANT: + la sortie est déterministe et ne contient aucune commande libre. + """ + mappings: dict[str, int] = {} + live_id_by_con: dict[int, str] = {} + for action in plan.get("actions", []): + if not isinstance(action, Mapping) or action.get("action") != "reuse-window": + continue + source_id = action.get("source_window") + live_window = action.get("live_window") + runtime = live_window.get("runtime") if isinstance(live_window, Mapping) else None + con_id = runtime.get("con_id") if isinstance(runtime, Mapping) else None + _require( + isinstance(source_id, str) + and SAFE_SOURCE_ID.fullmatch(source_id) is not None + and isinstance(con_id, int) + and not isinstance(con_id, bool) + and con_id > 0, + "invalid-runtime-mapping", + ) + _require(con_id not in live_id_by_con, "duplicate-runtime-window") + mappings[source_id] = con_id + live_id_by_con[con_id] = source_id + + source_windows = { + str(item.get("window_id")): item + for item in source.get("windows", []) + if isinstance(item, Mapping) + } + live_windows = { + str(item.get("window_id")): item + for item in live.get("windows", []) + if isinstance(item, Mapping) + } + source_keys = {window_id: window_id for window_id in source_windows} + live_keys: dict[str, str] = {} + for live_id, window in live_windows.items(): + runtime = window.get("runtime") + con_id = runtime.get("con_id") if isinstance(runtime, Mapping) else None + if isinstance(con_id, int) and con_id in live_id_by_con: + live_keys[live_id] = live_id_by_con[con_id] + + all_operations: list[dict[str, Any]] = [] + workspaces: list[dict[str, Any]] = [] + for workspace in _workspace_names_for_tree_actions(source, plan): + target_nodes = normalized_workspace_tree(source, workspace, source_keys) + live_nodes = normalized_workspace_tree(live, workspace, live_keys) + target_workspace_layout = _workspace_layout(source, workspace) + live_workspace_layout = _workspace_layout(live, workspace) + target_leaves = _leaves(target_nodes) + live_leaves = _leaves(live_nodes) + _require(target_leaves, "empty-target-tree") + _require(set(target_leaves) <= set(mappings), "missing-target-window") + _require(set(live_leaves) == set(target_leaves), "unexpected-live-window") + _require(len(live_leaves) == len(set(live_leaves)), "duplicate-live-window") + target = _target_shape(target_nodes, target_workspace_layout) + identical = ( + _shape(target_nodes) == _shape(live_nodes) + and (target["style"] != "workspace-direct" or target_workspace_layout == live_workspace_layout) + ) + operations: list[dict[str, Any]] = [] + if not identical: + try: + live_layout, live_order, live_style = _flat_parent(live_nodes, live_workspace_layout) + except TreeRestoreError as exc: + raise TreeRestoreError("unsupported-live-nested-tree") from exc + _require(target["style"] == live_style, "unsupported-root-style-transition") + operations.extend(_bubble_order(workspace, list(live_order), target["order"])) + anchor = target["order"][0] + if live_layout != target["layout"]: + operations.append(_operation( + "set-container-layout", + workspace, + source_window=anchor, + layout=target["layout"], + )) + if target["kind"] == "two-level": + nested_first, nested_second = target["nested_order"] + operations.append(_operation( + "set-split-orientation", + workspace, + source_window=nested_first, + layout=target["nested_layout"], + )) + operations.append(_operation( + "move-relative", + workspace, + source_window=nested_second, + relation="after", + reference_window=nested_first, + )) + workspaces.append({ + "workspace": workspace, + "capability": "supported-subset", + "target_shape": copy.deepcopy(target), + "target_tree": _shape(target_nodes), + "live_tree": _shape(live_nodes), + "runtime_con_ids": {key: mappings[key] for key in target_leaves}, + "percent_capability": "unsupported", + "operations": copy.deepcopy(operations), + }) + all_operations.extend(operations) + return { + "capability": "supported-subset", + "percent_capability": "unsupported", + "workspaces": workspaces, + "operations": all_operations, + } + + +def validate_operation(operation: Mapping[str, Any]) -> None: + """Refuser toute valeur hors de la somme typée avant construction argv.""" + kind = operation.get("operation") + _require(kind in {"set-container-layout", "set-split-orientation", "move-relative"}, "invalid-operation") + workspace = operation.get("workspace") + source = operation.get("source_window") + _require(isinstance(workspace, str) and SAFE_WORKSPACE.fullmatch(workspace) is not None, "invalid-operation-workspace") + _require(isinstance(source, str) and SAFE_SOURCE_ID.fullmatch(source) is not None, "invalid-operation-window") + if kind in {"set-container-layout", "set-split-orientation"}: + layout = operation.get("layout") + allowed = SUPPORTED_LAYOUTS if kind == "set-container-layout" else SPLIT_LAYOUTS + _require(layout in allowed, "invalid-operation-layout") + else: + _require(operation.get("relation") == "after", "invalid-operation-relation") + reference = operation.get("reference_window") + _require(isinstance(reference, str) and SAFE_SOURCE_ID.fullmatch(reference) is not None, "invalid-operation-reference") + + +def validate_compilation(compilation: Mapping[str, Any]) -> None: + _require(compilation.get("capability") == "supported-subset", "invalid-capability") + operations = compilation.get("operations") + _require(isinstance(operations, list), "invalid-operations") + for operation in operations: + _require(isinstance(operation, Mapping), "invalid-operation") + validate_operation(operation) + + +def tree_fingerprint( + live: Mapping[str, Any], + workspace: str, + runtime_by_source: Mapping[str, int], +) -> Any: + """Empreinte structurelle pour détecter une dérive entre deux mutations.""" + live_keys: dict[str, str] = {} + source_by_con = {con_id: source_id for source_id, con_id in runtime_by_source.items()} + for item in live.get("windows", []): + if not isinstance(item, Mapping): + continue + runtime = item.get("runtime") + con_id = runtime.get("con_id") if isinstance(runtime, Mapping) else None + window_id = item.get("window_id") + if isinstance(con_id, int) and con_id in source_by_con and isinstance(window_id, str): + live_keys[window_id] = source_by_con[con_id] + nodes = normalized_workspace_tree(live, workspace, live_keys) + return (_workspace_layout(live, workspace), _shape(nodes)) diff --git a/tests/test_session_executor.py b/tests/test_session_executor.py index e556b3b5b..60e18fb34 100644 --- a/tests/test_session_executor.py +++ b/tests/test_session_executor.py @@ -1,4 +1,4 @@ -"""STEP19B : exécuteur contrôlé, tous les effets externes sont simulés.""" +"""STEP19B/19C : exécuteur contrôlé, tous les effets externes sont simulés.""" from __future__ import annotations @@ -49,14 +49,25 @@ def kitty_saved(**kwargs): class FakeWorld: - def __init__(self, windows=(), *, launch_window=None): + def __init__(self, windows=(), *, launch_window=None, focused_workspace=None): self.windows = [copy.deepcopy(item) for item in windows] self.launch_window = copy.deepcopy(launch_window) + self.focused_workspace = focused_workspace self.commands = [] def collect(self): focused = next((item["window_id"] for item in self.windows if item.get("focused")), None) - return session(self.windows, focused=focused) + value = session(self.windows, focused=focused) + if self.focused_workspace is not None: + if not any(item.get("name") == self.focused_workspace for item in value["workspaces"]): + value["workspaces"].append({ + "name": self.focused_workspace, + "output": "eDP-1", + "layout": "splith", + "orientation": "horizontal", + }) + value["focus"]["workspace"] = self.focused_workspace + return value def run(self, argv, timeout): self.commands.append(list(argv)) @@ -79,6 +90,14 @@ class FakeWorld: target["fullscreen_mode"] = 1 elif operation == "fullscreen disable": target["fullscreen_mode"] = 0 + elif operation == "move scratchpad": + target["scratchpad"] = {"member": True, "visibility": "hidden", "state": "fresh"} + target["workspace"] = None + target["output"] = None + elif operation == "scratchpad show": + target["scratchpad"] = {"member": True, "visibility": "visible", "state": "changed"} + target["workspace"] = self.focused_workspace or "3" + target["output"] = "eDP-1" elif operation == "focus": for item in self.windows: item["focused"] = item is target @@ -466,6 +485,112 @@ class SessionExecutorTest(unittest.TestCase): self.assertNotIn("touch", serialized) self.assertNotIn("title", serialized) + def test_scratchpad_hidden_target_is_targeted_and_confirmed(self): + hidden = {"member": True, "visibility": "hidden", "state": "fresh"} + saved = kitty_saved(workspace=None, output=None, scratchpad=hidden) + live = kitty_saved(workspace="3", con_id=81) + world = FakeWorld([live], focused_workspace="3") + result, plan = execute(session([saved], persistent=True), world) + self.assertEqual(result["status"], "success") + self.assertIn("restore-scratchpad-hidden", [item["action"] for item in plan["actions"]]) + self.assertIn(["swaymsg", "-r", "[con_id=81] move scratchpad"], world.commands) + + def test_scratchpad_visible_target_uses_targeted_show(self): + visible = {"member": True, "visibility": "visible", "state": "changed"} + hidden = {"member": True, "visibility": "hidden", "state": "fresh"} + saved = kitty_saved(workspace="3", scratchpad=visible) + live = kitty_saved(workspace=None, output=None, con_id=82, scratchpad=hidden) + world = FakeWorld([live], focused_workspace="3") + result, plan = execute(session([saved], persistent=True), world) + self.assertEqual(result["status"], "success") + self.assertIn("restore-scratchpad-visible", [item["action"] for item in plan["actions"]]) + self.assertEqual(world.commands, [["swaymsg", "-r", "[con_id=82] scratchpad show"]]) + + def test_scratchpad_already_hidden_is_noop(self): + hidden = {"member": True, "visibility": "hidden", "state": "fresh"} + saved = kitty_saved(workspace=None, output=None, scratchpad=hidden) + live = kitty_saved(workspace=None, output=None, con_id=83, scratchpad=hidden) + world = FakeWorld([live], focused_workspace="3") + result, plan = execute(session([saved], persistent=True), world) + self.assertEqual(result["status"], "success") + self.assertNotIn("restore-scratchpad-hidden", [item["action"] for item in plan["actions"]]) + self.assertEqual(world.commands, []) + + def test_scratchpad_already_visible_is_noop(self): + visible = {"member": True, "visibility": "visible", "state": "changed"} + saved = kitty_saved(workspace="3", scratchpad=visible) + live = kitty_saved(workspace="3", con_id=84, scratchpad=visible) + world = FakeWorld([live], focused_workspace="3") + result, plan = execute(session([saved], persistent=True), world) + self.assertEqual(result["status"], "success") + self.assertNotIn("restore-scratchpad-visible", [item["action"] for item in plan["actions"]]) + self.assertEqual(world.commands, []) + + def test_unexpected_user_scratchpad_is_never_cycled(self): + hidden = {"member": True, "visibility": "hidden", "state": "fresh"} + saved = kitty_saved(workspace=None, output=None, scratchpad=hidden) + target = kitty_saved(workspace="3", con_id=85) + user = window( + "w9", "firefox", desktop_entry="firefox.desktop", executable="firefox", + workspace=None, output=None, con_id=99, pid=9999, scratchpad=hidden, + ) + world = FakeWorld([target, user], focused_workspace="3") + result, _plan = execute(session([saved], persistent=True), world) + self.assertEqual(result["status"], "success") + self.assertEqual(world.commands, [["swaymsg", "-r", "[con_id=85] move scratchpad"]]) + self.assertEqual(world.windows[1]["scratchpad"], hidden) + + def test_scratchpad_target_disappeared_stops(self): + hidden = {"member": True, "visibility": "hidden", "state": "fresh"} + saved = kitty_saved(workspace=None, output=None, scratchpad=hidden) + live = kitty_saved(workspace="3", con_id=86) + world = FakeWorld([live], focused_workspace="3") + plan = planner.build_restore_plan(session([saved], persistent=True), world.collect()) + world.windows.clear() + result = executor.execute_fresh_plan( + session([saved], persistent=True), plan, collector=world.collect, + planner_module=planner, desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, runner=world.run, + ) + self.assertEqual(result["status"], "failed") + self.assertIn("con_id disparu", result["reason"]) + self.assertEqual(world.commands, []) + + def test_scratchpad_identity_changed_stops(self): + hidden = {"member": True, "visibility": "hidden", "state": "fresh"} + saved = kitty_saved(workspace=None, output=None, scratchpad=hidden) + live = kitty_saved(workspace="3", con_id=87) + world = FakeWorld([live], focused_workspace="3") + plan = planner.build_restore_plan(session([saved], persistent=True), world.collect()) + world.windows[0]["app_id"] = "firefox" + world.windows[0]["restore_identity"]["app_id"] = "firefox" + world.windows[0]["restore_identity"]["desktop_entry"] = "firefox.desktop" + world.windows[0]["executable_basename"] = "firefox" + result = executor.execute_fresh_plan( + session([saved], persistent=True), plan, collector=world.collect, + planner_module=planner, desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, runner=world.run, + ) + self.assertEqual(result["status"], "failed") + self.assertIn("identité modifiée", result["reason"]) + self.assertEqual(world.commands, []) + + def test_multiple_scratchpad_identity_is_ambiguous(self): + visible = {"member": True, "visibility": "visible", "state": "changed"} + hidden = {"member": True, "visibility": "hidden", "state": "fresh"} + saved = kitty_saved(workspace="3", scratchpad=visible) + one = kitty_saved(workspace=None, output=None, con_id=88, pid=188, scratchpad=hidden) + two = kitty_saved(workspace=None, output=None, con_id=89, pid=189, scratchpad=hidden) + world = FakeWorld([one, two], focused_workspace="3") + plan = planner.build_restore_plan(session([saved], persistent=True), world.collect()) + with self.assertRaisesRegex(executor.RestoreExecutionError, "ambiguous-runtime-match"): + executor.execute_fresh_plan( + session([saved], persistent=True), plan, collector=world.collect, + planner_module=planner, desktop_entries=DESKTOP_ENTRIES, + resolve_desktop_entry=snapshot_backend.resolve_desktop_entry, runner=world.run, + ) + self.assertEqual(world.commands, []) + def test_source_contains_no_shell_execution_primitive(self): source = (SESSION_DIR / "session_executor.py").read_text(encoding="utf-8") self.assertNotIn("shell=True", source) diff --git a/tests/test_session_tree_restore.py b/tests/test_session_tree_restore.py new file mode 100644 index 000000000..42edeff8e --- /dev/null +++ b/tests/test_session_tree_restore.py @@ -0,0 +1,252 @@ +"""STEP19C : compilation pure du sous-ensemble d'arbre Sway prouvé.""" + +from __future__ import annotations + +import copy +import importlib.util +from pathlib import Path +from types import SimpleNamespace +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +MODULE = ROOT / "quickshell/.config/quickshell/labfy-sway/session/session_tree_restore.py" + + +def load_module(): + spec = importlib.util.spec_from_file_location("session_tree_restore_test", MODULE) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +tree = load_module() + + +def load_executor(): + path = ROOT / "quickshell/.config/quickshell/labfy-sway/session/session_executor.py" + spec = importlib.util.spec_from_file_location("session_tree_executor_test", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +executor = load_executor() + + +def make_snapshot( + *, + live=False, + layout="splith", + order=(1, 2, 3), + nested=False, + workspace_layout="splith", + floating_four=False, +): + prefix = "l" if live else "w" + con_base = 100 if live else 900 + windows = [] + for number in range(1, 5 if floating_four else 4): + window_id = f"{prefix}{number}" + parent = f"{prefix}root" + branch = "tiling" + index = order.index(number) if number in order else 0 + floating = False + if number == 4: + parent = None + branch = "floating" + index = 0 + floating = True + elif nested and number in {2, 3}: + parent = f"{prefix}nested" + index = number - 2 + windows.append({ + "window_id": window_id, + "runtime": {"con_id": con_base + number, "pid": 200 + number}, + "workspace": "8", + "floating": floating, + "percent": 0.5 if nested and number in {2, 3} else 1 / 3, + "tree_position": {"parent_container_id": parent, "branch": branch, "index": index}, + }) + children = [] + for number in order: + if nested and number == 2: + children.append({"kind": "container", "id": f"{prefix}nested"}) + continue + if nested and number == 3: + continue + children.append({"kind": "window", "id": f"{prefix}{number}"}) + containers = [{ + "container_id": f"{prefix}root", + "layout": layout, + "percent": 1.0, + "children": children, + "floating_children": [], + }] + if nested: + containers.append({ + "container_id": f"{prefix}nested", + "layout": "splitv" if layout == "splith" else "splith", + "percent": 2 / 3, + "children": [ + {"kind": "window", "id": f"{prefix}2"}, + {"kind": "window", "id": f"{prefix}3"}, + ], + "floating_children": [], + }) + roots = { + "tiling": [{"kind": "container", "id": f"{prefix}root"}], + "floating": ([{"kind": "window", "id": f"{prefix}4"}] if floating_four else []), + } + return { + "workspaces": [{"name": "8", "layout": workspace_layout, "roots": roots}], + "containers": containers, + "windows": windows, + } + + +def make_plan(source, live, *, omit=()): + actions = [] + for number in range(1, len(source["windows"]) + 1): + if number in omit: + continue + live_window = live["windows"][number - 1] + actions.append({ + "action": "reuse-window", + "source_window": f"w{number}", + "live_window": copy.deepcopy(live_window), + }) + actions.append({"action": "restore-tree-position", "source_window": "w1"}) + return {"actions": actions} + + +class TreeCompilerTest(unittest.TestCase): + def compile(self, source, live, **kwargs): + return tree.compile_tree_restore(source, live, make_plan(source, live, **kwargs)) + + def test_identical_tree_has_no_operations(self): + result = self.compile(make_snapshot(), make_snapshot(live=True)) + self.assertEqual(result["operations"], []) + + def test_root_splith(self): + result = self.compile(make_snapshot(layout="splith"), make_snapshot(live=True, layout="tabbed")) + self.assertEqual(result["operations"][-1]["layout"], "splith") + + def test_root_splitv(self): + result = self.compile(make_snapshot(layout="splitv"), make_snapshot(live=True, layout="splith")) + self.assertEqual(result["operations"][-1]["layout"], "splitv") + + def test_root_tabbed(self): + result = self.compile(make_snapshot(layout="tabbed"), make_snapshot(live=True, layout="splith")) + self.assertEqual(result["operations"][-1]["layout"], "tabbed") + + def test_root_stacked(self): + result = self.compile(make_snapshot(layout="stacked"), make_snapshot(live=True, layout="splith")) + self.assertEqual(result["operations"][-1]["layout"], "stacked") + + def test_child_order_diff_uses_typed_relative_moves(self): + result = self.compile(make_snapshot(order=(1, 2, 3)), make_snapshot(live=True, order=(3, 1, 2))) + self.assertEqual([item["operation"] for item in result["operations"]], ["move-relative"]) + self.assertEqual(result["operations"][0]["relation"], "after") + + def test_nested_supported_tree(self): + result = self.compile(make_snapshot(nested=True), make_snapshot(live=True)) + self.assertEqual( + [item["operation"] for item in result["operations"]], + ["set-split-orientation", "move-relative"], + ) + + def test_nested_unsupported_tree(self): + source = make_snapshot(nested=True) + nested = source["containers"][1] + nested["children"] = [{"kind": "container", "id": "wdeep"}] + source["containers"].append({ + "container_id": "wdeep", "layout": "splith", "percent": 1.0, + "children": [{"kind": "window", "id": "w2"}, {"kind": "window", "id": "w3"}], + "floating_children": [], + }) + with self.assertRaisesRegex(tree.TreeRestoreError, "unsupported-nested-tree"): + self.compile(source, make_snapshot(live=True)) + + def test_missing_target_window(self): + with self.assertRaisesRegex(tree.TreeRestoreError, "missing-target-window"): + self.compile(make_snapshot(), make_snapshot(live=True), omit=(3,)) + + def test_unexpected_live_window(self): + source = make_snapshot(floating_four=True) + live = make_snapshot(live=True, floating_four=True) + live["workspaces"][0]["roots"]["floating"] = [] + live["workspaces"][0]["roots"]["tiling"].append({"kind": "window", "id": "l4"}) + live["windows"][3]["floating"] = False + with self.assertRaisesRegex(tree.TreeRestoreError, "unexpected-live-window"): + self.compile(source, live) + + def test_floating_branch_is_preserved_and_excluded(self): + result = self.compile(make_snapshot(floating_four=True), make_snapshot(live=True, floating_four=True)) + self.assertEqual(result["operations"], []) + self.assertNotIn("w4", repr(result["workspaces"][0]["target_tree"])) + + def test_percent_is_explicitly_unsupported(self): + source = make_snapshot() + live = make_snapshot(live=True) + live["windows"][0]["percent"] = 0.9 + result = self.compile(source, live) + self.assertEqual(result["percent_capability"], "unsupported") + self.assertEqual(result["operations"], []) + + def test_determinism(self): + source = make_snapshot(layout="stacked", order=(1, 2, 3)) + live = make_snapshot(live=True, layout="splitv", order=(3, 2, 1)) + self.assertEqual(self.compile(source, live), self.compile(copy.deepcopy(source), copy.deepcopy(live))) + + def test_hostile_values_are_rejected(self): + with self.assertRaisesRegex(tree.TreeRestoreError, "invalid-operation-layout"): + tree.validate_operation({ + "operation": "set-container-layout", "workspace": "8", "source_window": "w1", + "layout": "tabbed; exec evil", + }) + with self.assertRaisesRegex(tree.TreeRestoreError, "invalid-operation-workspace"): + tree.validate_operation({ + "operation": "move-relative", "workspace": "8; exec evil", "source_window": "w1", + "relation": "after", "reference_window": "w2", + }) + + def test_runtime_tree_drift_stops_before_second_command(self): + source = make_snapshot(layout="tabbed") + live_before = make_snapshot(live=True, layout="splith") + live_after_first = make_snapshot(live=True, layout="tabbed") + live_drifted = make_snapshot(live=True, layout="tabbed", order=(2, 1, 3)) + states = iter((live_before, live_after_first, live_drifted)) + commands = [] + + class Planner: + @staticmethod + def match_windows(saved, candidates): + return {str(saved[0]["window_id"]): {"confidence": "exact", "live_index": 0}}, [] + + compilation = { + "operations": [ + {"operation": "set-container-layout", "workspace": "8", "source_window": "w1", "layout": "tabbed"}, + {"operation": "set-container-layout", "workspace": "8", "source_window": "w1", "layout": "stacked"}, + ] + } + with self.assertRaisesRegex(executor.RestoreExecutionError, "RESTORE_TREE_DRIFTED"): + executor._execute_tree_compilation( + source, + make_plan(source, live_before), + compilation, + {"w1": 101, "w2": 102, "w3": 103}, + collector=lambda: copy.deepcopy(next(states)), + planner_module=Planner, + tree_module=tree, + runner=lambda argv, _timeout: commands.append(list(argv)) or SimpleNamespace( + returncode=0, stdout='[{"success":true}]', stderr="" + ), + monotonic=lambda: 0.0, + sleeper=lambda _delay: None, + ) + self.assertEqual(len(commands), 1) + + +if __name__ == "__main__": + unittest.main()