kick/tests/test_smoke.py

#!/usr/bin/env python3
"""Smoke tests for kick.

The compositor is launched as a nested display (child of the host
Wayland session) and driven over the stdin event feed: raw keyboard
events are packed into binary records and dispatched into the *real*
host key handlers (host_kbd_key / host_kbd_modifiers), so the key
handler logic is under test without needing a focused window or wtype.

Each test launches its own instance (function-scoped fixture), so tests
are independent and can run in any order.

Requires: a live host Wayland session (WAYLAND_DISPLAY) and a prior
./build.sh. Missing prereqs skip the suite.

Known-flaky tests are marked @pytest.mark.flaky; pytest.ini's default
`-m "not flaky"` deselects them from every plain run (including direct
`pytest`). Run them explicitly with `./test.sh -m flaky`.
"""

import ctypes
import glob
import os
import re
import shutil
import signal
import struct
import subprocess
import tempfile
import time

import pytest

NEED_WL_CLIPBOARD = shutil.which("wl-copy") is None or shutil.which("wl-paste") is None

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
BIN = os.path.join(ROOT, "build", "kick")
WL = os.path.join(ROOT, "build", "wltest")
LOG = "/tmp/tabtest.log"
RED = "#ff0000"
BLUE = "#0000ff"
GREEN = "#00ff00"

# Raw keycodes used by the compositor's own key handler (see
# host_kbd_key in kick.c).  Names, not numbers, in tests.
F = 33
F11 = 87
UP = 103
DOWN = 108
DELETE = 111
KEY_PRESSED = 1
KEY_RELEASED = 0

# Ctrl modifier bit, as read from host_mods_depressed (1u << 2).
CTRL = 1 << 2
# Alt (Mod1) modifier bit (1u << 3).
ALT = 1 << 3
# Shift modifier bit (1u << 0).
SHIFT = 1 << 0
# Super/Mod4 modifier bit (1u << 6).
SUPER = 1 << 6

CTL_CMD_KEY = 1
CTL_CMD_MODS = 2
CTL_CMD_WARP = 3
CTL_CMD_PTR_MOTION = 4

ACTIVE_RE = re.compile(r"active app=(\d+)")
CLAIM_RE = re.compile(r"\[claim\] app=(\d+) pid=(\d+)")
SOCKET_RE = re.compile(r"Wayland socket: (kick-\d+)")


# Font paths the compositor searches (overlay_pick_font in kick.c);
# used to make the standalone headless session build a real glyph atlas.
FONT_PATHS = [
    "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
    "/usr/share/fonts/dejavu/DejaVuSans.ttf",
    "/usr/share/fonts/TTF/DejaVuSans.ttf",
    "/usr/share/fonts/truetype/liberation/LiberationSans-Regular.ttf",
    "/usr/share/fonts/truetype/liberation2/LiberationSans-Regular.ttf",
    "/usr/share/fonts/liberation/LiberationSans-Regular.ttf",
    "/usr/share/fonts/noto/NotoSans-Regular.ttf",
]


def _pick_font():
    """First existing font path, from $KICK_FONT or the default list."""
    env_font = os.environ.get("KICK_FONT")
    if env_font and os.path.isfile(env_font):
        return env_font
    for path in FONT_PATHS:
        if os.path.isfile(path):
            return path
    return None


_libc = ctypes.CDLL(None)
_PR_SET_PDEATHSIG = 1


def _preexec_death(expected_ppid):
    """preexec_fn: have the kernel SIGTERM this child if the test process
    (its parent) ever dies, SIGKILL included.  The compositor's SIGTERM handler
    tears the whole session down (clients, socket), and the short-lived
    wl-copy/wl-paste helpers just die -- so an aborted pytest run can never
    leave strays behind.  Check getppid() after setting it: if the parent died
    between fork() and prctl() the signal is already pending, but our
    expectations (and the child) are meaningless anyway, so bail out."""
    def _fn():
        _libc.prctl(_PR_SET_PDEATHSIG, signal.SIGTERM)
        if os.getppid() != expected_ppid:
            os._exit(1)
    return _fn


class Session:
    """A running kick instance plus helpers."""

    def __init__(self, proc):
        self.proc = proc
        self._socket = None

    # -- log watching ----------------------------------------------------
    def read_log(self):
        """Whole current log contents (rescanned fresh each call)."""
        with open(LOG, "r", errors="replace") as f:
            return f.read()

    def wait_for(self, pattern, timeout=15.0):
        """Wait until `pattern` matches somewhere in the whole log.

        Re-scans from the start each poll: kick's log lines can arrive
        out of order (e.g. `map app=1` before `map app=0`), so a line may
        enter the file before we happen to be looking for it.
        """
        rx = re.compile(pattern)
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            if rx.search(self.read_log()) is not None:
                return
            time.sleep(0.1)
        tail = self.read_log()[-1000:]
        pytest.fail(f"log never matched {rx.pattern};\n{tail}")

    @property
    def socket(self):
        if self._socket is None:
            self.wait_for(SOCKET_RE, timeout=60.0)
            m = SOCKET_RE.search(self.read_log())
            assert m, "no 'Wayland socket:' line in log"
            self._socket = m.group(1)
        return self._socket

    # -- input -----------------------------------------------------------
    def key(self, code, state):
        """Feed one key event into the compositor's real key handler."""
        record = struct.pack("<BII", CTL_CMD_KEY, code, state)
        self.proc.stdin.write(record)
        self.proc.stdin.flush()

    def mods(self, pattern):
        """Feed a keyboard-modifier set (e.g. CTRL, or 0 to clear)."""
        record = struct.pack("<BIIII", CTL_CMD_MODS, pattern, 0, 0, 0)
        self.proc.stdin.write(record)
        self.proc.stdin.flush()

    def warp(self, x, y):
        """Move the standalone cursor to an output pixel (x, y)."""
        record = struct.pack("<BII", CTL_CMD_WARP, x, y)
        self.proc.stdin.write(record)
        self.proc.stdin.flush()

    def ptr_motion(self, dx, dy):
        """Feed a relative pointer-motion delta into the same path libinput
        pointer devices take (wlr_cursor_move + seat forward)."""
        record = struct.pack("<BII", CTL_CMD_PTR_MOTION, dx, dy)
        self.proc.stdin.write(record)
        self.proc.stdin.flush()

    def tap(self, code):
        """Press then release one key."""
        self.key(code, KEY_PRESSED)
        self.key(code, KEY_RELEASED)

    def chord(self, mods, *keys):
        """Hold `mods`, tap each key in turn, then release `mods`."""
        self.mods(pattern=mods)
        for code in keys:
            self.tap(code)
        self.mods(pattern=0)

    def cycle(self, direction):
        """Swap to the next (direction>0) or previous tab via Alt+arrow."""
        key = DOWN if direction > 0 else UP
        self.chord(ALT, key)

    def cycle_ack(self, direction, timeout=5.0):
        """Tap Alt+arrow and block until the compositor logs a *new*
        active-tab line reflecting the cycle, then return the fresh value.

        cycle() alone races: `active()` reads the log file, which lags the
        compositor's async key processing, so a tight `while active() != N`
        loop can queue extra cycles past the target before its reads catch
        up.  Waiting for the log to grow guarantees the tap took effect
        before the caller makes its next decision.

        The count is only trustworthy once the fixture tabs have stopped
        emitting their own startup "active app=" lines (both clients map
        asynchronously, toggling active as they land); otherwise one of
        those can satisfy the "grew" check and hand back a stale value.
        So quiesce the log before tapping.
        """
        key = DOWN if direction > 0 else UP
        since = time.monotonic()
        before = None
        while time.monotonic() - since < timeout:
            before = len(ACTIVE_RE.findall(self.read_log()))
            time.sleep(0.2)
            if len(ACTIVE_RE.findall(self.read_log())) == before:
                break
        self.chord(ALT, key)
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            if len(ACTIVE_RE.findall(self.read_log())) > before:
                return self.active()
            time.sleep(0.05)
        pytest.fail("cycle tap never reflected in the log")

    def active(self):
        """Current active app index from the log, or None."""
        m = ACTIVE_RE.findall(self.read_log())
        return int(m[-1]) if m else None

    def wait_active(self, want, timeout=3.0):
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            if self.active() == want:
                return
            time.sleep(0.1)
        pytest.fail(f"never reached active app={want}")

    def wait_active_change(self, away_from, timeout=3.0):
        """Wait for active to differ from `away_from`; return the new value.

        Unlike wait_active(), this ignores historical log lines: it only
        returns once a *new* active tab shows up, so a stale launch-time
        line can't satisfy it before the key event is processed.
        """
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            cur = self.active()
            if cur is not None and cur != away_from:
                return cur
            time.sleep(0.1)
        pytest.fail(f"active app never changed away from {away_from}")

    def wait_any_active(self, timeout=10.0):
        """Wait until the startup-selected tab has activated (its first map),
        return its index.

        The last command-line command is active from the very first frame --
        black, titled by its command line -- and is the only tab that ever
        auto-activates on boot; every earlier tab maps hidden with no
        ``active app=`` line.  The first active line is therefore the settled
        tab, with no churn to wait out.  For the fixtures' N-tab sessions
        that tab is index N-1.
        """
        rx = re.compile(r"active app=(\d+)")
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            m = rx.findall(self.read_log())
            if m:
                return int(m[-1])
            time.sleep(0.1)
        pytest.fail("no tab was ever activated (startup tab never mapped)")

    def wait_active_frame(self, timeout=10.0):
        """Wait until the active tab has painted a real frame: the composed
        (or host-presented) frame's centre is no longer the black backdrop.
        Guards single-read pixel asserts that need black to mean "overlay
        shown" and probe areas to mean "pure client colour"."""
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            path = self.dump_host()
            if path:
                w, h, px = self.ppm_rgb(path)
                i = 3 * (h // 2 * w + w // 2)
                if (px[i], px[i + 1], px[i + 2]) != (0, 0, 0):
                    return
            time.sleep(0.1)
        pytest.fail("active tab never painted a frame (centre stayed black)")

    def wait_app_active(self, index, timeout=10.0):
        """Wait until the single app `index` reports itself active (for
        sessions launched with fewer than two tabs)."""
        rx = re.compile(r"active app=%d\b" % index)
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            if rx.search(self.read_log()):
                return True
            time.sleep(0.1)
        pytest.fail(f"app {index} never logged an active line")

    # -- frame dumps ------------------------------------------------------
    def dump_host(self, timeout=5.0):
        rx = re.compile(r"\[dump\] host back buffer -> (\S+\.ppm)")
        # The SIGUSR1 handler is only installed once the compositor is well
        # into startup; signaling too early kills it (default action), so
        # wait until the socket line exists first.
        self.socket
        before = len(rx.findall(self.read_log()))
        os.kill(self.proc.pid, signal.SIGUSR1)
        # Wait for a *new* "host back buffer -> ...ppm" line: it is logged
        # only after the file is fully written and closed.  The file
        # itself appears at fopen() while still mid-write, so file
        # appearance alone is not enough.
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            m = rx.findall(self.read_log())
            if len(m) > before:
                return m[-1]
            time.sleep(0.1)
        raise AssertionError("SIGUSR1 produced no 'host back buffer ->' line")

    def dump_scene(self, timeout=5.0):
        """SIGUSR1-request a scene dump; wait until the dump path has run.

        The scene buffer is a dmabuf and never CPU-readable, so no ppm is
        written; the `[dump] scene` line (either "not CPU-readable" or the
        stats line) proves the request reached the dump path.
        """
        rx = re.compile(r"\[dump\] scene")
        self.socket
        before = len(rx.findall(self.read_log()))
        os.kill(self.proc.pid, signal.SIGUSR1)
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            m = rx.findall(self.read_log())
            if len(m) > before:
                return
            time.sleep(0.1)
        raise AssertionError("SIGUSR1 produced no '[dump] scene' line")

    def active_count(self):
        """How many `active app=` transitions the log has seen so far."""
        return len(ACTIVE_RE.findall(self.read_log()))

    def launch_pids(self, timeout=10.0):
        """Map app index -> connected client PID from [claim] lines."""
        since = time.monotonic()
        while time.monotonic() - since < timeout:
            d = dict((int(a), int(p)) for a, p in CLAIM_RE.findall(self.read_log()))
            if {0, 1}.issubset(d):
                return d
            time.sleep(0.1)
        pytest.fail("both apps never logged a [claim] line")

    def pixel_at(self, path, x, y):
        """Pixel (x, y) of an 8-bit P6 ppm as an (R, G, B) tuple.

        The dump writes the back buffer top-down (window coords), so row 0
        of the file is the top of the window and column 0 is the left edge.
        """
        with open(path, "rb") as f:
            data = f.read()
        lines = data.split(b"\n", 3)
        if len(lines) < 3:
            raise AssertionError(f"bad PPM {path}")
        w, h = map(int, lines[1].split())
        off = (len(lines[0]) + 1 + len(lines[1]) + 1 + len(lines[2]) + 1)
        i = off + 3 * (y * w + x)
        return tuple(data[i:i + 3])

    def ppm_rgb(self, path):
        """(width, height, raw RGB bytes) of an 8-bit P6 ppm.

        Reading once lets a test scan many pixels without reopening the
        file for every sample.
        """
        with open(path, "rb") as f:
            data = f.read()
        lines = data.split(b"\n", 3)
        if len(lines) < 3:
            raise AssertionError(f"bad PPM {path}")
        w, h = map(int, lines[1].split())
        off = (len(lines[0]) + 1 + len(lines[1]) + 1 + len(lines[2]) + 1)
        return w, h, data[off:]

    def close(self):
        if self.proc.poll() is None:
            self.proc.terminate()
            try:
                self.proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.proc.kill()
                self.proc.wait()
        for ppm in glob.glob("/tmp/host_*.ppm") + glob.glob("/tmp/scene_*.ppm"):
            os.unlink(ppm)


@pytest.fixture(scope="function")
def session():
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    for path in (BIN, WL):
        if not os.access(path, os.X_OK):
            pytest.skip(f"build first: {path}")
    try:
        os.unlink(LOG)
    except FileNotFoundError:
        pass

    proc = subprocess.Popen(
        ["env", "KICK_LOG=2", "KICK_SYNTH_INPUT=1", "stdbuf", "-oL",
         BIN, f"{WL} '{RED}'", f"{WL} '{BLUE}'"],
        stdin=subprocess.PIPE,
        stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
        preexec_fn=_preexec_death(os.getpid()),
    )
    s = Session(proc)
    try:
        yield s
    finally:
        s.close()


@pytest.fixture(scope="function")
def standalone_session():
    """Standalone mode on a headless backend, exercised over the stdin feed.

    The native backend autocreate is bypassed with WLR_BACKENDS=headless (real
    DRM is unavailable in the test environment), so this checks the standalone
    render/input/present paths without a host window.  Requires a GPU renderer
    (PIXMAN/GBlob every time CPU backends), which the build always links.
    """
    for path in (BIN, WL):
        if not os.access(path, os.X_OK):
            pytest.skip(f"build first: {path}")
    try:
        os.unlink(LOG)
    except FileNotFoundError:
        pass

    # Clear any leftover XDG/compositor interaction: standalone must not
    # connect to the host compositor (it would pick the wayland backend and
    # open a second nested window).  WLR_BACKENDS=headless forces the
    # headless backend and the standalone output path.
    env = dict(os.environ)
    env["KICK_LOG"] = "2"
    env["KICK_MODE"] = "standalone"
    env["WLR_BACKENDS"] = "headless"
    env.pop("WAYLAND_DISPLAY", None)
    proc = subprocess.Popen(
        ["env", "-i",
         "PATH=" + env["PATH"],
         "HOME=" + env.get("HOME", "/"),
         "XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
         "XDG_DATA_HOME=" + env.get("XDG_DATA_HOME", ""),
         "XDG_DATA_DIRS=" + env.get("XDG_DATA_DIRS", ""),
         "KICK_LOG=2",
         "KICK_MODE=standalone",
         "WLR_BACKENDS=headless",
         "KICK_FONT=" + (env.get("KICK_FONT") or _pick_font() or ""),
         "XCURSOR_THEME=Adwaita",
         "stdbuf", "-oL",
         BIN, f"{WL} '{RED}'", f"{WL} '{BLUE}'"],
        stdin=subprocess.PIPE,
        stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
        preexec_fn=_preexec_death(os.getpid()),
    )
    s = Session(proc)
    try:
        yield s
    finally:
        s.close()


def test_standalone_headless_boot_and_cycle(standalone_session):
    """Standalone on the headless backend boots, maps both tabs, and cycles
    them with the same stdin-driven Alt+arrow handling as nested mode."""
    s = standalone_session
    assert s.proc.poll() is None, f"standalone exited early: {s.read_log()[-500:]}"
    start = s.wait_any_active()
    assert start in (0, 1)
    s.cycle(+1)
    assert s.wait_active_change(start) == 1 - start
    assert s.proc.poll() is None


def _has_dark(w, h, px):
    """True when the central region looks like the dark overlay box."""
    for y in range(h // 4, 3 * h // 4):
        row = y * w * 3
        for x in range(w // 4, 3 * w // 4):
            i = row + x * 3
            if px[i] < 40 and px[i + 1] < 40 and px[i + 2] < 40:
                return True
    return False


def test_standalone_headless_alt_overlay_scene(standalone_session):
    """Holding Alt in standalone mode routes through the shared hotkey path
    and draws the overlay (which in standalone enters the second wlr render
    pass before the frame is committed).  SIGUSR1 dumps the composited frame,
    whose centre darkens while Alt is held."""
    s = standalone_session
    s.wait_any_active()
    s.wait_active_frame()

    # Before Alt: the frame should be the client content (no dark overlay).
    w, h, px = s.ppm_rgb(s.dump_host())
    assert not _has_dark(w, h, px), "content already dark before Alt"

    s.mods(ALT)
    deadline = time.monotonic() + 15.0
    while time.monotonic() < deadline:
        w, h, px = s.ppm_rgb(s.dump_host())
        if _has_dark(w, h, px):
            break
    assert _has_dark(w, h, px), "standalone overlay box not visible while Alt"

    s.mods(0)
    assert s.proc.poll() is None


def test_standalone_headless_overlay_clears_on_alt_release(standalone_session):
    """Releasing Alt in standalone fully clears the overlay.  The overlay is
    drawn into the scene output buffer outside the scene graph, so without
    forcing whole-output damage on the transition a reused swapchain buffer
    keeps the stale overlay baked in and it reappears on alternating frames
    (until a client repaint covers it)."""
    s = standalone_session
    s.wait_any_active()

    s.mods(ALT)
    deadline = time.monotonic() + 15.0
    seen = False
    while time.monotonic() < deadline:
        w, h, px = s.ppm_rgb(s.dump_host())
        if _has_dark(w, h, px):
            seen = True
            break
    assert seen, "overlay did not appear while Alt"

    s.mods(0)
    deadline = time.monotonic() + 15.0
    while time.monotonic() < deadline:
        w, h, px = s.ppm_rgb(s.dump_host())
        if not _has_dark(w, h, px):
            break  # a freshly presented, overlay-free frame
        time.sleep(0.08)
    else:
        raise AssertionError("overlay remnant never cleared after Alt release")

    # Once clear, keep sampling: any reappearing dark frame is a stale buffer
    # slot echoing the baked-in overlay, i.e. the bug returning.
    deadline = time.monotonic() + 2.0
    while time.monotonic() < deadline:
        w, h, px = s.ppm_rgb(s.dump_host())
        assert not _has_dark(w, h, px), "overlay remnant reappeared after Alt release"
        time.sleep(0.08)
    assert s.proc.poll() is None


def test_standalone_headless_cursor_visible(standalone_session):
    """The standalone output shows a real cursor.  The headless backend has no
    cursor plane, so wlroots composites the software cursor into the frame
    buffer (the scene render pass picks it up).  Warping the pointer must
    change the composited pixels at the target -- the cursor visibly moves
    there -- while moving the cursor away again repaints the spot back to pure
    client colour without smearing.

    The GL dump is vertically flipped (bottom-up readback), so a cursor at
    compositor pixel (x, y) appears in the PPM at (x, H-1-y); the patch probes
    through the same flip."""
    s = standalone_session
    s.wait_any_active()
    s.wait_active_frame()
    assert s.proc.poll() is None

    def patch_at(path, cx, cy, r=22):
        """Sample the PPM around compositor pixel (cx, cy), mapping compositor
        row y to file row H-1-y for the bottom-up readback."""
        w, h, px = s.ppm_rgb(path)
        py = h - 1 - cy
        hits = []
        for y in range(max(0, py - r), min(h, py + r + 1)):
            row = y * w * 3
            for x in range(max(0, cx - r), min(w, cx + r + 1)):
                i = row + x * 3
                hits.append((px[i], px[i + 1], px[i + 2]))
        return hits

    def distinct(a, b):
        return any(ca != cb for ca, cb in zip(a, b))

    target = (320, 320)

    # Get the pointer clear of the probe area first, then snapshot: the
    # target patch must be pure client colour before we warp into it.
    s.warp(60, 60)
    time.sleep(0.3)
    before = patch_at(s.dump_host(), *target)
    assert len(set(before)) == 1, \
        "probe area should be pure client colour before the warp"

    s.warp(*target)
    deadline = time.monotonic() + 5.0
    while time.monotonic() < deadline:
        if distinct(patch_at(s.dump_host(), *target), before):
            break
        time.sleep(0.1)
    else:
        pytest.fail("cursor never became visible at the warp target")

    # Move away again: the target must end up back on pure client colour
    # (no cursor smeared into the composited buffer).
    s.warp(640, 400)
    deadline = time.monotonic() + 5.0
    while time.monotonic() < deadline:
        if not distinct(patch_at(s.dump_host(), *target), before):
            break
        time.sleep(0.1)
    else:
        pytest.fail("cursor left a remnant at the warp target")

    assert s.proc.poll() is None


def test_standalone_headless_cursor_relative_motion(standalone_session):
    """Relative pointer motion must move the visible cursor, exactly as libinput
    pointer devices do on a real host (native_ptr_motion -> wlr_cursor_move).  A
    regressed compositor leaves the pointer pinned at the warp target while the
    seat still moves, so this fails on code that forwards motion but never moves
    the cursor."""
    s = standalone_session
    s.wait_any_active()
    s.wait_active_frame()
    assert s.proc.poll() is None

    def patch_at(path, cx, cy, r=22):
        w, h, px = s.ppm_rgb(path)
        py = h - 1 - cy
        hits = []
        for y in range(max(0, py - r), min(h, py + r + 1)):
            row = y * w * 3
            for x in range(max(0, cx - r), min(w, cx + r + 1)):
                i = row + x * 3
                hits.append((px[i], px[i + 1], px[i + 2]))
        return hits

    def distinct(a, b):
        return any(ca != cb for ca, cb in zip(a, b))

    start = (320, 320)
    delta = (200, 120)
    moved = (start[0] + delta[0], start[1] + delta[1])

    # Clear the pointer, snapshot both probe areas as pure client colour.
    s.warp(40, 40)
    time.sleep(0.3)
    before_start = patch_at(s.dump_host(), *start)
    before_moved = patch_at(s.dump_host(), *moved)
    assert len(set(before_start)) == 1, \
        "start probe area should be pure client colour"
    assert len(set(before_moved)) == 1, \
        "moved probe area should be pure client colour"

    # Place the cursor at `start` and confirm it is painted there.
    s.warp(*start)
    deadline = time.monotonic() + 5.0
    while time.monotonic() < deadline:
        if distinct(patch_at(s.dump_host(), *start), before_start):
            break
        time.sleep(0.1)
    else:
        pytest.fail("cursor never became visible at the start target")

    # Relative motion (+dx, +dy) must carry the pointer to `moved`: the pixels
    # there change, and `start` returns to pure client colour.
    s.ptr_motion(*delta)
    deadline = time.monotonic() + 5.0
    while time.monotonic() < deadline:
        path = s.dump_host()
        if distinct(patch_at(path, *moved), before_moved) and \
                not distinct(patch_at(path, *start), before_start):
            break
        time.sleep(0.1)
    else:
        pytest.fail("relative motion did not move the cursor %+d,%+d" % delta)

    assert s.proc.poll() is None


def test_standalone_headless_locked_pointer_holds():
    """A LOCKED pointer constraint pins the visible cursor: once the
    constraint activates on pointer enter, relative motion must not carry the
    cursor away from the lock position."""
    s = _spawn_standalone({}, "%s --constrain locked '#ff0000'" % WL)
    try:
        s.wait_app_active(0)
        s.wait_active_frame()
        assert s.proc.poll() is None

        def patch_at(path, cx, cy, r=22):
            w, h, px = s.ppm_rgb(path)
            py = h - 1 - cy
            hits = []
            for y in range(max(0, py - r), min(h, py + r + 1)):
                row = y * w * 3
                for x in range(max(0, cx - r), min(w, cx + r + 1)):
                    i = row + x * 3
                    hits.append((px[i], px[i + 1], px[i + 2]))
            return hits

        def distinct(a, b):
            return any(ca != cb for ca, cb in zip(a, b))

        lock = (300, 300)
        target = (lock[0] + 90, lock[1] + 60)

        pure_lock = patch_at(s.dump_host(), *lock)
        pure_target = patch_at(s.dump_host(), *target)
        assert len(set(pure_lock)) == 1, "lock probe should be pure client colour"
        assert len(set(pure_target)) == 1, "target probe should be pure client colour"

        # Warp onto the surface: the constraint activates and the cursor is
        # painted at the lock position.
        s.warp(*lock)
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            if "[constraint] locked on surface" in s.read_log():
                break
            time.sleep(0.1)
        else:
            pytest.fail("locked constraint never activated")
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            if distinct(patch_at(s.dump_host(), *lock), pure_lock):
                break
            time.sleep(0.1)
        else:
            pytest.fail("cursor never appeared at the lock position")

        # Relative motion must not move the pinned cursor.
        s.ptr_motion(*[target[0] - lock[0], target[1] - lock[1]])
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            path = s.dump_host()
            if distinct(patch_at(path, *lock), pure_lock) and \
                    not distinct(patch_at(path, *target), pure_target):
                break
            time.sleep(0.1)
        else:
            pytest.fail("locked pointer moved away from the lock position")
        assert s.proc.poll() is None
    finally:
        s.close()


def test_standalone_headless_confined_pointer_clamps():
    """A CONFINED pointer constraint keeps the visible cursor inside the
    constraint region even when relative motion would carry it outside."""
    s = _spawn_standalone({}, "%s --constrain confined "
                          "--confine-rect 100 100 200 150 '#00ff00'" % WL)
    try:
        s.wait_app_active(0)
        s.wait_active_frame()
        assert s.proc.poll() is None

        def patch_at(path, cx, cy, r=22):
            w, h, px = s.ppm_rgb(path)
            py = h - 1 - cy
            hits = []
            for y in range(max(0, py - r), min(h, py + r + 1)):
                row = y * w * 3
                for x in range(max(0, cx - r), min(w, cx + r + 1)):
                    i = row + x * 3
                    hits.append((px[i], px[i + 1], px[i + 2]))
            return hits

        def distinct(a, b):
            return any(ca != cb for ca, cb in zip(a, b))

        start = (150, 120)      # inside the 100..300 x 100..250 region
        clamped = (299, 249)    # far motion clamps to the region's edge
        outside = (650, 520)    # where unconstrained motion would land

        pure_start = patch_at(s.dump_host(), *start)
        pure_clamped = patch_at(s.dump_host(), *clamped)
        assert len(set(pure_start)) == 1, "start probe should be pure client colour"
        assert len(set(pure_clamped)) == 1, \
            "clamped probe should be pure client colour"

        s.warp(*start)
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            if "[constraint] confined on surface" in s.read_log():
                break
            time.sleep(0.1)
        else:
            pytest.fail("confined constraint never activated")
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            if distinct(patch_at(s.dump_host(), *start), pure_start):
                break
            time.sleep(0.1)
        else:
            pytest.fail("cursor never appeared at the start position")

        # A big relative motion clamps to the region edge, not the raw target.
        s.ptr_motion(500, 400)
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            path = s.dump_host()
            if distinct(patch_at(path, *clamped), pure_clamped) and \
                    not distinct(patch_at(path, *outside), pure_start):
                break
            time.sleep(0.1)
        else:
            pytest.fail("confined pointer escaped the constraint region")
        assert s.proc.poll() is None
    finally:
        s.close()


def test_standalone_headless_screen_mode_cursor_stays_aligned():
    """--screen-mode + pointer motion must keep the visible cursor exactly on
    the warp target plus the relative delta, on a non-default output size.  The
    seat forward and the visible arrow must both derive from the same
    wlr_cursor position (single source of truth), not a parallel hand-clamped
    copy -- on a DRM monitor a requested mode that differs from the geometry
    the layout derives from the committed state would otherwise let the two
    split diagonally."""
    s = _spawn_standalone({}, "--screen-mode", "480x320",
                          "build/wltest '#ff0000'")
    try:
        s.wait_app_active(0)
        s.wait_active_frame()
        assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"

        def patch_at(path, cx, cy, r=22):
            w, h, px = s.ppm_rgb(path)
            py = h - 1 - cy
            hits = []
            for y in range(max(0, py - r), min(h, py + r + 1)):
                row = y * w * 3
                for x in range(max(0, cx - r), min(w, cx + r + 1)):
                    i = row + x * 3
                    hits.append((px[i], px[i + 1], px[i + 2]))
            return hits

        def distinct(a, b):
            return any(ca != cb for ca, cb in zip(a, b))

        start = (120, 120)
        delta = (100, 60)
        moved = (start[0] + delta[0], start[1] + delta[1])

        s.warp(30, 30)
        time.sleep(0.3)
        before_start = patch_at(s.dump_host(), *start)
        before_moved = patch_at(s.dump_host(), *moved)
        assert len(set(before_start)) == 1, \
            "start probe area should be pure client colour"
        assert len(set(before_moved)) == 1, \
            "moved probe area should be pure client colour"

        s.warp(*start)
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            if distinct(patch_at(s.dump_host(), *start), before_start):
                break
            time.sleep(0.1)
        else:
            pytest.fail("cursor never became visible at the start target")

        s.ptr_motion(*delta)
        deadline = time.monotonic() + 5.0
        while time.monotonic() < deadline:
            path = s.dump_host()
            if distinct(patch_at(path, *moved), before_moved) and \
                    not distinct(patch_at(path, *start), before_start):
                break
            time.sleep(0.1)
        else:
            pytest.fail("relative motion did not move the cursor %+d,%+d" % delta)

        assert s.proc.poll() is None
    finally:
        s.close()


def _spawn_standalone(extra_env, *apps):
    """Standalone compositor on the headless backend with a custom env for
    cursor-theme tests; XCURSOR_THEME/XCURSOR_PATH come only from extra_env."""
    try:
        os.unlink(LOG)
    except FileNotFoundError:
        pass
    env = dict(os.environ)
    env["KICK_LOG"] = "2"
    env["KICK_MODE"] = "standalone"
    env["WLR_BACKENDS"] = "headless"
    env.pop("WAYLAND_DISPLAY", None)
    env.pop("XCURSOR_THEME", None)
    env.pop("XCURSOR_PATH", None)
    env.update(extra_env or {})
    cmd = ["env", "-i",
           "PATH=" + env["PATH"],
           "HOME=" + env.get("HOME", "/"),
           "XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
           "XDG_DATA_HOME=" + env.get("XDG_DATA_HOME", ""),
           "XDG_DATA_DIRS=" + env.get("XDG_DATA_DIRS", ""),
           "KICK_LOG=2",
           "KICK_MODE=standalone",
           "WLR_BACKENDS=headless",
           "KICK_FONT=" + (env.get("KICK_FONT") or _pick_font() or "")]
    for key in ("XCURSOR_PATH", "XCURSOR_THEME"):
        if env.get(key):
            cmd.append("%s=%s" % (key, env[key]))
    cmd += ["stdbuf", "-oL", BIN]
    cmd += list(apps)
    proc = subprocess.Popen(
        cmd, stdin=subprocess.PIPE,
        stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
        preexec_fn=_preexec_death(os.getpid()))
    return Session(proc)


def test_standalone_headless_cursor_theme_missing_uses_builtin(tmp_path):
    """A theme that does not exist anywhere must not stop the compositor:
    wlroots substitutes a builtin arrow, so the pointer stays visible.  We
    merely log the substitution, naming the requested theme."""
    empty = tmp_path / "empty-cursors"
    empty.mkdir()
    s = _spawn_standalone({"XCURSOR_PATH": str(empty),
                           "XCURSOR_THEME": "DefinitelyMissingTheme"},
                          "build/wltest '#ff0000'")
    try:
        s.wait_app_active(0)
        assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
        log = s.read_log()
        assert "theme 'DefinitelyMissingTheme' not found; wlroots builtin " \
            "cursor in use" in log, \
            f"substitution should be logged naming the theme:\n{log[-400:]}"
    finally:
        s.close()


def test_standalone_headless_protocol_globals(tmp_path):
    """The standalone compositor advertises the passive protocol globals
    that clients rely on to run at their best: presentation-time for frame
    pacing, xdg-output for logical output geometry, viewporter for
    wl_viewport buffer scaling.  wltest --globals-out dumps the advertised
    interface list to a file (CLI-tab stdout is /dev/null'd by kick)."""
    out = tmp_path / "globals.txt"
    s = _spawn_standalone({}, "%s --globals-out '%s' '#112233'"
                          % (WL, out))
    try:
        deadline = time.monotonic() + 15.0
        while time.monotonic() < deadline:
            if out.exists():
                break
            time.sleep(0.1)
        assert out.exists(), "wltest --globals-out never wrote its file"
        names = {line.split()[0] for line in out.read_text().splitlines()}
        for want in ("wl_compositor", "wl_shm", "xdg_wm_base", "wl_seat",
                     "wp_presentation", "zxdg_output_manager_v1",
                     "wp_viewporter"):
            assert want in names, \
                f"missing global {want}; advertised: {sorted(names)}"
    finally:
        s.close()


def test_standalone_headless_primary_selection_roundtrip(tmp_path):
    """zwp_primary_selection_v1: a tab's primary selection must reach a
    second tab.  The device-manager global plus the request handler install
    the source on the seat, which fans the offer out to the keyboard-focused
    tab's device; wltest --primary-set installs a source (the first time a
    seat-granted serial arrives) and --primary-demand pulls the "text/plain"
    payload into a file."""
    out = tmp_path / "primary.out"
    s = _spawn_standalone({}, "%s --primary-set 'hello' '#ff0000'" % WL,
                          "%s --primary-demand '%s' '#00ff00'" % (WL, out))
    try:
        s.wait_any_active()
        s.wait_active_frame()
        assert s.proc.poll() is None
        # Active on boot is the last tab (the demander).  Give the setter
        # keyboard focus so it receives a seat-granted serial to set with --
        # wlroots rejects a set_selection serial it never handed out.
        if s.active() != 0:
            s.cycle(-1)
            assert s.wait_active_change(1) == 0
        # Any key press delivers a serial to the focused setter, which
        # installs the primary selection on the receipt.
        s.tap(23)  # 'i', harmless
        # Cycle back so the demander is (re)focused; the focus change flushes
        # the current primary selection to its device.  No active-change assert
        # here: the demander may exit as soon as it has written the file, which
        # flips active back to the setter -- so just wait for the payload.
        s.cycle(+1)
        deadline = time.monotonic() + 15.0
        while time.monotonic() < deadline:
            if out.exists():
                break
            time.sleep(0.1)
        assert out.exists(), "primary-demand tab never received the selection"
        assert out.read_text() == "hello", \
            f"primary round-trip payload mismatch: {out.read_text()!r}"
    finally:
        s.close()


@pytest.mark.flaky
def test_nested_primary_selection_bridged(tmp_path):
    """Nested mode: a tab's primary selection is fanned out to a second tab
    exactly as standalone, and the host-side bridge publishes it over
    wlr-data-control (when the host implements v2) without echo-looping."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    out = tmp_path / "primary.out"
    s = _nested_session("%s --primary-set 'hello' '#ff0000'" % WL,
                        "%s --primary-demand '%s' '#00ff00'" % (WL, out))
    try:
        s.wait_any_active()
        s.wait_active_frame()
        assert s.proc.poll() is None
        if s.active() != 0:
            s.cycle(-1)
            assert s.wait_active_change(1) == 0
        s.tap(23)  # 'i', harmless: grants the focused setter a serial
        s.cycle(+1)
        deadline = time.monotonic() + 15.0
        while time.monotonic() < deadline:
            assert s.proc.poll() is None, \
                f"compositor exited mid-test:\n{s.read_log()[-500:]}"
            if out.exists():
                break
            time.sleep(0.1)
        assert out.exists(), "nested primary-demand tab never received it"
        assert out.read_text() == "hello", \
            f"nested primary round-trip mismatch: {out.read_text()!r}"
        log = s.read_log()
        # The bridge must have run without looping: when the host advertises
        # data-control v2 the publish is logged once (a loop would recur).
        if "primary selection -> host" in log:
            assert log.count("primary selection -> host") == 1, \
                "primary bridge echoed/looked repeatedly:\n%s" % log[-500:]
        assert s.proc.poll() is None
    finally:
        s.close()


@pytest.mark.skipif(NEED_WL_CLIPBOARD, reason="wl-clipboard tools not installed")
def test_nested_host_primary_to_kick(tmp_path):
    """A host primary selection crosses the data-control bridge into kick
    (regression: host->nested primary paste used to hang the terminal waiting
    for EOF, and to never redraw).  The demander waits for the *expected
    payload* because a live host may already hold its own primary selection at
    boot, which kick faithfully mirrors first.

    Skipped on hosts without wlr-data-control v2 (e.g. GNOME/mutter): the
    host->nested primary mirror only exists on that backend, so this test
    needs a data-control host -- run it via ./test-headless.sh (whose headless
    host is kick itself) or on a compositor like labwc/sway."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    token = f"kick-h2n-{os.getpid()}-{time.monotonic_ns()}"
    out = tmp_path / "primary.out"
    s = _nested_session("%s --primary-demand-expect '%s' '%s' '#00ff00'"
                        % (WL, out, token),
                        f"{WL} '#ff0000'")
    try:
        s.wait_any_active()
        s.wait_active_frame()
        assert s.proc.poll() is None
        # The host->nested primary bridge only exists over wlr-data-control
        # v2 (a focus-based wl_data_device host has no primary channel at
        # all), which is decided at bridge_init before any client maps.
        s.wait_for(r"clipboard bridge:")
        if "clipboard bridge: host data-control backend" not in s.read_log():
            pytest.skip("host compositor lacks wlr-data-control v2; "
                        "host->nested primary selection is not bridged here")
        # Primary offers only reach the *focused* tab (wlroots delivers
        # device selection events to the focused client), so crown the
        # demand-expect tab (tab 0) before the host writes its primary.
        if s.active() != 0:
            s.cycle(-1)
            assert s.wait_active_change(1) == 0
        copy = subprocess.Popen(
            ["wl-copy", "-f", "--primary", token],
            env=dict(os.environ),
            stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            preexec_fn=_preexec_death(os.getpid()))
        try:
            deadline = time.monotonic() + 15.0
            while time.monotonic() < deadline:
                assert s.proc.poll() is None, \
                    f"compositor exited mid-test:\n{s.read_log()[-500:]}"
                if out.exists() and out.read_text() == token:
                    break
                time.sleep(0.1)
            assert out.exists(), \
                "nested demand-expect tab never received the host primary"
            assert out.read_text() == token, \
                f"host primary payload mismatch: {out.read_text()!r}"
        finally:
            copy.terminate()
            copy.wait(timeout=5)
        assert s.proc.poll() is None
    finally:
        s.close()


def test_standalone_headless_cursor_theme_none_boots(tmp_path):
    """No theme requested and no cursor theme installed: the builtin arrow is
    used silently (a request named literally 'default' is indistinguishable
    from the builtin), and there must be no warning and no failure."""
    empty = tmp_path / "empty-cursors"
    empty.mkdir()
    s = _spawn_standalone({"XCURSOR_PATH": str(empty)},
                          "build/wltest '#ff0000'")
    try:
        s.wait_app_active(0)
        assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
        log = s.read_log()
        assert "builtin cursor in use" not in log, \
            f"no warning expected for a 'default' request:\n{log[-400:]}"
    finally:
        s.close()


def test_standalone_headless_cursor_theme_installed_no_warning():
    """An installed theme must be used with no builtin-substitution warning.
    Uses the host's Adwaita theme as a real installed example."""
    if not os.path.exists("/usr/share/icons/Adwaita/cursors"):
        pytest.skip("host has no Adwaita cursor theme")
    s = _spawn_standalone({"XCURSOR_THEME": "Adwaita"},
                          "build/wltest '#ff0000'")
    try:
        s.wait_app_active(0)
        assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
        log = s.read_log()
        assert "builtin cursor in use" not in log, \
            f"installed theme must not be substituted:\n{log[-400:]}"
    finally:
        s.close()


def test_standalone_cursor_shape_request():
    """A client that asks for the 'text' cursor (e.g. galculator over its input
    field) has it applied to the compositor cursor: the request arrives via
    cursor-shape-v1 and wlroots sets the theme cursor accordingly."""
    s = _spawn_standalone({}, "build/wltest --cursor-shape text '#ff0000'")
    try:
        s.wait_app_active(0)
        s.ptr_motion(10, 10)  # give wltest a pointer enter
        s.wait_for(r"\[cursor\] shape text")
    finally:
        s.close()


def test_standalone_cursor_shape_resets_on_tab_switch():
    """The per-tab cursor shape is cached and the switch to another tab resets
    the compositor cursor to the default before its own shape is applied."""
    s = _spawn_standalone({}, "build/wltest '#0000ff'",
                          "build/wltest --cursor-shape text '#ff0000'")
    try:
        s.wait_app_active(1)
        s.ptr_motion(10, 10)
        s.wait_for(r"\[cursor\] shape text")
        baselines = s.read_log().count("[cursor] shape default")
        s.cycle(1)  # tab 1 -> tab 0 (plain client)

        shapes = []
        since = time.monotonic()
        while time.monotonic() - since < 15.0:
            shapes = re.findall(r"\[cursor\] shape \w+", s.read_log())
            if (shapes[-1] == "[cursor] shape default"
                    and s.read_log().count("[cursor] shape default") > baselines):
                return
            time.sleep(0.1)
        pytest.fail(f"shape not reset to default on tab switch: {shapes}")
    finally:
        s.close()


@pytest.mark.flaky
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 under
    ./test-headless.sh, whose headless host is kick itself and honours the
    forwarded set_shape; on a desktop host the outcome depends on the host's
    cursor-shape-v1 support (and on whether the real pointer has ever entered
    the window, which provides the serial the request is tagged with)."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("no host compositor for nested mode")
    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)
        s.warp(640, 400)
        if headless_host:
            s.wait_for(r"\[cursor\] host shape text")
        else:
            s.wait_for(r"\[cursor\] host shape (text|n/a)")
    finally:
        s.close()


def test_standalone_headless_last_tab_exits_cleanly(standalone_session):
    """Closing the final tab exits standalone with rc 0 instead of crashing
    on the NULL host-window proxies (which only exist in nested mode)."""
    s = standalone_session
    assert s.proc.poll() is None, f"standalone exited early: {s.read_log()[-500:]}"
    s.wait_any_active()
    for pid in s.launch_pids().values():
        os.kill(pid, signal.SIGTERM)
    rc = s.proc.wait(timeout=10)
    assert rc == 0, f"rc={rc} (139 = SIGSEGV); log tail:\n{s.read_log()[-800:]}"


def test_standalone_headless_screen_mode_sets_output_size():
    """--screen-mode WxH sizes the output; on the headless backend it becomes
    the add_output() size, observable through the dumped frame dimensions."""
    s = _spawn_standalone({}, "--screen-mode", "480x320", "build/wltest '#ff0000'")
    try:
        s.wait_app_active(0)
        assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
        w, h, _ = s.ppm_rgb(s.dump_host())
        assert (w, h) == (480, 320), \
            f"output should honour --screen-mode 480x320, got {w}x{h}"
    finally:
        s.close()


def test_standalone_headless_screen_mode_not_consumed_as_app():
    """--screen-mode is stripped from argv, so it must not become a tab
    launcher (no 'starts with -' warning) while both real apps still boot."""
    s = _spawn_standalone({}, "--screen-mode", "640x480",
                          "build/wltest '#ff0000'", "build/wltest '#0000ff'")
    try:
        s.wait_any_active()
        assert s.proc.poll() is None, f"log tail:\n{s.read_log()[-400:]}"
        assert "starts with '-'" not in s.read_log(), \
            f"--screen-mode leaked into the app list:\n{s.read_log()[-400:]}"
    finally:
        s.close()


def test_standalone_headless_screen_mode_bad_value_exits(tmp_path):
    """A malformed --screen-mode value is a startup error (rc 2) naming the
    problem, before any window or backend work."""
    env = dict(os.environ)
    for case in (["--screen-mode"], ["--screen-mode=abc"], ["--screen-mode", "1280"]):
        proc = subprocess.Popen(
            ["env", "-i",
             "PATH=" + env["PATH"],
             "HOME=" + env.get("HOME", "/"),
             "XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
             "KICK_MODE=standalone",
             "WLR_BACKENDS=headless",
             "stdbuf", "-oL",
             BIN] + case,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            preexec_fn=_preexec_death(os.getpid()),
        )
        out, _ = proc.communicate(timeout=15)
        assert proc.returncode == 2, \
            f"--screen-mode {case} should exit rc 2; rc={proc.returncode}\n{out}"
        assert b"--screen-mode" in out, \
            f"error should mention --screen-mode:\n{out}"


def test_standalone_headless_overlay_font_initialized(standalone_session):
    """Standalone builds the overlay glyph atlas with the wlr renderer, so tab
    titles are drawn next to their icons.  Regressed when overlay_font_init()
    only ever ran via the nested init_egl() path."""
    if _pick_font() is None:
        pytest.skip("no usable font on this machine")
    s = standalone_session
    assert s.proc.poll() is None, f"standalone exited early: {s.read_log()[-500:]}"
    s.wait_any_active()
    assert "[overlay] font " in s.read_log(), (
        "no standalone font atlas; overlay shows icons without text:\n"
        f"{s.read_log()[-800:]}")


def test_standalone_headless_overlay_no_needless_ellipsis(standalone_session):
    """Titles that fit their overlay row whole must not be shrunk to "..." —
    a row's text budget is its title's natural width, so forcing "..." space
    chopped the last characters (e.g. "foot" -> "fo...")."""
    if _pick_font() is None:
        pytest.skip("no usable font on this machine")
    s = standalone_session
    s.wait_any_active()

    s.mods(ALT)
    deadline = time.monotonic() + 15.0
    while time.monotonic() < deadline:
        w, h, px = s.ppm_rgb(s.dump_host())
        if _has_dark(w, h, px):
            break
    assert _has_dark(w, h, px), "standalone overlay box not visible while Alt"

    log = s.read_log()
    assert "[overlay] ellipsis" not in log, (
        "overlay ellipsized a title that fit its row:\n"
        f"{log[-800:]}")

    s.mods(0)


# -- clipboard bridge (GNOME/Mutter focus-based wl_data_device backend) -----

def _run_display(display, args, timeout=6, **kw):
    """Run a command against a specific Wayland display, capturing output."""
    env = dict(os.environ)
    env["WAYLAND_DISPLAY"] = display
    return subprocess.run(
        args, env=env, capture_output=True, text=True,
        stdin=subprocess.DEVNULL, timeout=timeout,
        preexec_fn=_preexec_death(os.getpid()), **kw)


def _poll_paste(display, want, deadline=45.0):
    """Wait until `wl-paste` on `display` returns exactly `want`."""
    r = None
    since = time.monotonic()
    while time.monotonic() - since < deadline:
        try:
            r = _run_display(display, ["wl-paste", "-n"])
        except subprocess.TimeoutExpired:
            # A single wl-paste can stall while the host clipboard daemon is
            # sluggish; that's "not there yet", not a failure.
            continue
        if r.stdout == want:
            return
        time.sleep(0.2)
    raise AssertionError(
        f"wl-paste on {display} never returned {want!r}; "
        + (f"last={r.stdout!r} err={r.stderr!r}"
           if r is not None else "every poll timed out"))


@pytest.mark.skipif(NEED_WL_CLIPBOARD, reason="wl-clipboard tools not installed")
def test_clipboard_nested_to_host(session):
    token = f"kick-n2h-{os.getpid()}-{time.monotonic_ns()}"
    copy = subprocess.Popen(
        ["wl-copy", "-f", token],
        env={**os.environ, "WAYLAND_DISPLAY": session.socket},
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
        preexec_fn=_preexec_death(os.getpid()))
    try:
        _poll_paste(os.environ.get("WAYLAND_DISPLAY", "wayland-0"), token)
    finally:
        copy.terminate()
        copy.wait(timeout=5)


@pytest.mark.skipif(NEED_WL_CLIPBOARD, reason="wl-clipboard tools not installed")
def test_clipboard_host_to_nested(session):
    token = f"kick-h2n-{os.getpid()}-{time.monotonic_ns()}"
    copy = subprocess.Popen(
        ["wl-copy", "-f", token],
        env=dict(os.environ),
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
        preexec_fn=_preexec_death(os.getpid()))
    try:
        _poll_paste(session.socket, token)
    finally:
        copy.terminate()
        copy.wait(timeout=5)


def test_tabs_map(session):
    session.wait_for(r"\[diag\] map app=0")
    session.wait_for(r"\[diag\] map app=1")


def test_cycle_tabs(session):
    start = session.wait_any_active()
    assert start in (0, 1)
    session.cycle(+1)
    other = session.wait_active_change(start)
    assert other == 1 - start
    session.cycle(-1)
    assert session.wait_active_change(other) == start


def test_function_keys_pass_through(session):
    """F9/F10/F11 are not grabbed by the compositor: they reach the focused
    client, so tapping them must not toggle fullscreen or trigger dumps."""
    start = session.wait_any_active()
    base_active = session.active_count()
    for code in (F11, 67, 68):  # F11, F9, F10
        session.tap(code)
    time.sleep(0.3)
    log = session.read_log()
    assert "host fullscreen=" not in log, "function key toggled fullscreen"
    assert "[dump] scene" not in log, "function key triggered a scene dump"
    assert "[dump] host" not in log, "function key triggered a host dump"
    assert session.active_count() == base_active, "function key cycled tabs"
    assert session.proc.poll() is None


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."""
    session.wait_any_active()
    session.tap(F)
    session.wait_for("host fullscreen=1")
    session.wait_for("clients fullscreen=1")
    session.tap(F)
    session.wait_for("host fullscreen=0")
    session.wait_for("clients fullscreen=0")
    assert session.proc.poll() is None


def test_sigterm_removes_socket(session):
    sock_path = os.path.join(
        os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}"),
        session.socket,
    )
    session.proc.terminate()
    session.proc.wait(timeout=5)
    assert session.proc.returncode == 0
    assert not os.path.exists(sock_path), f"socket left behind: {sock_path}"


def test_cycle_wraps(session):
    start = session.wait_any_active()
    assert start in (0, 1)
    session.cycle(+1)
    other = session.wait_active_change(start)
    assert other == 1 - start
    session.cycle(+1)
    assert session.wait_active_change(other) == start
    session.cycle(-1)
    assert session.wait_active_change(start) == other
    session.cycle(-1)
    assert session.wait_active_change(other) == start


def test_hold_alt_down_repeats(session):
    start = session.wait_any_active()
    assert start in (0, 1)
    # Press and hold Alt+Down and wait for the compositor's own repeat
    # timer (600ms delay, then 25Hz) to cycle several tabs.  base is counted
    # BEFORE the press so the press's own immediate cycle always counts toward
    # the >=3 transitions: the count can't race with the press being processed.
    # We still poll for the transitions rather than sleeping a fixed 1.3s so
    # any scheduling slack never pushes the repeats past the window.
    session.mods(ALT)
    base = session.active_count()
    session.key(DOWN, KEY_PRESSED)
    deadline = time.monotonic() + 5.0
    while time.monotonic() < deadline and session.active_count() - base < 3:
        time.sleep(0.05)
    transitions = session.active_count() - base
    # Clean release: let go of the key and the modifier.
    session.key(DOWN, KEY_RELEASED)
    session.mods(0)
    assert transitions >= 3, (
        f"held Alt+Down produced only {transitions} active transitions "
        "(expected repeats to cycle several tabs)"
    )
    settled = session.active_count()
    time.sleep(0.3)
    assert session.active_count() == settled, "repeats continued after release"


def test_alt_holds_tab_overlay(session):
    """Holding Alt shows the tab-title overlay (a centered box darkening the
    content); releasing Alt hides it again.  The overlay is drawn into the
    host frame in the same blit that produces the SIGUSR1 dump, so a dark
    box pixel appears in the centered window while Alt is held and reverts
    after release."""
    session.wait_any_active()
    session.wait_active_frame()

    def has_dark(w, h, px):
        for y in range(h // 4, 3 * h // 4):
            row = y * w * 3
            for x in range(w // 4, 3 * w // 4):
                i = row + x * 3
                if px[i] < 40 and px[i + 1] < 40 and px[i + 2] < 40:
                    return True
        return False

    w, h, px = session.ppm_rgb(session.dump_host())
    assert not has_dark(w, h, px), "content already dark before Alt"

    session.mods(ALT)
    deadline = time.monotonic() + 15.0
    while time.monotonic() < deadline:
        w, h, px = session.ppm_rgb(session.dump_host())
        if has_dark(w, h, px):
            break
    assert has_dark(w, h, px), "overlay box not visible while Alt is held"

    session.mods(0)
    deadline = time.monotonic() + 15.0
    while time.monotonic() < deadline:
        w, h, px = session.ppm_rgb(session.dump_host())
        if not has_dark(w, h, px):
            break
    assert not has_dark(w, h, px), "overlay did not hide on Alt release"
    assert session.proc.poll() is None


def test_non_ctrl_key_ignored(session):
    start = session.wait_any_active()
    assert start in (0, 1)
    base_active = session.active_count()
    session.key(30, KEY_PRESSED)  # 'A'
    session.key(30, KEY_RELEASED)
    time.sleep(0.3)
    assert session.active_count() == base_active, "'A' without ctrl cycled tabs"
    assert "host fullscreen=" not in session.read_log(), "'A' toggled fullscreen"
    assert session.proc.poll() is None
    session.tap(F11)
    time.sleep(0.3)
    assert "host fullscreen=" not in session.read_log(), "F11 toggled fullscreen"
    assert session.proc.poll() is None


def test_active_switch_on_client_exit(session):
    start = session.wait_any_active()
    assert start in (0, 1)
    pids = session.launch_pids()
    other = 1 - start

    # Kill the active tab: the compositor must fall through to the other.
    # [claim] pid is the connected client (wltest); SIGTERM closes its
    # Wayland socket, which is what makes the compositor destroy the tab.
    os.kill(pids[start], signal.SIGTERM)
    assert session.wait_active_change(start) == other, \
        "active did not fall through to the other tab after its tab died"
    assert session.proc.poll() is None, "compositor exited after one tab died"

    # Kill the last tab: the compositor should exit cleanly on its own.
    os.kill(pids[other], signal.SIGTERM)
    session.proc.wait(timeout=5)
    assert session.proc.returncode == 0, \
        f"compositor exited with {session.proc.returncode} after last tab closed"


def test_tab_death_activates_left_neighbor(session):
    """With three tabs [0,1,2], killing index 2 must activate its left
    neighbor (index 1), not the leftmost tab; killing the first tab (index 0)
    must fall through to index 1 rather than wrapping around to the right."""
    start = session.wait_any_active()
    assert start in (0, 1)
    # Ensure the tab at index 1 is active BEFORE the third client connects,
    # so the anchored new tab lands at index 2 (not stealing slot 1).
    if start == 0:
        session.cycle(+1)
        session.wait_active(1)
    # Both fixture tabs must have claimed slots 0 and 1 already, otherwise a
    # slot steal reorders things and the "index 2" assertion below lies.
    pids = session.launch_pids()

    third = subprocess.Popen(
        [WL, GREEN],
        env=dict(os.environ, WAYLAND_DISPLAY=session.socket),
        stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
        preexec_fn=_preexec_death(os.getpid()))
    try:
        # The new tab is anchored after the active tab (index 1): claim app=2.
        session.wait_for(r"\[claim\] app=2 pid=(\d+)")
        pid2 = int(re.search(r"\[claim\] app=2 pid=(\d+)",
                             session.read_log()).group(1))
        pid0 = pids[0]

        # The externally-spawned client is born hidden (no map-autofocus
        # anymore), so cycle to it: active is 1 (the crown), +1 reaches 2.
        session.cycle(+1)
        session.wait_active(2)

        # Rightmost tab dies: fall to its left neighbor (index 1), not 0.
        os.kill(pid2, signal.SIGTERM)
        assert session.wait_active_change(2) == 1, \
            "active did not fall to index 1 (left neighbor) after index 2 died"
        assert session.proc.poll() is None, \
            "compositor exited after middle tab died"

        # First tab dies: no left neighbor, fall right to index 1 (no wrap).
        session.cycle(-1)
        session.wait_active(0)
        os.kill(pid0, signal.SIGTERM)
        assert session.wait_active_change(0) == 1, \
            "active did not fall to index 1 after first tab died"
        assert session.proc.poll() is None, \
            "compositor exited after first tab died"
    finally:
        if third.poll() is None:
            third.terminate()
            third.wait(timeout=5)


def _nested_session(*tabs, extra_args=()):
    """Launch a nested compositor with an arbitrary tab list (each tab is a
    shell command string, as on the CLI).  extra_args are passed on the CLI
    before the tabs (e.g. --modifier super).  The `session` fixture exists for
    the common two-tab case; tests needing a specific tab lineup use this."""
    for path in (BIN, WL):
        if not os.access(path, os.X_OK):
            pytest.skip(f"build first: {path}")
    try:
        os.unlink(LOG)
    except FileNotFoundError:
        pass
    proc = subprocess.Popen(
        ["env", "KICK_LOG=2", "KICK_SYNTH_INPUT=1", "stdbuf", "-oL",
         BIN, *extra_args] + list(tabs),
        stdin=subprocess.PIPE,
        stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
        preexec_fn=_preexec_death(os.getpid()),
    )
    return Session(proc)


def test_alt_shift_t_becomes_active(session):
    """Alt+Shift+T's fresh terminal still becomes the active tab on map
    (the add_tab exception to the no-steal rule), while other new tabs never
    steal focus."""
    old_term = os.environ.get("TERMINAL")
    os.environ["TERMINAL"] = f"{WL} '#ffff00'"
    try:
        session.wait_any_active()
        before = session.active_count()
        session.chord(ALT | SHIFT, 20)
        deadline = time.monotonic() + 15.0
        while session.active_count() <= before:
            assert session.proc.poll() is None, \
                "compositor exited after Alt+Shift+T"
            assert time.monotonic() < deadline, \
                "Alt+Shift+T's terminal never became active"
            time.sleep(0.1)
        assert session.active_count() == before + 1, \
            "Alt+Shift+T should produce exactly one focus transition"
        assert session.proc.poll() is None
    finally:
        if old_term is None:
            os.environ.pop("TERMINAL", None)
        else:
            os.environ["TERMINAL"] = old_term


@pytest.mark.flaky
def test_sub_window_hidden_when_parent_not_active():
    """A second toplevel from an already-running client is a sub-window of
    that tab: it gets its own slot glued right after the parent tab, and
    because the parent is not the tab being shown it must NOT steal focus
    (no churn -- the crowned tab stays active)."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    # Tab 0 (RED) opens a green extra toplevel from the same client.
    s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'", f"{WL} '{BLUE}'")
    try:
        s.wait_any_active()
        p0 = s.launch_pids()[0]
        # The extra claims the slot right after tab 0 (app=1) with the same
        # client pid as RED.  BLUE can also claim app=1 (before the extra
        # inserts and shifts it), so match app=1 by pid on the extra.
        deadline = time.monotonic() + 15.0
        extra_claimed = False
        while time.monotonic() < deadline:
            for a, p in re.findall(r"\[claim\] app=(\d+) pid=(\d+)",
                                   s.read_log()):
                if a == "1" and int(p) == p0:
                    extra_claimed = True
                    break
            if extra_claimed:
                break
            assert s.proc.poll() is None, "compositor exited while extra mapped"
            time.sleep(0.1)
        assert extra_claimed, \
            "sub-window never claimed a slot right of its parent"
        # The extra must have appeared hidden: only the crown's active line
        # may exist.
        time.sleep(0.5)
        assert len(ACTIVE_RE.findall(s.read_log())) == 1, \
            "sub-window stole focus from the crowned tab"
        assert s.proc.poll() is None
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


@pytest.mark.flaky
def test_sub_window_activates_when_parent_active():
    """A sub-window whose parent tab is the one being shown becomes the
    active tab on map (the second exception to the no-steal rule)."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'")
    try:
        s.wait_any_active()  # the single tab is crowned, then ...
        s.wait_for(r"active app=1")  # ... its sub-window activates on map.
        assert s.proc.poll() is None
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


@pytest.mark.flaky
def test_sub_windows_stack_mode_never_claims_a_slot():
    """--sub-windows stack: a second toplevel from an already-running client
    rides on its owner's tab -- it logs [sub] stack, never claims a slot of
    its own, and takes the keyboard only because its parent tab is the one
    being shown."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'",
                        extra_args=("--sub-windows", "stack"))
    try:
        s.wait_any_active()               # app 0 crowned on its first map
        s.wait_for(r"\[sub\] stack app=0\b")
        s.wait_for(r"\[sub\] focus app=0\b")
        time.sleep(0.5)
        claims = CLAIM_RE.findall(s.read_log())
        assert len(claims) == 1 and claims[0][0] == "0", \
            f"stacked sub-window claimed a slot: {claims}"
        assert len(ACTIVE_RE.findall(s.read_log())) == 1, \
            "stacked sub-window stole focus via an active transition"
        assert s.proc.poll() is None
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


@pytest.mark.flaky
def test_sub_windows_stack_mode_focus_follows_tab():
    """--sub-windows stack: a stacked window of a background tab maps
    without taking the keyboard; showing its tab hands focus to it."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    # Tab 0 (RED) opens a green extra; tab 1 (BLUE) is the startup-selected
    # active tab, so the extra must map hidden.
    s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'", f"{WL} '{BLUE}'",
                        extra_args=("--sub-windows", "stack"))
    try:
        s.wait_any_active()               # app 1 crowned
        s.wait_for(r"\[sub\] stack app=0\b")
        s.wait_for(r"\[sub\] map app=0\b")   # mapped while its tab is hidden
        time.sleep(0.3)
        assert "[sub] focus app=0" not in s.read_log(), \
            "hidden tab's stacked window took the keyboard"
        s.cycle_ack(-1)                   # show tab 0
        s.wait_for(r"\[sub\] focus app=0\b")
        assert s.proc.poll() is None
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


@pytest.mark.flaky
def test_sub_windows_stack_mode_unstack_refocuses_then_close():
    """--sub-windows stack: closing the stacked window ('X' in wltest)
    refocuses the tab's own surface without killing the tab; Alt+Delete on
    the tab then tears the whole unit down."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'",
                        extra_args=("--sub-windows", "stack"))
    try:
        s.wait_any_active()
        s.wait_for(r"\[sub\] focus app=0\b")  # extra holds the keyboard
        s.tap(45)                             # 'X': close the extra
        s.wait_for(r"\[sub\] unstack app=0\b")
        s.wait_for(r"\[sub\] refocus app=0\b")
        time.sleep(0.5)
        claims = CLAIM_RE.findall(s.read_log())
        assert len(claims) == 1, f"tab lost or gained slots: {claims}"
        assert s.proc.poll() is None, "compositor died when the extra closed"
        s.chord(ALT, DELETE)                  # close the tab itself
        deadline = time.monotonic() + 10.0
        while s.proc.poll() is None:
            assert time.monotonic() < deadline, \
                "Alt+Delete never tore the tab with stacked windows down"
            time.sleep(0.1)
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


@pytest.mark.flaky
def test_sub_windows_stack_mode_row_title_follows_top():
    """--sub-windows stack: the overlay row shows the topmost stacked
    window's title while it is mapped, and falls back to the tab's own
    surface title when it closes ('X' in wltest)."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} --extra '{GREEN}' '{RED}'",
                        extra_args=("--sub-windows", "stack"))
    try:
        s.wait_any_active()
        s.wait_for(r"\[sub\] focus app=0\b")
        # The resolved-title log only fires while overlay rows are drawn,
        # so hold Alt for the whole check.
        s.mods(ALT)
        try:
            s.wait_for(r"\[title\] app=0 'extra'")
            s.tap(45)   # 'X': close the extra
            s.wait_for(r"\[title\] app=0 '#ff0000'")
        finally:
            s.mods(0)
        assert s.proc.poll() is None
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


def test_alt_delete_activates_right_neighbor():
    """Alt+Delete closes the active tab, making the tab to its right active
    (SIGTERM, escalated to SIGKILL only if the client lingers).  With three
    tabs [0,1,2] and 1 active, closing 1 must activate 2, not 0."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} '{RED}'", f"{WL} '{GREEN}'", f"{WL} '{BLUE}'")
    try:
        start = s.wait_any_active()
        assert start in (0, 1, 2)
        # Ensure the third tab has claimed a slot so index 2 really exists.
        s.wait_for(r"\[claim\] app=2 pid=(\d+)")
        # Make tab 1 active so a right neighbor exists.
        for _ in range(3):
            if s.active() == 1:
                break
            if s.cycle_ack(+1) == 1:
                break
            if s.proc.poll() is not None:
                pytest.fail("compositor exited while cycling to tab 1")
        s.wait_active(1)

        s.chord(ALT, DELETE)
        assert s.wait_active_change(1) == 2, \
            "Alt+Delete did not activate the right neighbor"
        assert s.proc.poll() is None, "compositor exited after Alt+Delete"
        s.wait_for(r"kick: closing tab 1")
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


def test_alt_delete_rightmost_falls_left():
    """Closing the rightmost tab has no right neighbor to hand active off to,
    so it falls back to the left neighbor (index 1 of [0,1,2])."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} '{RED}'", f"{WL} '{GREEN}'", f"{WL} '{BLUE}'")
    try:
        s.wait_any_active()
        # Ensure the third tab has claimed a slot so index 2 really exists.
        s.wait_for(r"\[claim\] app=2 pid=(\d+)")
        while s.active() != 2:
            if s.cycle_ack(+1) == 2:
                break
            if s.proc.poll() is not None:
                pytest.fail("compositor exited while cycling to tab 2")
        s.wait_active(2)

        s.chord(ALT, DELETE)
        assert s.wait_active_change(2) == 1, \
            "rightmost Alt+Delete did not fall back to the left neighbor"
        assert s.proc.poll() is None, \
            "compositor exited after closing rightmost tab"
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


def test_alt_delete_second_tab_right_neighbor_two_tabs():
    """Two-tab session: closing tab 0 makes its right neighbor (1) active."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} '{RED}'", f"{WL} '{BLUE}'")
    try:
        s.wait_any_active()
        if s.active() == 1:
            if s.cycle_ack(-1) != 0:
                s.wait_active(0)
        s.chord(ALT, DELETE)
        assert s.wait_active_change(0) == 1, \
            "closing tab 0 did not activate tab 1"
        assert s.proc.poll() is None, "compositor exited after closing tab 0"
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


def test_alt_shift_arrows_reorder_and_wrap():
    """Alt+Shift+Up/Down move the active tab within the overlay list without
    switching the active surface; moving off an end wraps to the other end and
    the whole list shifts (a rightmost tab moved off the end lands first with
    every other tab pushed one place, not swapped into place).  The resulting
    order is provable behaviorally: an Alt+Shift+T terminal claims the slot
    right after the active tab's new position."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    old_term = os.environ.get("TERMINAL")
    os.environ["TERMINAL"] = f"{WL} '{RED}'"
    try:
        s = _nested_session(f"{WL} '{RED}'", f"{WL} '{GREEN}'",
                            f"{WL} '{BLUE}'")
        try:
            s.wait_any_active()
            # Make the third tab (index 2) active so there is a middle and an
            # end to move through.
            for _ in range(3):
                if s.active() == 2:
                    break
                if s.cycle_ack(+1) == 2:
                    break
                if s.proc.poll() is not None:
                    pytest.fail("compositor exited while cycling to tab 2")
            s.wait_active(2)
            before = s.active_count()

            # Adjacent move UP: [R,G,B] -> [R,B,G].
            s.chord(ALT | SHIFT, UP)
            s.wait_for(r"\[reorder\] app=1 dir=-1")
            # Adjacent move DOWN: -> [R,G,B].
            s.chord(ALT | SHIFT, DOWN)
            s.wait_for(r"\[reorder\] app=2 dir=1")
            # Move DOWN off the end: wraps to the front and rotates the rest,
            # [R,G,B] -> [B,R,G] (B first, R and G shifted, not swapped).
            s.chord(ALT | SHIFT, DOWN)
            s.wait_for(r"\[reorder\] app=0 dir=1")
            time.sleep(0.3)
            assert s.active() == 2, "reorder must not switch the active tab"
            assert s.active_count() == before, \
                "reorder must not emit an active-tab transition"

            # With the active tab wrapped to slot 0, a new Alt+Shift+T tab
            # is anchored right after it: the newest claim must be app=1.
            claims = re.findall(r"\[claim\] app=(\d+) pid=(\d+)", s.read_log())
            s.chord(ALT | SHIFT, 20)
            deadline = time.monotonic() + 15.0
            while len(re.findall(r"\[claim\] app=(\d+) pid=(\d+)",
                                 s.read_log())) <= len(claims):
                assert s.proc.poll() is None, \
                    "compositor exited after reorder + new tab"
                assert time.monotonic() < deadline, \
                    "Alt+Shift+T never claimed a slot"
                time.sleep(0.1)
            newest = re.findall(r"\[claim\] app=(\d+) pid=(\d+)",
                                s.read_log())[-1]
            assert newest[0] == "1", \
                f"new tab claimed slot {newest[0]}, expected 1 (after B->front)"
        finally:
            s.close()
            if s.proc.poll() is None:
                s.proc.terminate()
    finally:
        if old_term is None:
            os.environ.pop("TERMINAL", None)
        else:
            os.environ["TERMINAL"] = old_term


def test_modifier_super_flag():
    """--modifier super moves the tab-management chords to Super/Mod4: cycle,
    reorder and close all work, while the Alt chords stop doing anything."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} '{RED}'", f"{WL} '{BLUE}'",
                        extra_args=["--modifier", "super"])
    try:
        start = s.wait_any_active()
        assert s.proc.poll() is None, "compositor exited with --modifier super"

        # Super+Down cycles; a plain Alt+Down must not.
        s.chord(SUPER, DOWN)
        other = s.wait_active_change(start, timeout=5.0)
        assert other == 1 - start, "Super+Down did not cycle the other tab"
        s.chord(ALT, DOWN)
        time.sleep(0.3)
        assert s.active() == other, \
            "Alt+Down still cycled in --modifier super mode"

        # Super+Shift+Up reorders the active tab.
        s.chord(SUPER | SHIFT, UP)
        s.wait_for(r"\[reorder\] app=\d+ dir=-1")

        # Super+Delete closes the active tab: the compositor logs the close
        # and hands active off (a NEW active-tab line appears).  Can't use
        # wait_active_change here: reorder changed the active tab's slot
        # without logging one, so the post-close index can match the
        # pre-close value even though the tab underneath changed.
        count = s.active_count()
        s.chord(SUPER, DELETE)
        s.wait_for(r"kick: closing tab")
        deadline = time.monotonic() + 5.0
        while s.active_count() <= count:
            assert s.proc.poll() is None, \
                "compositor exited after Super+Delete"
            assert time.monotonic() < deadline, \
                "active never handed off after Super+Delete"
            time.sleep(0.1)
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


def test_alt_delete_escalates_sigkill(tmp_path):
    """A client that traps SIGTERM must still be reaped by the SIGKILL
    escalation, so Alt+Delete never strands a lingering tab."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    # A wrapper that ignores SIGTERM, then execs the test client so the
    # tab's process group carries SIG_IGN for TERM across the exec.
    wrapper = tmp_path / "trap-term.sh"
    wrapper.write_text(
        "trap '' TERM\nexec '{}' '{}'\n".format(WL, GREEN))
    wrapper.chmod(0o755)
    s = _nested_session(f"{WL} '{RED}'", str(wrapper))
    try:
        pids = s.launch_pids()
        assert 1 in pids, "second tab never claimed a slot"
        trap_pid = pids[1]
        # Wait for the startup active-tab lines to settle (both clients map
        # asynchronously and toggle `active app=` as they land), otherwise
        # s.active() can return a transient value and the Alt+Delete below
        # closes the wrong tab.
        s.wait_any_active()
        # The trap tab is the second; bring it active and close it.
        while s.active() != 1:
            if s.cycle_ack(+1) == 1:
                break
            if s.proc.poll() is not None:
                pytest.fail("compositor exited while cycling to the trap tab")
        s.wait_active(1)

        s.chord(ALT, DELETE)
        deadline = time.monotonic() + 5.0
        alive = True
        while time.monotonic() < deadline:
            try:
                os.kill(trap_pid, 0)
                time.sleep(0.1)
            except ProcessLookupError:
                alive = False
                break
        assert not alive, "SIGTERM-trapping tab survived Alt+Delete (no SIGKILL escalation)"
        # It must have moved active away from the closing tab, and stayed up.
        assert s.active() != 1, "active still on the closing trap tab"
        assert s.proc.poll() is None, "compositor exited after closing trap tab"
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


def test_plain_delete_not_grabbed(session):
    """Plain Delete (no Alt) must not close the active tab."""
    session.wait_any_active()
    base_active = session.active()
    session.tap(DELETE)
    time.sleep(0.3)
    assert session.active() == base_active, \
        "Delete without Alt closed the active tab"
    assert session.proc.poll() is None


def test_standalone_headless_alt_delete_closes_tab(standalone_session):
    """Alt+Delete routes through the shared hotkey path in standalone mode:
    the active tab closes and the neighbor becomes active, then closing the
    last tab exits cleanly."""
    s = standalone_session
    start = s.wait_any_active()
    other = 1 - start
    s.chord(ALT, DELETE)
    assert s.wait_active_change(start) == other, \
        "standalone Alt+Delete did not activate the other tab"
    assert s.proc.poll() is None, "standalone compositor exited mid-test"

    # Close the final tab: the compositor should exit cleanly on its own.
    s.chord(ALT, DELETE)
    s.proc.wait(timeout=5)
    assert s.proc.returncode == 0, \
        f"standalone compositor exited with {s.proc.returncode} on last tab close"


def test_sigusr1_dumps(session):
    scene_rx = re.compile(r"\[dump\] scene")
    host_rx = re.compile(r"\[dump\] host back buffer -> (\S+\.ppm)")
    # Warm the dump machinery so both baselines exist, then assert SIGUSR1
    # produces a *new* scene dump attempt and a *new* host ppm.
    session.dump_host()
    session.wait_for(r"\[dump\] scene")
    scene_before = len(scene_rx.findall(session.read_log()))
    host_before = len(host_rx.findall(session.read_log()))

    os.kill(session.proc.pid, signal.SIGUSR1)
    since = time.monotonic()
    while time.monotonic() - since < 5.0:
        log = session.read_log()
        if len(scene_rx.findall(log)) > scene_before and \
           len(host_rx.findall(log)) > host_before:
            return
        time.sleep(0.1)
    pytest.fail("SIGUSR1 produced no new scene+host dump lines")


def test_dash_h_prints_usage():
    """ `-h` is the way to see usage: it prints it and exits 0 without
    starting a session. """
    r = subprocess.run([BIN, "-h"], stdout=subprocess.PIPE,
                       stderr=subprocess.STDOUT,
                       preexec_fn=_preexec_death(os.getpid()))
    assert r.returncode == 0
    assert b"usage:" in r.stdout


@pytest.mark.flaky
def test_no_args_opens_terminal():
    """With no command-line tabs, a single terminal tab opens and maps."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    for path in (BIN, WL):
        if not os.access(path, os.X_OK):
            pytest.skip(f"build first: {path}")
    try:
        os.unlink(LOG)
    except FileNotFoundError:
        pass

    # The one tab is the terminal named by TERMINAL (a wltest, so no real
    # terminal emulator has to be installed on the test host).
    proc = subprocess.Popen(
        ["env", "KICK_LOG=2", "KICK_SYNTH_INPUT=1",
         f"TERMINAL={WL} '{RED}'", "stdbuf", "-oL", BIN],
        stdin=subprocess.PIPE,
        stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
        preexec_fn=_preexec_death(os.getpid()),
    )
    s = Session(proc)
    try:
        s.wait_for(r"\[diag\] map app=0")
    finally:
        s.close()


def test_no_args_missing_terminal_fails():
    """A default terminal that is not on PATH aborts before any window."""
    r = subprocess.run(
        ["env", "TERMINAL=/nonexistent/tw-term", BIN],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        preexec_fn=_preexec_death(os.getpid()))
    assert r.returncode != 0
    assert b"not found in PATH" in (r.stdout + r.stderr)


def test_missing_arg_fails():
    """A command-line tab whose program is not on PATH aborts up front."""
    r = subprocess.run(
        [BIN, "definitely-no-such-cmd-xyz"],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        preexec_fn=_preexec_death(os.getpid()))
    assert r.returncode != 0
    assert b"definitely-no-such-cmd-xyz" in (r.stdout + r.stderr)
    assert b"not found in PATH" in (r.stdout + r.stderr)


def test_close_black_tab_active_last():
    """Alt+Delete on a black tab (its command has not connected yet) closes
    it outright: the surviving tab becomes active, the black slot is removed
    (closed tabs leave no hole), and closing the survivor exits cleanly."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    # CLI #0 (RED) maps hidden; the active tab is the still-black `sleep`.
    s = _nested_session(f"{WL} '{RED}'", "sleep 999")
    try:
        s.wait_for(r"\[diag\] map app=0")
        s.chord(ALT, DELETE)
        s.wait_for(r"kick: closing tab 1")
        # The black tab is gone; the RED survivor becomes active in its place.
        s.wait_for(r"active app=0")
        assert s.proc.poll() is None, \
            "compositor exited after closing the black tab"

        # Closing the last live tab tears the session down cleanly.
        m = re.search(r"\[claim\] app=0 pid=(\d+)", s.read_log())
        assert m, "RED tab never claimed a slot"
        os.kill(int(m.group(1)), signal.SIGTERM)
        s.proc.wait(timeout=5)
        assert s.proc.returncode == 0, \
            f"compositor exited with {s.proc.returncode} after last tab closed"
    finally:
        s.close()


def test_start_cmd_index_selects_command():
    """--start-cmd-index N (1-based) starts focused on command N instead of
    the default last command: index 1 makes the first command the active,
    black tab from the very first frame, revealed when it maps."""
    if not os.environ.get("WAYLAND_DISPLAY"):
        pytest.skip("needs a live host Wayland session (WAYLAND_DISPLAY)")
    s = _nested_session(f"{WL} '{RED}'", f"{WL} '{GREEN}'", f"{WL} '{BLUE}'",
                        extra_args=["--start-cmd-index", "1"])
    try:
        assert s.wait_any_active() == 0, \
            "--start-cmd-index 1 should make the first command active"
        assert s.proc.poll() is None
    finally:
        s.close()
        if s.proc.poll() is None:
            s.proc.terminate()


def test_start_cmd_index_errors():
    """A malformed or out-of-range --start-cmd-index is a startup error (rc 2)
    naming the flag, before any window or backend work.  The index is 1-based,
    so 0 is invalid and a value beyond the command count is out of range."""
    env = dict(os.environ)
    cases = (["--start-cmd-index"],
             ["--start-cmd-index=abc"],
             ["--start-cmd-index", "0"],
             ["--start-cmd-index", "4", "build/wltest '#ff0000'",
              "build/wltest '#00ff00'", "build/wltest '#0000ff'"])
    for case in cases:
        proc = subprocess.Popen(
            ["env", "-i",
             "PATH=" + env["PATH"],
             "HOME=" + env.get("HOME", "/"),
             "XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
             "KICK_MODE=standalone",
             "WLR_BACKENDS=headless",
             "stdbuf", "-oL",
             BIN] + case,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            preexec_fn=_preexec_death(os.getpid()),
        )
        out, _ = proc.communicate(timeout=15)
        assert proc.returncode == 2, \
            f"--start-cmd-index {case} should exit rc 2; rc={proc.returncode}\n{out}"
        assert b"--start-cmd-index" in out, \
            f"error should mention --start-cmd-index:\n{out}"


def test_sub_windows_flag_errors():
    """A missing or bad --sub-windows value is a startup error (rc 2) naming
    the flag, before any window or backend work."""
    env = dict(os.environ)
    cases = (["--sub-windows"],
             ["--sub-windows", "float"],
             ["--sub-windows=float"])
    for case in cases:
        proc = subprocess.Popen(
            ["env", "-i",
             "PATH=" + env["PATH"],
             "HOME=" + env.get("HOME", "/"),
             "XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
             "KICK_MODE=standalone",
             "WLR_BACKENDS=headless",
             "stdbuf", "-oL",
             BIN] + case,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            preexec_fn=_preexec_death(os.getpid()),
        )
        out, _ = proc.communicate(timeout=15)
        assert proc.returncode == 2, \
            f"--sub-windows {case} should exit rc 2; rc={proc.returncode}\n{out}"
        assert b"--sub-windows" in out, \
            f"error should mention --sub-windows:\n{out}"


def test_dead_command_explains_and_exits():
    """A sole command that exits before opening a window is dropped with a
    plain-stderr explanation (command + exit status) and a clean exit(0) --
    never silence.  Regression: bare `xmessage` exits 1 right after opening
    the display, which used to end the session with no word from kick."""
    if not os.access(BIN, os.X_OK):
        pytest.skip("build first")
    s = _spawn_kick_standalone(["/bin/false"])
    try:
        s.proc.wait(timeout=10)
        assert s.proc.returncode == 0, \
            f"expected clean exit(0), got {s.proc.returncode}"
        log = s.read_log()
        assert "command exited without opening a window" in log
        assert "exit code 1" in log
        assert "no tabs left to show, exiting" in log
    finally:
        s.close()


# ===================== XWayland (X11 tabs) =====================

# X11 client used for XWayland tests: xmessage pops a plain window that
# accepts our full-output configure (no fixed size hints), so it fills
# the tab and renders a non-black frame.
XMSG = shutil.which("xmessage")
HAVE_XWAYLAND = shutil.which("Xwayland") is not None and XMSG is not None


def _spawn_kick_standalone(commands):
    """Standalone headless kick with an arbitrary per-tab command list.

    Same environment discipline as standalone_session: no WAYLAND_DISPLAY
    leak, headless backend forced, KICK_LOG=2 so client stderr (and any
    echo tricks) land in the shared log.
    """
    env = dict(os.environ)
    try:
        os.unlink(LOG)
    except FileNotFoundError:
        pass
    proc = subprocess.Popen(
        ["env", "-i",
         "PATH=" + env["PATH"],
         "HOME=" + env.get("HOME", "/"),
         "XDG_RUNTIME_DIR=" + env.get("XDG_RUNTIME_DIR", "/tmp"),
         "XDG_DATA_HOME=" + env.get("XDG_DATA_HOME", ""),
         "XDG_DATA_DIRS=" + env.get("XDG_DATA_DIRS", ""),
         "KICK_LOG=2",
         "KICK_MODE=standalone",
         "WLR_BACKENDS=headless",
         "KICK_FONT=" + (env.get("KICK_FONT") or _pick_font() or ""),
         "XCURSOR_THEME=Adwaita",
         "stdbuf", "-oL",
         BIN] + list(commands),
        stdin=subprocess.PIPE,
        stdout=open(LOG, "ab"), stderr=subprocess.STDOUT,
        preexec_fn=_preexec_death(os.getpid()),
    )
    return Session(proc)


@pytest.fixture(scope="function")
def xwayland_session():
    """Mixed session: tab 0 is an X11 client (xmessage), tab 1 is Wayland.

    The compositor starts focused on the *last* command, so the session
    boots on the Wayland tab and Alt+Down wraps to the X11 one.
    """
    if not HAVE_XWAYLAND:
        pytest.skip("needs Xwayland and xmessage on PATH")
    s = _spawn_kick_standalone([f"{XMSG} hi", f"{WL} '{RED}'"])
    try:
        yield s
    finally:
        s.close()


def test_xwayland_ready_and_claim(xwayland_session):
    """XWayland comes up in lazy mode, and the X11 command's window is
    claimed by its own process group into tab slot 0."""
    s = xwayland_session
    assert s.proc.poll() is None, f"kick exited early: {s.read_log()[-500:]}"
    # Lazy start still assigns the display name synchronously at create
    # time; the ready line proves the X server actually spawned.
    s.wait_for(r"\[xwayland\] ready display=:(\d+)")
    m = re.search(r"\[xwayland\] ready display=:(\d+)", s.read_log())
    assert int(m.group(1)) > 0, "Xwayland display number must be positive"
    pids = {}
    since = time.monotonic()
    while time.monotonic() - since < 15.0:
        d = dict((int(a), int(p)) for a, p in CLAIM_RE.findall(s.read_log()))
        if {0, 1}.issubset(d):
            pids = d
            break
        time.sleep(0.1)
    else:
        pytest.fail(f"both tabs never claimed:\n{s.read_log()[-800:]}")
    # The X11 claim must be xmessage itself (its pid), not some unrelated
    # process -- i.e. pgid matching picked the right group.
    with open(f"/proc/{pids[0]}/comm") as f:
        assert f.read().strip() == "xmessage", \
            f"tab 0 claimed wrong process: pid {pids[0]}"


def test_xwayland_display_env_exported():
    """Children see DISPLAY pointing at the embedded Xwayland; without
    XWayland support the variable would be unset entirely."""
    if not HAVE_XWAYLAND:
        pytest.skip("needs Xwayland and xmessage on PATH")
    s = _spawn_kick_standalone(["sh -c 'echo DISPLAY=$DISPLAY'"])
    try:
        s.wait_for(r"DISPLAY=:(\d+)", timeout=15.0)
        m = re.search(r"DISPLAY=:(\d+)", s.read_log())
        assert int(m.group(1)) > 0
    finally:
        s.close()


def test_xwayland_tab_renders_and_cycles(xwayland_session):
    """The X11 tab cycles like any other and its surface actually reaches
    the screen: after wrapping to it, the host back buffer shows xmessage's
    non-black content filling the output."""
    s = xwayland_session
    start = s.wait_any_active()
    assert start == 1, f"expected to boot on the Wayland tab, got {start}"
    # Wait until the X11 surface has mapped before cycling, so the wrap
    # lands on a live tab rather than a black placeholder.
    s.wait_for(r"\[diag\] map app=0")
    got = s.cycle_ack(+1)
    assert got == 0, f"Alt+Down did not wrap to the X11 tab (got {got})"
    ppm = s.dump_host()
    w, h, px = s.ppm_rgb(ppm)
    # Sample the centre plus quarter points; xmessage paints its whole
    # window grey/white, so every sample must be non-black.
    for cx, cy in ((w // 2, h // 2), (w // 4, h // 4),
                   (3 * w // 4, 3 * h // 4)):
        r, g, b = s.pixel_at(ppm, cx, cy)
        assert (r, g, b) != (0, 0, 0), \
            f"X11 tab rendered black at ({cx},{cy})"


def test_xwayland_alt_delete_closes_x11_tab(xwayland_session):
    """Alt+Delete on the active X11 tab sends WM_DELETE_WINDOW; xmessage
    honors it, exits, and the escalation never needs to fire."""
    s = xwayland_session
    s.wait_any_active()
    s.wait_for(r"\[diag\] map app=0")
    got = s.cycle_ack(+1)
    assert got == 0
    d = dict((int(a), int(p)) for a, p in CLAIM_RE.findall(s.read_log()))
    xpid = d[0]
    s.chord(ALT, DELETE)
    since = time.monotonic()
    while time.monotonic() - since < 10.0:
        try:
            os.kill(xpid, 0)
        except ProcessLookupError:
            break  # xmessage exited from the graceful close
        time.sleep(0.1)
    else:
        pytest.fail(f"xmessage (pid {xpid}) survived Alt+Delete")
    # The compositor stays up with the remaining Wayland tab.
    assert s.proc.poll() is None, "compositor exited after closing X11 tab"
    assert s.active() == 1