mplex: MplexServer port with differential socket test against tai.py

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-25 01:30:17 +02:00
co-authored by Claude Sonnet 5.5
parent 82c0d42625
commit 796b0b6ef7
2 changed files with 2689 additions and 0 deletions
+777
View File
@@ -0,0 +1,777 @@
// 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;
}