test: migrate host-timed tests to private kick-in-kick hosts; flock clipboard

The 7 host-timed tests failed not (only) on timeouts: autopsy showed
parallel workers stealing each other's live-host clipboard (wl-paste
returns another test's token -- no timeout survives that).

Clipboard x3: the failure is cross-talk on ONE shared host clipboard.
Serialize just those sections with an flock file across xdist workers
(the 45s polls stay as backstops for genuine host slowness -- marks
kept; focus-based backend still needs the real host).

Fullscreen x3 + cursor-shape: migrate to per-test outer-kick hosts
(same xdg/fullscreen/shape protocols, deterministic compositor instead
of GNOME).  No coverage lost (single code path; timing was the only
host variable).  cursor-shape uses deterministic warp-enter; fullscreen
uses tap/chord acks.  Added KICK_REPEAT_DELAY_MS + extra_args to
_spawn_nested_in.  Unmarked after -n15 validation.

hold_alt stays (wall-clock timer).

Evidence: -n15 gate 3x clean (152 passed), serial gate clean (152),
migrated tests 6/6 at -n15.  12 -> 7 flaky.

tests/test_smoke.py

@@ -19,6 +19,7 @@ Known-flaky tests are marked @pytest.mark.flaky; pytest.ini's default
 """
 
 import ctypes
+import fcntl
 import os
 import re
 import shutil
@@ -1214,7 +1215,7 @@ def _spawn_standalone(extra_env, *apps, log=None):
     return Session(proc, log, dump_dir)
 
 
-def _spawn_nested_in(outer, *apps, log=None):
+def _spawn_nested_in(outer, *apps, extra_args=(), log=None):
     """A nested kick whose host compositor is another running kick (e.g. a
     standalone outer): the inner bridge binds the outer's data-control
     backend, so data-control/primary bridge paths are testable without a
@@ -1225,6 +1226,7 @@ def _spawn_nested_in(outer, *apps, log=None):
     env["KICK_LOG"] = "2"
     env["KICK_SYNTH_INPUT"] = "1"
     env["KICK_DUMP_DIR"] = dump_dir
+    env["KICK_REPEAT_DELAY_MS"] = KICK_REPEAT_TEST_DELAY
     env["WAYLAND_DISPLAY"] = outer.socket
     # The outer kick advertises no server decorations; take the direct
     # xdg-shell path instead of libdecor (which also needs KICK_LIBDECOR
@@ -1233,7 +1235,7 @@ def _spawn_nested_in(outer, *apps, log=None):
     env.pop("KICK_MODE", None)
     env.pop("WLR_BACKENDS", None)
     proc = subprocess.Popen(
-        [BIN, *apps],
+        [BIN, *extra_args] + list(apps),
         env=env, stdin=subprocess.PIPE,
         stdout=open(log, "ab"), stderr=subprocess.STDOUT,
         preexec_fn=_preexec_death(os.getpid()))
@@ -2226,34 +2228,38 @@ def test_clipboard_host_steal_cancels_nested_source(session):
     the bridge's host-side source is cancelled (the offering wl-copy exits)
     and the session survives."""
     token = f"kick-cancel-{os.getpid()}-{time.monotonic_ns()}"
-    copy = _dc_setter(session.socket, token)
+    lock = _host_clipboard_locked()
     try:
-        session.wait_any_active()
+        copy = _dc_setter(session.socket, token)
-        # The mirror is live once the host pastes what the tab offered.
-        _poll_paste(os.environ.get("WAYLAND_DISPLAY", "wayland-0"), token)
-        steal = subprocess.Popen(
-            ["wl-copy", "-f", "stolen"],
-            env=dict(os.environ),
-            stdin=subprocess.DEVNULL,
-            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-            preexec_fn=_preexec_death(os.getpid()))
         try:
-            # The data-control setter exits once its source is cancelled
+            session.wait_any_active()
-            # (like an offering wl-copy): the wait proves the bridge let
+            # The mirror is live once the host pastes what the tab offered.
-            # go of it.
+            _poll_paste(os.environ.get("WAYLAND_DISPLAY", "wayland-0"), token)
-            copy.wait(timeout=10.0)
+            steal = subprocess.Popen(
-            assert copy.returncode == 0, \
+                ["wl-copy", "-f", "stolen"],
-                f"cancelled setter should exit 0, rc={copy.returncode}"
+                env=dict(os.environ),
+                stdin=subprocess.DEVNULL,
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                preexec_fn=_preexec_death(os.getpid()))
+            try:
+                # The data-control setter exits once its source is cancelled
+                # (like an offering wl-copy): the wait proves the bridge let
+                # go of it.
+                copy.wait(timeout=10.0)
+                assert copy.returncode == 0, \
+                    f"cancelled setter should exit 0, rc={copy.returncode}"
+            finally:
+                steal.terminate()
+                steal.wait(timeout=5)
+            assert session.proc.poll() is None, \
+                f"compositor died on source cancel; log tail:\n{session.read_log()[-400:]}"
         finally:
-            steal.terminate()
+            try:
-            steal.wait(timeout=5)
+                copy.kill()
-        assert session.proc.poll() is None, \
+            except Exception:
-            f"compositor died on source cancel; log tail:\n{session.read_log()[-400:]}"
+                pass
     finally:
-        try:
+        lock.close()
-            copy.kill()
-        except Exception:
-            pass
 
 
 def test_standalone_headless_cursor_theme_none_boots(tmp_path):
@@ -2327,27 +2333,38 @@ def test_standalone_cursor_shape_resets_on_tab_switch():
         s.close()
 
 
-@pytest.mark.flaky
+def test_nested_cursor_shape_forwarded(tmp_path):
-def test_nested_cursor_shape_forwarded():
     """In nested mode a client's cursor-shape request is forwarded to the host
-    pointer (the host draws the cursor).  Deterministic when the host honours
+    pointer.  The host is a private outer kick (not the live desktop), so no
-    the forwarded set_shape; on a desktop host the outcome depends on the
+    real pointer or host cursor-shape support is needed: warps enter
-    host's cursor-shape-v1 support (and on whether the real pointer has ever
+    deterministically and the forward line proves the request went out."""
-    entered the window, which provides the serial the request is tagged
+    outer_log = tmp_path / "outer.log"
-    with)."""
+    inner_log = tmp_path / "inner.log"
-    if not os.environ.get("WAYLAND_DISPLAY"):
+    outer = _spawn_standalone(
-        pytest.skip("no host compositor for nested mode")
+        {}, "%s --app-id keep '#0000ff' 320x240" % WL, log=str(outer_log))
-    headless_host = os.environ["WAYLAND_DISPLAY"].startswith("kick-")
-    s = _nested_session("build/wltest '#0000ff'",
-                        "build/wltest --cursor-shape text '#ff0000'")
     try:
-        s.wait_app_active(1)
+        outer.wait_any_active()
-        if headless_host:
+        inner = _spawn_nested_in(
-            s.pointer_enter_until(r"\[cursor\] host shape text")
+            outer, "build/wltest '#0000ff'",
-        else:
+            "build/wltest --cursor-shape text '#ff0000'",
-            s.pointer_enter_until(r"\[cursor\] host shape (text|n/a)")
+            log=str(inner_log))
+        try:
+            inner.wait_app_active(1)
+            inner.wait_surface_mapped(1)
+            # Deterministic enter (replaces pointer_enter_until's re-warp
+            # loop, which exists for races with surface readiness): the
+            # surface is mapped, so one nested warp enters it and the
+            # client requests its shape.  (Nested warps log no [cursor]
+            # line -- no wlr_cursor there -- so wait for the forward line
+            # itself.)
+            inner.warp(400, 400)
+            inner.wait_for(r"\[cursor\] host shape text", timeout=30.0)
+            assert inner.proc.poll() is None
+            assert outer.proc.poll() is None
+        finally:
+            inner.close()
     finally:
-        s.close()
+        outer.close()
 
 
 def test_standalone_headless_last_tab_exits_cleanly(standalone_session):
@@ -2473,9 +2490,30 @@ def _run_display(display, args, timeout=6, **kw):
         preexec_fn=_preexec_death(os.getpid()), **kw)
 
 
+def _host_clipboard_locked():
+    """Exclusive cross-worker lock for the shared live-host clipboard.
+
+    Autopsied failure: parallel clipboard tests steal each other's host
+    selections (wl-paste returns another test's token), so the loser's
+    poll fails after 45s no matter the timeout.  Serializing just these
+    sections across xdist workers removes the cross-talk; genuine host
+    slowness still fails honestly inside the (generous) polls, which is
+    why these tests stay marked flaky.  Returns the lock file; close it
+    to release (use try/finally)."""
+    f = open("/tmp/kick-host-clipboard.lock", "w")
+    fcntl.flock(f, fcntl.LOCK_EX)
+    return f
+
+
 def _poll_paste(display, want, deadline=45.0, extra=()):
     """Wait until `wl-paste` on `display` returns exactly `want`
-    (extra args, e.g. ("--primary",), select the selection)."""
+    (extra args, e.g. ("--primary",), select the selection).
+
+    Callers driving the shared live-host clipboard must hold
+    _host_clipboard_locked() for the whole set-then-paste sequence:
+    parallel workers otherwise steal each other's selections and the
+    loser's token never appears no matter how long the poll runs."""
+
     r = None
     since = time.monotonic()
     while time.monotonic() - since < deadline:
@@ -2567,30 +2605,38 @@ def _poll_dc_demand(sock, out_file, token, primary=False, deadline=45.0):
 @pytest.mark.flaky
 def test_clipboard_nested_to_host(session):
     token = f"kick-n2h-{os.getpid()}-{time.monotonic_ns()}"
-    copy = _dc_setter(session.socket, token)
+    lock = _host_clipboard_locked()
     try:
-        _poll_paste(os.environ.get("WAYLAND_DISPLAY", "wayland-0"), token)
+        copy = _dc_setter(session.socket, token)
+        try:
+            _poll_paste(os.environ.get("WAYLAND_DISPLAY", "wayland-0"), token)
+        finally:
+            copy.terminate()
+            copy.wait(timeout=5)
     finally:
-        copy.terminate()
+        lock.close()
-        copy.wait(timeout=5)
 
 
 @pytest.mark.skipif(NEED_WL_CLIPBOARD, reason="wl-clipboard tools not installed")
 @pytest.mark.flaky
 def test_clipboard_host_to_nested(session, tmp_path):
     token = f"kick-h2n-{os.getpid()}-{time.monotonic_ns()}"
-    copy = subprocess.Popen(
+    lock = _host_clipboard_locked()
-        ["wl-copy", "-f", token],
-        env=dict(os.environ),
-        stdin=subprocess.DEVNULL,
-        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-        preexec_fn=_preexec_death(os.getpid()))
     try:
-        out = tmp_path / "h2n.out"
+        copy = subprocess.Popen(
-        _poll_dc_demand(session.socket, out, token)
+            ["wl-copy", "-f", token],
+            env=dict(os.environ),
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+            preexec_fn=_preexec_death(os.getpid()))
+        try:
+            out = tmp_path / "h2n.out"
+            _poll_dc_demand(session.socket, out, token)
+        finally:
+            copy.terminate()
+            copy.wait(timeout=5)
     finally:
-        copy.terminate()
+        lock.close()
-        copy.wait(timeout=5)
 
 
 def test_tabs_map(session):
@@ -2625,69 +2671,105 @@ def test_function_keys_pass_through(session):
     assert session.proc.poll() is None
 
 
-@pytest.mark.flaky
+def test_client_fullscreen_round_trip(tmp_path):
-def test_client_fullscreen_round_trip(session):
     """A client's fullscreen request is honored and mirrored onto the host
     window; the host configure then propagates fullscreen to every tab.
     wltest toggles xdg_toplevel fullscreen on the 'F' key, so tap it and
-    watch both directions of the mirror."""
+    watch both directions of the mirror.  The host is a private outer kick
-    idx = session.wait_any_active()
+    (not the live desktop), so configure roundtrips are deterministic."""
-    session.wait_surface_mapped(idx)
+    outer_log = tmp_path / "outer.log"
-    # Tapping F before seat keyboard focus reaches the surface drops the
+    inner_log = tmp_path / "inner.log"
-    # key silently; map alone does not guarantee enter under load.
+    outer = _spawn_standalone(
-    session.wait_for(r"\[diag\] kbd-enter app=%d\b" % idx)
+        {}, "%s --app-id keep '#0000ff' 320x240" % WL, log=str(outer_log))
-    session.tap_ack(F)
+    try:
-    session.wait_for("host fullscreen=1", timeout=30.0)
+        outer.wait_any_active()
-    session.wait_for("clients fullscreen=1")
+        inner = _spawn_nested_in(
-    session.tap_ack(F)
+            outer, "%s '#ff0000'" % WL, "%s '#00ff00'" % WL,
-    session.wait_for("host fullscreen=0", timeout=30.0)
+            log=str(inner_log))
-    session.wait_for("clients fullscreen=0")
+        try:
-    assert session.proc.poll() is None
+            idx = inner.wait_any_active()
+            inner.wait_surface_mapped(idx)
+            # Tapping F before seat keyboard focus reaches the surface drops
+            # the key silently; map alone does not guarantee enter under load.
+            inner.wait_for(r"\[diag\] kbd-enter app=%d\b" % idx)
+            inner.tap_ack(F)
+            inner.wait_for("host fullscreen=1", timeout=30.0)
+            inner.wait_for("clients fullscreen=1")
+            inner.tap_ack(F)
+            inner.wait_for("host fullscreen=0", timeout=30.0)
+            inner.wait_for("clients fullscreen=0")
+            assert inner.proc.poll() is None
+            assert outer.proc.poll() is None
+        finally:
+            inner.close()
+    finally:
+        outer.close()
 
 
-@pytest.mark.flaky
+def test_alt_f11_toggles_host_fullscreen(tmp_path):
-def test_alt_f11_toggles_host_fullscreen(session):
     """prefix+F11 toggles the kick window's fullscreen state in the host;
     the resulting configure propagates to every tab exactly like a
     host-compositor-initiated fullscreen.  Tapping plain F11 (no prefix)
-    must still pass through to the client untouched."""
+    must still pass through to the client untouched.  The host is a private
-    idx = session.wait_any_active()
+    outer kick, so configure roundtrips are deterministic."""
-    session.wait_surface_mapped(idx)
+    outer_log = tmp_path / "outer.log"
-    session.chord_ack(ALT, F11)
+    inner_log = tmp_path / "inner.log"
-    session.wait_for("host fullscreen=1", timeout=30.0)
+    outer = _spawn_standalone(
-    session.wait_for("clients fullscreen=1")
+        {}, "%s --app-id keep '#0000ff' 320x240" % WL, log=str(outer_log))
-    session.chord_ack(ALT, F11)
+    try:
-    session.wait_for("host fullscreen=0", timeout=30.0)
+        outer.wait_any_active()
-    session.wait_for("clients fullscreen=0")
+        inner = _spawn_nested_in(
-    assert session.proc.poll() is None
+            outer, "%s '#ff0000'" % WL, "%s '#00ff00'" % WL,
+            log=str(inner_log))
+        try:
+            idx = inner.wait_any_active()
+            inner.wait_surface_mapped(idx)
+            inner.chord_ack(ALT, F11)
+            inner.wait_for("host fullscreen=1", timeout=30.0)
+            inner.wait_for("clients fullscreen=1")
+            inner.chord_ack(ALT, F11)
+            inner.wait_for("host fullscreen=0", timeout=30.0)
+            inner.wait_for("clients fullscreen=0")
+            assert inner.proc.poll() is None
+            assert outer.proc.poll() is None
+        finally:
+            inner.close()
+    finally:
+        outer.close()
 
 
-@pytest.mark.flaky
+def test_alt_f11_propagates_to_stack_members(tmp_path):
-def test_alt_f11_propagates_to_stack_members():
     """prefix+F11's host fullscreen reaches not just each tab's main window
     but every stacked tab-group member too, and leaving fullscreen clears
-    both again."""
+    both again.  The host is a private outer kick, so configure roundtrips
-    if not os.environ.get("WAYLAND_DISPLAY"):
+    are deterministic."""
-        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
+    outer_log = tmp_path / "outer.log"
-    s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'", f"{WL} '{BLUE}'",
+    inner_log = tmp_path / "inner.log"
-                        extra_args=("--tab-groups", "on"))
+    outer = _spawn_standalone(
+        {}, "%s --app-id keep '#0000ff' 320x240" % WL, log=str(outer_log))
     try:
-        idx = s.wait_any_active()
+        outer.wait_any_active()
-        s.wait_surface_mapped(idx)
+        inner = _spawn_nested_in(
-        s.wait_for(r"\[sub\] stack app=0\b")
+            outer, f"{WL} --extra '{GREEN}' '{RED}'", f"{WL} '{BLUE}'",
-        s.chord_ack(ALT, F11)
+            extra_args=("--tab-groups", "on"), log=str(inner_log))
-        s.wait_for("host fullscreen=1", timeout=30.0)
+        try:
-        s.wait_for("clients fullscreen=1")
+            idx = inner.wait_any_active()
-        s.wait_for("members fullscreen=1")
+            inner.wait_surface_mapped(idx)
-        s.chord_ack(ALT, F11)
+            inner.wait_for(r"\[sub\] stack app=0\b")
-        s.wait_for("host fullscreen=0", timeout=30.0)
+            inner.chord_ack(ALT, F11)
-        s.wait_for("clients fullscreen=0")
+            inner.wait_for("host fullscreen=1", timeout=30.0)
-        s.wait_for("members fullscreen=0")
+            inner.wait_for("clients fullscreen=1")
-        assert s.proc.poll() is None
+            inner.wait_for("members fullscreen=1")
+            inner.chord_ack(ALT, F11)
+            inner.wait_for("host fullscreen=0", timeout=30.0)
+            inner.wait_for("clients fullscreen=0")
+            inner.wait_for("members fullscreen=0")
+            assert inner.proc.poll() is None
+            assert outer.proc.poll() is None
+        finally:
+            inner.close()
     finally:
-        s.close()
+        outer.close()
-        if s.proc.poll() is None:
-            s.proc.terminate()
 
 
 def test_sigterm_removes_socket(session):