tai/taicpp/tests/mplex_server_test.cpp

// Differential tests for MplexServer: the same raw-socket scenario (control protocol, frames, key
// handling, UI flows) is run against the C++ server and against tai.py's Python server in two
// throwaway TAI_HOMEs, and the normalized transcripts must be byte for byte equal. Real sockets,
// real bash panes, no mocks. Modeled on tests/plexer.py and tests/mplex.py.
//
//   g++ -std=c++23 -Wall -Wextra -Isrc/mplex tests/mplex_server_test.cpp src/mplex/*.cpp -o build/mplex_server_test
//   build/mplex_server_test [--tai ../tai.py] [--slow]
#include <fcntl.h>
#include <poll.h>
#include <signal.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/un.h>
#include <sys/wait.h>
#include <unistd.h>

#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <filesystem>
#include <fstream>
#include <functional>
#include <iostream>
#include <optional>
#include <sstream>
#include <string>
#include <vector>

#include "mplex_json.h"
#include "mplex_server.h"

using namespace detz::mplex;

namespace {

int g_checks = 0;
int g_failed = 0;

void check(bool ok, const std::string& what) {
  ++g_checks;
  if (!ok) {
    ++g_failed;
    std::cerr << "FAIL: " << what << "\n";
  }
}

void sleep_ms(int ms) { ::usleep(static_cast<useconds_t>(ms) * 1000); }

std::string replace_all(std::string text, const std::string& from, const std::string& to) {
  if (from.empty()) return text;
  size_t pos = 0;
  while ((pos = text.find(from, pos)) != std::string::npos) {
    text.replace(pos, from.size(), to);
    pos += to.size();
  }
  return text;
}

struct Frame {
  char kind;
  std::string payload;
};

// One unix socket connection.
class Conn {
 public:
  Conn() = default;
  Conn(const Conn&) = delete;
  Conn(Conn&& o) noexcept : fd(o.fd), buf(std::move(o.buf)) { o.fd = -1; }
  ~Conn() { close(); }
  int fd = -1;
  std::string buf;

  bool open(const std::string& path) {
    fd = ::socket(AF_UNIX, SOCK_STREAM, 0);
    sockaddr_un addr{};
    addr.sun_family = AF_UNIX;
    std::strncpy(addr.sun_path, path.c_str(), sizeof addr.sun_path - 1);
    if (::connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof addr) != 0) {
      ::close(fd);
      fd = -1;
      return false;
    }
    return true;
  }
  void close() {
    if (fd >= 0) ::close(fd);
    fd = -1;
  }
  void send_all(const std::string& data) {
    size_t off = 0;
    while (off < data.size()) {
      ssize_t n = ::send(fd, data.data() + off, data.size() - off, MSG_NOSIGNAL);
      if (n <= 0) return;
      off += static_cast<size_t>(n);
    }
  }
  // Read what arrives within timeout_ms into buf; returns false when the peer closed.
  bool fill(int timeout_ms) {
    pollfd p{fd, POLLIN, 0};
    if (::poll(&p, 1, timeout_ms) <= 0) return true;
    char tmp[65536];
    ssize_t n = ::recv(fd, tmp, sizeof tmp, 0);
    if (n <= 0) return false;
    buf.append(tmp, static_cast<size_t>(n));
    return true;
  }
  // JSON line reply (control protocol): until newline or close.
  std::string read_line(int timeout_ms = 5000) {
    int waited = 0;
    while (buf.find('\n') == std::string::npos && waited < timeout_ms) {
      if (!fill(50)) break;
      waited += 50;
    }
    size_t nl = buf.find('\n');
    std::string line = nl == std::string::npos ? buf : buf.substr(0, nl);
    buf.clear();
    return line;
  }
  // Frames: kind byte, u32 big-endian length, payload.
  std::optional<Frame> frame(int timeout_ms) {
    int waited = 0;
    for (;;) {
      if (buf.size() >= 5) {
        uint32_t len = (static_cast<unsigned char>(buf[1]) << 24) | (static_cast<unsigned char>(buf[2]) << 16) |
                       (static_cast<unsigned char>(buf[3]) << 8) | static_cast<unsigned char>(buf[4]);
        if (buf.size() >= 5 + len) {
          Frame f{buf[0], buf.substr(5, len)};
          buf.erase(0, 5 + len);
          return f;
        }
      }
      if (waited >= timeout_ms) return std::nullopt;
      if (!fill(50)) {
        waited = timeout_ms;
        continue;
      }
      waited += 50;
    }
  }
  bool closed_by_peer(int timeout_ms) {
    int waited = 0;
    while (waited < timeout_ms) {
      pollfd p{fd, POLLIN, 0};
      if (::poll(&p, 1, 50) > 0) {
        char tmp[4096];
        ssize_t n = ::recv(fd, tmp, sizeof tmp, 0);
        if (n == 0) return true;
        if (n < 0) return true;
      }
      waited += 50;
    }
    return false;
  }
};

std::string make_frame(char kind, const std::string& payload) {
  std::string out(1, kind);
  uint32_t n = static_cast<uint32_t>(payload.size());
  out += static_cast<char>(n >> 24);
  out += static_cast<char>(n >> 16);
  out += static_cast<char>(n >> 8);
  out += static_cast<char>(n);
  return out + payload;
}

// Everything one scenario needs about one server.
struct Env {
  std::string label;
  std::string home;
  std::string sock;
  std::string transcript;
  void note(const std::string& key, std::string value) {
    value = replace_all(value, home, "<HOME>");
    // stamp: 8 digits T 6 digits
    for (size_t i = 0; i + 15 <= value.size(); ++i) {
      bool ok = value[i + 8] == 'T';
      for (size_t k = 0; ok && k < 8; ++k) ok = std::isdigit(static_cast<unsigned char>(value[i + k])) != 0;
      for (size_t k = 9; ok && k < 15; ++k) ok = std::isdigit(static_cast<unsigned char>(value[i + k])) != 0;
      if (ok) value.replace(i, 15, "<STAMP>");
    }
    // "pid": N
    for (const char* tag : {"\"pid\": ", "pid\":"}) {
      size_t pos = 0;
      std::string t = tag;
      while ((pos = value.find(t, pos)) != std::string::npos) {
        size_t s = pos + t.size();
        size_t e = s;
        while (e < value.size() && std::isdigit(static_cast<unsigned char>(value[e]))) ++e;
        if (e > s) value.replace(s, e - s, "<PID>");
        pos = s + 1;
      }
    }
    std::string clean;
    for (char c : value) {
      if (c == '\x1b') clean += "<ESC>";
      else if (c == '\r') clean += "<CR>";
      else clean += c;
    }
    transcript += key + " => " + clean + "\n";
  }
  std::string control(const std::string& line) {
    Conn c;
    if (!c.open(sock)) return "<no connect>";
    c.send_all(line);
    return c.read_line();
  }
  std::string ctl(const std::string& key, const std::string& line) {
    std::string reply = control(line + "\n");
    note(key, reply);
    return reply;
  }
};

std::string hello_json(const std::vector<std::vector<std::string>>& commands, const std::string& pane = "") {
  Json info = Json::object();
  info.set("rows", Json::integer(24));
  info.set("cols", Json::integer(80));
  info.set("term", Json::string("xterm-256color"));
  info.set("cwd", Json::string("/tmp"));
  info.set("pid", Json::integer(4242));
  info.set("tty", Json::string("/dev/pts/99"));
  Json cmds = Json::array();
  for (const auto& words : commands) {
    Json w = Json::array();
    for (const auto& word : words) w.push(Json::string(word));
    cmds.push(w);
  }
  info.set("commands", cmds);
  Json env = Json::object();
  env.set("SHELL", Json::string("/bin/bash"));
  env.set("TERM", Json::string("xterm-256color"));
  env.set("PS1", Json::string("$ "));
  env.set("PATH", Json::string(std::getenv("PATH") ? std::getenv("PATH") : "/usr/bin:/bin"));
  env.set("HOME", Json::string("/tmp"));
  info.set("env", env);
  info.set("pane", Json::string(pane));
  return info.dumps();
}

// Run one hello conversation; returns "<kind>:<payload>" lines for the frames until close/X/D/E.
std::string run_command(Env& e, const std::vector<std::vector<std::string>>& commands, int wait_ms = 3000) {
  Conn c;
  if (!c.open(e.sock)) return "<no connect>";
  c.send_all(std::string("\x01") + make_frame('H', hello_json(commands)));
  std::string out;
  int waited = 0;
  while (waited < wait_ms) {
    auto f = c.frame(100);
    waited += 100;
    if (!f) continue;
    out += std::string(1, f->kind) + ":" + f->payload + "|";
    if (f->kind == 'X' || f->kind == 'D' || f->kind == 'E') break;
  }
  return out;
}

// An attached raw client. pump() drains frames; the screen text of the last O frames is kept.
struct Attached {
  Conn conn;
  std::string last_kind;
  std::string seen;  // concatenated O payloads
  bool attach(Env& e, const std::vector<std::vector<std::string>>& commands) {
    if (!conn.open(e.sock)) return false;
    conn.send_all(std::string("\x01") + make_frame('H', hello_json(commands)));
    for (int i = 0; i < 60; ++i) {
      auto f = conn.frame(100);
      if (!f) continue;
      last_kind += f->kind;
      if (f->kind == 'A') return true;
      if (f->kind == 'E' || f->kind == 'X' || f->kind == 'D') return false;
    }
    return false;
  }
  void pump(int ms) {
    int waited = 0;
    while (waited < ms) {
      auto f = conn.frame(50);
      waited += 50;
      if (f) {
        last_kind += f->kind;
        if (f->kind == 'O') seen += f->payload;
      }
    }
  }
  void keys(const std::string& data) { conn.send_all(make_frame('I', data)); }
  void resize(int rows, int cols) {
    conn.send_all(make_frame('R', "{\"rows\": " + std::to_string(rows) + ", \"cols\": " + std::to_string(cols) + "}"));
  }
};

// ---- the scenario ---------------------------------------------------------------------------------

void scenario(Env& e, bool slow) {
  // 1. frames: a detached base session (bash without rc files so both servers see the same shell).
  e.note("new-session -d", run_command(e, {{"new-session", "-d", "-s", "base", "bash", "--norc", "--noprofile"}}));
  sleep_ms(700);
  e.note("has-session", run_command(e, {{"has-session", "-t", "base"}}));
  e.note("has-session missing", run_command(e, {{"has-session", "-t", "nosuch"}}));
  e.note("unknown command", run_command(e, {{"zzz"}}));
  e.note("ambiguous command", run_command(e, {{"list-"}}));
  e.note("unsupported command", run_command(e, {{"display-menu"}}));
  e.note("bad flag", run_command(e, {{"list-sessions", "-Q"}}));
  e.note("list-sessions -F", run_command(e, {{"list-sessions", "-F", "#{session_name}:#{session_windows}:#{session_attached}"}}));
  e.note("list-windows -F", run_command(e, {{"list-windows", "-F", "#{window_index}:#{window_name}:#{window_flags}:#{window_panes}"}}));
  e.note("list-panes -F", run_command(e, {{"list-panes", "-F", "#{pane_index}:#{pane_id}:#{pane_width}x#{pane_height}:#{pane_active}"}}));
  e.note("two commands", run_command(e, {{"list-sessions", "-F", "a#{session_name}"}, {"list-sessions", "-F", "b#{session_name}"}}));
  e.note("empty commands", run_command(e, {}));
  {
    Conn c;
    c.open(e.sock);
    c.send_all(std::string("\x01") + make_frame('H', "{not json"));
    auto f = c.frame(3000);
    e.note("bad hello json", f ? std::string(1, f->kind) + ":" + f->payload : "<none>");
  }
  {
    Conn c;
    c.open(e.sock);
    c.send_all(std::string("\x01") + make_frame('H', "[1]"));
    auto f = c.frame(3000);
    e.note("hello not object", f ? std::string(1, f->kind) + ":" + f->payload : "<none>");
  }
  {
    Conn c;
    c.open(e.sock);
    c.send_all(std::string("\x01") + make_frame('H', "{\"rows\": \"x\"}"));
    auto f = c.frame(3000);
    e.note("hello bad rows", f ? std::string(1, f->kind) + ":" + f->payload : "<none>");
  }
  {
    Conn c;  // a frame length over MPLEX_FRAME_MAX drops the client
    c.open(e.sock);
    std::string bad = "\x01H";
    bad += std::string("\x7f\xff\xff\xff", 4);
    c.send_all(bad);
    e.note("oversized frame closes", c.closed_by_peer(2000) ? "closed" : "open");
  }

  // 2. control protocol
  e.ctl("status", "{\"cmd\": \"status\"}");
  e.ctl("send", "{\"cmd\": \"send\", \"text\": \"echo marker_one\", \"enter\": true}");
  sleep_ms(500);
  {
    std::string reply = e.control("{\"cmd\": \"read\", \"lines\": 50}\n");
    Json parsed = Json::loads(reply);
    const Json* data = parsed.get("data");
    bool has = data && data->is_string() && data->as_string().find("marker_one") != std::string::npos;
    e.note("read has marker", has ? "yes" : "no: " + reply);
  }
  e.ctl("send nonascii", "{\"cmd\": \"send\", \"text\": \"echo \\u00e9\\u4e2d\", \"enter\": true}");
  sleep_ms(300);
  e.ctl("send pane 0", "{\"cmd\": \"send\", \"text\": \"\", \"pane\": 0}");
  e.ctl("send pane %0", "{\"cmd\": \"send\", \"text\": \"\", \"pane\": \"%0\"}");
  e.ctl("send pane 999", "{\"cmd\": \"send\", \"pane\": \"999\"}");
  e.ctl("send pane x", "{\"cmd\": \"send\", \"pane\": \"x\"}");
  e.ctl("send pane list", "{\"cmd\": \"send\", \"pane\": [1]}");
  e.ctl("send pane empty", "{\"cmd\": \"send\", \"text\": \"\", \"pane\": \"\"}");
  e.ctl("read lines abc", "{\"cmd\": \"read\", \"lines\": \"abc\"}");
  e.ctl("read lines list", "{\"cmd\": \"read\", \"lines\": [1]}");
  for (const char* variant : {"null", "NaN", "2.9", "true", "\"3\"", "-4", "0"}) {
    std::string reply = e.control(std::string("{\"cmd\": \"read\", \"lines\": ") + variant + "}\n");
    // pane output carries per-run uuids: only the envelope is compared
    e.note(std::string("read lines ") + variant, reply.substr(0, 20) + (reply.find("\"error\"") != std::string::npos ? reply : ""));
  }
  e.ctl("not an object", "[1, 2, 3]");
  e.ctl("string request", "\"hello\"");
  e.ctl("number request", "7");
  e.ctl("null request", "null");
  e.ctl("empty object", "{}");
  e.ctl("unknown cmd", "{\"cmd\": \"nope\"}");
  e.ctl("cmd null", "{\"cmd\": null}");
  e.ctl("cmd number", "{\"cmd\": 5}");
  e.ctl("cmd list", "{\"cmd\": [1]}");
  e.ctl("not json", "not json at all");
  e.ctl("json trailing", "{\"cmd\": \"status\"} x");
  e.ctl("empty line", "");
  e.ctl("result no id", "{\"cmd\": \"result\"}");
  e.ctl("result bad id", "{\"cmd\": \"result\", \"id\": \"zz\"}");
  e.ctl("result id list", "{\"cmd\": \"result\", \"id\": [1]}");
  e.ctl("result unknown", "{\"cmd\": \"result\", \"id\": 99}");
  e.ctl("result unknown str", "{\"cmd\": \"result\", \"id\": \"99\"}");
  e.ctl("cancel no id", "{\"cmd\": \"cancel\"}");
  e.ctl("cancel unknown", "{\"cmd\": \"cancel\", \"id\": 99}");
  e.ctl("show no screen", "{\"cmd\": \"show\"}");
  e.ctl("show screen list", "{\"cmd\": \"show\", \"screen\": []}");
  e.ctl("show bad kind", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"x\"}}");
  e.ctl("show kind number", "{\"cmd\": \"show\", \"screen\": {\"kind\": 1}}");
  e.ctl("show bad title", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"title\": 1}}");
  e.ctl("show long title", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"title\": \"" + std::string(201, 'a') + "\"}}");
  e.ctl("show long text", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"" + std::string(2001, 'a') + "\"}}");
  e.ctl("show choice no options", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\"}}");
  e.ctl("show choice empty options", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"options\": []}}");
  {
    std::string many = "[";
    for (int i = 0; i < 13; ++i) many += std::string(i ? "," : "") + "\"o" + std::to_string(i) + "\"";
    many += "]";
    e.ctl("show choice 13 options", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"options\": " + many + "}}");
  }
  e.ctl("show choice blank label", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"options\": [\"  \"]}}");
  e.ctl("show choice number label", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"options\": [1]}}");
  e.ctl("show choice bad default", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"options\": [\"a\"], \"default\": \"x\"}}");
  e.ctl("show choice default range", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"options\": [\"a\"], \"default\": 5}}");
  e.ctl("show choice default null", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"options\": [\"a\"], \"default\": null}}");
  e.ctl("show bad initial", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"text\", \"initial\": 5}}");
  e.ctl("show long initial", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"text\", \"initial\": \"" + std::string(2001, 'a') + "\"}}");
  e.ctl("show bad timeout", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\"}, \"timeout\": \"x\"}");
  e.ctl("show timeout list", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\"}, \"timeout\": []}");
  e.ctl("show no client", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"hi\"}}");
  e.ctl("show pane 999", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\"}, \"pane\": \"999\"}");
  {
    // a request larger than one recv and larger than PLEXER_RECV_MAX without a newline
    Conn c;
    c.open(e.sock);
    c.send_all("{\"cmd\": \"send\", \"text\": \"" + std::string(1100000, 'a'));
    e.note("recv max without newline", c.read_line());
  }
  {
    // a large well formed request (> 65536 bytes) lands as one request
    std::string big = "{\"cmd\": \"send\", \"text\": \"echo BIG_START_" + std::string(90000, 'x') + "_BIG_END\", \"enter\": true}";
    e.ctl("send large", big);
    sleep_ms(800);
    std::string reply = e.control("{\"cmd\": \"read\", \"lines\": 2000}\n");
    Json parsed = Json::loads(reply);
    const Json* data = parsed.get("data");
    bool ok = data && data->as_string().find("BIG_START_") != std::string::npos && data->as_string().find("_BIG_END") != std::string::npos;
    e.note("large payload intact", ok ? "yes" : "no");
  }
  {
    Conn junk;  // binary garbage, then the server must still answer
    junk.open(e.sock);
    junk.send_all(std::string("\xff\xff\xff not json at all"));
    junk.close();
    e.ctl("status after junk", "{\"cmd\": \"status\"}");
  }

  // 3. attached client: UI flows driven by real keystrokes
  Attached a;
  bool attached = a.attach(e, {{"attach-session", "-t", "base"}});
  e.note("attach", attached ? "A" : "failed " + a.last_kind);
  if (attached) {
    a.pump(300);
    e.note("list-clients", run_command(e, {{"list-clients", "-F", "#{client_width}x#{client_height} #{client_session} #{client_key_table}"}}));
    e.note("session_attached", run_command(e, {{"list-sessions", "-F", "#{session_name}:#{session_attached}"}}));
    // choice: Down, Enter -> index 1
    std::string reply = e.ctl("show choice", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"title\": \"T\", \"text\": \"Pick one\", \"options\": [\"alpha\", \"beta\"]}, \"timeout\": 10}");
    e.ctl("poll choice", "{\"cmd\": \"result\", \"id\": 0}");
    e.ctl("show while busy", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"two\"}}");
    a.pump(200);
    a.keys("\x1b[B");
    a.pump(200);
    a.keys("\r");
    a.pump(200);
    e.ctl("result choice", "{\"cmd\": \"result\", \"id\": 0}");
    e.ctl("result collected once", "{\"cmd\": \"result\", \"id\": 0}");
    // text: type then Enter
    e.ctl("show text", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"text\", \"title\": \"T\", \"text\": \"Name?\", \"initial\": \"ab\"}, \"timeout\": 10}");
    a.pump(200);
    a.keys("c\x01X\r");
    a.pump(200);
    e.ctl("result text", "{\"cmd\": \"result\", \"id\": 1}");
    // message: any key
    e.ctl("show message", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"hi\"}, \"timeout\": 10}");
    a.pump(200);
    a.keys("x");
    a.pump(200);
    e.ctl("result message", "{\"cmd\": \"result\", \"id\": 2}");
    // cancel by Escape (escape-time)
    e.ctl("show cancel", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"hi\"}, \"timeout\": 10}");
    a.pump(200);
    a.keys("\x1b");
    a.pump(300);
    e.ctl("result cancelled", "{\"cmd\": \"result\", \"id\": 3}");
    // cancel via control
    e.ctl("show stuck", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"stuck\"}}");
    e.ctl("cancel stuck", "{\"cmd\": \"cancel\", \"id\": 4}");
    e.ctl("result after cancel", "{\"cmd\": \"result\", \"id\": 4}");
    // digit picks a choice, default selected
    e.ctl("show digit", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"choice\", \"options\": [\"a\", \"b\", \"c\"], \"default\": 2}, \"timeout\": 10}");
    a.pump(200);
    a.keys("2");
    a.pump(200);
    e.ctl("result digit", "{\"cmd\": \"result\", \"id\": 5}");
    // text prompt cancelled
    e.ctl("show text cancel", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"text\", \"text\": \"x\"}, \"timeout\": 10}");
    a.pump(200);
    a.keys("\x03");
    a.pump(200);
    e.ctl("result text cancel", "{\"cmd\": \"result\", \"id\": 6}");
    // timeout of a flow
    e.ctl("show short", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"t\"}, \"timeout\": 1}");
    sleep_ms(1500);
    a.pump(100);
    e.ctl("result timeout", "{\"cmd\": \"result\", \"id\": 7}");
    // timeout floor / ceiling are clamped, not rejected
    e.ctl("show timeout clamp", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"t\"}, \"timeout\": -5}");
    e.ctl("cancel clamp", "{\"cmd\": \"cancel\", \"id\": 8}");

    // 4. key handling through raw input frames
    a.keys("\x02" "c");  // C-b c: new window
    a.pump(500);
    e.note("after C-b c", run_command(e, {{"list-windows", "-F", "#{window_index}:#{window_flags}:#{window_panes}"}}));
    a.keys("\x02%");  // split horizontally
    a.pump(500);
    e.note("after C-b %", run_command(e, {{"list-panes", "-F", "#{pane_index}:#{pane_width}x#{pane_height}:#{pane_active}"}}));
    a.keys("\x02\"");  // split vertically
    a.pump(500);
    e.note("after C-b quote", run_command(e, {{"list-panes", "-F", "#{pane_index}:#{pane_width}x#{pane_height}:#{pane_active}"}}));
    a.keys("\x02" "o");  // next pane
    a.pump(200);
    e.note("after C-b o", run_command(e, {{"list-panes", "-F", "#{pane_index}:#{pane_active}"}}));
    // repeat table: C-b Up Up (repeatable) then a plain key leaves the table
    a.keys("\x02\x1b[A\x1b[A");
    a.pump(300);
    e.note("after repeat", run_command(e, {{"list-panes", "-F", "#{pane_index}:#{pane_active}"}}));
    // prompt: C-b : rename-window
    a.keys("\x02:");
    a.pump(200);
    a.keys("rename-window zed\r");
    a.pump(400);
    e.note("after prompt rename", run_command(e, {{"list-windows", "-F", "#{window_index}:#{window_name}"}}));
    // prefix then unbound key is dropped
    a.keys("\x02" "\x01");
    a.pump(200);
    // zoom, window switching
    a.keys("\x02z");
    a.pump(300);
    e.note("after zoom", run_command(e, {{"list-windows", "-F", "#{window_index}:#{window_flags}:#{window_zoomed_flag}"}}));
    a.keys("\x02p");
    a.pump(300);
    e.note("after C-b p", run_command(e, {{"list-windows", "-F", "#{window_index}:#{window_flags}"}}));
    a.keys("\x02" "l");
    a.pump(300);
    e.note("after C-b l", run_command(e, {{"list-windows", "-F", "#{window_index}:#{window_flags}"}}));
    // typing reaches the shell
    a.keys("echo typed_ok\r");
    a.pump(500);
    {
      std::string reply = e.control("{\"cmd\": \"read\", \"lines\": 100}\n");
      e.note("typed reaches pane", reply.find("typed_ok") != std::string::npos ? "yes" : "no");
    }
    // resize frame: bad ones ignored, good ones applied
    a.conn.send_all(make_frame('R', "not json"));
    a.conn.send_all(make_frame('R', "{\"rows\": 30}"));
    a.resize(30, 100);
    a.pump(300);
    e.note("after resize", run_command(e, {{"list-clients", "-F", "#{client_width}x#{client_height}"}}));
    e.note("window size", run_command(e, {{"list-windows", "-F", "#{window_width}x#{window_height}"}}));
    // detach
    a.keys("\x02" "d");
    bool got_d = false;
    for (int i = 0; i < 30 && !got_d; ++i) {
      auto f = a.conn.frame(100);
      if (f && f->kind == 'D') {
        got_d = true;
        e.note("detach frame", std::string("D:") + f->payload);
      }
    }
    if (!got_d) e.note("detach frame", "missing");
    e.note("closed after detach", a.conn.closed_by_peer(2000) ? "closed" : "open");
  }

  // 5. attached via new-session: the session count and a second client sharing a session
  {
    Attached b;
    bool ok = b.attach(e, {{"new-session", "-s", "second", "bash", "--norc", "--noprofile"}});
    e.note("attach new-session", ok ? "A" : "failed");
    b.pump(500);
    e.note("sessions now", run_command(e, {{"list-sessions", "-F", "#{session_name}:#{session_attached}"}}));
    e.note("attach to attached -d", run_command(e, {{"list-clients", "-F", "#{client_session}"}}));
    // no client for a pane whose session is not attached
    e.ctl("show other session pane", "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"hi\"}, \"pane\": \"0\"}");
    // killing the session a client is attached to ends the client with X [exited]
    e.note("kill-session", run_command(e, {{"kill-session", "-t", "second"}}));
    bool got_x = false;
    for (int i = 0; i < 30 && !got_x; ++i) {
      auto f = b.conn.frame(100);
      if (f && (f->kind == 'X' || f->kind == 'D')) {
        got_x = true;
        e.note("kill-session client frame", std::string(1, f->kind) + ":" + f->payload);
      }
    }
    if (!got_x) e.note("kill-session client frame", "missing");
  }

  // 6. flow cap: one attached client per session, 17 shows
  {
    std::vector<std::unique_ptr<Attached>> clients;
    std::vector<std::string> panes;
    for (int i = 0; i < 17; ++i) {
      auto c = std::make_unique<Attached>();
      bool ok = c->attach(e, {{"new-session", "-s", "f" + std::to_string(i), "bash", "--norc", "--noprofile"}});
      if (!ok) {
        e.note("flow client " + std::to_string(i), "attach failed");
        break;
      }
      clients.push_back(std::move(c));
    }
    sleep_ms(500);
    std::string listing = run_command(e, {{"list-panes", "-a", "-F", "#{pane_id}:#{session_name}"}});
    e.note("flow panes", listing);
    std::string last;
    for (int i = 0; i < 17 && i < static_cast<int>(clients.size()); ++i) {
      // pane ids are assigned in order: base session pane %0, then second, f0.. each own pane
      std::string session = "f" + std::to_string(i);
      size_t pos = listing.find(":" + session + "\n");
      size_t pos2 = listing.find(":" + session + "|");
      (void)pos;
      (void)pos2;
      // find "%N:fI" in listing
      size_t at = listing.find(":" + session);
      std::string id;
      if (at != std::string::npos) {
        size_t s = listing.rfind('%', at);
        size_t e2 = at;
        // listing is "%N:name\n%M:name" joined with newline inside M payload
        id = listing.substr(s + 1, e2 - s - 1);
      }
      last = e.ctl("flow show " + std::to_string(i), "{\"cmd\": \"show\", \"screen\": {\"kind\": \"message\", \"text\": \"x\"}, \"timeout\": 30, \"pane\": \"" + id + "\"}");
    }
    check(last.find("too many pending flows (16)") != std::string::npos, "17th flow refused with the cap text: " + last);
    // clients detach: results report detached
    for (auto& c : clients) c->conn.close();
    sleep_ms(500);
    e.ctl("flow result detached", "{\"cmd\": \"result\", \"id\": 9}");
    e.ctl("flow result detached 2", "{\"cmd\": \"result\", \"id\": 10}");
  }

  // 7. lock behaviour: a second server for the same home gives up
  {
    pid_t pid = ::fork();
    if (pid == 0) {
      // the grandchild exits quickly: serve() returns 1 when the lock is held
      MplexServer second(e.home);
      int code = second.serve();
      ::_exit(code);
    }
    int raw = 0;
    ::waitpid(pid, &raw, 0);
    e.note("second server exit code", std::to_string(WIFEXITED(raw) ? WEXITSTATUS(raw) : -1));
    e.ctl("still serving", "{\"cmd\": \"status\"}");
  }

  if (slow) {
    // session cap
    std::string last;
    for (int i = 0; i < 70; ++i) {
      last = run_command(e, {{"new-session", "-d", "-s", "cap" + std::to_string(i), "sleep", "300"}}, 2000);
      if (last.find("E:") != std::string::npos) break;
    }
    e.note("session cap", last);
  }

  // 8. kill-server ends everything and removes the socket
  e.note("kill-server", run_command(e, {{"kill-server"}}));
  sleep_ms(1500);
  e.note("socket gone", std::filesystem::exists(e.sock) ? "present" : "removed");
}

std::string make_home(const std::string& tag) {
  std::string pattern = "/tmp/mplex-server-test-" + tag + "-XXXXXX";
  std::vector<char> buf(pattern.begin(), pattern.end());
  buf.push_back('\0');
  char* dir = ::mkdtemp(buf.data());
  if (dir == nullptr) {
    std::perror("mkdtemp");
    std::exit(2);
  }
  return dir;
}

bool wait_socket(const std::string& sock) {
  for (int i = 0; i < 100; ++i) {
    if (std::filesystem::exists(sock)) {
      Conn c;
      if (c.open(sock)) return true;
    }
    sleep_ms(100);
  }
  return false;
}

}  // namespace

int main(int argc, char** argv) {
  std::string tai = "../tai.py";
  bool slow = false;
  for (int i = 1; i < argc; ++i) {
    if (std::string(argv[i]) == "--tai" && i + 1 < argc) tai = argv[++i];
    else if (std::string(argv[i]) == "--slow") slow = true;
  }
  ::setenv("SHELL", "/bin/bash", 1);
  ::setenv("TERM", "xterm-256color", 1);
  ::unsetenv("TAI_PLEXER_ACTIVE");
  ::unsetenv("TAI_PLEXER_SOCKET");
  ::unsetenv("TAI_PLEXER_PANE");
  ::unsetenv("TMUX");
  ::unsetenv("TAI_HOME");

  // ---- C++ server
  Env cpp;
  cpp.label = "cpp";
  cpp.home = make_home("cpp");
  cpp.sock = cpp.home + "/plexer/" + MPLEX_SOCKET_NAME;
  check(MplexServer::ensure_running(cpp.home), "C++ server started by ensure_running");
  check(wait_socket(cpp.sock), "C++ server socket answers");
  {
    struct stat st {};
    ::stat((cpp.home + "/plexer").c_str(), &st);
    check((st.st_mode & 0777) == 0700, "plexer dir is 0700");
    ::stat(cpp.sock.c_str(), &st);
    check((st.st_mode & 0777) == 0600, "socket is 0600");
  }
  scenario(cpp, slow);

  // ---- Python server (the reference)
  Env py;
  py.label = "python";
  py.home = make_home("py");
  py.sock = py.home + "/plexer/server.sock";
  {
    pid_t pid = ::fork();
    if (pid == 0) {
      ::setenv("TAI_HOME", py.home.c_str(), 1);
      int devnull = ::open("/dev/null", O_RDWR);
      ::dup2(devnull, 0);
      ::dup2(devnull, 1);
      ::dup2(devnull, 2);
      ::execlp("python3", "python3", tai.c_str(), "--mplex", "start-server", static_cast<char*>(nullptr));
      ::_exit(127);
    }
    int raw = 0;
    ::waitpid(pid, &raw, 0);
  }
  if (!wait_socket(py.sock)) {
    std::cerr << "python server did not start (tai.py at " << tai << ")\n";
    return 2;
  }
  scenario(py, slow);

  // ---- compare
  {
    std::ofstream("build/mplex_server_cpp.txt") << cpp.transcript;
    std::ofstream("build/mplex_server_py.txt") << py.transcript;
    std::istringstream a(cpp.transcript), b(py.transcript);
    std::string la, lb;
    int line = 0;
    int diffs = 0;
    while (true) {
      bool ga = static_cast<bool>(std::getline(a, la));
      bool gb = static_cast<bool>(std::getline(b, lb));
      if (!ga && !gb) break;
      ++line;
      ++g_checks;
      if (la != lb) {
        ++g_failed;
        ++diffs;
        std::cerr << "DIFF line " << line << "\n  cpp: " << (ga ? la : "<end>") << "\n  py : " << (gb ? lb : "<end>") << "\n";
      }
    }
    std::cerr << "transcript lines compared: " << line << ", differing: " << diffs << "\n";
  }
  if (slow) {
    // a server that never had a session exits after MPLEX_IDLE_EXIT seconds
    Env idle;
    idle.home = make_home("idle");
    idle.sock = idle.home + "/plexer/" + MPLEX_SOCKET_NAME;
    check(MplexServer::ensure_running(idle.home), "idle server starts");
    check(wait_socket(idle.sock), "idle server answers");
    sleep_ms(static_cast<int>(MPLEX_IDLE_EXIT * 1000) + 3000);
    check(!std::filesystem::exists(idle.sock), "idle server exited and removed its socket");
    std::filesystem::remove_all(idle.home);
  }
  std::cerr << "checks: " << g_checks << ", failed: " << g_failed << "\n";
  std::filesystem::remove_all(cpp.home);
  std::filesystem::remove_all(py.home);
  return g_failed == 0 ? 0 : 1;
}