tai/tests/mplex.py

#!/usr/bin/env python3
"""tests/mplex.py - integration tests for the mplex terminal multiplexer (tai.py --mplex).

Deterministic throughout, same reasoning as tests/plexer.py: no model decision exists
anywhere in this path, so the "real LLM" rule in tests/README.md has nothing to apply
to. What is real instead is everything else: the real tai.py binary as a client under
a real pty, a real detached server process, real bash panes, and the user's actual
key bytes (prefix, arrows, Alt-digits) written to the outer pty exactly as a terminal
would deliver them. State is verified two ways: what the attached terminal shows
(the client's output is fed through TuiVtScreen), and what `tai.py --mplex <command>`
reports about the server, which is the same command layer the key bindings call.

Every test runs in a throwaway TAI_HOME, so a real ~/.tai/mplex.toml never leaks in
and a real session is never touched. teardown() always kills the server it started.
"""
import asyncio
import fcntl
import os
import shutil
import struct
import subprocess
import sys
import tempfile
import termios
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from tai_test import TestCase, TestAssertionFailed, main_cli  # noqa: E402
import tai  # noqa: E402

PREFIX = b"\x02"
UP = b"\x1b[A"
DOWN = b"\x1b[B"
RIGHT = b"\x1b[C"
LEFT = b"\x1b[D"
CTRL_UP = b"\x1b[1;5A"


def alt(char):
    return b"\x1b" + char.encode()


class MplexTestBase(TestCase):
    """Throwaway TAI_HOME, an attachable client with a decoded screen, and a CLI helper."""

    mode = "pty"
    timeout = 90.0
    rows = 24
    cols = 80
    config = None

    async def setup(self):
        self.tmp_home = tempfile.mkdtemp(prefix="tai-mplex-test-")
        self.env_overrides = {"TAI_HOME": self.tmp_home, "TAI_PASSPHRASE": "", "TERM": "xterm-256color", "SHELL": "/bin/bash"}
        self.env_overrides.pop("TAI_PLEXER_ACTIVE", None)
        if self.config is not None:
            with open(os.path.join(self.tmp_home, "mplex.toml"), "w", encoding="utf-8") as handle:
                handle.write(self.config)
        self.clients = []

    def cli_env(self):
        env = dict(os.environ)
        env.update(self.env_overrides)
        for name in (tai.PLEXER_ACTIVE_ENV, tai.PLEXER_SOCKET_ENV, tai.PLEXER_PANE_ENV, "TMUX"):
            env.pop(name, None)
        return env

    def cli(self, *words, check=True):
        done = subprocess.run([sys.executable, str(self.tai_path), "--mplex"] + list(words), env=self.cli_env(),
                              stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=30)
        if check and done.returncode != 0:
            raise TestAssertionFailed("mplex %s failed (%d): %s" % (" ".join(words), done.returncode, (done.stdout + done.stderr).strip()))
        return (done.stdout + done.stderr).strip() if not check else done.stdout.strip()

    async def attach(self, *words):
        """Spawn a client; returns (proc, screen) with a drain task decoding its output."""
        proc = await self.spawn(extra_args=["--mplex"] + list(words), env_overrides={})
        fcntl.ioctl(proc._master_fd, termios.TIOCSWINSZ, struct.pack("HHHH", self.rows, self.cols, 0, 0))
        screen = tai.TuiVtScreen(self.rows, self.cols)
        task = asyncio.ensure_future(self.drain(proc, screen))
        self.clients.append((proc, screen, task))
        return proc, screen

    @staticmethod
    async def drain(proc, screen):
        try:
            while True:
                chunk = await proc._read_chunk(65536)
                if not chunk:
                    await asyncio.sleep(0.05)
                else:
                    screen.feed(chunk)
        except asyncio.CancelledError:
            pass

    @staticmethod
    def status(screen):
        return screen.text().split("\n")[-1]

    async def wait_for(self, what, predicate, timeout=10.0):
        deadline = time.monotonic() + timeout
        last = None
        while time.monotonic() < deadline:
            try:
                last = predicate()
            except TestAssertionFailed:
                last = None
            if last:
                return last
            await asyncio.sleep(0.05)
        raise TestAssertionFailed("timed out waiting for: %s" % what)

    async def type_keys(self, proc, data, settle=0.15):
        os.write(proc._master_fd, data)
        await asyncio.sleep(settle)

    async def teardown(self):
        for _proc, _screen, task in self.clients:
            task.cancel()
        try:
            self.cli("kill-server", check=False)
        except Exception:
            pass
        await asyncio.sleep(0.3)
        shutil.rmtree(self.tmp_home, ignore_errors=True)

    def expect(self, label, actual, expected):
        if actual != expected:
            raise TestAssertionFailed("%s: expected %r, got %r" % (label, expected, actual))


class MplexLifecycleTest(MplexTestBase):
    """Start, work, detach, survive, reattach, exit, server gone."""

    name = "mplex-lifecycle"

    async def run(self):
        proc, screen = await self.attach()
        await self.wait_for("status bar with a window list", lambda: "0:bash*" in self.status(screen))
        sock = os.path.join(self.tmp_home, "plexer", tai.MPLEX_SOCKET_NAME)
        await self.wait_for("server socket", lambda: os.path.exists(sock))
        os.write(proc._master_fd, b"echo keepalive-$((20+22))\r")
        await self.wait_for("command output in the pane", lambda: "keepalive-42" in screen.text())
        listing_attached = self.cli("list-sessions")

        await self.type_keys(proc, PREFIX + b"d")
        code = await proc.wait(timeout=10.0)
        detached_text = screen.text()
        server_alive = os.path.exists(sock)
        listing_detached = self.cli("list-sessions")

        proc2, screen2 = await self.attach("attach")
        await self.wait_for("the reattached screen", lambda: "keepalive-42" in screen2.text())
        await self.type_keys(proc2, b"exit\r")
        code2 = await proc2.wait(timeout=10.0)
        await self.wait_for("server to exit after its last pane", lambda: not os.path.exists(sock), timeout=8.0)
        return {"code": code, "detached_text": detached_text, "server_alive": server_alive, "attached": listing_attached,
                "detached": listing_detached, "code2": code2, "exit_text": screen2.text()}

    async def verify(self, result):
        self.expect("detach exit code", result["code"], 0)
        if "[detached (from session 0)]" not in result["detached_text"]:
            raise TestAssertionFailed("detach message missing: %r" % result["detached_text"][-200:])
        if not result["server_alive"]:
            raise TestAssertionFailed("server died when the only client detached")
        if "(attached)" not in result["attached"] or "(attached)" in result["detached"]:
            raise TestAssertionFailed("attached flag wrong: %r / %r" % (result["attached"], result["detached"]))
        self.expect("exit code after last pane", result["code2"], 0)
        if "[exited]" not in result["exit_text"]:
            raise TestAssertionFailed("[exited] message missing: %r" % result["exit_text"][-200:])


class MplexShortcutsTest(MplexTestBase):
    """Every default prefix shortcut that has a keyboard effect, checked against the command layer."""

    name = "mplex-shortcuts"
    timeout = 150.0

    def q(self, fmt, target=None):
        words = ["display-message", "-p"] + (["-t", target] if target else []) + [fmt]
        return self.cli(*words)

    async def run(self):
        proc, screen = await self.attach()
        await self.wait_for("first prompt", lambda: "0:bash*" in self.status(screen))
        failures = []

        async def press(keys, settle=0.25):
            await self.type_keys(proc, keys, settle)

        def check(label, actual, expected):
            if actual != expected:
                failures.append("%s: expected %r, got %r" % (label, expected, actual))

        windows = lambda: self.cli("list-windows", "-F", "#{window_index}#{window_flags}").replace("\n", " ")
        panes = lambda: self.cli("list-panes", "-F", "#{pane_index}:#{pane_width}x#{pane_height}").replace("\n", " ")

        await press(PREFIX + b"c")
        await press(PREFIX + b"c")
        check("c twice", windows(), "0 1- 2*")
        await press(PREFIX + b"p")
        check("p previous", windows(), "0 1* 2-")
        await press(PREFIX + b"n")
        check("n next", windows(), "0 1- 2*")
        await press(PREFIX + b"l")
        check("l last", windows(), "0 1* 2-")
        await press(PREFIX + b"0")
        check("0 select", windows(), "0* 1- 2")
        await press(PREFIX + alt("n"))
        check("M-n next -a (no alerts, no change)", windows(), "0* 1- 2")
        await press(PREFIX + b"&")
        await press(b"y", 0.4)
        check("& confirm kill-window", len(windows().split()), 2)

        await press(PREFIX + b"'")
        await press(b"2\r", 0.4)
        check("' prompt select-window", self.q("#{window_index}"), "2")
        await press(PREFIX + b",")
        await press(b"\x15renamed\r", 0.4)
        check(", rename-window", self.q("#{window_name}"), "renamed")
        await press(PREFIX + b"$")
        await press(b"\x15sess\r", 0.4)
        check("$ rename-session", self.q("#{session_name}"), "sess")
        await press(PREFIX + b".")
        await press(b"9\r", 0.4)
        check(". move-window", self.q("#{window_index}"), "9")

        await press(PREFIX + b"%")
        check("% split-window -h", panes(), "0:40x23 1:39x23")
        await press(PREFIX + b"\"")
        check('" split-window', len(panes().split()), 3)
        await press(PREFIX + b";")
        check("; last-pane", self.q("#{pane_index}"), "1")
        await press(PREFIX + b"o")
        check("o next pane", self.q("#{pane_index}"), "2")
        await press(PREFIX + UP)
        check("Up select-pane -U", self.q("#{pane_index}"), "1")
        await press(PREFIX + LEFT)
        check("Left select-pane -L", self.q("#{pane_index}"), "0")
        await press(PREFIX + RIGHT)
        check("Right select-pane -R", self.q("#{pane_index}"), "1")
        await press(PREFIX + DOWN)
        check("Down select-pane -D", self.q("#{pane_index}"), "2")
        before = self.q("#{pane_id}")
        await press(PREFIX + b"{")
        order = self.cli("list-panes", "-F", "#{pane_id}").split()
        check("{ swap-pane -U moves the pane up", order.index(before), 1)
        await press(PREFIX + b"}")
        order = self.cli("list-panes", "-F", "#{pane_id}").split()
        check("} swap-pane -D moves it back", order.index(before), 2)
        first = self.cli("list-panes", "-F", "#{pane_id}").split()
        await press(PREFIX + b"\x0f")
        rotated = self.cli("list-panes", "-F", "#{pane_id}").split()
        check("C-o rotate-window", rotated, first[1:] + first[:1])
        await press(PREFIX + alt("o"))
        check("M-o rotate-window -D", self.cli("list-panes", "-F", "#{pane_id}").split(), first)

        for key, name in (("1", "even-horizontal"), ("2", "even-vertical"), ("3", "main-horizontal"), ("4", "main-vertical"), ("5", "tiled")):
            await press(PREFIX + alt(key))
            sizes = self.cli("list-panes", "-F", "#{pane_width}x#{pane_height}").split()
            if name == "even-horizontal":
                check("M-1 " + name, sizes, ["26x23", "26x23", "26x23"])
            elif name == "even-vertical":
                check("M-2 " + name, sizes, ["80x7", "80x7", "80x7"])
            elif name == "tiled":
                check("M-5 " + name, sizes, ["39x11", "40x11", "80x11"])
        await press(PREFIX + b" ")
        check("Space next-layout", len(self.cli("list-panes").splitlines()), 3)

        await press(PREFIX + b"x")
        await press(b"n", 0.3)
        check("x then n keeps the pane", len(self.cli("list-panes").splitlines()), 3)
        count_before = len(self.cli("list-panes").splitlines())
        await press(PREFIX + b"x")
        await press(b"y", 0.5)
        check("x then y kill-pane", len(self.cli("list-panes").splitlines()), count_before - 1)

        await press(PREFIX + b"\"")
        check("window back to 3 panes", len(self.cli("list-panes").splitlines()), 3)
        await press(PREFIX + b"z")
        check("z zoom on", self.q("#{window_zoomed_flag}"), "1")
        check("zoomed status flag", "Z" in self.status(screen), True)
        await press(PREFIX + b"z")
        check("z zoom off", self.q("#{window_zoomed_flag}"), "0")

        await press(PREFIX + alt("5"))
        self.cli("select-pane", "-t", ":.0")
        check("tiled pane 0 height", self.q("#{pane_height}"), "11")
        await press(PREFIX + b"\x1b[1;5B" + b"\x1b[1;5B")
        check("C-Down twice (repeat) grows by 2", self.q("#{pane_height}"), "13")
        await asyncio.sleep(0.8)
        await press(b"\x1b[1;5B")
        check("repeat window expired: C-Down no longer resizes", self.q("#{pane_height}"), "13")
        check("pane 0 width before M-Right (tiled gives the odd cell to the last column)", self.q("#{pane_width}"), "39")
        await press(PREFIX + b"\x1b[1;3C")
        check("M-Right resize-pane -R 5", self.q("#{pane_width}"), "44")

        await press(PREFIX + b"!")
        check("! break-pane", len(windows().split()), 3)
        await press(PREFIX + b"m")
        check("m mark pane", self.q("#{pane_marked}"), "1")
        await press(PREFIX + b"M")
        check("M clear mark", self.q("#{pane_marked}"), "0")

        await press(PREFIX + b"i")
        check("i display-message", ", current pane" in self.status(screen), True)
        await press(PREFIX + b"~")
        check("~ show-messages opens a view", self.q("#{pane_mode}"), "view-mode")
        await press(b"q", 0.3)
        check("q leaves the view", self.q("#{pane_in_mode}"), "0")
        await press(PREFIX + b"?")
        check("? list-keys -N opens a view", self.q("#{pane_mode}"), "view-mode")
        check("? lists the key notes", "C-b Space" in screen.text(), True)
        await press(b"q", 0.3)

        await press(PREFIX + b"t")
        check("t clock-mode", self.q("#{pane_mode}"), "clock-mode")
        await press(b"q", 0.3)
        check("any key leaves clock-mode", self.q("#{pane_in_mode}"), "0")

        await press(PREFIX + b"q")
        await press(b"0", 0.4)
        check("q display-panes digit selects", self.q("#{pane_index}"), "0")

        await press(PREFIX + b"w")
        check("w choose-tree", self.q("#{pane_mode}"), "tree-mode")
        await press(b"q", 0.3)
        await press(PREFIX + b"s")
        check("s choose-tree -s", self.q("#{pane_mode}"), "tree-mode")
        await press(b"q", 0.3)
        await press(PREFIX + b"D")
        check("D choose-client", self.q("#{pane_mode}"), "client-mode")
        await press(b"q", 0.3)
        await press(PREFIX + b"f")
        await press(b"renamed\r", 0.5)
        check("f find-window selects the match", self.q("#{window_name}"), "renamed")
        await press(PREFIX + b":")
        await press(b"new-session -d -s second\r", 0.6)
        check(": command prompt runs a command", "second" in self.cli("list-sessions"), True)
        await press(PREFIX + b")")
        check(") next session", self.q("#{session_name}"), "second")
        await press(PREFIX + b"(")
        check("( previous session", self.q("#{session_name}"), "sess")
        await press(PREFIX + b"L")
        check("L last session", self.q("#{session_name}"), "second")
        return failures

    async def verify(self, result):
        if result:
            raise TestAssertionFailed("%d shortcut check(s) failed:\n  %s" % (len(result), "\n  ".join(result)))


class MplexCopyModeTest(MplexTestBase):
    """Scrollback, search, selection, copy into a buffer, paste back into the shell."""

    name = "mplex-copy-mode"
    timeout = 60.0

    async def run(self):
        proc, screen = await self.attach()
        await self.wait_for("first prompt", lambda: "0:bash*" in self.status(screen))
        os.write(proc._master_fd, b"seq 1 200; echo needle-in-haystack; seq 300 340\r")
        await self.wait_for("output", lambda: "340" in screen.text())
        await self.type_keys(proc, PREFIX + b"[", 0.3)
        mode = self.cli("display-message", "-p", "#{pane_mode}")
        await self.type_keys(proc, b"\x1b[5~", 0.3)
        pos_after_pgup = self.cli("display-message", "-p", "#{scroll_position}")
        await self.type_keys(proc, b"\x1b<", 0.3)
        top = self.cli("display-message", "-p", "#{scroll_position}")
        history = self.cli("display-message", "-p", "#{history_size}")
        await self.type_keys(proc, b"\x1b>", 0.3)
        bottom = self.cli("display-message", "-p", "#{scroll_position}")
        await self.type_keys(proc, b"\x12", 0.3)
        os.write(proc._master_fd, b"needle\r")
        await asyncio.sleep(0.5)
        found = "needle-in-haystack" in screen.text()
        await self.type_keys(proc, b"\x01", 0.2)
        await self.type_keys(proc, b"\x00", 0.2)
        await self.type_keys(proc, b"\x05", 0.2)
        await self.type_keys(proc, b"\x1bw", 0.4)
        buffer_text = self.cli("show-buffer")
        mode_after = self.cli("display-message", "-p", "#{pane_in_mode}")
        os.write(proc._master_fd, b"echo pasted:")
        await asyncio.sleep(0.2)
        await self.type_keys(proc, PREFIX + b"]", 0.4)
        os.write(proc._master_fd, b"\r")
        await asyncio.sleep(0.5)
        captured = self.cli("capture-pane", "-p")
        return {"mode": mode, "pgup": pos_after_pgup, "top": top, "history": history, "bottom": bottom, "found": found,
                "buffer": buffer_text, "mode_after": mode_after, "captured": captured}

    async def verify(self, result):
        self.expect("prefix [ enters copy-mode", result["mode"], "copy-mode")
        if int(result["pgup"]) <= 0:
            raise TestAssertionFailed("PageUp did not scroll back: %r" % result["pgup"])
        self.expect("M-< goes to the top of history", result["top"], result["history"])
        self.expect("M-> returns to the bottom", result["bottom"], "0")
        if not result["found"]:
            raise TestAssertionFailed("C-r incremental search did not bring the match on screen")
        if "needle-in-haystack" not in result["buffer"]:
            raise TestAssertionFailed("C-Space ... M-w did not copy the line: %r" % result["buffer"])
        self.expect("M-w leaves copy-mode", result["mode_after"], "0")
        if "pasted:" not in result["captured"] or result["captured"].count("needle-in-haystack") < 2:
            raise TestAssertionFailed("paste-buffer did not insert the copied text: %r" % result["captured"][-400:])


class MplexConfigTest(MplexTestBase):
    """~/.tai/mplex.toml: another prefix, new and removed bindings, options, vi copy keys, loud errors."""

    name = "mplex-config"
    config = """
[options]
prefix = "C-a"
prefix2 = "F12"
base-index = 1
status-left = "<#S> "
mode-keys = "vi"
status-position = "top"
no-such-option = 1

[keys.prefix]
"|" = "split-window -h"
"-" = "split-window -v"
"C-b" = false
c = { command = "new-window -n custom", note = "custom window" }
Q = "kill-server"
broken = "no-such-command"

[keys.root]
"M-Right" = "next-window"
"""
    timeout = 60.0

    async def run(self):
        proc, screen = await self.attach()
        await self.wait_for("the config problem notice", lambda: "problem(s) skipped" in screen.text())
        notice_seen = True
        await self.wait_for("first prompt", lambda: "1:bash*" in screen.text(), timeout=10.0)
        top_line = screen.text().split("\n")[0]
        await self.type_keys(proc, b"\x01c", 0.4)
        names = self.cli("list-windows", "-F", "#{window_index}:#{window_name}").replace("\n", " ")
        await self.type_keys(proc, b"\x01|", 0.4)
        panes_after_pipe = len(self.cli("list-panes").splitlines())
        await self.type_keys(proc, b"\x02c", 0.4)
        windows_after_old_prefix = len(self.cli("list-windows").splitlines())
        await self.type_keys(proc, b"\x1b[24~c", 0.4)
        windows_after_prefix2 = len(self.cli("list-windows").splitlines())
        await self.type_keys(proc, b"\x1b[1;3C", 0.4)
        current_after_root = self.cli("display-message", "-p", "#{window_index}")
        await self.type_keys(proc, b"\x01[", 0.3)
        mode_keys = self.cli("show-options", "-gv", "mode-keys")
        await self.type_keys(proc, b"k", 0.2)
        vi_up = self.cli("display-message", "-p", "#{pane_in_mode}")
        await self.type_keys(proc, b"q", 0.3)
        messages = self.cli("show-messages")
        keys_c = self.cli("list-keys", "-T", "prefix", "c")
        notes = self.cli("list-keys", "-N")
        return {"top": top_line, "notice": notice_seen, "names": names, "pipe": panes_after_pipe, "old": windows_after_old_prefix,
                "prefix2": windows_after_prefix2, "root": current_after_root, "mode_keys": mode_keys, "vi_up": vi_up,
                "messages": messages, "keys_c": keys_c, "notes": notes, "cli_prefix": self.cli("show-options", "-gv", "prefix")}

    async def verify(self, result):
        if "<0>" not in result["top"]:
            raise TestAssertionFailed("status-left / status-position top not applied: %r" % result["top"])
        if not result["notice"]:
            raise TestAssertionFailed("config problems were not announced on the status line")
        self.expect("prefix changed", result["cli_prefix"], "C-a")
        if "2:custom" not in result["names"]:
            raise TestAssertionFailed("custom c binding with base-index 1 did not create window 2:custom: %r" % result["names"])
        self.expect("| split binding", result["pipe"], 2)
        self.expect("old prefix C-b is unbound (no new window)", result["old"], 2)
        self.expect("prefix2 F12 works", result["prefix2"], 3)
        self.expect("root-table M-Right moves to the next window (wraps to 1)", result["root"], "1")
        self.expect("mode-keys vi", result["mode_keys"], "vi")
        self.expect("vi k stays in copy mode", result["vi_up"], "1")
        for needle in ("no-such-option", "keys.prefix.broken"):
            if needle not in result["messages"]:
                raise TestAssertionFailed("show-messages does not name the bad config entry %r: %r" % (needle, result["messages"]))
        if "new-window -n custom" not in result["keys_c"]:
            raise TestAssertionFailed("list-keys does not show the rebound c: %r" % result["keys_c"])
        if "custom window" not in result["notes"]:
            raise TestAssertionFailed("list-keys -N does not show the binding note")


class MplexMultiClientTest(MplexTestBase):
    """Two terminals on one session: both repaint, the smaller one sets the size, input from either lands."""

    name = "mplex-multi-client"
    timeout = 60.0

    async def run(self):
        proc, screen = await self.attach()
        await self.wait_for("first prompt", lambda: "0:bash*" in self.status(screen))
        self.rows, self.cols = 18, 60
        proc2, screen2 = await self.attach("attach")
        await self.wait_for("second client painted", lambda: "0:bash*" in self.status(screen2))
        await asyncio.sleep(0.4)
        size = self.cli("display-message", "-p", "#{window_width}x#{window_height}")
        attached = self.cli("display-message", "-p", "#{session_attached}")
        os.write(proc2._master_fd, b"echo from-second-$((1+1))\r")
        await self.wait_for("first client sees the second one's command", lambda: "from-second-2" in screen.text())
        await self.type_keys(proc2, PREFIX + b"c", 0.4)
        first_follows = await self.wait_for("first client follows the new window", lambda: "1:bash*" in self.status(screen))
        await self.type_keys(proc2, PREFIX + b"d")
        await proc2.wait(timeout=10.0)
        await asyncio.sleep(0.5)
        size_after = self.cli("display-message", "-p", "#{window_width}x#{window_height}")
        return {"size": size, "attached": attached, "follows": first_follows, "size_after": size_after}

    async def verify(self, result):
        self.expect("window size is the smallest client", result["size"], "60x17")
        self.expect("session_attached", result["attached"], "2")
        self.expect("size grows back after the small client detaches", result["size_after"], "80x23")


class MplexCliTest(MplexTestBase):
    """The non-attached surface: detached sessions, targets, send-keys, capture-pane, errors with exit codes."""

    name = "mplex-cli"
    timeout = 60.0

    async def run(self):
        out = {}
        out["new"] = self.cli("new-session", "-d", "-s", "work", "-x", "100", "-y", "30", "-P", "-F", "#{session_name}:#{window_width}x#{window_height}")
        self.cli("split-window", "-h", "-t", "work")
        self.cli("new-window", "-t", "work", "-n", "logs")
        self.cli("send-keys", "-t", "work:logs", "echo cli-$((6*7))", "Enter")
        await asyncio.sleep(0.8)
        out["capture"] = self.cli("capture-pane", "-p", "-t", "work:logs")
        out["windows"] = self.cli("list-windows", "-t", "work", "-F", "#{window_index}:#{window_name}:#{window_panes}").replace("\n", " ")
        out["panes"] = self.cli("list-panes", "-t", "work:0", "-F", "#{pane_index}:#{pane_width}").replace("\n", " ")
        self.cli("set-option", "-g", "status-left", "[X] ")
        out["option"] = self.cli("show-options", "-gv", "status-left")
        self.cli("set-buffer", "-b", "named", "buffer text")
        out["buffer"] = self.cli("show-buffer", "-b", "named")
        self.cli("bind-key", "-T", "prefix", "F9", "display-message", "hi", "\\;", "list-windows")
        out["bound"] = self.cli("list-keys", "-T", "prefix", "F9")
        self.cli("unbind-key", "-T", "prefix", "F9")
        out["unbound"] = self.cli("list-keys", "-T", "prefix", "F9")
        out["bad_cmd"] = self.cli("no-such-command", check=False)
        out["unsupported"] = self.cli("display-menu", check=False)
        out["missing"] = self.cli("has-session", "-t", "nope", check=False)
        done = subprocess.run([sys.executable, str(self.tai_path), "--mplex", "has-session", "-t", "nope"], env=self.cli_env(),
                              stdin=subprocess.DEVNULL, capture_output=True, text=True)
        out["rc_missing"] = done.returncode
        self.cli("new-window", "-d", "-t", "work", "-n", "first", ";", "new-window", "-d", "-t", "work", "-n", "second")
        out["sequence"] = self.cli("list-windows", "-t", "work", "-F", "#{window_name}").replace("\n", " ")
        self.cli("kill-window", "-t", "work:first", ";", "kill-window", "-t", "work:second")
        self.cli("kill-window", "-t", "work:logs")
        out["after_kill"] = self.cli("list-windows", "-t", "work", "-F", "#{window_index}")
        self.cli("kill-session", "-t", "work")
        out["empty"] = self.cli("list-sessions", check=False)
        return out

    async def verify(self, result):
        self.expect("new-session -d -x 100 -y 30 gives a 100x30 window, as tmux does", result["new"], "work:100x30")
        if "cli-42" not in result["capture"]:
            raise TestAssertionFailed("send-keys / capture-pane round trip failed: %r" % result["capture"][-300:])
        self.expect("windows", result["windows"], "0:bash:2 1:logs:1")
        self.expect("split widths", result["panes"], "0:50 1:49")
        self.expect("set-option / show-options", result["option"], "[X]")
        self.expect("buffers", result["buffer"], "buffer text")
        if "display-message hi" not in result["bound"] or "list-windows" not in result["bound"]:
            raise TestAssertionFailed("bind-key with ; did not keep both commands: %r" % result["bound"])
        self.expect("unbind", result["unbound"], "")
        for key, needle in (("bad_cmd", "unknown command"), ("unsupported", "not supported"), ("missing", "can't find session")):
            if needle not in result[key]:
                raise TestAssertionFailed("%s did not report %r: %r" % (key, needle, result[key]))
        self.expect("exit code of a failing command", result["rc_missing"], 1)
        self.expect("a lone ; runs two commands in order", result["sequence"], "bash logs first second")
        self.expect("kill-window", result["after_kill"], "0")


ALL_CASES = [MplexLifecycleTest(), MplexShortcutsTest(), MplexCopyModeTest(), MplexConfigTest(), MplexMultiClientTest(), MplexCliTest()]

if __name__ == "__main__":
    sys.exit(main_cli(ALL_CASES, default_log_dir=str(Path(__file__).resolve().parent / "logs")))