feat(session): restore Sway tree and scratchpad state
This commit is contained in:
parent
5a14abe323
commit
5567a76738
6 changed files with 1372 additions and 15 deletions
|
|
@ -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>
|
||||
mark --add|--replace [--toggle] <identifier>
|
||||
unmark [<identifier>]
|
||||
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=<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_<token_hex_aléatoire>
|
||||
[source] move container to mark __labfy_restore_<token_hex_aléatoire>
|
||||
[reference] unmark __labfy_restore_<token_hex_aléatoire>
|
||||
```
|
||||
|
||||
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=<numérique>, 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=<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.
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
252
tests/test_session_tree_restore.py
Normal file
252
tests/test_session_tree_restore.py
Normal file
|
|
@ -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()
|
||||
Loading…
Reference in a new issue