diff --git a/taicpp/docs/wiring/mplex.md b/taicpp/docs/wiring/mplex.md new file mode 100644 index 0000000..eb79765 --- /dev/null +++ b/taicpp/docs/wiring/mplex.md @@ -0,0 +1,84 @@ +# Wiring: mplex and uiflow (hooks for the integrator) + +Everything below is applied outside `src/mplex` and `src/plexer/plexer_flow.*`, which the mplex +port does not own. Nothing here changes behavior of the ported classes. + +## Build + +Add `src/mplex/mplex.mk` to the module list (the top Makefile already has `-include src/*/*.mk`, +so the file is picked up as is). It appends every mplex source and `src/plexer/plexer_flow.cpp` +to `SRCS`. `ALL_SRCS = $(shell find src -name '*.cpp')` already compiles them into `build/detz` +and `build/detz_test`; there is no duplicate `main` in `src/` (the standalone binary +`tests/mplex_cli_main.cpp` is compiled only by the `$(BUILD)/mplex_cli` rule inside `mplex.mk`). + +Includes: the headers use flat names prefixed `mplex_` / `plexer_flow.h`; the Makefile's +`-I` for every `src/*` directory is enough. Link needs only libc (no sqlite, no curl). + +Extra targets defined in `mplex.mk`: `make build/mplex_cli`, `make build/mplex_flow_test`, +`make mplex-flow-test`, `make mplex-e2e`. + +## `--mplex [COMMAND...]` + +tai.py declares `--mplex` with `nargs=argparse.REMAINDER, default=None`: everything after the flag +belongs to mplex, bare `--mplex` means a new session. `src/cli/cli.cpp:130` already parses this into +`o.mplex` (`std::optional>`, set to an empty vector for bare `--mplex`). + +Dispatch (tai.py 40125-40128): after `--version`, `--cap*`, `--serve-mcp`, `--update-models`, +`--demo*` and the theme check, before `--plexer-wrap` and `--tui`: + +```cpp +#include "mplex_client.h" +... +if (opts.mplex.has_value()) { + std::string home = Env("TAI_HOME").empty() ? HomeDir() + "/.tai" : Env("TAI_HOME"); + return detz::mplex::MplexCli::main(home, *opts.mplex); +} +``` + +`MplexCli::main` returns the process exit code and writes its own diagnostics to stderr, exactly +like the Python (`sessions should be nested with care, unset TAI_PLEXER_ACTIVE to force`, +`open terminal failed: not a terminal`, `no server running on `, ...). + +## `--plexer-wrap` + +tai.py 40129-40133: same `home` computation, `profile = args.profile or "default"`: + +```cpp +if (opts.plexer_wrap) { + std::string home = ...; // same as above + detz::mplex::NativePlexerBackend backend(home, opts.profile.empty() ? "default" : opts.profile); + return backend.run(); +} +``` + +`NativePlexerBackend::run()` starts or joins the server and attaches this terminal to a fresh +session (`MplexCli::main(home, {"new-session"})`). + +## Shell hook text and detection + +`detz::mplex::PLEXER_BASHRC_BLOCK`, `FISH_PLEXER_BLOCK`, `PLEXER_BASHRC_MARK_BEGIN/END` and +`plexer_should_wrap(env)` (header `mplex_client.h`) are byte-identical to the Python constants. The +installer (`--install`) must write these strings, not copy its own. + +## Agent tools (`plexer_send`, `plexer_read`, `plexer_type`, `plexer_status`, `plexer_show`) + +Use `detz::plexer::PlexerControllerFactory::detect()` from `plexer_flow.h` (native when +`TAI_PLEXER_SOCKET` is set, tmux when `TMUX_PANE` is set, null otherwise, which hides the tools). +`plexer_show` is `validate_flow_spec(spec)` followed by `PlexerFlow(app).run(*controller)`; its +result JSON is the return value. The existing skeleton `detz::plexer::Controller/Native/Tmux` in +`src/plexer/plexer.h` is superseded by `PlexerController` and friends and can be deleted by the +foundation owner once the tool handlers call the new classes (both can coexist: names differ). + +## Server process + +`MplexServer::ensure_running(home)` double-forks a detached daemon that runs the select loop in the +same binary, so `detz` needs no extra executable. Its stdout/stderr go to +`/plexer/server.log`. The daemon inherits the environment; `TERM`, `SHELL` and `TAI_HOME` +behave as in tai.py. + +## Verified + +See the report of the porting run: `tests/mplex_flow_test.cpp` (spec/flow corpus against tai.py), +`tests/mplex_e2e.py` (six scenarios run through a pty against both `tai.py --mplex` and +`build/mplex_cli --mplex`, observations compared), `tests/mplex_tmux_oracle.py` (tmux 3.6 oracle for +the tmux controller). diff --git a/taicpp/src/mplex/mplex.mk b/taicpp/src/mplex/mplex.mk new file mode 100644 index 0000000..5f1d93c --- /dev/null +++ b/taicpp/src/mplex/mplex.mk @@ -0,0 +1,25 @@ +SRCS += src/mplex/mplex_args.cpp src/mplex/mplex_canvas.cpp src/mplex/mplex_chooser.cpp src/mplex/mplex_client.cpp +SRCS += src/mplex/mplex_clientconn.cpp src/mplex/mplex_clockmode.cpp src/mplex/mplex_command_specs.cpp src/mplex/mplex_commands.cpp +SRCS += src/mplex/mplex_common.cpp src/mplex/mplex_config.cpp src/mplex/mplex_copymode.cpp src/mplex/mplex_ctx.cpp +SRCS += src/mplex/mplex_default_keys.cpp src/mplex/mplex_flowscreen.cpp src/mplex/mplex_format.cpp src/mplex/mplex_input.cpp +SRCS += src/mplex/mplex_json.cpp src/mplex/mplex_keys.cpp src/mplex/mplex_layout.cpp src/mplex/mplex_mode.cpp +SRCS += src/mplex/mplex_options.cpp src/mplex/mplex_pane.cpp src/mplex/mplex_panenumbers.cpp src/mplex/mplex_prompt.cpp +SRCS += src/mplex/mplex_py.cpp src/mplex/mplex_renderer.cpp src/mplex/mplex_server.cpp src/mplex/mplex_style.cpp +SRCS += src/mplex/mplex_toml.cpp src/mplex/mplex_vtscreen.cpp +SRCS += src/plexer/plexer_flow.cpp + +# Standalone client/server binary and its tests; every output lands in $(BUILD). +MPLEX_LIB_SRCS = $(filter src/mplex/%.cpp src/plexer/plexer_flow.cpp,$(SRCS)) +MPLEX_INC = -Isrc/mplex -Isrc/plexer +$(BUILD)/mplex_cli: tests/mplex_cli_main.cpp $(MPLEX_LIB_SRCS) + @mkdir -p $(BUILD) + $(CXX) $(CXXSTD) $(WARN) -O2 $(MPLEX_INC) tests/mplex_cli_main.cpp $(MPLEX_LIB_SRCS) -o $@ +$(BUILD)/mplex_flow_test: tests/mplex_flow_test.cpp $(MPLEX_LIB_SRCS) + @mkdir -p $(BUILD) + $(CXX) $(CXXSTD) $(WARN) -O2 $(MPLEX_INC) tests/mplex_flow_test.cpp $(MPLEX_LIB_SRCS) -o $@ +mplex-flow-test: $(BUILD)/mplex_flow_test + python3 -I tests/mplex_flow_gen.py .. $(BUILD)/flow_corpus.jsonl + $(BUILD)/mplex_flow_test $(BUILD)/flow_corpus.jsonl +mplex-e2e: $(BUILD)/mplex_cli + python3 -I tests/mplex_e2e.py .. +.PHONY: mplex-flow-test mplex-e2e diff --git a/taicpp/src/mplex/mplex_client.cpp b/taicpp/src/mplex/mplex_client.cpp new file mode 100644 index 0000000..687c78e --- /dev/null +++ b/taicpp/src/mplex/mplex_client.cpp @@ -0,0 +1,432 @@ +// MplexClient, MplexCli, NativePlexerBackend, plexer_should_wrap: literal port of tai.py:16804-17062. +#include "mplex_client.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "mplex_args.h" +#include "mplex_commands.h" +#include "mplex_json.h" +#include "mplex_py.h" +#include "mplex_server.h" + +extern char** environ; + +namespace detz::mplex { + +namespace { + +int g_wake_w = -1; + +void on_winch_signal(int) { + if (g_wake_w >= 0) { + int saved = errno; + ssize_t ignored = ::write(g_wake_w, "w", 1); + (void)ignored; + errno = saved; + } +} + +void write_all_fd(int fd, const char* data, size_t size) { + size_t done = 0; + while (done < size) { + ssize_t n = ::write(fd, data + done, size - done); + if (n < 0) { + if (errno == EINTR) continue; + return; + } + done += static_cast(n); + } +} + +void out_text(FILE* stream, const std::string& text) { + std::fwrite(text.data(), 1, text.size(), stream); + std::fflush(stream); +} + +bool send_all(int fd, const Bytes& data) { + size_t done = 0; + while (done < data.size()) { + ssize_t n = ::send(fd, data.data() + done, data.size() - done, MSG_NOSIGNAL); + if (n < 0) { + if (errno == EINTR) continue; + return false; + } + done += static_cast(n); + } + return true; +} + +bool is_digits(const std::string& text) { + if (text.empty()) return false; + for (char c : text) + if (c < '0' || c > '9') return false; + return true; +} + +} // namespace + +struct MplexClient::Impl { + std::string home; + std::vector> commands; + std::string sock_path; + int stdin_fd = -1; + bool has_saved = false; + struct termios saved {}; + bool attached = false; + int wake_r = -1, wake_w = -1; + + std::pair size() const { + struct winsize ws {}; + if (::ioctl(1, TIOCGWINSZ, &ws) == 0 && ws.ws_row && ws.ws_col) return {ws.ws_row, ws.ws_col}; + return {24, 80}; + } + + Bytes hello() const { + auto [rows, cols] = size(); + const char* pane_env = std::getenv(PLEXER_PANE_ENV); + std::string pane = pane_env ? pane_env : ""; + Json info = Json::object(); + info.set("rows", Json::integer(rows)); + info.set("cols", Json::integer(cols)); + const char* term = std::getenv("TERM"); + info.set("term", Json::string(term ? term : "")); + char cwd[4096]; + info.set("cwd", Json::string(::getcwd(cwd, sizeof cwd) ? cwd : "")); + info.set("pid", Json::integer(::getpid())); + std::string tty; + if (::isatty(0)) { + const char* name = ::ttyname(0); + if (name) tty = name; + } + info.set("tty", Json::string(tty)); + Json cmds = Json::array(); + for (const auto& words : commands) { + Json item = Json::array(); + for (const auto& word : words) item.push(Json::string(word)); + cmds.push(std::move(item)); + } + info.set("commands", std::move(cmds)); + Json env = Json::object(); + int count = 0; + for (char** cur = environ; cur && *cur && count < 512; ++cur, ++count) { + std::string entry = *cur; + size_t eq = entry.find('=', 1); + if (eq == std::string::npos) continue; + env.set(entry.substr(0, eq), Json::string(entry.substr(eq + 1))); + } + info.set("env", std::move(env)); + info.set("pane", Json::string(is_digits(pane) ? pane : "")); + return MplexClient::frame("H", info.dumps()); + } + + void enter() { + stdin_fd = 0; + has_saved = false; + struct termios attrs {}; + if (::tcgetattr(stdin_fd, &attrs) == 0) { + saved = attrs; + has_saved = true; + // tty.setraw + attrs.c_iflag &= ~static_cast(BRKINT | ICRNL | INPCK | ISTRIP | IXON); + attrs.c_oflag &= ~static_cast(OPOST); + attrs.c_cflag &= ~static_cast(CSIZE | PARENB); + attrs.c_cflag |= CS8; + attrs.c_lflag &= ~static_cast(ECHO | ICANON | IEXTEN | ISIG); + attrs.c_cc[VMIN] = 1; + attrs.c_cc[VTIME] = 0; + if (::tcsetattr(stdin_fd, TCSAFLUSH, &attrs) != 0) has_saved = false; + } + const std::string seq = "\x1b[?1049h\x1b[0m\x1b[2J\x1b[H"; + write_all_fd(1, seq.data(), seq.size()); + attached = true; + } + + void leave() { + if (!attached) return; + attached = false; + const std::string seq = "\x1b[0m\x1b[?25h\x1b[?1049l"; + write_all_fd(1, seq.data(), seq.size()); + if (has_saved) ::tcsetattr(stdin_fd, TCSADRAIN, &saved); + } + + void write_out(const Bytes& data) { + size_t done = 0; + while (done < data.size()) { + ssize_t n = ::write(1, data.data() + done, data.size() - done); + if (n < 0) { + if (errno == EINTR) continue; + if (errno == EAGAIN || errno == EWOULDBLOCK) { + fd_set wfds; + FD_ZERO(&wfds); + FD_SET(1, &wfds); + struct timeval tv {1, 0}; + ::select(2, nullptr, &wfds, nullptr, &tv); + continue; + } + return; + } + done += static_cast(n); + } + } + + void send_size(int sock) { + auto [rows, cols] = size(); + Json size_json = Json::object(); + size_json.set("rows", Json::integer(rows)); + size_json.set("cols", Json::integer(cols)); + send_all(sock, MplexClient::frame("R", size_json.dumps())); + } + + void suspend(int sock) { + leave(); + ::kill(::getpid(), SIGTSTP); + enter(); + send_size(sock); + } + + // Returns true with code/message set when the session is over. + bool handle_frames(int sock, Bytes& buf, int& code, std::string& message) { + while (buf.size() >= 5) { + uint32_t length = 0; + std::memcpy(&length, buf.data() + 1, 4); + length = ntohl(length); + if (buf.size() < 5 + static_cast(length)) return false; + std::string kind = buf.substr(0, 1); + Bytes payload = buf.substr(5, length); + buf.erase(0, 5 + static_cast(length)); + if (kind == "A") { + enter(); + } else if (kind == "O") { + write_out(payload); + } else if (kind == "M") { + out_text(stdout, to_utf8(to_ustr(payload)) + "\n"); + } else if (kind == "E") { + code = 1; + message = to_utf8(to_ustr(payload)); + return true; + } else if (kind == "D" || kind == "X") { + code = 0; + message = to_utf8(to_ustr(payload)); + return true; + } else if (kind == "Z") { + suspend(sock); + } + } + return false; + } + + int run() { + int sock = ::socket(AF_UNIX, SOCK_STREAM, 0); + struct sockaddr_un addr {}; + addr.sun_family = AF_UNIX; + bool connected = false; + if (sock >= 0 && sock_path.size() < sizeof(addr.sun_path)) { + std::memcpy(addr.sun_path, sock_path.c_str(), sock_path.size() + 1); + connected = ::connect(sock, reinterpret_cast(&addr), sizeof addr) == 0; + } + if (!connected) { + if (sock >= 0) ::close(sock); + out_text(stderr, "no server running on " + sock_path + "\n"); + return 1; + } + send_all(sock, std::string(MPLEX_MAGIC, 1) + hello()); + int fds[2]; + if (::pipe(fds) != 0) return 1; + wake_r = fds[0]; + wake_w = fds[1]; + ::fcntl(wake_w, F_SETFL, ::fcntl(wake_w, F_GETFL) | O_NONBLOCK); + g_wake_w = wake_w; + struct sigaction action {}; + action.sa_handler = on_winch_signal; + sigemptyset(&action.sa_mask); + struct sigaction previous {}; + ::sigaction(SIGWINCH, &action, &previous); + Bytes buf; + int code = 0; + std::string message; + for (;;) { + fd_set rfds; + FD_ZERO(&rfds); + FD_SET(sock, &rfds); + FD_SET(wake_r, &rfds); + int maxfd = sock > wake_r ? sock : wake_r; + if (attached) { + FD_SET(stdin_fd, &rfds); + if (stdin_fd > maxfd) maxfd = stdin_fd; + } + int ready = ::select(maxfd + 1, &rfds, nullptr, nullptr, nullptr); + if (ready < 0) { + if (errno == EINTR) continue; + break; + } + if (FD_ISSET(wake_r, &rfds)) { + char scratch[4096]; + ssize_t ignored = ::read(wake_r, scratch, sizeof scratch); + (void)ignored; + send_size(sock); + } + if (attached && FD_ISSET(stdin_fd, &rfds)) { + char data[4096]; + ssize_t n = ::read(stdin_fd, data, sizeof data); + if (n <= 0) { + message = "[detached]"; + break; + } + send_all(sock, MplexClient::frame("I", Bytes(data, static_cast(n)))); + } + if (FD_ISSET(sock, &rfds)) { + static char chunk[262144]; + ssize_t n = ::recv(sock, chunk, sizeof chunk, 0); + if (n <= 0) { + code = 1; + message = "[lost server]"; + break; + } + buf.append(chunk, static_cast(n)); + if (handle_frames(sock, buf, code, message)) break; + } + } + leave(); + ::sigaction(SIGWINCH, &previous, nullptr); + g_wake_w = -1; + ::close(sock); + ::close(wake_r); + ::close(wake_w); + if (!message.empty()) out_text(code ? stderr : stdout, message + "\n"); + return code; + } +}; + +MplexClient::MplexClient(const std::string& home, const std::vector>& commands) + : impl_(std::make_shared()) { + impl_->home = home; + impl_->commands = commands; + impl_->sock_path = plexer_dir(home) + "/" + MPLEX_SOCKET_NAME; +} + +Bytes MplexClient::frame(const std::string& kind, const Bytes& payload) { + uint32_t length = htonl(static_cast(payload.size())); + Bytes out = kind; + out.append(reinterpret_cast(&length), 4); + out += payload; + return out; +} + +int MplexClient::run() { return impl_->run(); } + +// ---- MplexCli -------------------------------------------------------------- + +bool MplexCli::attaches(const std::string& name, const std::vector& rest) { + if (name == "attach-session") return true; + if (name != "new-session") return false; + try { + const MplexCommandSpec* spec = mplex_find_spec(name); + return !MplexArgs::parse(name, spec->spec, rest, 0, -1).has('d'); + } catch (const MplexError&) { + return true; + } +} + +std::vector> MplexCli::split_commands(const std::vector& words) { + std::vector> commands(1); + for (const auto& word : words) { + if (word == ";") + commands.emplace_back(); + else + commands.back().push_back(word == "\\;" ? ";" : word); + } + std::vector> out; + for (auto& item : commands) + if (!item.empty()) out.push_back(std::move(item)); + return out; +} + +int MplexCli::main(const std::string& home, const std::vector& words) { + auto commands = split_commands(words); + if (commands.empty()) commands.push_back({"new-session"}); + std::string name; + try { + name = MplexCommands::resolve(commands[0][0]); + } catch (const MplexError& exc) { + out_text(stderr, std::string(exc.what()) + "\n"); + return 1; + } + commands[0][0] = name; + std::vector first = commands[0]; + std::vector rest(first.begin() + 1, first.end()); + if (attaches(name, rest)) { + const char* active = std::getenv(PLEXER_ACTIVE_ENV); + if (active && *active) { + out_text(stderr, std::string("sessions should be nested with care, unset ") + PLEXER_ACTIVE_ENV + " to force\n"); + return 1; + } + if (!(::isatty(0) && ::isatty(1))) { + out_text(stderr, "open terminal failed: not a terminal\n"); + return 1; + } + } + if (name == "new-session" || name == "start-server") { + if (!MplexServer::ensure_running(home)) { + out_text(stderr, "could not start the mplex server, see " + plexer_dir(home) + "/" + MPLEX_LOG_NAME + "\n"); + return 1; + } + } + return MplexClient(home, commands).run(); +} + +int NativePlexerBackend::run() { + std::vector words = {"new-session"}; + for (const auto& item : shell_cmd) words.push_back(item); + return MplexCli::main(home, words); +} + +bool plexer_should_wrap(const std::vector>& env) { + auto get = [&](const std::string& key) -> std::string { + for (const auto& item : env) + if (item.first == key) return item.second; + return ""; + }; + if (!get(PLEXER_ACTIVE_ENV).empty()) return false; + if (!get("TMUX").empty()) return false; + struct stat st {}; + if (::stat((py::expanduser("~") + "/.tai-no-plexer").c_str(), &st) == 0) return false; + return true; +} + +const char* const PLEXER_BASHRC_MARK_BEGIN = "# tai plexer hook - start"; +const char* const PLEXER_BASHRC_MARK_END = "# tai plexer hook - end"; +const char* const PLEXER_BASHRC_BLOCK = + "# tai plexer hook - start\n" + "if [ -z \"$TAI_PLEXER_ACTIVE\" ] && [ -z \"$TMUX\" ] && [ ! -e \"$HOME/.tai-no-plexer\" ] && case \"$-\" in *i*) true;; *) false;; esac; then\n" + " if [ -t 0 ] && [ -t 1 ]; then\n" + " exec \"$HOME/.local/bin/tai.py\" --plexer-wrap\n" + " fi\n" + "fi\n" + "# tai plexer hook - end\n"; +const char* const FISH_PLEXER_BLOCK = + "# tai plexer hook - start\n" + "if status is-interactive\n" + "and test -z \"$TAI_PLEXER_ACTIVE\"\n" + "and test -z \"$TMUX\"\n" + "and not test -e \"$HOME/.tai-no-plexer\"\n" + "and isatty stdin\n" + "and isatty stdout\n" + " exec \"$HOME/.local/bin/tai.py\" --plexer-wrap\n" + "end\n" + "# tai plexer hook - end\n"; + +} // namespace detz::mplex diff --git a/taicpp/src/plexer/plexer_flow.cpp b/taicpp/src/plexer/plexer_flow.cpp new file mode 100644 index 0000000..8aed8f1 --- /dev/null +++ b/taicpp/src/plexer/plexer_flow.cpp @@ -0,0 +1,1049 @@ +// Port of tai.py:9302-9985: PlexerController family, plexer_show spec validation, PlexerFlow. +#include "plexer_flow.h" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "mplex_common.h" +#include "mplex_py.h" + +namespace detz::plexer { + +namespace py = detz::mplex::py; +using detz::mplex::to_ustr; +using detz::mplex::UStr; + +const std::vector PLEXER_FLOW_TYPES = {"choice", "text", "info"}; +const std::vector PLEXER_FLOW_KINDS = {"message", "choice", "text"}; + +namespace { + +// ---- Python value helpers --------------------------------------------------------------- + +std::string repr_json(const Json& v) { + switch (v.kind()) { + case Json::Kind::Null: return "None"; + case Json::Kind::Bool: return v.as_bool() ? "True" : "False"; + case Json::Kind::Int: return std::to_string(v.as_int()); + case Json::Kind::Float: return py::format_float(v.as_float()); + case Json::Kind::String: return py::repr(v.as_string()); + case Json::Kind::Array: { + std::string out = "["; + bool first = true; + for (const auto& item : v.as_array()) { + if (!first) out += ", "; + first = false; + out += repr_json(item); + } + return out + "]"; + } + case Json::Kind::Object: { + std::string out = "{"; + bool first = true; + for (const auto& item : v.as_object()) { + if (!first) out += ", "; + first = false; + out += py::repr(item.first) + ": " + repr_json(item.second); + } + return out + "}"; + } + } + return "None"; +} + +// str(value) +std::string str_json(const Json& v) { return v.is_string() ? v.as_string() : repr_json(v); } + +// "%d" in error texts +std::string d(long long n) { return std::to_string(n); } + +bool is_scalar(const Json& v) { return v.is_string() || v.is_number() || v.is_bool() || v.is_null(); } + +// int(value) with Python's TypeError/ValueError turned into nullopt. +std::optional py_int(const Json& v) { + switch (v.kind()) { + case Json::Kind::Bool: return v.as_bool() ? 1 : 0; + case Json::Kind::Int: return v.as_int(); + case Json::Kind::Float: { + double f = v.as_float(); + if (f != f || f > 9e18 || f < -9e18) return std::nullopt; + return static_cast(f); + } + case Json::Kind::String: + try { + return py::to_int(v.as_string()); + } catch (const std::exception&) { + return std::nullopt; + } + default: return std::nullopt; + } +} + +const Json null_json; + +const Json& jget(const Json& obj, std::string_view key) { + const Json* hit = obj.get(std::string(key)); + return hit ? *hit : null_json; +} + +Json jget_or(const Json& obj, const std::string& key, const Json& fallback) { + const Json* hit = obj.get(key); + return hit ? *hit : fallback; +} + +bool in_list(const std::vector& list, const std::string& value) { + return std::find(list.begin(), list.end(), value) != list.end(); +} + +std::string errno_text(int err) { + return "[Errno " + std::to_string(err) + "] " + std::strerror(err); +} + +std::string cap300(const std::string& text) { return py::slice(text, 0, 300); } + +std::string universal_newlines(const std::string& raw) { + std::string out; + out.reserve(raw.size()); + for (size_t i = 0; i < raw.size(); ++i) { + if (raw[i] == '\r') { + out.push_back('\n'); + if (i + 1 < raw.size() && raw[i + 1] == '\n') ++i; + } else { + out.push_back(raw[i]); + } + } + return out; +} + +std::string decode_replace(const std::string& raw) { return detz::mplex::to_utf8(to_ustr(raw)); } + +// "text" % shlex-quoted arg, for the reader fragments: handles %% and a single %s. +std::string format_s(const std::string& fmt, const std::string& arg) { + std::string out; + for (size_t i = 0; i < fmt.size(); ++i) { + if (fmt[i] == '%' && i + 1 < fmt.size()) { + if (fmt[i + 1] == '%') { + out.push_back('%'); + ++i; + continue; + } + if (fmt[i + 1] == 's') { + out += arg; + ++i; + continue; + } + } + out.push_back(fmt[i]); + } + return out; +} + +std::string temp_dir() { + for (const char* name : {"TMPDIR", "TEMP", "TMP"}) { + const char* v = std::getenv(name); + if (v && *v) return v; + } + return "/tmp"; +} + +// tempfile.mkstemp(prefix="tai-flow-", suffix=...): returns (fd, path); fd -1 on failure. +std::pair make_temp(const std::string& suffix) { + std::string pattern = temp_dir() + "/tai-flow-XXXXXX" + suffix; + std::vector buf(pattern.begin(), pattern.end()); + buf.push_back('\0'); + int fd = ::mkstemps(buf.data(), static_cast(suffix.size())); + return {fd, std::string(buf.data())}; +} + +} // namespace + +std::string plexer_dir(const std::string& home) { + std::string path = home + "/" + PLEXER_DIR_NAME; + std::string partial; + size_t pos = 0; + while (pos <= path.size()) { + size_t next = path.find('/', pos); + if (next == std::string::npos) next = path.size(); + partial = path.substr(0, next); + if (!partial.empty()) ::mkdir(partial.c_str(), 0777); + pos = next + 1; + } + ::chmod(path.c_str(), 0700); + return path; +} + +// ---- PlexerController --------------------------------------------------------------------- + +void PlexerController::type_text(const std::string& text, double delay, bool enter) { + UStr chars = to_ustr(text); + for (wchar_t ch : chars) { + send_keys(detz::mplex::to_utf8(ch)); + ::usleep(static_cast(delay * 1e6)); + } + if (enter) send_keys("", true); +} + +// ---- NativePlexerController ----------------------------------------------------------------- + +NativePlexerController::NativePlexerController(const std::string& socket_path_in, const std::optional& pane_in) + : socket_path(socket_path_in), pane(pane_in) { + kind = "native"; +} + +Json NativePlexerController::call(const Json& payload_in) { + Json payload = payload_in; + if (pane && !pane->empty()) payload.set("pane", Json::string(*pane)); + int sock = ::socket(AF_UNIX, SOCK_STREAM, 0); + struct timeval tv {10, 0}; + if (sock >= 0) { + ::setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv); + ::setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof tv); + } + struct sockaddr_un addr {}; + addr.sun_family = AF_UNIX; + int err = 0; + if (sock < 0) { + err = errno; + } else if (socket_path.size() >= sizeof(addr.sun_path)) { + err = ENAMETOOLONG; + } else { + std::memcpy(addr.sun_path, socket_path.c_str(), socket_path.size() + 1); + if (::connect(sock, reinterpret_cast(&addr), sizeof addr) != 0) err = errno; + } + if (err) { + if (sock >= 0) ::close(sock); + throw PlexerError("cannot reach native plexer backend: " + cap300(err == EAGAIN ? "timed out" : errno_text(err))); + } + std::string chunks; + try { + std::string line = payload.dumps() + "\n"; + size_t done = 0; + while (done < line.size()) { + ssize_t n = ::send(sock, line.data() + done, line.size() - done, MSG_NOSIGNAL); + if (n < 0) { + if (errno == EINTR) continue; + int e = errno; + throw PlexerError("native plexer backend read/write failed: " + + cap300(e == EAGAIN ? "timed out" : errno_text(e))); + } + done += static_cast(n); + } + for (;;) { + char chunk[65536]; + ssize_t n = ::recv(sock, chunk, sizeof chunk, 0); + if (n < 0) { + if (errno == EINTR) continue; + int e = errno; + throw PlexerError("native plexer backend read/write failed: " + + cap300(e == EAGAIN ? "timed out" : errno_text(e))); + } + if (n == 0) break; + chunks.append(chunk, static_cast(n)); + if (chunk[n - 1] == '\n') break; + } + } catch (...) { + ::close(sock); + throw; + } + ::close(sock); + std::string raw = py::strip(decode_replace(chunks)); + Json reply = Json::object(); + if (!raw.empty()) { + try { + reply = Json::loads(raw); + } catch (const std::exception&) { + throw PlexerError("native plexer backend sent an unparseable reply"); + } + } + if (!reply.is_object()) throw PlexerError("native plexer backend refused the command"); + const Json* ok = reply.get("ok"); + if (!(ok && ok->truthy())) { + const Json* error = reply.get("error"); + if (error && error->truthy()) throw PlexerError(str_json(*error)); + throw PlexerError("native plexer backend refused the command"); + } + return jget(reply, "data"); +} + +Json NativePlexerController::send_keys(const std::string& text, bool enter) { + Json p = Json::object(); + p.set("cmd", Json::string("send")); + p.set("text", Json::string(text)); + p.set("enter", Json::boolean(enter)); + return call(p); +} + +Json NativePlexerController::read_pane(int lines) { + Json p = Json::object(); + p.set("cmd", Json::string("read")); + p.set("lines", Json::integer(lines)); + return call(p); +} + +Json NativePlexerController::status() { + Json p = Json::object(); + p.set("cmd", Json::string("status")); + return call(p); +} + +Json NativePlexerController::ask(const Json& screen, double timeout) { + Json show = Json::object(); + show.set("cmd", Json::string("show")); + show.set("screen", screen); + show.set("timeout", Json::number(timeout)); + Json handle = call(show); + if (!handle.is_object() || !handle.has("id")) throw PlexerError("native plexer backend sent no flow id"); + Json flow_id = jget(handle, "id"); + double deadline = py::monotonic() + std::max(1.0, timeout); + for (;;) { + Json q = Json::object(); + q.set("cmd", Json::string("result")); + q.set("id", flow_id); + Json reply = call(q); + const Json* done = reply.is_object() ? reply.get("done") : nullptr; + if (done && done->truthy()) { + const Json* status = reply.get("status"); + std::string st = (status && status->is_string()) ? status->as_string() : ""; + if (st == "ok") { + Json out = Json::object(); + out.set("status", Json::string("ok")); + out.set("value", jget(reply, "value")); + return out; + } + if (st == "cancelled" || st == "timeout") { + Json out = Json::object(); + out.set("status", Json::string(st)); + return out; + } + std::string shown = (status && status->truthy()) ? str_json(*status) : "unknown"; + throw PlexerError("flow " + str_json(flow_id) + " ended: " + shown); + } + if (py::monotonic() >= deadline) { + try { + Json c = Json::object(); + c.set("cmd", Json::string("cancel")); + c.set("id", flow_id); + call(c); + } catch (const PlexerError&) { + } + Json out = Json::object(); + out.set("status", Json::string("timeout")); + return out; + } + ::usleep(static_cast(PLEXER_FLOW_POLL * 1e6)); + } +} + +// ---- TmuxPlexerController --------------------------------------------------------------------- + +TmuxPlexerController::TmuxPlexerController(const std::string& pane_in) : pane(pane_in) { kind = "tmux"; } + +std::string TmuxPlexerController::tmux(const std::vector& args) { + std::vector argv = {"tmux"}; + argv.insert(argv.end(), args.begin(), args.end()); + py::RunResult done; + try { + done = py::run_argv(argv, true, true, 15); + } catch (const std::exception& exc) { + throw PlexerError("tmux call failed: " + cap300(exc.what())); + } + if (done.timed_out) { + std::string cmd = "["; + for (size_t i = 0; i < argv.size(); ++i) cmd += (i ? ", " : "") + py::repr(argv[i]); + cmd += "]"; + throw PlexerError("tmux call failed: " + cap300("Command '" + cmd + "' timed out after 15 seconds")); + } + if (done.returncode != 0) + throw PlexerError("tmux error: " + py::slice(py::strip(universal_newlines(decode_replace(done.err))), 0, 300)); + return universal_newlines(decode_replace(done.out)); +} + +Json TmuxPlexerController::send_keys(const std::string& text, bool enter) { + if (!text.empty()) tmux({"send-keys", "-t", pane, "-l", text}); + if (enter) tmux({"send-keys", "-t", pane, "Enter"}); + return Json::boolean(true); +} + +Json TmuxPlexerController::read_pane(int lines) { + return Json::string(tmux({"capture-pane", "-p", "-t", pane, "-S", "-" + std::to_string(std::max(1, lines))})); +} + +Json TmuxPlexerController::status() { + std::string info = tmux({"list-panes", "-t", pane, "-F", "#{session_name}:#{window_index}.#{pane_index} #{pane_current_command}"}); + Json out = Json::object(); + out.set("backend", Json::string("tmux")); + out.set("pane", Json::string(pane)); + out.set("info", Json::string(py::strip(info))); + return out; +} + +Json TmuxPlexerController::ask(const Json& screen, double timeout) { + const Json& kind_json = jget(screen, "kind"); + std::string k = kind_json.is_string() ? kind_json.as_string() : ""; + if (kind_json.is_string()) { + if (k == "choice") return ask_choice(screen, timeout); + if (k == "text") return ask_text(screen, timeout); + if (k == "message") return ask_message(screen, timeout); + } + throw PlexerError("unknown screen kind " + repr_json(kind_json)); +} + +std::string TmuxPlexerController::ask_result_path() { + auto made = make_temp(".answer"); + if (made.first >= 0) ::close(made.first); + ::unlink(made.second.c_str()); + return made.second; +} + +std::optional TmuxPlexerController::await_answer(const std::string& path, double budget) { + double deadline = py::monotonic() + std::max(0.0, budget); + for (;;) { + std::ifstream in(path, std::ios::binary); + if (in) { + std::stringstream ss; + ss << in.rdbuf(); + return decode_replace(ss.str()); + } + if (py::monotonic() >= deadline) return std::nullopt; + ::usleep(50000); + } +} + +void TmuxPlexerController::ask_dismiss() { + try { + py::run_argv({"tmux", "display-popup", "-C"}, true, true, 5); + } catch (const std::exception&) { + } +} + +TmuxPlexerController::RunOutcome TmuxPlexerController::ask_run(const std::vector& argv, const std::string& path, + double timeout) { + std::vector full = {"tmux"}; + full.insert(full.end(), argv.begin(), argv.end()); + py::RunResult done; + try { + done = py::run_argv(full, true, true, std::max(1.0, timeout)); + } catch (const std::exception& exc) { + throw PlexerError("tmux call failed: " + cap300(exc.what())); + } + RunOutcome out; + if (done.timed_out) { + ask_dismiss(); + out.timed_out = true; + out.content = "timeout"; + return out; + } + std::string err = py::strip(universal_newlines(decode_replace(done.err))); + if (done.returncode != 0 && !err.empty()) + throw PlexerError("tmux " + argv[0] + " failed: " + py::slice(err, 0, 300)); + if (done.returncode != 0) return out; + out.content = await_answer(path); + return out; +} + +Json TmuxPlexerController::ask_choice(const Json& screen, double timeout) { + const Json& opts_json = jget(screen, "options"); + std::vector options; + if (opts_json.truthy() && opts_json.is_array()) + for (const auto& o : opts_json.as_array()) options.push_back(str_json(o)); + std::string path = ask_result_path(); + RunOutcome raw; + try { + const Json& title = jget(screen, "title"); + std::vector argv = {"display-menu", "-T", title.truthy() ? str_json(title) : "tai"}; + static const std::string keys = "123456789abcdefghijklmnopqrstuvwxyz"; + for (size_t index = 0; index < options.size(); ++index) { + if (index >= keys.size()) throw PlexerError("string index out of range"); + argv.push_back(options[index]); + argv.push_back(std::string(1, keys[index])); + argv.push_back("run-shell \"printf '%s' '" + std::to_string(index) + "' > " + py::shlex_quote(path) + "\""); + } + raw = ask_run(argv, path, timeout); + } catch (...) { + ::unlink(path.c_str()); + throw; + } + ::unlink(path.c_str()); + Json out = Json::object(); + if (raw.content && *raw.content == "timeout") { + out.set("status", Json::string("timeout")); + return out; + } + std::string stripped = raw.content ? py::strip(*raw.content) : ""; + if (!raw.content || !py::all_digits(to_ustr(stripped))) { + out.set("status", Json::string("cancelled")); + return out; + } + long long index = py::to_int(stripped); + if (index >= 0 && index < static_cast(options.size())) { + out.set("status", Json::string("ok")); + out.set("value", Json::integer(index)); + return out; + } + out.set("status", Json::string("cancelled")); + return out; +} + +TmuxPlexerController::RunOutcome TmuxPlexerController::ask_popup(const std::string& title, const std::string& body, + const std::string& reader, double timeout) { + auto made = make_temp(".prompt"); + if (made.first < 0) throw PlexerError("cannot stage prompt text: " + errno_text(errno)); + std::string prompt_path = made.second; + { + size_t done = 0; + while (done < body.size()) { + ssize_t n = ::write(made.first, body.data() + done, body.size() - done); + if (n < 0) { + if (errno == EINTR) continue; + std::string text = errno_text(errno); + ::close(made.first); + throw PlexerError("cannot stage prompt text: " + cap300(text)); + } + done += static_cast(n); + } + ::close(made.first); + } + std::string path = ask_result_path(); + RunOutcome raw; + try { + std::string command = "cat " + py::shlex_quote(prompt_path) + "; " + format_s(reader, py::shlex_quote(path)); + raw = ask_run({"display-popup", "-w", "62", "-h", "10", "-T", title.empty() ? "tai" : title, "-E", command}, path, timeout); + } catch (...) { + ::unlink(prompt_path.c_str()); + ::unlink(path.c_str()); + throw; + } + ::unlink(prompt_path.c_str()); + ::unlink(path.c_str()); + return raw; +} + +Json TmuxPlexerController::ask_text(const Json& screen, double timeout) { + const Json& text = jget(screen, "text"); + std::string body = (text.truthy() ? str_json(text) : std::string()) + "\n\n> "; + const Json& title = jget(screen, "title"); + RunOutcome raw = ask_popup(title.truthy() ? str_json(title) : "", body, "read -r TAI_V && printf '%%s' \"$TAI_V\" > %s", timeout); + Json out = Json::object(); + if (raw.content && *raw.content == "timeout") { + out.set("status", Json::string("timeout")); + return out; + } + if (!raw.content) { + out.set("status", Json::string("cancelled")); + return out; + } + out.set("status", Json::string("ok")); + out.set("value", Json::string(py::slice(*raw.content, 0, PLEXER_FLOW_ANSWER_MAX))); + return out; +} + +Json TmuxPlexerController::ask_message(const Json& screen, double timeout) { + const Json& text = jget(screen, "text"); + std::string body = (text.truthy() ? str_json(text) : std::string()) + "\n\n[Enter] "; + const Json& title = jget(screen, "title"); + RunOutcome raw = ask_popup(title.truthy() ? str_json(title) : "", body, "read -r _ && printf ok > %s", timeout); + Json out = Json::object(); + if (raw.content && *raw.content == "timeout") { + out.set("status", Json::string("timeout")); + return out; + } + if (!raw.content) { + out.set("status", Json::string("cancelled")); + return out; + } + out.set("status", Json::string("ok")); + out.set("value", Json::boolean(true)); + return out; +} + +std::unique_ptr PlexerControllerFactory::detect() { + const char* socket_path = std::getenv(PLEXER_SOCKET_ENV); + if (socket_path && *socket_path) { + const char* pane = std::getenv(PLEXER_PANE_ENV); + return std::make_unique(socket_path, pane ? std::optional(pane) : std::nullopt); + } + const char* pane = std::getenv("TMUX_PANE"); + if (pane && *pane) return std::make_unique(pane); + return nullptr; +} + +// ---- validation -------------------------------------------------------------------------------- + +std::pair validate_flow_id(const Json& raw, const std::string& what) { + std::string text = py::strip(raw.truthy() ? str_json(raw) : std::string()); + if (text.empty()) return {"", what + " id must be a non-empty string"}; + if (py::len(text) > static_cast(PLEXER_FLOW_ID_MAX)) + return {"", what + " id " + py::repr(py::slice(text, 0, 72)) + " is longer than " + d(PLEXER_FLOW_ID_MAX) + " chars"}; + for (wchar_t ch : to_ustr(text)) + if (py::isspace(ch)) return {"", what + " id " + py::repr(text) + " must not contain whitespace"}; + return {text, ""}; +} + +std::pair validate_flow_text(const Json& raw, const std::string& what, size_t limit) { + if (!raw.is_string()) return {"", what + " must be a string"}; + if (py::len(raw.as_string()) > limit) return {"", what + " is longer than " + d(static_cast(limit)) + " chars"}; + return {raw.as_string(), ""}; +} + +std::pair validate_flow_options(const Json& raw, const std::string& what) { + if (!raw.is_array() || raw.as_array().empty()) + return {Json::array(), what + " needs 1 to " + d(PLEXER_FLOW_OPTIONS_MAX) + " options"}; + if (raw.as_array().size() > static_cast(PLEXER_FLOW_OPTIONS_MAX)) + return {Json::array(), what + " has " + d(static_cast(raw.as_array().size())) + " options, at most " + + d(PLEXER_FLOW_OPTIONS_MAX)}; + Json options = Json::array(); + long long pos = 0; + for (const auto& item : raw.as_array()) { + Json label, value; + if (item.is_string()) { + label = item; + value = item; + } else if (item.is_object()) { + label = jget(item, "label"); + value = jget_or(item, "value", jget(item, "label")); + } else { + return {Json::array(), what + " option " + d(pos + 1) + " must be a string or {label, value}"}; + } + if (!label.is_string() || py::strip(label.as_string()).empty()) + return {Json::array(), what + " option " + d(pos + 1) + " needs a non-empty label"}; + if (py::len(label.as_string()) > static_cast(PLEXER_FLOW_LABEL_MAX)) + return {Json::array(), what + " option " + d(pos + 1) + " label is longer than " + d(PLEXER_FLOW_LABEL_MAX) + " chars"}; + if (!is_scalar(value)) return {Json::array(), what + " option " + d(pos + 1) + " value must be a JSON scalar"}; + Json entry = Json::object(); + entry.set("label", label); + entry.set("value", value); + options.push(entry); + ++pos; + } + return {options, ""}; +} + +std::pair validate_flow_condition(const Json& raw, const std::string& what, + const std::vector& known) { + if (raw.is_null()) return {Json::object(), ""}; + if (!raw.is_object() || raw.as_object().empty()) + return {Json::object(), what + " show_if must be a non-empty {field id: expected} object"}; + Json clauses = Json::object(); + for (const auto& entry : raw.as_object()) { + const std::string& key = entry.first; + const Json& expected = entry.second; + if (!in_list(known, key)) return {Json::object(), what + " show_if refers to unknown field " + py::repr(key)}; + Json accepted = expected.is_array() ? expected : Json::array({expected}); + if (accepted.as_array().empty()) + return {Json::object(), what + " show_if for " + py::repr(key) + " needs at least one expected value"}; + for (const auto& value : accepted.as_array()) + if (!is_scalar(value)) return {Json::object(), what + " show_if for " + py::repr(key) + " must hold JSON scalars"}; + clauses.set(key, accepted); + } + return {clauses, ""}; +} + +namespace { + +using Fail = std::pair; + +Fail fail(const std::string& reason) { return {Json(), reason}; } + +} // namespace + +std::pair validate_flow_spec(const Json& spec) { + if (!spec.is_object()) return fail("spec must be a JSON object"); + Json version = jget_or(spec, "v", Json::integer(PLEXER_FLOW_VERSION)); + if (!version.equals(Json::integer(PLEXER_FLOW_VERSION))) + return fail("spec v must be " + d(PLEXER_FLOW_VERSION) + ", got " + repr_json(version)); + auto [title, error] = validate_flow_text(jget_or(spec, "title", Json::string("")), "title", PLEXER_FLOW_TITLE_MAX); + if (!error.empty()) return fail(error); + auto timeout_opt = py_int(jget_or(spec, "timeout", Json::integer(PLEXER_FLOW_TIMEOUT_DEFAULT))); + if (!timeout_opt) return fail("timeout must be an integer number of seconds"); + long long timeout = *timeout_opt; + if (!(PLEXER_FLOW_TIMEOUT_MIN <= timeout && timeout <= PLEXER_FLOW_TIMEOUT_MAX)) + return fail("timeout must be " + d(PLEXER_FLOW_TIMEOUT_MIN) + " to " + d(PLEXER_FLOW_TIMEOUT_MAX) + " seconds"); + const Json& raw_screens = jget(spec, "screens"); + if (!raw_screens.is_array() || raw_screens.as_array().empty()) return fail("screens must be a non-empty list"); + if (raw_screens.as_array().size() > static_cast(PLEXER_FLOW_SCREENS_MAX)) + return fail(d(static_cast(raw_screens.as_array().size())) + " screens, at most " + d(PLEXER_FLOW_SCREENS_MAX)); + Json screens = Json::array(); + std::vector screen_ids, field_ids; + long long pos = 0; + for (const auto& item : raw_screens.as_array()) { + std::string what = "screen " + d(pos + 1); + ++pos; + if (!item.is_object()) return fail(what + " must be an object"); + auto [ident, e1] = validate_flow_id(jget(item, "id"), what); + if (!e1.empty()) return fail(e1); + if (in_list(screen_ids, ident)) return fail("duplicate screen id " + py::repr(ident)); + screen_ids.push_back(ident); + auto [title_s, e2] = validate_flow_text(jget_or(item, "title", Json::string("")), ident + " title", PLEXER_FLOW_TITLE_MAX); + if (!e2.empty()) return fail(e2); + auto [body, e3] = validate_flow_text(jget_or(item, "body", Json::string("")), ident + " body", PLEXER_FLOW_TEXT_MAX); + if (!e3.empty()) return fail(e3); + Json raw_fields = jget_or(item, "fields", Json::array()); + if (!raw_fields.is_array()) return fail(ident + " fields must be a list"); + if (raw_fields.as_array().size() > static_cast(PLEXER_FLOW_FIELDS_MAX)) + return fail(ident + " has " + d(static_cast(raw_fields.as_array().size())) + " fields, at most " + + d(PLEXER_FLOW_FIELDS_MAX)); + Json fields = Json::array(); + long long fpos = 0; + for (const auto& field : raw_fields.as_array()) { + std::string fwhat = ident + " field " + d(fpos + 1); + ++fpos; + if (!field.is_object()) return fail(fwhat + " must be an object"); + Json kind = jget(field, "type"); + if (!(kind.is_string() && in_list(PLEXER_FLOW_TYPES, kind.as_string()))) { + std::string listed; + for (size_t i = 0; i < PLEXER_FLOW_TYPES.size(); ++i) listed += (i ? "/" : "") + PLEXER_FLOW_TYPES[i]; + return fail(fwhat + " type must be one of " + listed + ", got " + repr_json(kind)); + } + auto [fid, e4] = validate_flow_id(jget(field, "id"), fwhat); + if (!e4.empty()) return fail(e4); + if (in_list(field_ids, fid)) return fail("duplicate field id " + py::repr(fid)); + field_ids.push_back(fid); + auto [label, e5] = validate_flow_text(jget_or(field, "label", Json::string("")), fid + " label", PLEXER_FLOW_LABEL_MAX); + if (!e5.empty()) return fail(e5); + Json entry = Json::object(); + entry.set("id", Json::string(fid)); + entry.set("type", kind); + entry.set("label", Json::string(label)); + entry.set("show_if", jget(field, "show_if")); + if (kind.as_string() == "choice") { + auto [options, e6] = validate_flow_options(jget(field, "options"), fid); + if (!e6.empty()) return fail(e6); + entry.set("options", options); + auto def = py_int(jget_or(field, "default", Json::integer(0))); + if (!def) return fail(fid + " default must be an option index"); + long long count = static_cast(options.as_array().size()); + if (!(0 <= *def && *def < count)) + return fail(fid + " default " + d(*def) + " is outside its " + d(count) + " options"); + entry.set("default", Json::integer(*def)); + } else if (kind.as_string() == "text") { + auto [initial, e7] = validate_flow_text(jget_or(field, "initial", Json::string("")), fid + " initial", PLEXER_FLOW_ANSWER_MAX); + if (!e7.empty()) return fail(e7); + entry.set("initial", Json::string(initial)); + } else { + auto [text, e8] = validate_flow_text(jget_or(field, "text", Json::string("")), fid + " text", PLEXER_FLOW_TEXT_MAX); + if (!e8.empty()) return fail(e8); + entry.set("text", Json::string(text)); + } + fields.push(entry); + } + Json raw_actions = jget_or(item, "actions", Json::array()); + if (!raw_actions.is_array()) return fail(ident + " actions must be a list"); + if (raw_actions.as_array().size() > static_cast(PLEXER_FLOW_ACTIONS_MAX)) + return fail(ident + " has " + d(static_cast(raw_actions.as_array().size())) + " actions, at most " + + d(PLEXER_FLOW_ACTIONS_MAX)); + Json actions = Json::array(); + std::vector action_ids; + long long apos = 0; + for (const auto& action : raw_actions.as_array()) { + std::string awhat = ident + " action " + d(apos + 1); + ++apos; + if (!action.is_object()) return fail(awhat + " must be an object"); + auto [aid, e9] = validate_flow_id(jget(action, "id"), awhat); + if (!e9.empty()) return fail(e9); + if (in_list(action_ids, aid)) return fail(ident + " has a duplicate action id " + py::repr(aid)); + action_ids.push_back(aid); + auto [label, e10] = validate_flow_text(jget_or(action, "label", Json::string("")), aid + " label", PLEXER_FLOW_LABEL_MAX); + if (!e10.empty()) return fail(e10); + if (py::strip(label).empty()) return fail(aid + " needs a non-empty label"); + Json entry = Json::object(); + entry.set("id", Json::string(aid)); + entry.set("label", Json::string(label)); + actions.push(entry); + } + if (py::strip(body).empty() && fields.as_array().empty() && actions.as_array().empty()) + return fail(ident + " has nothing to show (no body, fields or actions)"); + Json screen = Json::object(); + screen.set("id", Json::string(ident)); + screen.set("title", Json::string(title_s)); + screen.set("body", Json::string(body)); + screen.set("fields", fields); + screen.set("actions", actions); + screen.set("next", jget(item, "next")); + screen.set("show_if", jget(item, "show_if")); + screens.push(screen); + } + std::vector known_fields; + for (const auto& screen : screens.as_array()) + for (const auto& field : jget(screen, "fields").as_array()) { + const std::string& t = jget(field, "type").as_string(); + if (t == "choice" || t == "text") known_fields.push_back(jget(field, "id").as_string()); + } + for (auto& screen : screens.as_array()) { + for (auto& field : screen.get("fields")->as_array()) { + auto [clauses, err] = validate_flow_condition(jget(field, "show_if"), jget(field, "id").as_string(), known_fields); + if (!err.empty()) return fail(err); + field.set("show_if", clauses); + } + const std::string sid = jget(screen, "id").as_string(); + { + auto [clauses, err] = validate_flow_condition(jget(screen, "show_if"), sid, known_fields); + if (!err.empty()) return fail(err); + screen.set("show_if", clauses); + } + Json target = jget(screen, "next"); + auto known_target = [&](const Json& dest) { + return dest.is_string() && (dest.as_string() == "$done" || dest.as_string() == "$cancel" || in_list(screen_ids, dest.as_string())); + }; + if (target.is_null()) { + screen.set("next", Json()); + } else if (target.is_string()) { + if (!known_target(target)) return fail(sid + " next refers to unknown screen " + repr_json(target)); + screen.set("next", target); + } else if (target.is_object()) { + if (jget(screen, "actions").as_array().empty()) return fail(sid + " next is a map but the screen has no actions"); + for (const auto& entry : target.as_object()) { + bool found = false; + for (const auto& action : jget(screen, "actions").as_array()) + if (jget(action, "id").as_string() == entry.first) found = true; + if (!found) return fail(sid + " next maps unknown action " + py::repr(entry.first)); + if (!known_target(entry.second)) + return fail(sid + " next for " + py::repr(entry.first) + " refers to unknown screen " + repr_json(entry.second)); + } + screen.set("next", target); + } else { + return fail(sid + " next must be a screen id or an {action: screen} map"); + } + } + Json start = jget_or(spec, "start", jget(screens.as_array()[0], "id")); + if (!(start.is_string() && in_list(screen_ids, start.as_string()))) + return fail("start refers to unknown screen " + repr_json(start)); + Json app = Json::object(); + app.set("title", Json::string(title)); + app.set("timeout", Json::integer(timeout)); + app.set("start", start); + app.set("screens", screens); + return {app, ""}; +} + +// ---- PlexerFlow ----------------------------------------------------------------------------------- + +PlexerFlow::PlexerFlow(const Json& app_in) : app(app_in) { + for (const auto& screen : app.get("screens")->as_array()) { + by_id[jget(screen, "id").as_string()] = screen; + order.push_back(jget(screen, "id").as_string()); + } +} + +bool PlexerFlow::condition_holds(const Json& clauses, const Json& answers) { + if (!clauses.is_object()) return true; + for (const auto& entry : clauses.as_object()) { + const Json* have = answers.get(entry.first); + if (!have) return false; + bool hit = false; + for (const auto& accepted : entry.second.as_array()) + if (have->equals(accepted)) hit = true; + if (!hit) return false; + } + return true; +} + +std::string PlexerFlow::follow_up(const Json& screen) { + const Json& target = jget(screen, "next"); + if (target.is_string()) return target.as_string(); + size_t pos = 0; + const std::string& id = jget(screen, "id").as_string(); + for (size_t i = 0; i < order.size(); ++i) + if (order[i] == id) pos = i; + if (pos + 1 < order.size()) return order[pos + 1]; + return "$done"; +} + +const Json* PlexerFlow::first_askable(const Json& screen) { + for (const auto& field : jget(screen, "fields").as_array()) { + const std::string& t = jget(field, "type").as_string(); + if ((t == "choice" || t == "text") && condition_holds(jget(field, "show_if"), answers)) return &field; + } + return nullptr; +} + +std::string PlexerFlow::screen_text(const Json& screen, const Json* field) { + auto or_title = [&](const Json& first) { + const std::string& s = first.as_string(); + return s.empty() ? jget(app, "title").as_string() : s; + }; + if (field != nullptr && field != first_askable(screen)) { + const std::string& label = jget(*field, "label").as_string(); + if (!label.empty()) return label; + return or_title(jget(screen, "title")); + } + std::vector parts = {or_title(jget(screen, "title")), jget(screen, "body").as_string()}; + for (const auto& item : jget(screen, "fields").as_array()) + if (jget(item, "type").as_string() == "info" && condition_holds(jget(item, "show_if"), answers)) + parts.push_back(jget(item, "text").as_string()); + if (field != nullptr && !jget(*field, "label").as_string().empty()) parts.push_back(jget(*field, "label").as_string()); + std::string out; + bool first = true; + for (const auto& part : parts) { + if (part.empty() || py::strip(part).empty()) continue; + if (!first) out += "\n\n"; + first = false; + out += part; + } + return out; +} + +double PlexerFlow::remaining() const { return deadline - py::monotonic(); } + +namespace { + +Json outcome_for(const std::string& status, const std::string& at, const Json& answers, const std::string& field = "") { + Json out = Json::object(); + out.set("status", Json::string(status)); + out.set("at", Json::string(at)); + if (!field.empty()) out.set("field", Json::string(field)); + out.set("answers", answers); + return out; +} + +} // namespace + +Json PlexerFlow::run(PlexerController& controller) { + deadline = py::monotonic() + static_cast(jget(app, "timeout").as_int()); + std::string current = jget(app, "start").as_string(); + for (;;) { + ++visits; + if (visits > PLEXER_FLOW_VISITS_MAX) { + Json out = Json::object(); + out.set("status", Json::string("error")); + out.set("error", Json::string("flow passed " + d(PLEXER_FLOW_VISITS_MAX) + " screens (loop?), aborting")); + out.set("at", Json::string(current)); + out.set("answers", answers); + return out; + } + const Json& screen = by_id.at(current); + std::string target; + if (condition_holds(jget(screen, "show_if"), answers)) { + ScreenResult result = run_screen(controller, screen); + if (result.kind == ScreenResult::Kind::Outcome) return result.outcome; + target = result.kind == ScreenResult::Kind::Jump ? result.target : follow_up(screen); + } else { + target = follow_up(screen); + } + if (target == "$done") { + Json out = Json::object(); + out.set("status", Json::string("done")); + out.set("answers", answers); + out.set("action", action); + return out; + } + if (target == "$cancel") return outcome_for("cancelled", current, answers); + current = target; + } +} + +PlexerFlow::ScreenResult PlexerFlow::run_screen(PlexerController& controller, const Json& screen) { + auto outcome = [](Json value) { + ScreenResult r; + r.kind = ScreenResult::Kind::Outcome; + r.outcome = std::move(value); + return r; + }; + const std::string sid = jget(screen, "id").as_string(); + const std::string screen_title = [&] { + const std::string& t = jget(screen, "title").as_string(); + return t.empty() ? jget(app, "title").as_string() : t; + }(); + bool asked = false; + for (const auto& field : jget(screen, "fields").as_array()) { + const std::string& type = jget(field, "type").as_string(); + if (type == "info" || !condition_holds(jget(field, "show_if"), answers)) continue; + if (remaining() <= 0) return outcome(outcome_for("timeout", sid, answers)); + const std::string fid = jget(field, "id").as_string(); + if (type == "choice") { + Json ask = Json::object(); + ask.set("title", Json::string(screen_title)); + ask.set("text", Json::string(screen_text(screen, &field))); + ask.set("kind", Json::string("choice")); + Json labels = Json::array(); + for (const auto& opt : jget(field, "options").as_array()) labels.push(jget(opt, "label")); + ask.set("options", labels); + ask.set("default", jget(field, "default")); + Json reply = controller.ask(ask, remaining()); + if (jget(reply, "status").as_string() != "ok") + return outcome(outcome_for(jget(reply, "status").as_string(), sid, answers, fid)); + const Json& opts = jget(field, "options"); + long long idx = jget(reply, "value").is_number() ? jget(reply, "value").as_int() : -1; + if (idx < 0 || idx >= static_cast(opts.as_array().size())) throw PlexerError("list index out of range"); + answers.set(fid, jget(opts.as_array()[static_cast(idx)], "value")); + } else { + Json ask = Json::object(); + ask.set("title", Json::string(screen_title)); + ask.set("text", Json::string(screen_text(screen, &field))); + ask.set("kind", Json::string("text")); + ask.set("initial", jget(field, "initial")); + Json reply = controller.ask(ask, remaining()); + if (jget(reply, "status").as_string() != "ok") + return outcome(outcome_for(jget(reply, "status").as_string(), sid, answers, fid)); + answers.set(fid, jget(reply, "value")); + } + asked = true; + } + const Json& actions = jget(screen, "actions"); + if (actions.as_array().empty()) { + if (!asked) { + if (remaining() <= 0) return outcome(outcome_for("timeout", sid, answers)); + Json ask = Json::object(); + ask.set("title", Json::string(screen_title)); + ask.set("text", Json::string(screen_text(screen))); + ask.set("kind", Json::string("message")); + Json reply = controller.ask(ask, remaining()); + if (jget(reply, "status").as_string() != "ok") + return outcome(outcome_for(jget(reply, "status").as_string(), sid, answers)); + } + return ScreenResult{}; + } + if (remaining() <= 0) return outcome(outcome_for("timeout", sid, answers)); + std::string text; + if (first_askable(screen) == nullptr && py::strip(jget(screen, "body").as_string()).empty()) + text = screen_title; + else if (first_askable(screen) == nullptr) + text = screen_text(screen); + else + text = screen_title; + Json ask = Json::object(); + ask.set("title", Json::string(screen_title)); + ask.set("text", Json::string(text)); + ask.set("kind", Json::string("choice")); + Json labels = Json::array(); + for (const auto& a : actions.as_array()) labels.push(jget(a, "label")); + ask.set("options", labels); + ask.set("default", Json::integer(0)); + Json reply = controller.ask(ask, remaining()); + if (jget(reply, "status").as_string() != "ok") + return outcome(outcome_for(jget(reply, "status").as_string(), sid, answers)); + long long idx = jget(reply, "value").is_number() ? jget(reply, "value").as_int() : -1; + if (idx < 0 || idx >= static_cast(actions.as_array().size())) throw PlexerError("list index out of range"); + action = jget(actions.as_array()[static_cast(idx)], "id"); + const Json& target = jget(screen, "next"); + if (target.is_object()) { + const Json* dest = target.get(action.as_string()); + if (dest == nullptr || dest->is_null()) return ScreenResult{}; + if (dest->as_string() == "$done") { + Json out = Json::object(); + out.set("status", Json::string("done")); + out.set("answers", answers); + out.set("action", action); + return outcome(out); + } + if (dest->as_string() == "$cancel") return outcome(outcome_for("cancelled", sid, answers)); + ScreenResult r; + r.kind = ScreenResult::Kind::Jump; + r.target = dest->as_string(); + return r; + } + return ScreenResult{}; +} + +} // namespace detz::plexer diff --git a/taicpp/tests/mplex_cli_main.cpp b/taicpp/tests/mplex_cli_main.cpp new file mode 100644 index 0000000..c8712fa --- /dev/null +++ b/taicpp/tests/mplex_cli_main.cpp @@ -0,0 +1,30 @@ +// Test artifact: runs the C++ mplex client/server as a standalone binary, mirroring what +// tai.py main does for `--mplex [COMMAND...]` and `--plexer-wrap`. Not part of the detz build. +#include +#include +#include +#include + +#include "mplex_client.h" + +int main(int argc, char** argv) { + using namespace detz::mplex; + std::string home; + const char* env_home = std::getenv("TAI_HOME"); + if (env_home && *env_home) { + home = env_home; + } else { + const char* user_home = std::getenv("HOME"); + home = std::string(user_home ? user_home : "") + "/.tai"; + } + if (argc >= 2 && std::strcmp(argv[1], "--mplex") == 0) { + std::vector words(argv + 2, argv + argc); + return MplexCli::main(home, words); + } + if (argc >= 2 && std::strcmp(argv[1], "--plexer-wrap") == 0) { + NativePlexerBackend backend(home, "default"); + return backend.run(); + } + std::fputs("usage: mplex_cli --mplex [COMMAND...] | --plexer-wrap\n", stderr); + return 2; +} diff --git a/taicpp/tests/mplex_e2e.py b/taicpp/tests/mplex_e2e.py new file mode 100644 index 0000000..9d48e70 --- /dev/null +++ b/taicpp/tests/mplex_e2e.py @@ -0,0 +1,387 @@ +"""End-to-end driver for the mplex client/server: runs the same scenarios against tai.py --mplex +(the oracle) and against build/mplex_cli --mplex, through a real pty, a real detached server and +real bash panes, then compares what the attached terminal shows and what the command line reports. + +usage: python3 -I mplex_e2e.py [--only NAME] [--cpp PATH] [--python-only] +""" +import fcntl +import importlib.util +import json +import os +import pty +import select +import shutil +import signal +import struct +import subprocess +import sys +import tempfile +import termios +import time + +ROOT = sys.argv[1] +ARGS = sys.argv[2:] +ONLY = ARGS[ARGS.index("--only") + 1] if "--only" in ARGS else None +CPP = ARGS[ARGS.index("--cpp") + 1] if "--cpp" in ARGS else os.path.join(ROOT, "taicpp", "build", "mplex_cli") +PYTHON_ONLY = "--python-only" in ARGS + +spec = importlib.util.spec_from_file_location("tai_mod", ROOT + "/tai.py") +tai = importlib.util.module_from_spec(spec) +sys.modules["tai_mod"] = tai +spec.loader.exec_module(tai) + +PREFIX = b"\x02" +UP, DOWN, RIGHT, LEFT = b"\x1b[A", b"\x1b[B", b"\x1b[C", b"\x1b[D" +ROWS, COLS = 24, 80 + + +class Harness: + """One implementation (argv prefix) in a throwaway TAI_HOME with a throwaway bash home.""" + + def __init__(self, label, argv, config=None): + self.label = label + self.argv = argv + self.tmp = tempfile.mkdtemp(prefix="tai-mplex-e2e-") + self.home = os.path.join(self.tmp, "tai") + os.makedirs(self.home) + self.bash_home = os.path.join(self.tmp, "home") + os.makedirs(self.bash_home) + with open(os.path.join(self.bash_home, ".bash_profile"), "w") as handle: + handle.write("PS1='$ '\nunset PROMPT_COMMAND\nHISTFILE=/dev/null\n") + base = 'status-right = "R"\nautomatic-rename = false\n' if config is None else config + with open(os.path.join(self.home, "mplex.toml"), "w") as handle: + handle.write("[options]\n" + base if "[options]" not in base else base) + self.env = dict(os.environ) + for name in ("TAI_PLEXER_ACTIVE", "TAI_PLEXER_SOCKET", "TAI_PLEXER_PANE", "TMUX", "VISUAL", "EDITOR"): + self.env.pop(name, None) + self.env.update({"TAI_HOME": self.home, "HOME": self.bash_home, "TERM": "xterm-256color", "SHELL": "/bin/bash", + "TAI_PASSPHRASE": ""}) + self.clients = [] + + def cli(self, *words): + done = subprocess.run(self.argv + list(words), env=self.env, stdin=subprocess.DEVNULL, capture_output=True, + text=True, timeout=30) + return done.returncode, done.stdout, done.stderr + + def attach(self, *words): + client = Client(self, list(words)) + self.clients.append(client) + return client + + def close(self): + for client in self.clients: + client.kill() + try: + self.cli("kill-server") + except Exception: # noqa: BLE001 + pass + time.sleep(0.3) + shutil.rmtree(self.tmp, ignore_errors=True) + + +class Client: + def __init__(self, harness, words): + self.harness = harness + master, slave = pty.openpty() + fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", ROWS, COLS, 0, 0)) + self.master = master + self.proc = subprocess.Popen(harness.argv + words, stdin=slave, stdout=slave, stderr=slave, env=harness.env, + start_new_session=True, close_fds=True) + os.close(slave) + self.screen = tai.TuiVtScreen(ROWS, COLS) + self.raw = b"" + + def pump(self, seconds=0.2): + end = time.monotonic() + seconds + while time.monotonic() < end: + ready, _w, _x = select.select([self.master], [], [], 0.05) + if ready: + try: + data = os.read(self.master, 65536) + except OSError: + return + if not data: + return + self.raw += data + self.screen.feed(data) + + def send(self, data, settle=0.25): + os.write(self.master, data) + self.pump(settle) + + def text(self): + return self.screen.text() + + def wait_for(self, label, predicate, timeout=10.0): + end = time.monotonic() + timeout + while time.monotonic() < end: + self.pump(0.1) + if predicate(self.text()): + return + raise AssertionError("%s: timed out waiting for %s; screen:\n%s" % (self.harness.label, label, self.text())) + + def alive(self): + return self.proc.poll() is None + + def wait_exit(self, timeout=10.0): + end = time.monotonic() + timeout + while time.monotonic() < end: + self.pump(0.1) + if self.proc.poll() is not None: + self.pump(0.2) + return self.proc.returncode + raise AssertionError("%s: client did not exit" % self.harness.label) + + def kill(self): + if self.proc.poll() is None: + try: + os.killpg(self.proc.pid, signal.SIGKILL) + except OSError: + pass + try: + os.close(self.master) + except OSError: + pass + + +def screen_body(client): + """The screen minus the status row, trailing blanks stripped, for comparison.""" + lines = client.text().split("\n") + return "\n".join(line.rstrip() for line in lines) + + +# ---- scenarios: each returns a list of (label, observation) ---------------------------------- + +def sc_lifecycle(h): + seen = [] + client = h.attach() + client.wait_for("shell prompt and window list", lambda t: "0:bash*" in t and "$" in t) + seen.append(("fresh", screen_body(client))) + client.send(b"echo hello-$((40+2))\r", 0.6) + seen.append(("echo", screen_body(client))) + rc, out, err = h.cli("list-sessions", "-F", "#{session_name}:#{session_windows}") + seen.append(("ls", (rc, out, err))) + client.send(PREFIX + b"d", 0.5) + code = client.wait_exit() + seen.append(("detach exit", code)) + rc, out, err = h.cli("list-sessions", "-F", "#{session_name}:#{session_attached}") + seen.append(("ls after detach", (rc, out, err))) + again = h.attach("attach") + again.wait_for("reattach", lambda t: "hello-42" in t) + seen.append(("reattached", screen_body(again))) + again.send(b"exit\r", 0.8) + code = again.wait_exit() + seen.append(("exit code", code)) + time.sleep(0.5) + seen.append(("server gone", h.cli("list-sessions")[0])) + return seen + + +def sc_panes_windows(h): + seen = [] + client = h.attach() + client.wait_for("start", lambda t: "0:bash*" in t) + client.send(PREFIX + b"%", 0.6) + client.send(b"echo right-pane\r", 0.5) + seen.append(("split h", screen_body(client))) + client.send(PREFIX + b'"', 0.6) + seen.append(("split v", screen_body(client))) + client.send(PREFIX + b"o", 0.3) + client.send(PREFIX + b"c", 0.7) + client.wait_for("second window", lambda t: "1:bash*" in t) + seen.append(("new window", screen_body(client))) + client.send(PREFIX + b"p", 0.4) + seen.append(("previous", screen_body(client))) + client.send(PREFIX + b"z", 0.4) + seen.append(("zoom", screen_body(client))) + client.send(PREFIX + b"z", 0.4) + client.send(PREFIX + b"," + b"\x15" + b"renamed\r", 0.5) + seen.append(("rename", screen_body(client))) + for cmd in (["list-windows"], ["list-panes"], ["list-panes", "-a", "-F", "#{pane_index} #{pane_width}x#{pane_height}"], + ["display-message", "-p", "#{window_name} #{window_panes}"], ["select-layout", "even-vertical"], + ["list-panes", "-F", "#{pane_index} #{pane_width}x#{pane_height} #{pane_top} #{pane_left}"]): + seen.append((" ".join(cmd), tuple(x.replace(h.home, "HOME") if isinstance(x, str) else x for x in h.cli(*cmd)))) + client.send(PREFIX + b"x", 0.4) + client.send(b"y", 0.6) + seen.append(("kill pane", screen_body(client))) + return seen + + +def sc_commands(h): + seen = [] + assert h.cli("new-session", "-d", "-s", "work")[0] == 0 + time.sleep(0.5) + for cmd in (["list-sessions", "-F", "#{session_name}"], ["has-session", "-t", "work"], ["has-session", "-t", "nope"], + ["show-options", "-g"], ["show-options", "-g", "status"], ["set-option", "-g", "status-right", "X"], + ["show-options", "-gv", "status-right"], ["set-option", "-g", "nosuch", "1"], ["bogus-command"], + ["list-commands"], ["list-keys", "-T", "root"], ["list-keys", "-N"], ["list-keys", "-T", "copy-mode-vi"], + ["bind-key", "-n", "M-x", "display-message", "hi"], ["list-keys", "-T", "root"], + ["send-keys", "-t", "work", "echo marker-$((1+2))", "Enter"], ["new-window", "-t", "work", "-n", "two"], + ["list-windows", "-t", "work", "-F", "#{window_index}:#{window_name}:#{window_flags}"], + ["rename-session", "-t", "work", "renamed"], ["list-sessions", "-F", "#{session_name}"], + ["split-window", "-t", "renamed:two", "-h"], ["list-panes", "-t", "renamed:two", "-F", "#{pane_index}:#{pane_width}"], + ["resize-pane", "-t", "renamed:two.0", "-x", "20"], ["list-panes", "-t", "renamed:two", "-F", "#{pane_index}:#{pane_width}"], + ["set-buffer", "hello"], ["show-buffer"], ["list-buffers", "-F", "#{buffer_name}:#{buffer_size}"], + ["capture-pane", "-p", "-t", "renamed:0"], ["kill-window", "-t", "renamed:two"], + ["list-windows", "-t", "renamed"], ["display-message", "-p", "-t", "renamed", "#S"], + ["new-session", "-d", "-s", "work2", "-n", "w", "sleep 30"], ["list-windows", "-t", "work2", "-F", "#{window_name}"], + ["kill-session", "-t", "work2"], ["list-sessions", "-F", "#{session_name}"]): + time.sleep(0.15) + rc, out, err = h.cli(*cmd) + seen.append((" ".join(cmd), (rc, out.replace(h.home, "HOME"), err.replace(h.home, "HOME")))) + return seen + + +def sc_copy_mode(h): + seen = [] + client = h.attach() + client.wait_for("start", lambda t: "0:bash*" in t) + client.send(b"seq 1 60\r", 0.8) + client.send(PREFIX + b"[", 0.4) + seen.append(("enter", screen_body(client))) + client.send(b"\x1b[5~", 0.4) + seen.append(("page up", screen_body(client))) + client.send(b"\x02", 0.1) + client.send(b"q", 0.4) + client.send(PREFIX + b"[", 0.3) + client.send(b"\x00", 0.2) + client.send(UP + UP + UP, 0.3) + client.send(b"\x1bw", 0.4) + seen.append(("after copy", screen_body(client))) + rc, out, err = h.cli("show-buffer") + seen.append(("buffer", (rc, out, err))) + client.send(b"echo $((6*7))\r", 0.6) + client.send(PREFIX + b"]", 0.5) + seen.append(("paste", screen_body(client))) + return seen + + +def sc_config(h): + seen = [] + client = h.attach() + client.wait_for("start", lambda t: "1:bash*" in t) + seen.append(("status", client.text().split("\n")[-1].rstrip())) + client.send(b"\x01c", 0.7) + client.wait_for("new window by C-a c", lambda t: "2:bash*" in t) + seen.append(("prefix C-a c", client.text().split("\n")[-1].rstrip())) + client.send(b"\x1b[1;3D", 0.4) + seen.append(("M-Left", client.text().split("\n")[-1].rstrip())) + import re + seen.append(("show-messages", tuple(re.sub(r"\d\d:\d\d:\d\d ", "", x) if isinstance(x, str) else x for x in h.cli("show-messages")))) + return seen + + +CONFIG_TOML = '''status-right = "R" +automatic-rename = false +prefix = "C-a" +base-index = 1 +status-position = "top" +status-left = "<#S> " + +[keys.root] +"M-Left" = "previous-window" + +[keys.prefix] +"b" = false +"nosuchkey!!" = "new-window" +"x" = "nosuchcommand" +''' + + +def sc_errors(h): + seen = [] + real_cli = h.cli + h.cli = lambda *w: tuple(x.replace(h.tmp, "TMP") if isinstance(x, str) else x for x in real_cli(*w)) + seen.append(("unknown", h.cli("nosuchcommand"))) + seen.append(("ambiguous", h.cli("l"))) + seen.append(("no server", h.cli("list-sessions"))) + seen.append(("flags", h.cli("new-session", "-Z"))) + seen.append(("noterm", h.cli("new-session"))) + seen.append(("attach noterm", h.cli("attach-session"))) + return seen + + +def sc_native_flow(h): + """The control socket: both controllers (tai.py's and the C++ one) against this server.""" + seen = [] + client = h.attach() + client.wait_for("start", lambda t: "0:bash*" in t) + cpp_test = os.path.join(ROOT, "taicpp", "build", "mplex_flow_test") + helpers = [("py-controller", [sys.executable, os.path.join(ROOT, "taicpp", "tests", "mplex_native_oracle.py"), ROOT]), + ("cpp-controller", [cpp_test, "--native"])] + for label, argv in helpers: + for answer in ("go", "cancel"): + env = dict(h.env) + env["TAI_PLEXER_SOCKET"] = os.path.join(h.home, "plexer", "server.sock") + env["TAI_PLEXER_PANE"] = "0" + proc = subprocess.Popen(argv, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + client.wait_for("%s first dialog" % label, lambda t: "Environment" in t and "staging" in t) + seen.append((label + " " + answer + " dialog 1", screen_body(client))) + if answer == "cancel": + client.send(b"\x1b", 0.5) + else: + client.send(DOWN, 0.3) + seen.append((label + " " + answer + " moved", screen_body(client))) + client.send(b"\r", 0.6) + client.wait_for("text prompt", lambda t: "Tag" in t) + seen.append((label + " " + answer + " prompt", client.text().split("\n")[-1].rstrip())) + client.send(b"\x15x1\r", 0.6) + client.wait_for("actions", lambda t: "Deploy" in t) + seen.append((label + " " + answer + " actions", screen_body(client))) + client.send(b"\r", 0.6) + out, _ = proc.communicate(timeout=30) + seen.append((label + " " + answer + " output", out)) + client.send(b"\x15", 0.3) + return seen + + +SCENARIOS = [("lifecycle", sc_lifecycle, None), ("panes_windows", sc_panes_windows, None), ("commands", sc_commands, None), + ("copy_mode", sc_copy_mode, None), ("config", sc_config, CONFIG_TOML), ("native_flow", sc_native_flow, None), ("errors", sc_errors, None)] + + +def run_all(label, argv): + results = {} + for name, fn, config in SCENARIOS: + if ONLY and name != ONLY: + continue + harness = Harness(label, argv, config) + try: + results[name] = fn(harness) + except Exception as exc: # noqa: BLE001 + results[name] = [("EXCEPTION", repr(exc))] + finally: + harness.close() + return results + + +def main(): + python_argv = [sys.executable, ROOT + "/tai.py", "--mplex"] + oracle = run_all("python", python_argv) + if PYTHON_ONLY: + for name, rows in oracle.items(): + print("==", name) + for label, value in rows: + print("--", label, "\n", value) + return 0 + cpp = run_all("cpp", [CPP, "--mplex"]) + bad = 0 + for name in oracle: + a, b = oracle[name], cpp.get(name, []) + if a == b: + print("PASS", name, "(%d observations)" % len(a)) + continue + bad += 1 + print("FAIL", name) + for (la, va), (lb, vb) in zip(a, b): + if (la, va) != (lb, vb): + print(" step:", la, "/", lb) + print(" python:", json.dumps(va, default=str)[:1500]) + print(" cpp :", json.dumps(vb, default=str)[:1500]) + if len(a) != len(b): + print(" length differs: python %d, cpp %d" % (len(a), len(b))) + print("%d of %d scenarios identical" % (len(oracle) - bad, len(oracle))) + return 1 if bad else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/taicpp/tests/mplex_flow_gen.py b/taicpp/tests/mplex_flow_gen.py new file mode 100644 index 0000000..371638a --- /dev/null +++ b/taicpp/tests/mplex_flow_gen.py @@ -0,0 +1,172 @@ +"""Corpus generator for mplex_flow_test.cpp: runs the real tai.py validate_flow_spec and +PlexerFlow over generated specs and writes the results as JSON lines. + +usage: python3 -I mplex_flow_gen.py +""" +import copy +import importlib.util +import json +import random +import sys + +root, out_path = sys.argv[1], sys.argv[2] +spec = importlib.util.spec_from_file_location("tai_mod", root + "/tai.py") +tai = importlib.util.module_from_spec(spec) +sys.modules["tai_mod"] = tai +spec.loader.exec_module(tai) + +rng = random.Random(20260101) + +BASES = [ + {"v": 1, "title": "Deploy", "timeout": 300, "start": "main", "screens": [ + {"id": "main", "title": "Where", "body": "Pick a target.", + "fields": [ + {"id": "env", "type": "choice", "label": "Environment", "options": ["staging", "prod"], "default": 0}, + {"id": "tag", "type": "text", "label": "Tag", "initial": "v1", "show_if": {"env": "prod"}}, + {"id": "note", "type": "info", "text": "Prod deploys page the team."}], + "actions": [{"id": "go", "label": "Deploy"}, {"id": "back", "label": "Back"}], + "next": {"go": "$done", "back": "$cancel"}}]}, + {"screens": [{"id": "a", "body": "hello"}, {"id": "b", "title": "B", "fields": [ + {"id": "x", "type": "choice", "options": [{"label": "One", "value": 1}, {"label": "Two", "value": 2.5}, "three"]}, + {"id": "y", "type": "text", "label": "Name é中", "show_if": {"x": [2.5, "three"]}}], + "next": "c"}, {"id": "c", "title": "C", "show_if": {"x": 1}, "body": "only one"}, + {"id": "d", "actions": [{"id": "ok", "label": "OK"}, {"id": "again", "label": "Again"}], + "next": {"again": "a"}}]}, + {"title": "T", "timeout": 5, "screens": [{"id": "only", "fields": [{"id": "i", "type": "info", "text": "just info"}]}]}, + {"start": "second", "screens": [{"id": "first", "body": "1"}, {"id": "second", "body": "2", "next": "$cancel"}]}, +] + +ODD = [None, True, False, 0, 1, -3, 7, 1.0, 2.5, "", "x", " ", "a b", "$done", "$cancel", [], {}, [1], ["a"], {"a": 1}, "é" * 80, "x" * 65, "y" * 201, "z" * 2001] + + +def walk(node, path=()): + yield node, path + if isinstance(node, dict): + for key, value in list(node.items()): + yield from walk(value, path + (key,)) + elif isinstance(node, list): + for index, value in enumerate(node): + yield from walk(value, path + (index,)) + + +def at(node, path): + for step in path: + node = node[step] + return node + + +def mutate(base): + spec = copy.deepcopy(base) + for _ in range(rng.choice([1, 1, 1, 1, 2, 3])): + places = [path for _node, path in walk(spec) if path] + if not places: + break + path = rng.choice(places) + parent = at(spec, path[:-1]) + key = path[-1] + action = rng.choice(["replace", "delete", "dupe", "replace"]) + if action == "delete": + if isinstance(parent, dict): + del parent[key] + else: + parent.pop(key) + elif action == "dupe" and isinstance(parent, list) and parent: + parent.append(copy.deepcopy(parent[rng.randrange(len(parent))])) + else: + parent[key] = copy.deepcopy(rng.choice(ODD)) + return spec + + +def run_validate(candidate): + try: + app, err = tai.validate_flow_spec(copy.deepcopy(candidate)) + except Exception: # noqa: BLE001 python crashes are not part of the contract + return None + return app, err + + +class Script(tai.PlexerController): + kind = "script" + + def __init__(self, replies): + self.replies = list(replies) + self.asks = [] + + def ask(self, screen, timeout): + self.asks.append(screen) + if not self.replies: + return {"status": "cancelled"} + return self.replies.pop(0) + + +def random_replies(): + out = [] + for _ in range(rng.randrange(0, 70)): + roll = rng.random() + if roll < 0.04: + out.append({"status": "cancelled"}) + elif roll < 0.07: + out.append({"status": "timeout"}) + elif roll < 0.55: + out.append({"status": "ok", "value": rng.randrange(0, 3)}) + else: + out.append({"status": "ok", "value": rng.choice(["", "v2", "café", True])}) + return out + + +def fix_replies(app, replies): + """Index replies must be valid for the screen they answer; replay with a guard controller.""" + guard = [] + + class Guard(tai.PlexerController): + kind = "guard" + + def __init__(self): + self.todo = list(replies) + + def ask(self, screen, timeout): + if not self.todo: + reply = {"status": "cancelled"} + else: + reply = self.todo.pop(0) + if reply.get("status") == "ok": + if screen["kind"] == "choice": + reply = {"status": "ok", "value": reply["value"] % len(screen["options"]) if isinstance(reply["value"], int) and not isinstance(reply["value"], bool) else 0} + elif screen["kind"] == "text": + reply = {"status": "ok", "value": reply["value"] if isinstance(reply["value"], str) else "t"} + else: + reply = {"status": "ok", "value": True} + guard.append(reply) + return reply + + return Guard(), guard + + +lines = [] +cases = 0 +seen_specs = [] +for base in BASES: + seen_specs.append(base) +for _ in range(4000): + seen_specs.append(mutate(rng.choice(BASES))) +for _ in range(60): + seen_specs.append(rng.choice(ODD)) +for candidate in seen_specs: + result = run_validate(candidate) + if result is None: + continue + app, err = result + lines.append(json.dumps({"kind": "spec", "spec": candidate, "app": app, "err": err})) + cases += 1 + if app is not None: + for _round in range(3): + controller, used = fix_replies(app, random_replies()) + try: + outcome = tai.PlexerFlow(copy.deepcopy(app)).run(controller) + except Exception: # noqa: BLE001 + continue + lines.append(json.dumps({"kind": "flow", "app": app, "replies": used, "outcome": outcome})) + cases += 1 +with open(out_path, "w") as handle: + handle.write("\n".join(lines) + "\n") +print("cases", cases) diff --git a/taicpp/tests/mplex_flow_test.cpp b/taicpp/tests/mplex_flow_test.cpp new file mode 100644 index 0000000..07e7aec --- /dev/null +++ b/taicpp/tests/mplex_flow_test.cpp @@ -0,0 +1,167 @@ +// Differential test: validate_flow_spec and PlexerFlow.run against tai.py. +// usage: mplex_flow_test +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "plexer_flow.h" + +using detz::mplex::Json; +using namespace detz::plexer; + +namespace { + +class Script : public PlexerController { + public: + explicit Script(std::vector replies_in) : replies(std::move(replies_in)) { kind = "script"; } + std::vector replies; + size_t next = 0; + Json send_keys(const std::string&, bool) override { return Json(); } + Json read_pane(int) override { return Json(); } + Json status() override { return Json(); } + Json ask(const Json&, double) override { + if (next >= replies.size()) { + Json out = Json::object(); + out.set("status", Json::string("cancelled")); + return out; + } + return replies[next++]; + } +}; + +} // namespace + +// Real tmux 3.6 round trip (tmux is the oracle): prints one line per observation so the same +// steps run through tai.py's TmuxPlexerController can be diffed (see mplex_tmux_oracle.py). +int tmux_mode() { + char dir[] = "/tmp/mplex-tmux-XXXXXX"; + if (!mkdtemp(dir)) return 2; + setenv("TMUX_TMPDIR", dir, 1); + unsetenv("TMUX"); + if (std::system("tmux new-session -d -s t -x 80 -y 24 'cat' >/dev/null 2>&1") != 0) return 2; + FILE* pipe = popen("tmux list-panes -t t -F '#{pane_id}'", "r"); + char buf[64] = {0}; + if (!pipe || !std::fgets(buf, sizeof buf, pipe)) return 2; + pclose(pipe); + std::string pane(buf); + while (!pane.empty() && (pane.back() == '\n' || pane.back() == '\r')) pane.pop_back(); + TmuxPlexerController c(pane); + std::printf("send %s\n", c.send_keys("hello world", true).dumps().c_str()); + usleep(300000); + std::string content = c.read_pane(5).as_string(); + std::printf("read %s\n", Json::string(content).dumps().c_str()); + std::printf("status %s\n", c.status().dumps().c_str()); + try { + TmuxPlexerController bad("%99"); + bad.read_pane(3); + } catch (const PlexerError& e) { + std::printf("error %s\n", e.what()); + } + Json screen = Json::object(); + screen.set("kind", Json::string("weird")); + try { + c.ask(screen, 5); + } catch (const PlexerError& e) { + std::printf("ask %s\n", e.what()); + } + screen.set("kind", Json::string("choice")); + screen.set("title", Json::string("T")); + screen.set("options", Json::loads("[\"a\", \"b\"]")); + try { + std::printf("choice %s\n", c.ask(screen, 5).dumps().c_str()); + } catch (const PlexerError& e) { + std::printf("choice error %s\n", e.what()); + } + screen.set("kind", Json::string("text")); + screen.set("text", Json::string("question")); + try { + std::printf("text %s\n", c.ask(screen, 5).dumps().c_str()); + } catch (const PlexerError& e) { + std::printf("text error %s\n", e.what()); + } + if (std::system("tmux kill-server >/dev/null 2>&1") != 0) return 3; + return 0; +} + +// Drives the native controller against a running mplex server (env TAI_PLEXER_SOCKET/PANE): +// status keys, send, read, then a whole PlexerFlow app answered by the user at the terminal. +int native_mode() { + auto controller = PlexerControllerFactory::detect(); + if (!controller) { + std::puts("no controller"); + return 2; + } + Json status = controller->status(); + std::string keys; + for (const auto& item : status.as_object()) keys += item.first + ","; + std::printf("status-keys %s kind %s\n", keys.c_str(), controller->kind.c_str()); + std::printf("send %s\n", controller->send_keys("echo native-$((20+22))", true).dumps().c_str()); + usleep(600000); + std::string tail = controller->read_pane(30).as_string(); + std::printf("read-has-marker %d\n", tail.find("native-42") != std::string::npos); + Json spec = Json::loads( + "{\"title\": \"Deploy\", \"timeout\": 30, \"screens\": [{\"id\": \"main\", \"title\": \"Where\", \"body\": \"Pick.\"," + " \"fields\": [{\"id\": \"env\", \"type\": \"choice\", \"label\": \"Environment\", \"options\": [\"staging\", \"prod\"]}," + " {\"id\": \"tag\", \"type\": \"text\", \"label\": \"Tag\", \"initial\": \"v1\", \"show_if\": {\"env\": \"prod\"}}]," + " \"actions\": [{\"id\": \"go\", \"label\": \"Deploy\"}, {\"id\": \"back\", \"label\": \"Back\"}]}]}"); + auto [app, err] = validate_flow_spec(spec); + if (!err.empty()) { + std::printf("spec error %s\n", err.c_str()); + return 2; + } + std::fflush(stdout); + PlexerFlow flow(app); + std::printf("outcome %s\n", flow.run(*controller).dumps().c_str()); + return 0; +} + +int main(int argc, char** argv) { + if (argc >= 2 && std::strcmp(argv[1], "--tmux") == 0) return tmux_mode(); + if (argc >= 2 && std::strcmp(argv[1], "--native") == 0) return native_mode(); + if (argc < 2) { + std::fputs("usage: mplex_flow_test corpus.jsonl\n", stderr); + return 2; + } + std::ifstream in(argv[1]); + std::string line; + int specs = 0, flows = 0, accepted = 0, failures = 0; + while (std::getline(in, line)) { + if (line.empty()) continue; + Json item = Json::loads(line); + const std::string kind = item.get("kind")->as_string(); + if (kind == "spec") { + ++specs; + auto [app, err] = validate_flow_spec(*item.get("spec")); + const Json& want_app = *item.get("app"); + const std::string& want_err = item.get("err")->as_string(); + std::string got_app = app.is_null() ? "null" : app.dumps(); + if (!app.is_null()) ++accepted; + if (got_app != want_app.dumps() || err != want_err) { + ++failures; + if (failures < 15) + std::cerr << "SPEC MISMATCH\n spec: " << item.get("spec")->dumps() << "\n want: " << want_err << " | " + << want_app.dumps() << "\n got: " << err << " | " << got_app << "\n"; + } + } else if (kind == "flow") { + ++flows; + std::vector replies = item.get("replies")->as_array(); + Script script(replies); + PlexerFlow flow(*item.get("app")); + Json outcome = flow.run(script); + if (outcome.dumps() != item.get("outcome")->dumps()) { + ++failures; + if (failures < 15) + std::cerr << "FLOW MISMATCH\n app: " << item.get("app")->dumps() << "\n want: " << item.get("outcome")->dumps() + << "\n got: " << outcome.dumps() << "\n"; + } + } + } + std::printf("spec cases %d (accepted %d), flow cases %d, failures %d\n", specs, accepted, flows, failures); + return failures ? 1 : 0; +} diff --git a/taicpp/tests/mplex_native_oracle.py b/taicpp/tests/mplex_native_oracle.py new file mode 100644 index 0000000..3418a9f --- /dev/null +++ b/taicpp/tests/mplex_native_oracle.py @@ -0,0 +1,28 @@ +"""Same steps as `mplex_flow_test --native`, through tai.py's own NativePlexerController and PlexerFlow. + +usage: python3 -I mplex_native_oracle.py (env TAI_PLEXER_SOCKET / TAI_PLEXER_PANE set) +""" +import importlib.util +import json +import sys +import time + +root = sys.argv[1] +spec = importlib.util.spec_from_file_location("tai_mod", root + "/tai.py") +tai = importlib.util.module_from_spec(spec) +sys.modules["tai_mod"] = tai +spec.loader.exec_module(tai) +controller = tai.PlexerControllerFactory.detect() +status = controller.status() +print("status-keys %s kind %s" % ("".join(key + "," for key in status), controller.kind)) +print("send", json.dumps(controller.send_keys("echo native-$((20+22))", True))) +time.sleep(0.6) +print("read-has-marker", int("native-42" in controller.read_pane(30))) +app, err = tai.validate_flow_spec({ + "title": "Deploy", "timeout": 30, "screens": [{ + "id": "main", "title": "Where", "body": "Pick.", + "fields": [{"id": "env", "type": "choice", "label": "Environment", "options": ["staging", "prod"]}, + {"id": "tag", "type": "text", "label": "Tag", "initial": "v1", "show_if": {"env": "prod"}}], + "actions": [{"id": "go", "label": "Deploy"}, {"id": "back", "label": "Back"}]}]}) +sys.stdout.flush() +print("outcome", json.dumps(tai.PlexerFlow(app).run(controller))) diff --git a/taicpp/tests/mplex_tmux_oracle.py b/taicpp/tests/mplex_tmux_oracle.py new file mode 100644 index 0000000..267cade --- /dev/null +++ b/taicpp/tests/mplex_tmux_oracle.py @@ -0,0 +1,42 @@ +"""Oracle for `mplex_flow_test --tmux`: the same steps through tai.py's TmuxPlexerController +against a real tmux (isolated TMUX_TMPDIR), printing the same lines. + +usage: python3 -I mplex_tmux_oracle.py +""" +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import time + +root = sys.argv[1] +spec = importlib.util.spec_from_file_location("tai_mod", root + "/tai.py") +tai = importlib.util.module_from_spec(spec) +sys.modules["tai_mod"] = tai +spec.loader.exec_module(tai) +os.environ["TMUX_TMPDIR"] = tempfile.mkdtemp(prefix="mplex-tmux-") +os.environ.pop("TMUX", None) +subprocess.run("tmux new-session -d -s t -x 80 -y 24 'cat' >/dev/null 2>&1", shell=True) +pane = subprocess.run("tmux list-panes -t t -F '#{pane_id}'", shell=True, capture_output=True, text=True).stdout.strip() +c = tai.TmuxPlexerController(pane) +print("send", json.dumps(c.send_keys("hello world", True))) +time.sleep(0.3) +print("read", json.dumps(c.read_pane(5))) +print("status", json.dumps(c.status())) +try: + tai.TmuxPlexerController("%99").read_pane(3) +except tai.PlexerError as exc: + print("error", exc) +try: + c.ask({"kind": "weird"}, 5) +except tai.PlexerError as exc: + print("ask", exc) +for kind in ("choice", "text"): + screen = {"kind": kind, "title": "T", "options": ["a", "b"], "text": "question"} + try: + print(kind, json.dumps(c.ask(screen, 5))) + except tai.PlexerError as exc: + print(kind, "error", exc) +subprocess.run("tmux kill-server >/dev/null 2>&1", shell=True)