feat(vdisplay): complete Stage 5 §6A group semantics — per-group restore, Mutter group-aware, gamescope groups

Host-side completion of Stage 5 (§6A many-clients-as-monitors), all unit-tested;
two-session on-glass validation still pending (no GPU on the dev VM):

- Per-group topology restore (§6.1): the KWin `exclusive` restore no longer rides
  the per-session StopGuard (which re-enabled the physical the moment the FIRST of
  several exclusive sessions dropped, under a live sibling). KWin hands its restore
  to the registry as a closure (new trait `take_topology_restore`); the registry
  keeps it in the display group (`Entry.topology_restore`) and, on teardown, floats
  it to a surviving same-group sibling (`hand_off_restore`) or runs it when the group
  empties — outside the lock, before the last output's keepalive drops, so the
  compositor never sees zero outputs. All three teardown paths (lease drop / linger
  expiry / mgmt release) honor it. Single-display path byte-for-byte unchanged.
  Unit-tested: float / run-on-last / non-carrier-first / never-cross-backend.

- Mutter group-aware (new trait `set_first_in_group`): the registry tells each
  backend whether it's the first display of its group; a non-first Mutter session
  EXTENDS into the already-exclusive desktop instead of re-applying a sole-monitor
  ApplyMonitorsConfig that would disable the first session's virtual. (Mutter
  connectors are un-nameable, so it can't build a keep-all-virtuals config; skipping
  is the safe equivalent.) Single-session unchanged. Residual APPLY_TEMPORARY revert
  documented.

- gamescope groups (§6.1): `registry::group_key` makes each gamescope spawn its own
  group (independent nested session, no shared desktop) — never auto-rowed against or
  restore-/topology-grouped with another gamescope. Applied in both the /display/state
  assembly and the acquire-time position computation. Unit-tested.

Remaining Stage 5: the web console arrangement table, on-glass validation, and the
documented residuals (wlroots exclusive, Mutter APPLY_TEMPORARY). design doc updated.

cargo build/test (214)/clippy --all-targets/fmt green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/punktfunk-host/src/vdisplay.rs

@@ -81,6 +81,26 @@ pub trait VirtualDisplay: Send {
     /// single-display / first-of-group session issues no positioning at all. Best-effort — a failure
     /// leaves the compositor's default placement.
     fn apply_position(&mut self, _x: i32, _y: i32) {}
+    /// Take the topology **restore** action this [`create`](Self::create) prepared — the work that
+    /// un-does an `exclusive`/`primary` topology change (e.g. re-enable the physical outputs KWin
+    /// disabled). The registry lifts it into the display **group** so it runs **once, when the group's
+    /// last display is torn down** (design §6.1 — per-group restore), not when this one session's
+    /// display drops: a sibling `exclusive` session must not have the physical re-enabled under it.
+    /// Called right after `create`; the backend must not also run it itself. Default `None` — a backend
+    /// whose topology auto-reverts (Mutter `APPLY_TEMPORARY`) or that changes nothing has nothing to
+    /// hand off.
+    fn take_topology_restore(&mut self) -> Option<Box<dyn FnOnce() + Send>> {
+        None
+    }
+    /// Tell the backend whether this create will be the **first** display in its group — i.e. no
+    /// sibling of the same backend is already live (design §6.1). A backend that *establishes* the
+    /// group's topology (Mutter's sole-monitor `exclusive` `ApplyMonitorsConfig`) applies it only when
+    /// first; a later sibling **extends** into the already-exclusive desktop instead of re-clobbering it
+    /// (a fresh sole-monitor config would disable the first session's virtual output). Set by the
+    /// registry right before [`create`](Self::create). Default no-op: KWin recognises siblings at
+    /// runtime by output name (first-slot-wins + a group-aware disable filter), and single-display
+    /// backends never have a sibling.
+    fn set_first_in_group(&mut self, _first: bool) {}
 }
 
 /// Compositors punktfunk knows how to drive (plan §6).

crates/punktfunk-host/src/vdisplay/linux/kwin.rs

@@ -82,6 +82,22 @@ pub struct KwinDisplay {
     /// The base output name the last `create` used (`punktfunk` / `punktfunk-<id>`) — so
     /// [`apply_position`](VirtualDisplay::apply_position) can address the KWin output `Virtual-<name>`.
     last_name: Option<String>,
+    /// The topology-restore action the last `create` prepared (re-enable the outputs an `exclusive`
+    /// topology disabled), pending pickup by the registry via [`take_topology_restore`] — so the
+    /// physical is re-enabled only when the display GROUP's last member drops (§6.1), not this session's.
+    /// A backstop [`Drop`] runs it if the registry never took it (so a physical is never left dark).
+    pending_restore: Option<Box<dyn FnOnce() + Send>>,
+}
+
+impl Drop for KwinDisplay {
+    fn drop(&mut self) {
+        // Backstop only: the registry takes the restore right after `create` (moving it into the group),
+        // so this is normally `None`. If some path skipped the take, re-enable here so a physical is
+        // never stranded dark.
+        if let Some(restore) = self.pending_restore.take() {
+            restore();
+        }
+    }
 }
 
 impl KwinDisplay {
@@ -103,6 +119,10 @@ impl VirtualDisplay for KwinDisplay {
         self.last_slot
     }
 
+    fn take_topology_restore(&mut self) -> Option<Box<dyn FnOnce() + Send>> {
+        self.pending_restore.take()
+    }
+
     fn apply_position(&mut self, x: i32, y: i32) {
         let Some(name) = self.last_name.clone() else {
             return;
@@ -173,7 +193,7 @@ impl VirtualDisplay for KwinDisplay {
         // plasmashell + windows land on the streamed surface, not the headless `kwin --virtual`
         // bootstrap output. Read from the policy (replacing the PUNKTFUNK_KWIN_VIRTUAL_PRIMARY boolean).
         use crate::vdisplay::policy::Topology;
-        let restore = match crate::vdisplay::effective_topology() {
+        let disabled = match crate::vdisplay::effective_topology() {
             Topology::Exclusive => apply_virtual_primary(&name),
             Topology::Primary => {
                 apply_virtual_primary_only(&name);
@@ -181,17 +201,44 @@ impl VirtualDisplay for KwinDisplay {
             }
             Topology::Extend | Topology::Auto => Vec::new(),
         };
+        // Per-group restore (§6.1): DON'T bind the re-enable to this session's keepalive (a per-session
+        // `StopGuard` restore would re-enable the physical the moment the FIRST of several exclusive
+        // sessions drops — under a still-live sibling). Instead stash it as a closure the registry lifts
+        // into the display group and runs once, when the group's LAST member is torn down (ordered before
+        // that display's output is reclaimed, so KWin never sees zero outputs). Empty ⇒ nothing to restore.
+        self.pending_restore = (!disabled.is_empty()).then(|| {
+            let disabled = disabled.clone();
+            Box::new(move || reenable_outputs(&disabled)) as Box<dyn FnOnce() + Send>
+        });
         // Layout position (§6.2) is applied by the registry via `apply_position` right after create
         // (it owns the display group, so it computes auto-row / manual placement over the whole group).
         Ok(VirtualOutput {
             node_id,
             remote_fd: None,
             preferred_mode: Some((mode.width, mode.height, achieved_hz)),
-            keepalive: Box::new(StopGuard { stop, restore }),
+            keepalive: Box::new(StopGuard { stop }),
         })
     }
 }
 
+/// Re-enable the outputs an `exclusive` topology disabled (bootstrap / physical), so KWin re-homes onto
+/// them. Called by the registry when the display group's last member is torn down (design §6.1), BEFORE
+/// that member's output is reclaimed — so KWin is never momentarily left with zero enabled outputs.
+fn reenable_outputs(outputs: &[String]) {
+    if outputs.is_empty() {
+        return;
+    }
+    let args: Vec<String> = outputs
+        .iter()
+        .map(|o| format!("output.{o}.enable"))
+        .collect();
+    let _ = std::process::Command::new("kscreen-doctor")
+        .args(&args)
+        .status();
+    std::thread::sleep(Duration::from_millis(200));
+    tracing::info!(reenabled = ?outputs, "KWin: restored the physical/bootstrap outputs (group empty)");
+}
+
 /// Best-effort: raise the just-created virtual output's refresh above KWin's default 60 Hz by
 /// installing + selecting a custom mode via `kscreen-doctor` (the output is `Virtual-<VOUT_NAME>`,
 /// refresh given in mHz), then **read back the active mode** and return the refresh KWin actually
@@ -396,28 +443,15 @@ fn apply_virtual_primary_only(name: &str) {
 }
 
 /// Dropping this releases the KWin virtual output: it flips the keepalive thread's `stop`, which
-/// drops the Wayland connection and makes KWin reclaim the output.
+/// drops the Wayland connection and makes KWin reclaim the output. The topology **restore** is no
+/// longer bound here — it moved to the registry's display group (§6.1, [`reenable_outputs`]), which
+/// runs it once when the group's last member drops, BEFORE this keepalive is dropped.
 struct StopGuard {
     stop: Arc<AtomicBool>,
-    /// Bootstrap output(s) `apply_virtual_primary` disabled to make our streamed output the sole
-    /// desktop — re-enabled here FIRST, so KWin is never left with zero enabled outputs as our
-    /// output is reclaimed. Empty unless PUNKTFUNK_KWIN_VIRTUAL_PRIMARY is set.
-    restore: Vec<String>,
 }
 
 impl Drop for StopGuard {
     fn drop(&mut self) {
-        if !self.restore.is_empty() {
-            let args: Vec<String> = self
-                .restore
-                .iter()
-                .map(|o| format!("output.{o}.enable"))
-                .collect();
-            let _ = std::process::Command::new("kscreen-doctor")
-                .args(&args)
-                .status();
-            std::thread::sleep(Duration::from_millis(200));
-        }
         self.stop.store(true, Ordering::Relaxed);
     }
 }

crates/punktfunk-host/src/vdisplay/linux/mutter.rs

@@ -42,11 +42,19 @@ const CURSOR_EMBEDDED: u32 = 1;
 
 /// The Mutter virtual-display driver. Each [`create`](VirtualDisplay::create) spins up a
 /// keepalive thread owning the D-Bus sessions behind the virtual monitor.
-pub struct MutterDisplay;
+pub struct MutterDisplay {
+    /// Whether this display is the FIRST of its group (§6.1) — set by the registry before `create`.
+    /// A later sibling **extends** into the already-exclusive desktop instead of re-applying the
+    /// sole-monitor config (which would disable the first session's virtual). Defaults true (a lone
+    /// session establishes topology as before).
+    first_in_group: bool,
+}
 
 impl MutterDisplay {
     pub fn new() -> Result<Self> {
-        Ok(MutterDisplay)
+        Ok(MutterDisplay {
+            first_in_group: true,
+        })
     }
 }
 
@@ -64,13 +72,18 @@ impl VirtualDisplay for MutterDisplay {
         "mutter"
     }
 
+    fn set_first_in_group(&mut self, first: bool) {
+        self.first_in_group = first;
+    }
+
     fn create(&mut self, mode: Mode) -> Result<VirtualOutput> {
         let (setup_tx, setup_rx) = std::sync::mpsc::channel::<Result<u32, String>>();
         let stop = Arc::new(AtomicBool::new(false));
         let stop_thread = stop.clone();
+        let first_in_group = self.first_in_group;
         thread::Builder::new()
             .name("punktfunk-mutter-vout".into())
-            .spawn(move || session_thread(setup_tx, stop_thread, mode))
+            .spawn(move || session_thread(setup_tx, stop_thread, mode, first_in_group))
             .context("spawn Mutter virtual-output thread")?;
 
         let node_id = match setup_rx.recv_timeout(Duration::from_secs(20)) {
@@ -104,8 +117,14 @@ impl Drop for StopGuard {
 }
 
 /// Keepalive thread: run the D-Bus handshake on a private tokio runtime, report the PipeWire
-/// node id, then hold the connection until stopped.
+/// node id, then hold the connection until stopped. `first_in_group` gates the topology change (a
-fn session_thread(setup_tx: Sender<Result<u32, String>>, stop: Arc<AtomicBool>, mode: Mode) {
+/// non-first sibling extends into the group's already-exclusive desktop instead of re-clobbering it).
+fn session_thread(
+    setup_tx: Sender<Result<u32, String>>,
+    stop: Arc<AtomicBool>,
+    mode: Mode,
+    first_in_group: bool,
+) {
     let rt = match tokio::runtime::Builder::new_multi_thread()
         .worker_threads(1)
         .enable_all()
@@ -122,12 +141,23 @@ fn session_thread(setup_tx: Sender<Result<u32, String>>, stop: Arc<AtomicBool>,
         // value. `Extend` leaves the virtual output an extension (no config change); `Primary` makes
         // it the primary monitor but keeps the physicals as secondaries; `Exclusive` makes it the
         // SOLE output (physicals disabled). `Auto` never reaches here — it's resolved upstream.
+        use crate::vdisplay::policy::Topology;
         let topo = crate::vdisplay::effective_topology();
-        let want_config = matches!(
+        let topo_policy = matches!(topo, Topology::Primary | Topology::Exclusive);
-            topo,
+        // Group-aware (§6.1): only the FIRST display of the group establishes the topology. A later
-            crate::vdisplay::policy::Topology::Primary | crate::vdisplay::policy::Topology::Exclusive
+        // sibling extends into the already-exclusive desktop — re-applying the sole-monitor config would
-        );
+        // disable the first session's virtual output (Mutter connectors are un-nameable, so we can't
-        let exclusive = matches!(topo, crate::vdisplay::policy::Topology::Exclusive);
+        // build a config that keeps all group virtuals; skipping is the safe choice). *Concurrent
+        // Mutter exclusive is on-glass-validation-pending; the APPLY_TEMPORARY revert when the FIRST
+        // session leaves under a live sibling is a documented residual (design §7).*
+        let want_config = first_in_group && topo_policy;
+        if topo_policy && !first_in_group {
+            tracing::info!(
+                "mutter: joining an existing display group — extending (the first session owns the \
+                 exclusive/primary topology)"
+            );
+        }
+        let exclusive = matches!(topo, Topology::Exclusive);
         // Snapshot the monitor layout BEFORE the virtual output exists (so we can tell the new
         // connector apart and restore on teardown) whenever we're going to touch the topology.
         let dc_pre = if want_config {

crates/punktfunk-host/src/vdisplay/registry.rs

@@ -190,11 +190,40 @@ mod linux {
         /// shared/anonymous or a backend with no per-client identity) — keys the group arrangement +
         /// the `/display/state` slot. Captured at create; kept across a keep-alive reuse.
         identity_slot: Option<u32>,
+        /// The topology-restore action for this display's GROUP (design §6.1): re-enable the physical
+        /// outputs an `exclusive` topology disabled. At most ONE entry per group carries it (the first
+        /// exclusive session); on teardown it hands off to a surviving sibling, and only runs when the
+        /// group's last member drops. `None` for extend/primary and non-first / non-exclusive members.
+        topology_restore: Option<Restore>,
         /// Generation stamp: a [`DisplayLease`] only releases if its gen still matches (a stale lease
         /// — its entry was reused + re-stamped — is a no-op).
         gen: u64,
     }
 
+    /// A per-group topology-restore action (see [`Entry::topology_restore`]).
+    type Restore = Box<dyn FnOnce() + Send>;
+
+    /// Hand off a torn-down display's topology restore (design §6.1 — per-group restore): if a
+    /// same-group (backend) sibling survives in `remaining`, MOVE the restore onto it (a later teardown
+    /// runs it); if the group is now empty, RETURN the action so the caller runs it (before dropping the
+    /// reclaimed display's keepalive, so the physical is re-enabled while our output still exists —
+    /// the compositor never sees zero outputs). `None` in → `None` out.
+    fn hand_off_restore(
+        remaining: &mut [Entry],
+        backend: &'static str,
+        restore: Option<Restore>,
+    ) -> Option<Restore> {
+        let action = restore?;
+        // At most one restore per group, so any surviving sibling has `None` to receive it.
+        match remaining.iter_mut().find(|e| e.backend == backend) {
+            Some(sibling) => {
+                sibling.topology_restore = Some(action);
+                None
+            }
+            None => Some(action), // group empty → run it now
+        }
+    }
+
     struct Reg {
         entries: Mutex<Vec<Entry>>,
         gen: AtomicU64,
@@ -220,18 +249,26 @@ mod linux {
 
     /// Remove entries whose linger deadline has passed, returning them so the caller drops (tears
     /// them down) *after* releasing the lock — a backend keepalive `Drop` (Mutter D-Bus Stop) can
-    /// block, and holding the pool lock across it would stall every other acquire/release.
+    /// block, and holding the pool lock across it would stall every other acquire/release. Each
-    fn take_expired(entries: &mut Vec<Entry>, now: Instant) -> Vec<Entry> {
+    /// expired entry's topology restore is [handed off](hand_off_restore) to a surviving group sibling,
+    /// or collected into the returned `restores` when its group empties (run before the entries drop).
+    fn take_expired(entries: &mut Vec<Entry>, now: Instant) -> (Vec<Entry>, Vec<Restore>) {
         let mut expired = Vec::new();
+        let mut restores = Vec::new();
         let mut i = 0;
         while i < entries.len() {
             if entries[i].life.poll_expiry(now) {
-                expired.push(entries.remove(i));
+                let mut e = entries.remove(i);
+                let backend = e.backend;
+                if let Some(r) = hand_off_restore(entries, backend, e.topology_restore.take()) {
+                    restores.push(r);
+                }
+                expired.push(e);
             } else {
                 i += 1;
             }
         }
-        expired
+        (expired, restores)
     }
 
     /// Background thread (started once): reap lingering displays past their deadline.
@@ -242,10 +279,14 @@ mod linux {
                 .name("vdisplay-linger".into())
                 .spawn(|| loop {
                     std::thread::sleep(Duration::from_millis(500));
-                    let expired = {
+                    let (expired, restores) = {
                         let mut es = reg().entries.lock().unwrap();
                         take_expired(&mut es, Instant::now())
                     };
+                    // Re-enable physicals (group emptied) BEFORE dropping the outputs — outside the lock.
+                    for restore in restores {
+                        restore();
+                    }
                     for e in expired {
                         tracing::info!(
                             backend = e.backend,
@@ -277,11 +318,14 @@ mod linux {
         let backend = vd.name();
         let r = reg();
 
-        // Reap expired first (drop outside the lock).
+        // Reap expired first (run any group restores + drop outside the lock).
-        let expired = {
+        let (expired, restores) = {
             let mut es = r.entries.lock().unwrap();
             take_expired(&mut es, Instant::now())
         };
+        for restore in restores {
+            restore();
+        }
         drop(expired);
 
         // Reuse: a kept (lingering/pinned) display of the same backend + mode. A reconnecting session
@@ -309,6 +353,16 @@ mod linux {
             }
         }
 
+        // Tell the backend whether it's the FIRST display of its group (no same-backend sibling live,
+        // §6.1) — so a topology-establishing backend (Mutter exclusive) extends into an already-exclusive
+        // desktop rather than re-clobbering the first session's virtual. Best-effort (a concurrent create
+        // is a narrow race); single-session is always `first == true` → today's behavior.
+        let first_in_group = {
+            let es = r.entries.lock().unwrap();
+            !es.iter().any(|e| e.backend == backend)
+        };
+        vd.set_first_in_group(first_in_group);
+
         // Create a fresh display (NOT under the lock — `vd.create` blocks + spawns threads).
         let real = vd.create(mode)?;
         // The identity slot the backend just resolved (KWin per-slot naming; `None` elsewhere) — keys
@@ -328,6 +382,10 @@ mod linux {
 
         let node_id = real.node_id;
         let preferred_mode = real.preferred_mode;
+        // The backend's topology-restore action (KWin `exclusive` → re-enable the disabled physicals),
+        // lifted into the group so it runs once when the group's last member drops (§6.1), not at this
+        // session's teardown. `None` for non-exclusive / non-first / backends whose topology auto-reverts.
+        let topology_restore = vd.take_topology_restore();
         let gen = r.gen.fetch_add(1, Ordering::Relaxed);
         let mut life = lifecycle::State::default();
         life.acquire(); // Idle → Active{refs:1} (Acquire::Create)
@@ -339,6 +397,7 @@ mod linux {
             mode,
             backend,
             identity_slot,
+            topology_restore,
             gen,
         };
 
@@ -353,9 +412,12 @@ mod linux {
                 .map(|e| e.layout)
                 .unwrap_or_default();
             let mut es = r.entries.lock().unwrap();
+            // Same-group members (design §6.1): same backend for a shared desktop, but each gamescope
+            // spawn is its own group, so a new gamescope never auto-rows against another.
+            let new_group = group_key(backend, gen);
             let existing: Vec<(u64, Member)> = es
                 .iter()
-                .filter(|e| e.backend == backend)
+                .filter(|e| group_key(e.backend, e.gen) == new_group)
                 .map(|e| {
                     (
                         e.gen,
@@ -390,31 +452,42 @@ mod linux {
     fn release(gen: u64) {
         let Some(r) = REG.get() else { return };
         let linger = linger();
-        let torn_down = {
+        let (torn_down, restore) = {
             let mut es = r.entries.lock().unwrap();
             let Some(idx) = es.iter().position(|e| e.gen == gen) else {
                 return; // stale lease (entry reused + re-stamped, or already gone) — no-op
             };
             match es[idx].life.release(Instant::now(), linger) {
-                Release::Teardown | Release::Noop => Some(es.remove(idx)),
+                Release::Teardown | Release::Noop => {
+                    let mut e = es.remove(idx);
+                    let backend = e.backend;
+                    // Per-group restore (§6.1): hand the physical re-enable to a surviving sibling, or run
+                    // it now if this was the group's last member.
+                    let restore = hand_off_restore(&mut es, backend, e.topology_restore.take());
+                    (Some(e), restore)
+                }
                 Release::Linger => {
                     tracing::info!(
                         backend = es[idx].backend,
                         "virtual display: last session left — lingering (keep-alive)"
                     );
-                    None
+                    (None, None)
                 }
                 Release::Pin => {
                     tracing::info!(
                         backend = es[idx].backend,
                         "virtual display: last session left — pinned (keep-alive forever)"
                     );
-                    None
+                    (None, None)
                 }
                 // Linux entries are single-session (refs == 1), so Decref never occurs; harmless.
-                Release::Decref => None,
+                Release::Decref => (None, None),
             }
         };
+        // Re-enable the physicals (group emptied) BEFORE dropping the output — outside the lock.
+        if let Some(restore) = restore {
+            restore();
+        }
         if let Some(e) = torn_down {
             tracing::info!(
                 backend = e.backend,
@@ -498,10 +571,23 @@ mod linux {
             .expect("members is non-empty (just pushed `new`)")
     }
 
-    /// Group the flattened rows into the mgmt `/display/state` view (design §6.1/§6.2): group = backend
+    /// The display **group** a backend+display belongs to (design §6.1). The desktop compositors
-    /// (one desktop per compositor session), ordered by acquire (`gen`), with each member's position
+    /// (KWin/Mutter/wlroots) put every managed output on ONE desktop → one group per backend. A
-    /// from the pure [`layout`] engine. Pure — no I/O, no global — so the grouping / ordering / position
+    /// gamescope **spawn** is an independent nested session per client (no shared desktop), so each
-    /// assignment is unit-tested against synthetic rows.
+    /// gamescope display is its OWN group — never auto-rowed against, or topology-/restore-grouped with,
+    /// another gamescope session.
+    fn group_key(backend: &str, gen: u64) -> String {
+        if backend == "gamescope" {
+            format!("gamescope#{gen}")
+        } else {
+            backend.to_string()
+        }
+    }
+
+    /// Group the flattened rows into the mgmt `/display/state` view (design §6.1/§6.2) by
+    /// [`group_key`], ordered by acquire (`gen`), with each member's position from the pure [`layout`]
+    /// engine. Pure — no I/O, no global — so the grouping / ordering / position assignment is
+    /// unit-tested against synthetic rows.
     fn assemble_displays(
         rows: Vec<Row>,
         layout_policy: &Layout,
@@ -509,19 +595,19 @@ mod linux {
     ) -> Vec<DisplayInfo> {
         use crate::vdisplay::layout::{self, Member};
 
-        // Small stable group ids by sorted backend name — deterministic; in practice a host runs one
+        // Small stable group ids by sorted group key — deterministic; in practice a host runs one live
-        // live backend → group 1.
+        // desktop backend → group 1 (with each gamescope spawn its own group).
-        let mut backends: Vec<&'static str> = rows.iter().map(|row| row.backend).collect();
+        let mut keys: Vec<String> = rows.iter().map(|r| group_key(r.backend, r.gen)).collect();
-        backends.sort_unstable();
+        keys.sort();
-        backends.dedup();
+        keys.dedup();
 
         let mut out: Vec<DisplayInfo> = Vec::new();
-        for (gi, backend) in backends.iter().enumerate() {
+        for (gi, key) in keys.iter().enumerate() {
             // This group's members in acquire order (gen ascending) → display_index + arrangement.
             let mut idx: Vec<usize> = rows
                 .iter()
                 .enumerate()
-                .filter(|(_, row)| row.backend == *backend)
+                .filter(|(_, row)| &group_key(row.backend, row.gen) == key)
                 .map(|(i, _)| i)
                 .collect();
             idx.sort_by_key(|&i| rows[i].gen);
@@ -557,21 +643,32 @@ mod linux {
 
     pub(super) fn force_release(slot: Option<u64>) -> usize {
         let Some(r) = REG.get() else { return 0 };
-        let released = {
+        let (released, restores) = {
             let mut es = r.entries.lock().unwrap();
             let mut out = Vec::new();
+            let mut restores = Vec::new();
             let mut i = 0;
             while i < es.len() {
                 let selected = slot.is_none_or(|s| es[i].gen == s);
                 if selected && es[i].life.force_release() {
-                    out.push(es.remove(i));
+                    let mut e = es.remove(i);
+                    let backend = e.backend;
+                    let restore = e.topology_restore.take();
+                    if let Some(rst) = hand_off_restore(&mut es, backend, restore) {
+                        restores.push(rst);
+                    }
+                    out.push(e);
                 } else {
                     i += 1;
                 }
             }
-            out
+            (out, restores)
         };
         let n = released.len();
+        // Re-enable physicals (group emptied) BEFORE dropping the outputs — outside the lock.
+        for restore in restores {
+            restore();
+        }
         for e in released {
             tracing::info!(
                 backend = e.backend,
@@ -599,6 +696,106 @@ mod linux {
         use super::*;
         use crate::vdisplay::policy::{Layout, LayoutMode, Position};
         use std::collections::BTreeMap;
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::Arc;
+
+        /// A minimal pool entry for the pure teardown/restore tests (dummy keepalive; the
+        /// `hand_off_restore` logic only reads `backend` + `topology_restore`).
+        fn test_entry(backend: &'static str, gen: u64, restore: Option<Restore>) -> Entry {
+            Entry {
+                life: lifecycle::State::default(),
+                keepalive: Box::new(()),
+                node_id: 0,
+                preferred_mode: None,
+                mode: Mode {
+                    width: 1920,
+                    height: 1080,
+                    refresh_hz: 60,
+                },
+                backend,
+                identity_slot: None,
+                topology_restore: restore,
+                gen,
+            }
+        }
+
+        /// A restore closure that flips `flag` when run — so a test can assert exactly WHEN it fires.
+        fn flag_restore(flag: &Arc<AtomicBool>) -> Restore {
+            let f = flag.clone();
+            Box::new(move || f.store(true, Ordering::SeqCst))
+        }
+
+        #[test]
+        fn topology_restore_floats_to_a_sibling_then_runs_on_the_last_teardown() {
+            let ran = Arc::new(AtomicBool::new(false));
+            // Two KWin displays in one group; the first (gen 1) carries the group's restore.
+            let mut pool = vec![
+                test_entry("kwin", 1, Some(flag_restore(&ran))),
+                test_entry("kwin", 2, None),
+            ];
+
+            // Tear down the restore-carrier while its sibling is still alive → transfer, don't run.
+            let mut e1 = pool.remove(0);
+            let out = hand_off_restore(&mut pool, "kwin", e1.topology_restore.take());
+            assert!(out.is_none(), "transferred, not run");
+            assert!(!ran.load(Ordering::SeqCst));
+            // The restore floated onto the surviving sibling.
+            assert!(pool[0].topology_restore.is_some());
+
+            // Tear down the last member → group empty → the restore is returned to run.
+            let mut e2 = pool.remove(0);
+            let out = hand_off_restore(&mut pool, "kwin", e2.topology_restore.take());
+            let action = out.expect("group empty → run the restore");
+            assert!(!ran.load(Ordering::SeqCst), "not run yet");
+            action();
+            assert!(ran.load(Ordering::SeqCst), "runs on the last drop");
+        }
+
+        #[test]
+        fn single_session_topology_restore_runs_on_its_own_teardown() {
+            // The validated single-display case: one exclusive session → restore runs at its teardown.
+            let ran = Arc::new(AtomicBool::new(false));
+            let mut pool = vec![test_entry("kwin", 1, Some(flag_restore(&ran)))];
+            let mut e = pool.remove(0);
+            let action = hand_off_restore(&mut pool, "kwin", e.topology_restore.take())
+                .expect("last (only) member → run");
+            action();
+            assert!(ran.load(Ordering::SeqCst));
+        }
+
+        #[test]
+        fn tearing_down_a_non_carrier_first_leaves_the_restore_for_last() {
+            let ran = Arc::new(AtomicBool::new(false));
+            // gen 2 carries the restore; gen 1 does not (a later exclusive session found the physical
+            // already disabled).
+            let mut pool = vec![
+                test_entry("kwin", 1, None),
+                test_entry("kwin", 2, Some(flag_restore(&ran))),
+            ];
+            // Tear down the non-carrier first → nothing to hand off, carrier untouched.
+            let mut e1 = pool.remove(0);
+            assert!(hand_off_restore(&mut pool, "kwin", e1.topology_restore.take()).is_none());
+            // The carrier (gen 2) still holds the group's restore.
+            assert!(pool[0].topology_restore.is_some());
+            // Now the carrier (last member) → run.
+            let mut e2 = pool.remove(0);
+            hand_off_restore(&mut pool, "kwin", e2.topology_restore.take())
+                .expect("last member → run")();
+            assert!(ran.load(Ordering::SeqCst));
+        }
+
+        #[test]
+        fn restore_never_floats_across_backends() {
+            // group = backend: a KWin restore must not land on a Mutter display (a different desktop).
+            let ran = Arc::new(AtomicBool::new(false));
+            let mut pool = vec![test_entry("mutter", 2, None)];
+            let out = hand_off_restore(&mut pool, "kwin", Some(flag_restore(&ran)));
+            assert!(out.is_some(), "no same-backend sibling → return to run");
+            assert!(
+                pool[0].topology_restore.is_none(),
+                "restore must not cross into another backend's group"
+            );
+        }
 
         fn row(gen: u64, backend: &'static str, w: u32, slot: Option<u32>) -> Row {
             Row {
@@ -677,6 +874,23 @@ mod linux {
             assert_eq!(pos, Placement { x: 100, y: 200 });
         }
 
+        #[test]
+        fn gamescope_spawns_are_separate_groups() {
+            // Two independent gamescope spawns must NOT share a group or auto-row against each other.
+            let rows = vec![
+                row(1, "gamescope", 1920, None),
+                row(2, "gamescope", 1280, None),
+            ];
+            let out = assemble_displays(rows, &Layout::default(), "extend");
+            assert_eq!(out.len(), 2);
+            assert_ne!(out[0].group, out[1].group, "distinct groups");
+            // Each is display 0 of its own group, at the origin (not auto-rowed against the other).
+            assert_eq!(out[0].display_index, 0);
+            assert_eq!(out[1].display_index, 0);
+            assert_eq!(out[0].position, (0, 0));
+            assert_eq!(out[1].position, (0, 0));
+        }
+
         #[test]
         fn manual_layout_keys_positions_by_identity_slot() {
             // Client 7 arranged to the LEFT of client 1 (reversed vs. auto-row).