From 796b0b6ef76dedc375eb982f554e1cc595ec6cdc Mon Sep 17 00:00:00 2001 From: retoor Date: Sun, 25 Oct 2026 01:30:17 +0200 Subject: [PATCH] mplex: MplexServer port with differential socket test against tai.py Co-Authored-By: Claude Sonnet 5.5 --- taicpp/src/mplex/mplex_server.cpp | 1912 ++++++++++++++++++++++++++++ taicpp/tests/mplex_server_test.cpp | 777 +++++++++++ 2 files changed, 2689 insertions(+) create mode 100644 taicpp/src/mplex/mplex_server.cpp create mode 100644 taicpp/tests/mplex_server_test.cpp diff --git a/taicpp/src/mplex/mplex_server.cpp b/taicpp/src/mplex/mplex_server.cpp new file mode 100644 index 0000000..7566510 --- /dev/null +++ b/taicpp/src/mplex/mplex_server.cpp @@ -0,0 +1,1912 @@ +// MplexServer: literal port of tai.py 13942-15330 (and plexer_dir, tai.py:9329). +// Owns every session, window, pane and client. One poll loop, no threads. +#include "mplex_server.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "mplex_chooser.h" +#include "mplex_clockmode.h" +#include "mplex_commands.h" +#include "mplex_copymode.h" +#include "mplex_keys.h" +#include "mplex_py.h" +#include "mplex_renderer.h" + +namespace detz::mplex { + +namespace { + +// tai.py VERSION: the `version` format variable. +constexpr const char* MPLEX_VERSION_TEXT = "1.44.2"; + +MplexServer* g_server = nullptr; + +void on_sigterm(int) { + if (g_server != nullptr) g_server->running = false; +} + +// Traceback-like text on stderr (tai.py writes traceback.format_exc()). +void write_traceback(const std::exception& exc) { + std::fprintf(stderr, "Traceback (most recent call last):\n File \"mplex_server.cpp\", in MplexServer\n%s\n", exc.what()); + std::fflush(stderr); +} + +std::string base64(const std::string& data) { + static const char* tbl = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + std::string out; + size_t i = 0; + while (i + 2 < data.size()) { + unsigned v = (static_cast(data[i]) << 16) | (static_cast(data[i + 1]) << 8) | + static_cast(data[i + 2]); + out += tbl[(v >> 18) & 63]; + out += tbl[(v >> 12) & 63]; + out += tbl[(v >> 6) & 63]; + out += tbl[v & 63]; + i += 3; + } + if (i + 1 == data.size()) { + unsigned v = static_cast(data[i]) << 16; + out += tbl[(v >> 18) & 63]; + out += tbl[(v >> 12) & 63]; + out += "=="; + } else if (i + 2 == data.size()) { + unsigned v = (static_cast(data[i]) << 16) | (static_cast(data[i + 1]) << 8); + out += tbl[(v >> 18) & 63]; + out += tbl[(v >> 12) & 63]; + out += tbl[(v >> 6) & 63]; + out += '='; + } + return out; +} + +// text.decode("utf-8", "replace") then back to UTF-8 (what str.encode would hold). +std::string scrub(const std::string& text) { return to_utf8(to_ustr(text)); } + +// str(value) for a JSON value. +std::string pystr(const Json& value) { + switch (value.kind()) { + case Json::Kind::Null: return "None"; + case Json::Kind::Bool: return value.as_bool() ? "True" : "False"; + case Json::Kind::Int: return std::to_string(value.as_int()); + case Json::Kind::Float: return py::format_float(value.as_float()); + case Json::Kind::String: return value.as_string(); + default: return value.dumps(); + } +} + +// repr(value) for a JSON value. +std::string pyrepr(const Json& value) { + if (value.is_string()) return py::repr(value.as_string()); + return pystr(value); +} + +const char* type_name(const Json& value) { + switch (value.kind()) { + case Json::Kind::Null: return "NoneType"; + case Json::Kind::Bool: return "bool"; + case Json::Kind::Int: return "int"; + case Json::Kind::Float: return "float"; + case Json::Kind::String: return "str"; + case Json::Kind::Array: return "list"; + default: return "dict"; + } +} + +// int(value) with Python's errors (ValueError and TypeError both surface as ValueError here). +long long json_int(const Json& value) { + switch (value.kind()) { + case Json::Kind::Bool: return value.as_bool() ? 1 : 0; + case Json::Kind::Int: return value.as_int(); + case Json::Kind::Float: { + double f = value.as_float(); + if (std::isnan(f)) throw ValueError("cannot convert float NaN to integer"); + if (std::isinf(f)) throw ValueError("cannot convert float infinity to integer"); + return static_cast(f); + } + case Json::Kind::String: return py::to_int(value.as_string()); + default: + throw ValueError(std::string("int() argument must be a string, a bytes-like object or a real number, not '") + + type_name(value) + "'"); + } +} + +// float(value) with Python's errors. +double json_float(const Json& value) { + switch (value.kind()) { + case Json::Kind::Bool: return value.as_bool() ? 1.0 : 0.0; + case Json::Kind::Int: return static_cast(value.as_int()); + case Json::Kind::Float: return value.as_float(); + case Json::Kind::String: return py::to_float(value.as_string()); + default: + throw ValueError(std::string("float() argument must be a string or a real number, not '") + type_name(value) + "'"); + } +} + +Json ok_reply(Json data) { + Json reply = Json::object(); + reply.set("ok", Json::boolean(true)); + reply.set("data", std::move(data)); + return reply; +} + +Json err_reply(const std::string& text) { + Json reply = Json::object(); + reply.set("ok", Json::boolean(false)); + reply.set("error", Json::string(text)); + return reply; +} + +void env_set(MplexPane::EnvList& env, const std::string& key, const std::string& value) { + for (auto& item : env) + if (item.first == key) { + item.second = value; + return; + } + env.emplace_back(key, value); +} + +void env_pop(MplexPane::EnvList& env, const std::string& key) { + for (size_t i = 0; i < env.size(); ++i) + if (env[i].first == key) { + env.erase(env.begin() + static_cast(i)); + return; + } +} + +// float -> poll milliseconds, rounded up so a deadline is never woken for early. +int to_ms(double seconds) { + if (seconds <= 0) return 0; + double ms = std::ceil(seconds * 1000.0); + if (ms > 2000000000.0) ms = 2000000000.0; + return static_cast(ms); +} + +} // namespace + +std::string plexer_dir(const std::string& home) { + std::string path = py::join_path(home, PLEXER_DIR_NAME); + std::filesystem::create_directories(path); + ::chmod(path.c_str(), 0700); + return path; +} + +// --------------------------------------------------------------------------------------------- + +MplexServer::MplexServer(const std::string& home_in) { + home = home_in; + dir = plexer_dir(home); + sock_path = py::join_path(dir, MPLEX_SOCKET_NAME); + config = MplexConfig::load(MplexConfig::default_path(home)); + options = &config->options; + renderer = std::make_unique(this); + commands = std::make_unique(this); + ids = {{"session", 0}, {"window", 0}, {"pane", 0}, {"client", 0}, {"buffer", 0}}; + started = py::monotonic(); + { + std::time_t t = std::time(nullptr); + std::tm tmv{}; + gmtime_r(&t, &tmv); + char buf[64]; + std::strftime(buf, sizeof buf, "%Y%m%dT%H%M%S", &tmv); + stamp = buf; + } + for (const auto& problem : config->errors) log_message("mplex.toml: " + problem); + if (config->skipped) + notice = "mplex.toml: " + std::to_string(config->skipped) + " problem(s) skipped, see show-messages"; +} + +MplexServer::~MplexServer() { + if (g_server == this) g_server = nullptr; +} + +// ---- bookkeeping ------------------------------------------------------------------------ + +int MplexServer::next_id(const std::string& kind) { + int value = ids[kind]; + ids[kind] += 1; + return value; +} + +void MplexServer::log_message(const std::string& text) { + messages.emplace_back(py::now(), text); + while (messages.size() > static_cast(MPLEX_MESSAGES_MAX)) messages.pop_front(); +} + +std::shared_ptr MplexServer::latest_session() const { + std::shared_ptr best; + for (const auto& entry : sessions.items()) { + const auto& item = entry.second; + if (!best || item->last_attached > best->last_attached || + (item->last_attached == best->last_attached && item->id > best->id)) + best = item; + } + return best; +} + +std::vector> MplexServer::attached_clients(const std::shared_ptr& session) const { + std::vector> out; + for (const auto& c : clients) + if (c->attached && (!session || c->session == session)) out.push_back(c); + return out; +} + +MplexOptions* MplexServer::pane_options(const std::shared_ptr& pane) { + auto window = pane->window(); + return window ? &window->options : options; +} + +void MplexServer::message(const std::shared_ptr& client, const std::string& text, std::optional milliseconds) { + log_message(text); + if (!client) return; + long ms; + if (milliseconds) { + ms = *milliseconds; + } else { + ms = (client->session == nullptr ? options->get("display-time") : client->session->options.get("display-time")).i; + } + double deadline = py::monotonic() + static_cast(ms) / 1000.0; + client->message = std::make_pair(text, deadline); + client->dirty = true; + std::shared_ptr keep = client; + add_timer(deadline + 0.01, [keep] { keep->dirty = true; }); +} + +void MplexServer::mark_dirty(const std::shared_ptr& window) { + for (const auto& client : clients) + if (client->attached && (!window || (client->session && client->session->window() == window))) client->dirty = true; +} + +void MplexServer::mark_all_dirty() { + for (const auto& client : clients) client->dirty = true; +} + +void MplexServer::add_timer(double deadline, std::function callback) { + if (timers.size() < 1024) timers.emplace_back(deadline, std::move(callback)); +} + +// ---- buffers and copy --------------------------------------------------------------------- + +std::string MplexServer::add_buffer(const std::string& text, const std::optional& name_in) { + std::string name = name_in ? *name_in : "buffer" + std::to_string(next_id("buffer")); + buffers.erase(std::remove_if(buffers.begin(), buffers.end(), [&](const MplexBuffer& b) { return b.name == name; }), + buffers.end()); + std::string data = text.size() > static_cast(MPLEX_PASTE_MAX) ? py::slice(text, 0, MPLEX_PASTE_MAX) : text; + buffers.insert(buffers.begin(), MplexBuffer{name, data}); + size_t limit = static_cast(options->get("buffer-limit").i); + if (buffers.size() > limit) buffers.resize(limit); + return name; +} + +std::string MplexServer::top_buffer_text() const { return buffers.empty() ? "" : buffers[0].data; } + +MplexBuffer& MplexServer::find_buffer(const std::string& name) { + if (name.empty()) { + if (buffers.empty()) throw MplexError("no buffers"); + return buffers[0]; + } + for (auto& item : buffers) + if (item.name == name) return item; + throw MplexError("unknown buffer: " + name); +} + +void MplexServer::run_shell_input(const std::string& command, const std::string& text) { + py::RunResult done; + try { + done = py::run_shell(command, scrub(text), false, true, MPLEX_SHELL_TIMEOUT, true); + } catch (const py::OsError& exc) { + throw MplexError("copy command failed: " + short_error(exc)); + } + if (done.timed_out) + throw MplexError("copy command failed: Command '" + command + "' timed out after " + + py::format_float(MPLEX_SHELL_TIMEOUT) + " seconds"); + if (done.returncode != 0) { + std::string err = py::strip(scrub(done.err)); + throw MplexError("copy command failed: " + py::slice(err, 0, 200)); + } +} + +void MplexServer::store_copy(const std::string& text_in, const std::string& command, bool append) { + std::string text = text_in; + if (append && !buffers.empty()) { + std::string joined = buffers[0].data + text; + if (joined.size() > static_cast(MPLEX_PASTE_MAX)) joined = py::slice(joined, 0, MPLEX_PASTE_MAX); + buffers[0].data = joined; + text = joined; + } else { + add_buffer(text); + } + std::string pipe = command.empty() ? options->get("copy-command").s : command; + if (!pipe.empty()) run_shell_input(pipe, text); + if (options->get("set-clipboard").s != "off") { + std::string sequence = "\x1b]52;c;" + base64(scrub(text)) + "\x07"; + for (const auto& client : attached_clients()) client->send("O", sequence); + } +} + +// ---- sizing -------------------------------------------------------------------------------- + +void MplexServer::resize_session(const std::shared_ptr& session) { + auto attached = attached_clients(session); + if (!attached.empty()) { + int c = attached[0]->cols, r = attached[0]->rows; + for (const auto& item : attached) { + c = std::min(c, item->cols); + r = std::min(r, item->rows); + } + session->size = std::make_pair(c, r); + } else if (!session->size) { + session->size = std::make_pair(80, 24); + } + int cols = session->size->first, rows = session->size->second; + for (const auto& window : session->windows.values()) { + window->width = std::max(1, cols); + window->height = std::max(1, rows - (window->options.get("status").truthy() ? 1 : 0)); + window->relayout(); + } + mark_dirty(); +} + +void MplexServer::resize_all() { + for (const auto& session : sessions.values()) resize_session(session); +} + +// ---- model: panes, windows, sessions -------------------------------------------------------- + +std::string MplexServer::default_shell(MplexOptions& opts) const { + std::string shell = opts.get("default-shell").s; + if (!shell.empty()) return shell; + std::string env = py::getenv_or("SHELL"); + if (!env.empty()) return env; + return "/bin/sh"; +} + +std::string MplexServer::resolve_term(const std::string& name) const { + std::vector wanted = {name, "screen-256color", "xterm-256color", "xterm", "vt100"}; + std::vector roots = {py::getenv_or("TERMINFO"), py::expanduser("~/.terminfo"), "/etc/terminfo", + "/lib/terminfo", "/usr/share/terminfo", "/usr/lib/terminfo"}; + for (const auto& term : wanted) { + for (const auto& root : roots) { + if (root.empty() || term.empty()) continue; + UStr u = to_ustr(term); + std::string first = to_utf8(u.substr(0, 1)); + char hex[16]; + std::snprintf(hex, sizeof hex, "%02x", static_cast(u[0])); + namespace fs = std::filesystem; + std::error_code ec; + if (fs::exists(fs::path(root) / first / term, ec) || fs::exists(fs::path(root) / hex / term, ec)) return term; + } + } + return "xterm"; +} + +std::shared_ptr MplexServer::spawn_pane(const std::shared_ptr& window, int width, int height, + const std::vector& words, const std::string& cwd, + const MplexPane::EnvList* env_extra) { + if (panes.size() >= static_cast(MPLEX_MAX_PANES)) + throw MplexError("too many panes (limit " + std::to_string(MPLEX_MAX_PANES) + ")"); + auto session = window->session(); + MplexOptions& opts = window->options; + auto pane = std::make_shared(next_id("pane"), width, height, static_cast(opts.get("history-limit").i), dir, stamp); + std::string shell = default_shell(opts); + std::string file; + std::vector argv; + if (!words.empty()) { + file = words.size() == 1 ? shell : words[0]; + argv = words.size() == 1 ? std::vector{shell, "-c", words[0]} : words; + } else if (!opts.get("default-command").s.empty()) { + file = shell; + argv = {shell, "-c", opts.get("default-command").s}; + } else { + file = shell; + argv = {"-" + py::basename(shell)}; + } + MplexPane::EnvList env = session->env; + env_pop(env, "LINES"); + env_pop(env, "COLUMNS"); + env_set(env, "TERM", resolve_term(opts.get("default-terminal").s)); + env_set(env, PLEXER_ACTIVE_ENV, "1"); + env_set(env, PLEXER_SOCKET_ENV, sock_path); + env_set(env, PLEXER_PANE_ENV, std::to_string(pane->id)); + if (env_extra) + for (const auto& item : *env_extra) env_set(env, item.first, item.second); + pane->set_window(window); + pane->spawn(file, argv, cwd.empty() ? session->cwd : cwd, env); + panes.set(pane->id, pane); + return pane; +} + +void MplexServer::register_window(const std::shared_ptr& session, const std::shared_ptr& window, + int index, bool select) { + window->set_session(session); + window->options.parent = &session->options; + session->windows.set(index, window); + windows.set(window->id, window); + if (select || !session->current) session->select(index); +} + +std::shared_ptr MplexServer::create_window(const std::shared_ptr& session, const std::string& name, + const std::string& cwd, const std::vector& words, + const MplexPane::EnvList* env_extra, int index, bool select) { + if (windows.size() >= static_cast(MPLEX_MAX_WINDOWS)) + throw MplexError("too many windows (limit " + std::to_string(MPLEX_MAX_WINDOWS) + ")"); + auto size = session->size ? *session->size : std::make_pair(80, 24); + int cols = size.first, rows = size.second; + auto window = std::make_shared(next_id("window"), name, std::max(1, cols), + std::max(1, rows - (options->get("status").truthy() ? 1 : 0)), &session->options); + window->set_session(session); + auto pane = spawn_pane(window, window->width, window->height, words, cwd, env_extra); + window->root = std::make_shared("pane", pane, window->width, window->height); + window->active = pane; + pane->last_active = py::monotonic(); + window->auto_rename = name.empty(); + if (name.empty()) { + if (words.empty()) { + window->name = pane->shell_name; + } else { + std::string base; + if (words.size() == 1) { + auto parts = py::split_ws(words[0]); + base = parts.empty() ? std::string() : parts[0]; + } else { + base = words[0]; + } + window->name = py::basename(base); + } + } + register_window(session, window, index, select); + window->relayout(); + mark_dirty(); + return window; +} + +std::shared_ptr MplexServer::create_session(const std::string& name, const std::string& cwd, + const MplexPane::EnvList& env, const std::vector& words, + const std::string& window_name, std::pair size) { + if (sessions.size() >= static_cast(MPLEX_MAX_SESSIONS)) + throw MplexError("too many sessions (limit " + std::to_string(MPLEX_MAX_SESSIONS) + ")"); + auto session = std::make_shared(next_id("session"), name, options, cwd, env); + session->size = size; + sessions.set(session->id, session); + had_session = true; + try { + create_window(session, window_name, cwd, words, nullptr, static_cast(options->get("base-index").i), true); + } catch (...) { + sessions.erase(session->id); + throw; + } + return session; +} + +std::string MplexServer::auto_session_name() const { + int number = 0; + for (;;) { + bool taken = false; + for (const auto& entry : sessions.items()) + if (entry.second->name == std::to_string(number)) taken = true; + if (!taken) return std::to_string(number); + ++number; + } +} + +std::shared_ptr MplexServer::split_pane(const std::shared_ptr& window, const std::shared_ptr& target, + const std::string& kind, bool before, const std::vector& words, + const std::string& cwd, const MplexPane::EnvList* env_extra, int size, bool select) { + auto leaf = window->leaf_of(target); + if (!leaf) throw MplexError("pane is not in this window"); + if (window->zoomed) { + window->zoomed = nullptr; + window->relayout(); + leaf = window->leaf_of(target); + } + int span = kind == "lr" ? leaf->w : leaf->h; + if (span < 3) throw MplexError("no space for new pane"); + auto pane = spawn_pane(window, 1, 1, words, cwd, env_extra); + try { + auto result = MplexLayout::split(window->root, leaf, kind, before, pane, size); + window->root = result.first; + } catch (const MplexError&) { + discard_pane(pane); + throw; + } + window->layout_index = -1; + window->relayout(); + if (select) window->set_active(pane); + mark_dirty(window); + return pane; +} + +void MplexServer::discard_pane(const std::shared_ptr& pane) { + pane->close(); + panes.erase(pane->id); + if (pane->pid) { + reaping.emplace_back(*pane->pid, py::monotonic()); + pane->pid.reset(); + } + pane->modes.clear(); // breaks the pane <-> mode ownership cycle +} + +void MplexServer::select_pane(const std::shared_ptr& window, const std::shared_ptr& pane) { + if (window->zoomed && pane != window->zoomed) { + window->zoomed = nullptr; + window->relayout(); + } + window->set_active(pane); + mark_dirty(window); +} + +void MplexServer::kill_pane(const std::shared_ptr& pane) { + auto window = pane->window(); + if (!window) return; + if (window->panes().size() == 1) { + kill_window(window); + return; + } + bool was_active = window->active == pane; + window->root = MplexLayout::remove(window->root, window->leaf_of(pane)); + if (window->zoomed == pane) window->zoomed = nullptr; + if (marked == pane) marked = nullptr; + discard_pane(pane); + auto rest = window->panes(); + if (window->last_pane == pane) window->last_pane = nullptr; + if (was_active) { + if (window->last_pane && std::find(rest.begin(), rest.end(), window->last_pane) != rest.end()) { + window->active = window->last_pane; + } else { + std::shared_ptr best; + for (const auto& item : rest) + if (!best || item->last_active > best->last_active) best = item; + window->active = best; + } + window->last_pane = nullptr; + } + window->layout_index = -1; + window->relayout(); + mark_dirty(window); +} + +void MplexServer::kill_window(const std::shared_ptr& window) { + auto session = window->session(); + if (!session) throw std::runtime_error("'NoneType' object has no attribute 'index_of'"); + auto index = session->index_of(window); + for (const auto& pane : window->panes()) { + if (marked == pane) marked = nullptr; + discard_pane(pane); + } + windows.erase(window->id); + if (index) session->windows.erase(*index); + if (session->windows.empty()) { + kill_session(session); + return; + } + session->history.erase(std::remove(session->history.begin(), session->history.end(), window), session->history.end()); + if (session->current == index) { + std::optional nxt = session->last(); + if (!nxt) { + auto idx = session->indexes(); + std::optional lower; + for (int i : idx) + if (index && i < *index) lower = i; + nxt = lower ? *lower : idx.back(); + } + session->current = nxt; + auto target = *session->windows.find(*nxt); + session->history.erase(std::remove(session->history.begin(), session->history.end(), target), session->history.end()); + } + if (session->options.get("renumber-windows").truthy()) renumber(session); + mark_dirty(); +} + +void MplexServer::renumber(const std::shared_ptr& session) { + int base = static_cast(options->get("base-index").i); + auto current = session->window(); + std::vector> ordered; + for (int i : session->indexes()) ordered.push_back(*session->windows.find(i)); + OrderedMap> fresh; + for (size_t n = 0; n < ordered.size(); ++n) fresh.set(base + static_cast(n), ordered[n]); + session->windows = fresh; + session->current = current ? session->index_of(current) : std::nullopt; +} + +void MplexServer::kill_session(const std::shared_ptr& session) { + for (const auto& window : session->windows.values()) { + for (const auto& pane : window->panes()) { + if (marked == pane) marked = nullptr; + discard_pane(pane); + } + windows.erase(window->id); + } + session->windows.clear(); + sessions.erase(session->id); + for (const auto& client : attached_clients(session)) { + if (!sessions.empty() && !options->get("detach-on-destroy").truthy()) + attach_client(client, latest_session()); + else + detach_client(client, "[exited]", "X"); + } + mark_dirty(); +} + +void MplexServer::attach_client(const std::shared_ptr& client, const std::shared_ptr& session) { + auto old = client->session; + client->session = session; + session->last_attached = py::now(); + client->table = "root"; + client->prompt = nullptr; + client->overlay = nullptr; + client->prev.reset(); + if (!client->attached) { + client->attached = true; + client->send("A"); + } + if (old && old != session) resize_session(old); + resize_session(session); + client->dirty = true; + if (notice) { + message(client, *notice, 4000); + notice.reset(); + } +} + +void MplexServer::detach_client(const std::shared_ptr& client, const std::string& text, const std::string& kind) { + auto old = client->session; + client->send(kind, text); + client->attached = false; + client->session = nullptr; + client->closing = true; + if (old && sessions.contains(old->id)) resize_session(old); +} + +// ---- formats --------------------------------------------------------------------------------- + +std::string MplexServer::window_flags(const std::shared_ptr& session, const std::shared_ptr& window) { + std::string flags; + auto index = session->index_of(window); + if (index == session->current) + flags += "*"; + else if (index == session->last()) + flags += "-"; + if (window->zoomed) flags += "Z"; + if (marked && marked->window() == window) flags += "M"; + return flags; +} + +MplexFormat::Resolver MplexServer::resolver_for(const std::shared_ptr& session, const std::shared_ptr& window, + const std::shared_ptr& pane, const std::shared_ptr& client) { + return [this, session, window, pane, client](const std::string& name) -> std::optional { + if (!name.empty() && name[0] == '(') return shell_output(name.substr(1)); + return variable(name, session, window, pane, client); + }; +} + +std::string MplexServer::shell_output(const std::string& command) { + double now = py::monotonic(); + auto it = shell_cache.find(command); + long interval = std::max(1, options->get("status-interval").i); + if (it != shell_cache.end() && now - it->second.first < static_cast(interval)) return it->second.second; + std::string text; + try { + auto done = py::run_shell(command, Bytes(), true, false, MPLEX_SHELL_TIMEOUT); + if (!done.timed_out) { + std::string out = done.out.substr(0, MPLEX_SHELL_OUTPUT_MAX); + text = py::strip(scrub(out)); + auto parts = py::split(text, "\n"); + text = parts.empty() ? "" : parts[0]; + } + } catch (const py::OsError&) { + text = ""; + } + if (shell_cache.size() > 64) shell_cache.clear(); + shell_cache[command] = std::make_pair(now, text); + return text; +} + +std::optional MplexServer::variable(const std::string& name, const std::shared_ptr& session, + const std::shared_ptr& window, const std::shared_ptr& pane, + const std::shared_ptr& client) { + if (MplexOptions::find_def(name) != nullptr) { + MplexOptions& scope = window ? window->options : (session ? session->options : *options); + OptValue value = scope.get(name); + if (value.is_bool()) return std::string(value.b ? "1" : "0"); + return value.str(); + } + if (name == "host") return py::gethostname(); + if (name == "host_short") return py::split(py::gethostname(), ".")[0]; + if (name == "version") return std::string(MPLEX_VERSION_TEXT); + if (session) { + if (name == "session_name") return session->name; + if (name == "session_id") return session->tag(); + if (name == "session_windows") return std::to_string(session->windows.size()); + if (name == "session_created") return std::to_string(static_cast(session->created)); + if (name == "session_attached") return std::to_string(attached_clients(session).size()); + } + if (window && session) { + auto index = session->index_of(window); + if (name == "window_index") return index ? std::to_string(*index) : std::string("None"); + if (name == "window_name") return window->name; + if (name == "window_id") return window->tag(); + if (name == "window_flags") return window_flags(session, window); + if (name == "window_active") return std::string(index == session->current ? "1" : "0"); + if (name == "window_last_flag") return std::string(index == session->last() ? "1" : "0"); + if (name == "window_panes") return std::to_string(window->panes().size()); + if (name == "window_zoomed_flag") return std::string(window->zoomed ? "1" : "0"); + if (name == "window_width") return std::to_string(window->width); + if (name == "window_height") return std::to_string(window->height); + if (name == "window_activity_flag" || name == "window_bell_flag" || name == "window_silence_flag" || name == "window_bigger") + return std::string("0"); + if (name == "window_marked_flag") return std::string(marked && marked->window() == window ? "1" : "0"); + } + if (pane) { + auto mode = pane->top_mode(); + MplexVtScreen& screen = pane->screen; + auto pane_window = pane->window(); + if (name == "pane_id") return pane->tag(); + if (name == "pane_index") { + if (!pane_window) return std::string("0"); + auto list = pane_window->panes(); + long pos = std::find(list.begin(), list.end(), pane) - list.begin(); + if (pos >= static_cast(list.size())) throw std::runtime_error("x not in list"); + return std::to_string(pos + pane_window->options.get("pane-base-index").i); + } + if (name == "pane_title") return pane->title(); + if (name == "pane_current_command") return pane->foreground_command(); + if (name == "pane_current_path") return pane->current_path(); + if (name == "pane_pid") return (pane->pid && *pane->pid != 0) ? std::to_string(*pane->pid) : std::string(); + if (name == "pane_width") return std::to_string(pane->w); + if (name == "pane_height") return std::to_string(pane->h); + if (name == "pane_active") return std::string(pane_window && pane_window->active == pane ? "1" : "0"); + if (name == "pane_last") return std::string(pane_window && pane_window->last_pane == pane ? "1" : "0"); + if (name == "pane_in_mode") return std::string(mode ? "1" : "0"); + if (name == "pane_mode") return mode ? mode->name : std::string(); + if (name == "pane_dead") return std::string(pane->dead ? "1" : "0"); + if (name == "pane_dead_status") return pane->exit_status ? std::to_string(*pane->exit_status) : std::string(); + if (name == "pane_left") return std::to_string(pane->x); + if (name == "pane_top") return std::to_string(pane->y); + if (name == "pane_right") return std::to_string(pane->x + pane->w - 1); + if (name == "pane_bottom") return std::to_string(pane->y + pane->h - 1); + if (name == "pane_synchronized") + return std::string(pane_window && pane_window->options.get("synchronize-panes").truthy() ? "1" : "0"); + if (name == "pane_marked") return std::string(marked == pane ? "1" : "0"); + if (name == "pane_marked_set") return std::string(marked ? "1" : "0"); + if (name == "alternate_on") return std::string(screen.alt_active() ? "1" : "0"); + if (name == "history_size") return std::to_string(pane->history_size()); + if (name == "history_limit") return std::to_string(screen.history ? screen.history->maxlen : 0); + if (name == "cursor_x") return std::to_string(screen.cx); + if (name == "cursor_y") return std::to_string(screen.cy); + auto copy = std::dynamic_pointer_cast(mode); + if (name == "pane_search_string") return (copy && copy->search) ? copy->search->text : std::string(); + if (copy) { + if (name == "scroll_position") return std::to_string(copy->oy); + if (name == "copy_cursor_x") return std::to_string(copy->cx); + if (name == "copy_cursor_y") return std::to_string(copy->cy); + if (name == "selection_present") return std::string(copy->anchor ? "1" : "0"); + if (name == "copy_cursor_line") return copy->text(copy->abs_y()); + if (name == "copy_cursor_word") { + UStr line = copy->utext(copy->abs_y()); + if (line.empty()) return std::string(); + long x = std::min(copy->cx, static_cast(line.size()) - 1); + UStr seps = to_ustr(pane_window->options.get("word-separators").s) + L" "; + long start = x, end = x; + while (start > 0 && seps.find(line[static_cast(start - 1)]) == UStr::npos) --start; + while (end + 1 < static_cast(line.size()) && seps.find(line[static_cast(end + 1)]) == UStr::npos) ++end; + return to_utf8(line.substr(static_cast(start), static_cast(end - start + 1))); + } + } + if (name == "scroll_position") return std::string(); + } + if (client) { + if (name == "client_width") return std::to_string(client->cols); + if (name == "client_height") return std::to_string(client->rows); + if (name == "client_session") return client->session ? client->session->name : std::string(); + if (name == "client_tty") return client->tty; + if (name == "client_termname") return client->term; + if (name == "client_pid") return std::to_string(client->pid); + if (name == "client_key_table") return client->table; + if (name == "client_prefix") return std::string(client->table == "prefix" ? "1" : "0"); + if (name == "client_created") return std::to_string(static_cast(client->created)); + } + return std::nullopt; +} + +std::string MplexServer::expand(const std::string& text, MplexCtx& ctx, const std::map* extra) { + MplexFormat::Resolver base = resolver_for(ctx.session, ctx.window, ctx.pane, ctx.client); + MplexFormat::Resolver resolve = base; + if (extra && !extra->empty()) { + resolve = [base, extra](const std::string& name) -> std::optional { + auto it = extra->find(name); + if (it != extra->end()) return it->second; + return base(name); + }; + } + return MplexFormat::with_time(MplexFormat::expand(text, resolve)); +} + +// ---- key handling ------------------------------------------------------------------------------- + +bool MplexServer::is_prefix(const std::shared_ptr& session, const std::string& name) { + auto& opts = session->options; + return (name == opts.get("prefix").s || name == opts.get("prefix2").s) && name != "None"; +} + +void MplexServer::input_bytes(const std::shared_ptr& client, const Bytes& data) { + client->keybuf += data; + if (client->keybuf.size() > 65536) client->keybuf = client->keybuf.substr(client->keybuf.size() - 65536); + process_keys(client, false); +} + +void MplexServer::process_keys(const std::shared_ptr& client, bool final) { + auto split = MplexKeys::split(client->keybuf, final); + client->keybuf = split.rest; + for (const auto& token : split.tokens) { + if (!client->attached) break; + handle_key(client, token.first, token.second); + } + if (!client->keybuf.empty() && client->attached) + client->esc_deadline = py::monotonic() + static_cast(client->session->options.get("escape-time").i) / 1000.0; + else + client->esc_deadline.reset(); +} + +void MplexServer::handle_key(const std::shared_ptr& client, const std::string& name, const Bytes& raw) { + if (!client->attached) return; + client->dirty = true; + if (client->prompt) { + auto prompt = client->prompt; + prompt->key(*this, *client, name, raw); + return; + } + if (client->overlay) { + auto overlay = client->overlay; + overlay->key(name, raw, *client); + return; + } + client->message.reset(); + auto session = client->session; + auto window = session->window(); + auto pane = window->active; + if (name.empty()) { + forward(pane, window, raw); + return; + } + std::string table = client->table; + if (table == "root") { + const MplexBinding* found = config->lookup("root", name); + if (found != nullptr) { + MplexBinding binding = *found; + run_binding(client, binding, pane); + return; + } + if (is_prefix(session, name)) { + client->table = "prefix"; + client->repeating = false; + return; + } + auto mode = pane->top_mode(); + if (mode) { + mode->key(name, raw, *client); + return; + } + forward(pane, window, raw); + return; + } + const MplexBinding* found = config->lookup(table, name); + std::optional binding; + if (found != nullptr) binding = *found; + if (client->repeating && !(binding && binding->repeat)) { + client->table = "root"; + client->repeating = false; + handle_key(client, name, raw); + return; + } + client->table = "root"; + client->repeating = false; + if (binding) { + run_binding(client, *binding, pane); + if (binding->repeat && client->attached && client->table == "root") { + client->table = table; + client->repeating = true; + client->repeat_until = py::monotonic() + static_cast(client->session->options.get("repeat-time").i) / 1000.0; + } + } +} + +void MplexServer::forward(const std::shared_ptr& pane, const std::shared_ptr& window, const Bytes& raw) { + Bytes data = pane->screen.app_cursor ? MplexKeys::app_cursor_form(raw) : raw; + std::vector> targets; + if (window->options.get("synchronize-panes").truthy()) + targets = window->panes(); + else + targets = {pane}; + for (const auto& target : targets) { + if (!target->dead) { + target->write(data); + target->input_bytes += static_cast(data.size()); + } + } +} + +void MplexServer::run_binding(const std::shared_ptr& client, const MplexBinding& binding, + const std::shared_ptr& pane) { + MplexCtx ctx(this, client, pane, true); + run_checked(ctx, binding.command); +} + +void MplexServer::run_checked(MplexCtx& ctx, const std::string& text) { + try { + std::string output = run_line(ctx, text); + if (!output.empty()) show_output(ctx, output); + } catch (const MplexError& exc) { + message(ctx.client, exc.what()); + } catch (const std::exception& exc) { // a bad command must never take the server down + write_traceback(exc); + message(ctx.client, "internal error: " + short_error(exc)); + } +} + +void MplexServer::show_output(MplexCtx& ctx, const std::string& text) { + auto pane = ctx.pane; + std::string trimmed = py::rstrip(text, "\n"); + auto lines = py::split(trimmed, "\n"); + if (!pane || (lines.size() == 1 && py::len(lines[0]) < 200)) { + message(ctx.client, lines.empty() ? "" : lines[0]); + return; + } + auto old = pane->find_mode(); + if (old && old->static_rows) old->close(); + pane->push_mode(std::make_shared(this, pane, &lines)); + mark_dirty(pane->window()); +} + +void MplexServer::ask_confirm(const std::shared_ptr& client, const std::string& text, std::function callback) { + MplexPromptOptions opts; + opts.confirm = true; + client->prompt = std::make_shared(text + " ", "", [callback](const std::string&) { callback(); }, opts); + client->dirty = true; +} + +void MplexServer::refresh_choosers() { + for (const auto& pane : panes.values()) { + auto modes = pane->modes; + for (const auto& mode : modes) + if (auto chooser = std::dynamic_pointer_cast(mode)) chooser->rebuild(); + } + mark_dirty(); +} + +// ---- command execution ---------------------------------------------------------------------------- + +std::string MplexServer::run_line(MplexCtx& ctx, const std::string& text, int depth) { + return run_commands(ctx, MplexCommandParser::parse(text), depth); +} + +std::string MplexServer::run_commands(MplexCtx& ctx, const std::vector>& cmds, int depth) { + if (depth > MPLEX_COMMAND_DEPTH_MAX) throw MplexError("commands nested too deeply"); + std::vector> flat; + for (const auto& words : cmds) { + std::vector piece; + for (const auto& word : words) { + if (word == ";") { + if (!piece.empty()) flat.push_back(piece); + piece.clear(); + } else { + piece.push_back(word); + } + } + if (!piece.empty()) flat.push_back(piece); + } + std::vector output; + for (const auto& words : flat) { + std::string text = run_one(ctx, words, depth); + if (!text.empty()) output.push_back(text); + } + return py::join("\n", output); +} + +std::string MplexServer::run_words(MplexCtx& ctx, const std::vector& words) { + return run_commands(ctx, {words}, 0); +} + +std::string MplexServer::run_one(MplexCtx& ctx, const std::vector& words, int depth) { + std::string name = MplexCommands::resolve(words[0]); + const MplexCommandSpec* spec = mplex_find_spec(name); + std::vector rest(words.begin() + 1, words.end()); + MplexArgs args = MplexArgs::parse(name, spec->spec, rest, spec->min_args, spec->max_args); + ctx.depth = depth; + return commands->call(name, ctx, args); +} + +// ---- sockets -------------------------------------------------------------------------------------------- + +int MplexServer::open_listener() { + int fd = ::open(py::join_path(dir, MPLEX_LOCK_NAME).c_str(), O_RDWR | O_CREAT | O_CLOEXEC, 0600); + if (fd < 0) throw std::runtime_error(std::string("cannot open lock file: ") + std::strerror(errno)); + if (::flock(fd, LOCK_EX | LOCK_NB) != 0) { + ::close(fd); + return -1; + } + lock_fd = fd; + ::unlink(sock_path.c_str()); + int listener_fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); + if (listener_fd < 0) throw std::runtime_error(std::string("socket: ") + std::strerror(errno)); + sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + if (sock_path.size() >= sizeof addr.sun_path) { + ::close(listener_fd); + throw std::runtime_error("AF_UNIX path too long"); + } + std::memcpy(addr.sun_path, sock_path.c_str(), sock_path.size() + 1); + if (::bind(listener_fd, reinterpret_cast(&addr), sizeof addr) != 0) { + int err = errno; + ::close(listener_fd); + throw std::runtime_error(std::string("[Errno ") + std::to_string(err) + "] " + std::strerror(err)); + } + ::chmod(sock_path.c_str(), 0600); + ::listen(listener_fd, 16); + ::fcntl(listener_fd, F_SETFL, ::fcntl(listener_fd, F_GETFL) | O_NONBLOCK); + return listener_fd; +} + +void MplexServer::accept_connection() { + int fd = ::accept4(listener, nullptr, nullptr, SOCK_NONBLOCK | SOCK_CLOEXEC); + if (fd < 0) return; + if (conns.size() + clients.size() >= static_cast(MPLEX_MAX_CLIENTS) * 2) { + ::close(fd); + return; + } + conns.push_back(std::make_shared(fd)); +} + +void MplexServer::pending_read(const std::shared_ptr& pending) { + char buf[65536]; + ssize_t n = ::recv(pending->fd, buf, sizeof buf, 0); + if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) return; + if (n <= 0) { + drop_pending(pending); + return; + } + pending->buf.append(buf, static_cast(n)); + if (!pending->buf.empty() && pending->buf[0] == MPLEX_MAGIC[0]) { + if (clients.size() >= static_cast(MPLEX_MAX_CLIENTS)) { + drop_pending(pending); + return; + } + conns.erase(std::remove(conns.begin(), conns.end(), pending), conns.end()); + auto client = std::make_shared(this, pending->fd, next_id("client")); + clients.push_back(client); + client_data(client, pending->buf.substr(1)); + return; + } + if (pending->buf.find('\n') != Bytes::npos || pending->buf.size() > static_cast(PLEXER_RECV_MAX)) control(pending); +} + +void MplexServer::drop_pending(const std::shared_ptr& pending) { + auto it = std::find(conns.begin(), conns.end(), pending); + if (it != conns.end()) conns.erase(it); + if (pending->fd >= 0) { + ::close(pending->fd); + pending->fd = -1; + } +} + +void MplexServer::control(const std::shared_ptr& pending) { + Json reply; + try { + Json request = Json::loads(scrub(pending->buf)); + if (!request.is_object()) throw ValueError("request must be a JSON object"); + reply = dispatch_control(request); + } catch (const std::exception& exc) { // one bad request must not end the session + reply = err_reply(short_error(exc)); + } + if (pending->fd >= 0) { + int flags = ::fcntl(pending->fd, F_GETFL); + ::fcntl(pending->fd, F_SETFL, flags & ~O_NONBLOCK); + timeval tv{2, 0}; + ::setsockopt(pending->fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof tv); + std::string data = reply.dumps() + "\n"; + size_t off = 0; + while (off < data.size()) { + ssize_t n = ::send(pending->fd, data.data() + off, data.size() - off, MSG_NOSIGNAL); + if (n <= 0) { + if (n < 0 && errno == EINTR) continue; + break; + } + off += static_cast(n); + } + } + drop_pending(pending); +} + +std::shared_ptr MplexServer::control_pane(const Json& request) { + const Json* ident = request.get("pane"); + if (ident != nullptr && !ident->is_null() && !(ident->is_string() && ident->as_string().empty())) { + std::string text = py::lstrip(pystr(*ident), "%"); + long long id = py::to_int(text); + auto* found = panes.find(static_cast(id)); + if (found == nullptr) throw ValueError("no such pane: " + pystr(*ident)); + return *found; + } + auto session = latest_session(); + if (!session || !session->window()) throw ValueError("no session"); + return session->window()->active; +} + +Json MplexServer::dispatch_control(const Json& request) { + const Json* cmd = request.get("cmd"); + auto is = [&](const char* name) { return cmd != nullptr && cmd->is_string() && cmd->as_string() == name; }; + if (is("send")) { + auto pane = control_pane(request); + const Json* t = request.get("text"); + std::string text = (t != nullptr && t->truthy()) ? pystr(*t) : std::string(); + pane->write(scrub(text)); + const Json* enter = request.get("enter"); + if (enter != nullptr && enter->truthy()) pane->write("\r"); + return ok_reply(Json::boolean(true)); + } + if (is("read")) { + auto pane = control_pane(request); + const Json* l = request.get("lines"); + long long lines = (l != nullptr && l->truthy()) ? json_int(*l) : 200; + lines = std::max(1, std::min(lines, 5000)); + return ok_reply(Json::string(pane->tail_log(static_cast(lines)))); + } + if (is("status")) { + auto pane = control_pane(request); + auto window = pane->window(); + if (!window) throw std::runtime_error("'NoneType' object has no attribute 'session'"); + auto session = window->session(); + if (!session) throw std::runtime_error("'NoneType' object has no attribute 'name'"); + Json data = Json::object(); + data.set("backend", Json::string("native")); + data.set("pid", pane->pid ? Json::integer(*pane->pid) : Json()); + data.set("log", Json::string(pane->log_path)); + data.set("socket", Json::string(sock_path)); + data.set("pane", Json::string(pane->tag())); + data.set("session", Json::string(session->name)); + auto idx = session->index_of(window); + data.set("window", idx ? Json::integer(*idx) : Json()); + return ok_reply(data); + } + if (is("show")) return flow_show(request); + if (is("result")) return flow_result(request); + if (is("cancel")) return flow_cancel(request); + return err_reply("unknown command " + (cmd ? pyrepr(*cmd) : std::string("None"))); +} + +void MplexServer::flow_prune(double now) { + for (int ident : flows.keys()) { + auto* found = flows.find(ident); + if (found == nullptr) continue; + std::optional settled = found->answer ? found->settled : std::nullopt; + if (settled && now - *settled > MPLEX_FLOW_KEEP) { + flows.erase(ident); + } else if (!settled && now > found->until + MPLEX_FLOW_KEEP) { + flow_ui_clear(*found); + flows.erase(ident); + } + } +} + +void MplexServer::flow_ui_clear(MplexFlowRecord& found) { + auto client = found.client; + if (found.overlay && client->overlay.get() == static_cast(found.overlay.get())) client->overlay = nullptr; + if (found.prompt && client->prompt == found.prompt) client->prompt = nullptr; + client->dirty = true; +} + +std::shared_ptr MplexServer::flow_client(const std::shared_ptr& pane) { + auto window = pane->window(); + std::shared_ptr session = window ? window->session() : latest_session(); + if (!session) throw ValueError("no session"); + auto attached = attached_clients(session); + if (attached.empty()) throw ValueError("no attached client for this pane's session"); + return attached[0]; +} + +MplexServer::FlowScreenSpec MplexServer::flow_check_screen(const Json& raw) { + if (!raw.is_object()) throw ValueError("screen must be an object"); + auto field = [&](const char* key, Json fallback) { + const Json* v = raw.get(key); + return v ? *v : fallback; + }; + Json kind = field("kind", Json()); + bool known = false; + if (kind.is_string()) + for (const char* k : {"message", "choice", "text"}) + if (kind.as_string() == k) known = true; + if (!known) throw ValueError("screen kind must be one of message/choice/text"); + Json title = field("title", Json::string("")); + Json text = field("text", Json::string("")); + if (!title.is_string() || !text.is_string()) throw ValueError("screen title and text must be strings"); + if (py::len(title.as_string()) > static_cast(PLEXER_FLOW_TITLE_MAX) || + py::len(text.as_string()) > static_cast(PLEXER_FLOW_TEXT_MAX)) + throw ValueError("screen title/text too long"); + FlowScreenSpec spec; + spec.kind = kind.as_string(); + spec.title = title.as_string(); + spec.text = text.as_string(); + if (spec.kind == "choice") { + const Json* listed = raw.get("options"); + if (listed == nullptr || !listed->is_array() || listed->as_array().empty() || + listed->as_array().size() > static_cast(PLEXER_FLOW_OPTIONS_MAX)) + throw ValueError("choice needs 1 to " + std::to_string(PLEXER_FLOW_OPTIONS_MAX) + " options"); + for (const auto& label : listed->as_array()) { + if (!label.is_string() || py::strip(label.as_string()).empty() || + py::len(label.as_string()) > static_cast(PLEXER_FLOW_LABEL_MAX)) + throw ValueError("choice options must be non-empty strings"); + spec.options.push_back(label.as_string()); + } + long long def; + try { + def = json_int(field("default", Json::integer(0))); + } catch (const std::exception&) { + throw ValueError("choice default must be an option index"); + } + if (!(0 <= def && def < static_cast(spec.options.size()))) throw ValueError("choice default is outside its options"); + spec.default_index = static_cast(def); + } + Json initial = field("initial", Json::string("")); + if (!initial.is_string() || py::len(initial.as_string()) > static_cast(PLEXER_FLOW_ANSWER_MAX)) + throw ValueError("initial text too long"); + spec.initial = initial.as_string(); + return spec; +} + +Json MplexServer::flow_show(const Json& request) { + auto pane = control_pane(request); + const Json* screen_raw = request.get("screen"); + FlowScreenSpec screen = flow_check_screen(screen_raw ? *screen_raw : Json()); + double timeout; + try { + const Json* t = request.get("timeout"); + timeout = t ? json_float(*t) : 60.0; + } catch (const std::exception&) { + throw ValueError("timeout must be a number of seconds"); + } + { + // max(1.0, min(timeout, MAX)) with Python's comparison semantics (NaN included) + double hi = static_cast(PLEXER_FLOW_TIMEOUT_MAX); + double lowered = (hi < timeout) ? hi : timeout; + timeout = (lowered > 1.0) ? lowered : 1.0; + } + double now = py::monotonic(); + flow_prune(now); + if (flows.size() >= static_cast(MPLEX_FLOW_MAX)) + throw ValueError("too many pending flows (" + std::to_string(MPLEX_FLOW_MAX) + ")"); + auto client = flow_client(pane); + if (client->overlay || client->prompt) throw ValueError("client is busy (a prompt or overlay is already open)"); + int flow_id = flow_seq; + flow_seq += 1; + MplexFlowRecord found; + found.id = flow_id; + found.client = client; + found.kind = screen.kind; + found.until = now + timeout; + found.started = now; + if (screen.kind == "text") { + std::string label = (screen.title.empty() ? std::string() : screen.title + ": ") + + py::slice(py::replace(screen.text, "\n", " "), 0, 200) + " "; + MplexPromptOptions opts; + opts.on_cancel = [this, flow_id] { flow_answered(flow_id, "cancelled"); }; + auto prompt = std::make_shared( + label, screen.initial, + [this, flow_id](const std::string& value) { + flow_answered(flow_id, "ok", Json::string(py::slice(value, 0, PLEXER_FLOW_ANSWER_MAX))); + }, + opts); + found.prompt = prompt; + client->prompt = prompt; + } else { + auto overlay = std::make_shared(this, flow_id, screen.title, screen.text, screen.options, + screen.default_index, now + timeout); + found.overlay = overlay; + client->overlay = overlay; + } + flows.set(flow_id, found); + client->dirty = true; + Json data = Json::object(); + data.set("id", Json::integer(flow_id)); + return ok_reply(data); +} + +void MplexServer::flow_answered(int flow_id, const std::string& status, const Json& value) { + auto* found = flows.find(flow_id); + if (found != nullptr && !found->answer) { + found->answer = MplexFlowAnswer{status, value}; + found->settled = py::monotonic(); + } +} + +Json MplexServer::flow_result(const Json& request) { + int flow_id; + const Json* id = request.get("id"); + try { + flow_id = static_cast(json_int(id ? *id : Json())); + } catch (const std::exception&) { + throw ValueError("result needs a flow id"); + } + auto* found = flows.find(flow_id); + if (found == nullptr) throw ValueError("unknown flow " + (id ? pyrepr(*id) : std::string("None"))); + auto client = found->client; + bool listed = std::find(clients.begin(), clients.end(), client) != clients.end(); + if (!listed || !client->attached) { + flows.erase(flow_id); + Json data = Json::object(); + data.set("done", Json::boolean(true)); + data.set("status", Json::string("detached")); + return ok_reply(data); + } + if (found->answer) { + MplexFlowAnswer done = *found->answer; + flow_ui_clear(*found); + flows.erase(flow_id); + Json data = Json::object(); + data.set("done", Json::boolean(true)); + data.set("status", Json::string(done.status)); + data.set("value", done.value); + return ok_reply(data); + } + if (py::monotonic() >= found->until) { + flow_ui_clear(*found); + flows.erase(flow_id); + Json data = Json::object(); + data.set("done", Json::boolean(true)); + data.set("status", Json::string("timeout")); + return ok_reply(data); + } + Json data = Json::object(); + data.set("done", Json::boolean(false)); + return ok_reply(data); +} + +Json MplexServer::flow_cancel(const Json& request) { + int flow_id; + const Json* id = request.get("id"); + try { + flow_id = static_cast(json_int(id ? *id : Json())); + } catch (const std::exception&) { + throw ValueError("cancel needs a flow id"); + } + auto* found = flows.find(flow_id); + if (found == nullptr) throw ValueError("unknown flow " + (id ? pyrepr(*id) : std::string("None"))); + flow_ui_clear(*found); + flows.erase(flow_id); + return ok_reply(Json::boolean(true)); +} + +void MplexServer::client_read(const std::shared_ptr& client) { + char buf[65536]; + ssize_t n = client->fd >= 0 ? ::recv(client->fd, buf, sizeof buf, 0) : -1; + if (n < 0 && client->fd >= 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) return; + if (n <= 0) { + drop_client(client); + return; + } + client_data(client, Bytes(buf, static_cast(n))); +} + +void MplexServer::client_data(const std::shared_ptr& client, const Bytes& data) { + auto frames = client->frames(data); + if (!frames) { + drop_client(client); + return; + } + for (const auto& frame : *frames) { + const std::string& kind = frame.first; + const Bytes& payload = frame.second; + if (kind == "H") { + on_hello(client, payload); + } else if (kind == "I" && client->attached) { + input_bytes(client, payload); + } else if (kind == "R") { + try { + Json size = Json::loads(scrub(payload)); + if (!size.is_object()) throw ValueError("not subscriptable"); + const Json* r = size.get("rows"); + if (r == nullptr) throw ValueError("KeyError"); + client->rows = static_cast(std::max(2, std::min(1000, json_int(*r)))); + const Json* c = size.get("cols"); + if (c == nullptr) throw ValueError("KeyError"); + client->cols = static_cast(std::max(2, std::min(2000, json_int(*c)))); + } catch (const std::exception&) { + continue; + } + client->prev.reset(); + if (client->attached) resize_session(client->session); + client->dirty = true; + } + } +} + +void MplexServer::on_hello(const std::shared_ptr& client, const Bytes& payload) { + std::vector> cmds; + try { + Json info = Json::loads(scrub(payload)); + if (!info.is_object()) throw ValueError("hello must be an object"); + auto get_or = [&](const char* key) -> Json { + const Json* v = info.get(key); + return v ? *v : Json(); + }; + Json v = get_or("rows"); + client->rows = static_cast(std::max(2, std::min(1000, v.truthy() ? json_int(v) : 24))); + v = get_or("cols"); + client->cols = static_cast(std::max(2, std::min(2000, v.truthy() ? json_int(v) : 80))); + v = get_or("term"); + client->term = py::slice(v.truthy() ? pystr(v) : std::string(), 0, 64); + v = get_or("tty"); + client->tty = py::slice(v.truthy() ? pystr(v) : std::string(), 0, 128); + v = get_or("cwd"); + client->cwd = py::slice(v.truthy() ? pystr(v) : std::string("/"), 0, 4096); + v = get_or("pid"); + client->pid = static_cast(v.truthy() ? json_int(v) : 0); + Json env = get_or("env"); + client->env.clear(); + if (env.truthy() && env.is_object()) { + size_t count = 0; + for (const auto& item : env.as_object()) { + if (count++ >= 512) break; + env_set(client->env, py::slice(item.first, 0, 256), py::slice(pystr(item.second), 0, 8192)); + } + } + Json pane = get_or("pane"); + if (!(pane.is_null() || (pane.is_string() && pane.as_string().empty()))) + client->env_pane = static_cast(json_int(pane)); + else + client->env_pane.reset(); + Json listed = get_or("commands"); + if (!listed.truthy()) { + cmds.push_back({"new-session"}); + } else { + std::vector items; + if (listed.is_array()) + items = listed.as_array(); + else if (listed.is_object()) + for (const auto& kv : listed.as_object()) items.push_back(Json::string(kv.first)); + else if (listed.is_string()) + for (wchar_t ch : to_ustr(listed.as_string())) items.push_back(Json::string(to_utf8(ch))); + else + throw ValueError(std::string("'") + type_name(listed) + "' object is not subscriptable"); + if (items.size() > 64) items.resize(64); + for (const auto& item : items) { + if (!item.truthy()) continue; + std::vector words; + if (item.is_array()) { + for (const auto& w : item.as_array()) words.push_back(pystr(w)); + } else if (item.is_object()) { + for (const auto& kv : item.as_object()) words.push_back(kv.first); + } else if (item.is_string()) { + for (wchar_t ch : to_ustr(item.as_string())) words.push_back(to_utf8(ch)); + } else { + throw ValueError(std::string("'") + type_name(item) + "' object is not iterable"); + } + if (words.size() > 256) words.resize(256); + cmds.push_back(words); + } + } + } catch (const std::exception& exc) { + client->send("E", "bad hello: " + short_error(exc)); + client->closing = true; + return; + } + MplexCtx ctx(this, client); + std::string output; + try { + std::vector outs; + for (const auto& words : cmds) { + std::string text = run_one(ctx, words, 0); + if (!text.empty()) outs.push_back(text); + } + output = py::join("\n", outs); + } catch (const MplexError& exc) { + client->send("E", exc.what()); + client->closing = true; + return; + } catch (const std::exception& exc) { + write_traceback(exc); + client->send("E", "internal error: " + short_error(exc)); + client->closing = true; + return; + } + if (!client->attached) { + if (!output.empty()) client->send("M", output); + client->send("X", ""); + client->closing = true; + } +} + +void MplexServer::drop_client(const std::shared_ptr& client) { + auto it = std::find(clients.begin(), clients.end(), client); + if (it != clients.end()) clients.erase(it); + auto session = client->session; + client->attached = false; + client->session = nullptr; + client->close(); + if (session && sessions.contains(session->id)) resize_session(session); +} + +// ---- main loop ------------------------------------------------------------------------------------------ + +int MplexServer::serve() { + listener = open_listener(); + if (listener < 0) return 1; + g_server = this; + struct sigaction term {}; + term.sa_handler = on_sigterm; + sigemptyset(&term.sa_mask); + ::sigaction(SIGTERM, &term, nullptr); + ::signal(SIGHUP, SIG_IGN); + ::signal(SIGINT, SIG_IGN); + ::signal(SIGPIPE, SIG_IGN); // Python ignores SIGPIPE at startup: a dead peer is an error, not a signal + running = true; + try { + loop(); + } catch (...) { + shutdown(); + throw; + } + shutdown(); + return 0; +} + +void MplexServer::stop() { running = false; } + +void MplexServer::loop() { + int failures = 0; + while (running) { + try { + iterate(); + failures = 0; + } catch (const std::exception& exc) { // one failing iteration is logged and survived + write_traceback(exc); + failures += 1; + if (failures >= MPLEX_LOOP_ERRORS_MAX) { + std::fprintf(stderr, "mplex server: %d failed iterations in a row, exiting\n", failures); + running = false; + } + ::usleep(50000); + } + } +} + +namespace { +enum class FdKind { Listen, Pending, Client, Pane }; +struct FdEntry { + FdKind kind; + std::shared_ptr pending; + std::shared_ptr client; + std::shared_ptr pane; +}; +} // namespace + +void MplexServer::iterate() { + double now = py::monotonic(); + run_deadlines(now); + std::vector> fd_map; + auto add = [&](int fd, FdEntry entry) { + for (auto& item : fd_map) + if (item.first == fd) { + item.second = std::move(entry); + return; + } + fd_map.emplace_back(fd, std::move(entry)); + }; + add(listener, FdEntry{FdKind::Listen, nullptr, nullptr, nullptr}); + for (const auto& pending : conns) add(pending->fd, FdEntry{FdKind::Pending, pending, nullptr, nullptr}); + for (const auto& client : clients) add(client->fd, FdEntry{FdKind::Client, nullptr, client, nullptr}); + for (const auto& pane : panes.values()) + if (pane->master_fd >= 0) add(pane->master_fd, FdEntry{FdKind::Pane, nullptr, nullptr, pane}); + std::vector> writers; + for (const auto& client : clients) + if (!client->out.empty()) writers.emplace_back(client->fd, FdEntry{FdKind::Client, nullptr, client, nullptr}); + for (const auto& pane : panes.values()) + if (!pane->pending_in.empty() && pane->master_fd >= 0) + writers.emplace_back(pane->master_fd, FdEntry{FdKind::Pane, nullptr, nullptr, pane}); + + std::vector pfds; + for (const auto& item : fd_map) pfds.push_back(pollfd{item.first, POLLIN, 0}); + for (const auto& item : writers) { + bool merged = false; + for (auto& p : pfds) + if (p.fd == item.first) { + p.events |= POLLOUT; + merged = true; + } + if (!merged) pfds.push_back(pollfd{item.first, POLLOUT, 0}); + } + int rc = ::poll(pfds.data(), pfds.size(), to_ms(wakeup(now))); + if (rc < 0) { + if (errno == EINTR) return; + sweep_bad_fds(); + return; + } + for (const auto& p : pfds) + if (p.revents & POLLNVAL) { + sweep_bad_fds(); + return; + } + for (const auto& p : pfds) { + if (!(p.revents & (POLLIN | POLLHUP | POLLERR))) continue; + for (const auto& item : fd_map) { + if (item.first != p.fd) continue; + const FdEntry& entry = item.second; + switch (entry.kind) { + case FdKind::Listen: accept_connection(); break; + case FdKind::Pending: pending_read(entry.pending); break; + case FdKind::Client: client_read(entry.client); break; + case FdKind::Pane: pane_read(entry.pane); break; + } + break; + } + } + for (const auto& p : pfds) { + if (!(p.revents & (POLLOUT | POLLERR))) continue; + for (const auto& item : writers) { + if (item.first != p.fd) continue; + if (item.second.kind == FdKind::Pane) + item.second.pane->drain(); + else + item.second.client->flush(); + break; + } + } + now = py::monotonic(); + reap_children(now); + render_clients(now); + close_finished(); + if (sessions.empty() && (had_session || now - started > MPLEX_IDLE_EXIT) && !clients_busy()) running = false; +} + +bool MplexServer::clients_busy() const { + for (const auto& client : clients) + if (!client->out.empty()) return true; + return false; +} + +void MplexServer::sweep_bad_fds() { + struct stat st {}; + for (const auto& pending : std::vector>(conns)) + if (pending->fd < 0 || ::fstat(pending->fd, &st) != 0) conns.erase(std::remove(conns.begin(), conns.end(), pending), conns.end()); + for (const auto& client : std::vector>(clients)) + if (client->fd < 0 || ::fstat(client->fd, &st) != 0) drop_client(client); + for (const auto& pane : panes.values()) + if (pane->master_fd >= 0 && ::fstat(pane->master_fd, &st) != 0) pane_exited(pane); +} + +double MplexServer::wakeup(double now) { + double lowest = now + 1.0; + auto take = [&](double v) { lowest = std::min(lowest, v); }; + for (const auto& t : timers) take(t.first); + for (const auto& client : clients) { + if (client->esc_deadline && *client->esc_deadline != 0.0) take(*client->esc_deadline); + if (client->repeating) take(client->repeat_until); + if (client->dirty && client->attached && client->out.size() <= static_cast(MPLEX_CLIENT_OUT_MAX / 4)) + take(client->last_render + MPLEX_RENDER_INTERVAL); + if (client->overlay) take(client->overlay->until); + } + return std::max(0.0, lowest - now); +} + +void MplexServer::run_deadlines(double now) { + if (!timers.empty()) { + std::vector>> ripe, keep; + for (auto& item : timers) (item.first <= now ? ripe : keep).push_back(std::move(item)); + timers = std::move(keep); + for (auto& item : ripe) item.second(); + } + for (const auto& client : std::vector>(clients)) { + if (client->esc_deadline && *client->esc_deadline != 0.0 && now >= *client->esc_deadline && client->attached) { + client->esc_deadline.reset(); + process_keys(client, true); + } + if (client->repeating && now >= client->repeat_until) { + client->repeating = false; + client->table = "root"; + client->dirty = true; + } + if (client->overlay && now >= client->overlay->until) { + client->overlay = nullptr; + client->dirty = true; + } + } + if (now >= next_tick) { + next_tick = now + 1.0; + tick(now); + } + for (const auto& pending : std::vector>(conns)) + if (now - pending->created > 5.0) drop_pending(pending); +} + +void MplexServer::tick(double now) { + if (!flows.empty()) flow_prune(now); + for (const auto& window : windows.values()) { + if (window->auto_rename && window->active && window->session()) { + std::string tmpl = window->options.get("automatic-rename-format").s; + if (window->options.get("automatic-rename").truthy()) { + auto resolver = resolver_for(window->session(), window, window->active); + std::string name = MplexFormat::expand(tmpl, resolver); + if (!name.empty() && name != window->name) { + window->name = py::slice(name, 0, 64); + mark_all_dirty(); + } + } + } + for (const auto& pane : window->panes()) + if (std::dynamic_pointer_cast(pane->top_mode())) mark_dirty(window); + } + long long interval = options->get("status-interval").i; + if (interval && static_cast(now) % interval == 0) mark_all_dirty(); +} + +void MplexServer::pane_read(const std::shared_ptr& pane) { + if (pane->master_fd < 0) throw std::runtime_error("an integer is required"); + char buf[65536]; + ssize_t n = ::read(pane->master_fd, buf, sizeof buf); + if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)) return; + if (n <= 0) { + pane_exited(pane); + return; + } + pane->feed(Bytes(buf, static_cast(n))); + mark_dirty(pane->window()); +} + +void MplexServer::pane_exited(const std::shared_ptr& pane) { + std::optional status; + if (pane->pid) { + int raw = 0; + pid_t done = ::waitpid(*pane->pid, &raw, WNOHANG); + if (done < 0) { + pane->pid.reset(); + } else if (done > 0) { + if (WIFEXITED(raw)) + status = WEXITSTATUS(raw); + else if (WIFSIGNALED(raw)) + status = -WTERMSIG(raw); + else + status = raw; + pane->pid.reset(); + } + } + auto window = pane->window(); + if (window && window->options.get("remain-on-exit").truthy() && !pane->dead) { + pane->dead = true; + pane->exit_status = status; + pane->screen.feed("\r\nPane is dead (status " + (status ? std::to_string(*status) : std::string("None")) + ")\r\n"); + if (pane->master_fd >= 0) { + ::close(pane->master_fd); + pane->master_fd = -1; + } + mark_dirty(window); + return; + } + if (window && panes.contains(pane->id)) kill_pane(pane); +} + +void MplexServer::reap_children(double now) { + std::vector> keep; + for (const auto& item : reaping) { + int raw = 0; + pid_t done = ::waitpid(item.first, &raw, WNOHANG); + if (done < 0) continue; + if (done > 0) continue; + if (now - item.second > MPLEX_REAP_KILL_AFTER) { + if (::killpg(item.first, SIGKILL) != 0) ::kill(item.first, SIGKILL); + } + keep.push_back(item); + } + reaping = keep; +} + +void MplexServer::render_clients(double now) { + for (const auto& client : std::vector>(clients)) { + if (!client->attached || !client->dirty) continue; + if (now - client->last_render < MPLEX_RENDER_INTERVAL) continue; + if (client->out.size() > static_cast(MPLEX_CLIENT_OUT_MAX / 4)) continue; + MplexCanvas canvas = renderer->frame(*client); + auto encoded = canvas.encode(client->prev ? &*client->prev : nullptr); + client->last_render = now; + client->dirty = false; + if (client->prev && encoded.second == *client->prev && canvas.cursor == client->cursor) continue; + client->prev = std::move(encoded.second); + client->cursor = canvas.cursor; + client->send("O", encoded.first); + } +} + +void MplexServer::close_finished() { + for (const auto& client : std::vector>(clients)) + if (client->closing && client->out.empty()) drop_client(client); +} + +void MplexServer::shutdown() { + for (const auto& client : std::vector>(clients)) { + if (client->attached) client->send("X", "[server exited]"); + for (int attempt = 0; attempt < 20; ++attempt) { + if (client->out.empty()) break; + client->flush(); + if (!client->out.empty()) ::usleep(5000); + } + client->close(); + } + clients.clear(); + for (const auto& pane : panes.values()) { + pane->close(true); + if (pane->pid) { + int raw = 0; + ::waitpid(*pane->pid, &raw, WNOHANG); + } + } + if (listener >= 0) { + ::close(listener); + listener = -1; + } + ::unlink(sock_path.c_str()); +} + +bool MplexServer::probe(const std::string& path) { + int fd = ::socket(AF_UNIX, SOCK_STREAM | SOCK_CLOEXEC, 0); + if (fd < 0) return false; + timeval tv{1, 0}; + ::setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv); + ::setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof tv); + sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + bool ok = false; + if (path.size() < sizeof addr.sun_path) { + std::memcpy(addr.sun_path, path.c_str(), path.size() + 1); + ok = ::connect(fd, reinterpret_cast(&addr), sizeof addr) == 0; + } + ::close(fd); + return ok; +} + +bool MplexServer::ensure_running(const std::string& home, double wait) { + std::string path = py::join_path(plexer_dir(home), MPLEX_SOCKET_NAME); + if (probe(path)) return true; + pid_t pid = ::fork(); + if (pid < 0) throw std::runtime_error(std::string("fork: ") + std::strerror(errno)); + if (pid == 0) { + int code = 1; + try { + ::setsid(); + pid_t second = ::fork(); + if (second > 0) ::_exit(0); + int devnull = ::open("/dev/null", O_RDWR); + int log = ::open(py::join_path(plexer_dir(home), MPLEX_LOG_NAME).c_str(), O_WRONLY | O_CREAT | O_APPEND, 0600); + ::dup2(devnull, 0); + ::dup2(log, 1); + ::dup2(log, 2); + if (::chdir("/") != 0) { + } + MplexServer server(home); + code = server.serve(); + } catch (const std::exception& exc) { + write_traceback(exc); + } + ::_exit(code); + } + int raw = 0; + ::waitpid(pid, &raw, 0); + double deadline = py::monotonic() + wait; + while (py::monotonic() < deadline) { + if (probe(path)) return true; + ::usleep(50000); + } + return false; +} + +} // namespace detz::mplex diff --git a/taicpp/tests/mplex_server_test.cpp b/taicpp/tests/mplex_server_test.cpp new file mode 100644 index 0000000..d5b6fd6 --- /dev/null +++ b/taicpp/tests/mplex_server_test.cpp @@ -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 +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#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(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(&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(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(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(int timeout_ms) { + int waited = 0; + for (;;) { + if (buf.size() >= 5) { + uint32_t len = (static_cast(buf[1]) << 24) | (static_cast(buf[2]) << 16) | + (static_cast(buf[3]) << 8) | static_cast(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(payload.size()); + out += static_cast(n >> 24); + out += static_cast(n >> 16); + out += static_cast(n >> 8); + out += static_cast(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, ""); + // 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(value[i + k])) != 0; + for (size_t k = 9; ok && k < 15; ++k) ok = std::isdigit(static_cast(value[i + k])) != 0; + if (ok) value.replace(i, 15, ""); + } + // "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(value[e]))) ++e; + if (e > s) value.replace(s, e - s, ""); + pos = s + 1; + } + } + std::string clean; + for (char c : value) { + if (c == '\x1b') clean += ""; + else if (c == '\r') clean += ""; + else clean += c; + } + transcript += key + " => " + clean + "\n"; + } + std::string control(const std::string& line) { + Conn c; + if (!c.open(sock)) return ""; + 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>& 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 ":" lines for the frames until close/X/D/E. +std::string run_command(Env& e, const std::vector>& commands, int wait_ms = 3000) { + Conn c; + if (!c.open(e.sock)) return ""; + 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>& 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 : ""); + } + { + 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 : ""); + } + { + 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 : ""); + } + { + 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> clients; + std::vector panes; + for (int i = 0; i < 17; ++i) { + auto c = std::make_unique(); + 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(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 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(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(std::getline(a, la)); + bool gb = static_cast(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 : "") << "\n py : " << (gb ? lb : "") << "\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(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; +}