tai/taicpp/tests/orch_test.cpp

// Orchestration: formatter, panel, board over a fake transport, CAP adapter, spec constants.
#include <atomic>
#include <chrono>
#include <cstdio>
#include <fstream>
#include <sstream>
#include <string>
#include <thread>
#include <vector>

#include "orch_board.h"
#include "orch_cap_adapter.h"
#include "orch_formatter.h"
#include "orch_panel.h"
#include "orch_policy.h"

using namespace detz::orch;
using Json = detz::wire::Json;

static int failures = 0;
#define CHECK(c) do { if (!(c)) { std::printf("FAIL %s:%d %s\n", __FILE__, __LINE__, #c); ++failures; } } while (0)

static double now_s() { return std::chrono::duration<double>(std::chrono::steady_clock::now().time_since_epoch()).count(); }

struct FakeTransport : OrchTransport {
  std::vector<std::vector<OrchOp>> writes;
  std::vector<int> opened;
  std::vector<int> to_close;          // ids reported as closed on the next write
  int next_id = 10;
  int max_open = 1000;
  bool fail_writes = false;
  bool available() const override { return true; }
  std::optional<int> open(const std::string&, const std::string&, std::string& error, int*) override {
    if (static_cast<int>(opened.size()) >= max_open) { error = "no space for another feed pane"; return std::nullopt; }
    opened.push_back(next_id);
    return next_id++;
  }
  bool write(const std::vector<OrchOp>& ops, std::vector<int>& closed, std::string& error, std::vector<std::pair<int, int>>*) override {
    if (fail_writes) { error = "down"; return false; }
    writes.push_back(ops);
    closed = to_close;
    to_close.clear();
    return true;
  }
  void close(int) override {}
};

static std::string all_data(const FakeTransport& t, int id) {
  std::string s;
  for (const auto& w : t.writes) for (const auto& op : w) if (op.id == id) s += op.data;
  return s;
}

int main() {
  // ---- formatter
  CHECK(OrchFormatter::sanitize("a\x1b[31mred\x1b[0m\x1b[2Jb\x1b]0;title\x07 c\x01\r") == "a\x1b[31mred\x1b[0mb c");
  CHECK(OrchFormatter::sanitize("x\x1b[10;5Hy") == "xy");           // cursor movement dropped
  CHECK(OrchFormatter::crlf("a\nb") == "a\r\nb");
  CHECK(OrchFormatter::clip("h\xC3\xA9llo world", 5) == "h\xC3\xA9llo...");   // by character, not byte
  CHECK(OrchFormatter::one_line("a \n\t b  c ") == "a b c");
  {
    OrchFormatter f;
    std::string t = f.render(OrchEvent::text_of("partial"));
    bool held = f.holding();
    std::string c = f.render(OrchEvent::tool_start("shell", "{\"command\":\"ls\"}"));
    CHECK(t.empty() && held);                                  // the trailing word is held: more of it may come
    CHECK(c.rfind("partial\r\n", 0) == 0);                            // a tool line flushes it and starts on a new line
    std::string r = f.render(OrchEvent::tool_end("shell", "l1\nl2\nl3\nl4\n", false, 5));
    CHECK(r.find("l3") != std::string::npos && r.find("l4") == std::string::npos);   // three lines
    CHECK(r.find("chars") != std::string::npos);
    CHECK(f.render(OrchEvent::error_of("boom")).find("boom") != std::string::npos);
    CHECK(f.render(OrchEvent::usage_of(5)).empty());
  }

  // ---- word wrap
  {
    OrchFormatter f;
    f.set_width(20);
    std::string out = f.render(OrchEvent::text_of("alpha beta gamma delta epsilon "));
    CHECK(out.find("alpha beta gamma\r\ndelta") != std::string::npos || out.find("\r\n") != std::string::npos);
    CHECK(out.find("gam\r\nma") == std::string::npos);                // never inside a word
    OrchFormatter g;
    g.set_width(20);
    std::string a = g.render(OrchEvent::text_of("tok1 tok2 to"));
    std::string b = g.render(OrchEvent::text_of("ken3 tok4 "));
    CHECK((a + b).find("token3") != std::string::npos);                // a word split across chunks is rejoined
    CHECK(g.settle().empty());
    OrchFormatter h;
    h.set_width(12);
    std::string longword = h.render(OrchEvent::text_of(std::string(30, 'z') + " end "));
    CHECK(longword.find("\r\n") != std::string::npos);                // a word longer than the line is broken
    std::string wide = OrchFormatter().render(OrchEvent::text_of("\xE2\x9C\x93\xE2\x9C\x93 \xE2\x9C\x93 "));
    CHECK(wide.find("\xE2\x9C\x93\xE2\x9C\x93") != std::string::npos);   // multibyte kept whole
  }

  // ---- CAP adapter
  {
    OrchCapAdapter a;
    Json start = Json::object();
    start.set("kind", "cap.tool_call.start"); start.set("call_id", "c1"); start.set("name", "shell"); start.set("input", Json::object());
    auto s = a.map(start);
    CHECK(s && s->kind == EventKind::ToolStart && s->name == "shell");
    Json end = Json::object();
    end.set("kind", "cap.tool_call.end"); end.set("call_id", "c1"); end.set("output", "ok"); end.set("is_error", true); end.set("duration_ms", 7);
    auto e = a.map(end);
    CHECK(e && e->kind == EventKind::ToolEnd && e->name == "shell" && e->failed && e->millis == 7);
    Json txt = Json::object();
    txt.set("kind", "cap.text_chunk"); txt.set("text", "hi"); txt.set("channel", "thought");
    auto t = a.map(txt);
    CHECK(t && t->kind == EventKind::Thought && t->text == "hi\n");
    Json use = Json::object();
    use.set("kind", "cap.usage"); use.set("input_tokens", 10); use.set("output_tokens", 5);
    CHECK(a.map(use) && a.map(use)->tokens == 15);
    Json other = Json::object();
    other.set("kind", "cap.fs.change");
    CHECK(!a.map(other));
  }

  // ---- panel: buffering, ceiling, watchdog, cancel
  {
    OrchPanel p("agent #1", "do the thing", 5);
    p.emit(OrchEvent::text_of("hello"));
    p.emit(OrchEvent::tool_start("shell", "x"));
    p.emit(OrchEvent::usage_of(1500));
    CHECK(p.tools() == 1 && p.tokens() == 1500);
    CHECK(p.header().find("agent #1") != std::string::npos && p.header().find("1.5k tok") != std::string::npos);
    OrchOp op;
    CHECK(p.collect(op) && op.id == 5 && op.data.find("do the thing") != std::string::npos && op.data.find("hello") != std::string::npos);
    CHECK(op.header && op.state && *op.state == "running");
    CHECK(!p.collect(op));                                           // quiet panel sends nothing
    std::atomic<int> fired{0};
    std::string why;
    p.set_cancel([&](const std::string& r) { why = r; ++fired; });
    p.watchdog(now_s() + ORCH_IDLE_WARN + 1);
    CHECK(p.state() == OrchPanel::State::Idle && fired == 0);
    p.emit(OrchEvent::text_of("again"));
    CHECK(p.state() == OrchPanel::State::Running);                   // an event is a heartbeat
    p.watchdog(now_s() + ORCH_IDLE_KILL + 1);
    CHECK(fired == 1 && why.find("no output") != std::string::npos);
    p.watchdog(now_s() + ORCH_IDLE_KILL + 2);
    CHECK(fired == 1);                                               // fires once
    p.finish(OrchPanel::State::Done, "42 chars");
    CHECK(p.finished());
    p.emit(OrchEvent::text_of("late"));                              // ignored after finish
    OrchOp last;
    CHECK(p.collect(last) && last.done && last.linger == ORCH_LINGER && last.data.find("late") == std::string::npos);
    CHECK(last.data.find("done") != std::string::npos && p.drained());
  }
  {
    OrchPanel flood("flood", "t", 1);
    std::string line(1000, 'x');
    line += "\n";
    for (int i = 0; i < 600; ++i) flood.emit(OrchEvent::text_of(line));
    OrchOp op;
    CHECK(flood.collect(op));
    CHECK(flood.dropped_bytes() > 0);                                 // over the ceiling: dropped loudly
    CHECK(op.data.size() <= ORCH_WRITE_MAX);
  }
  {
    OrchPanel hidden("hidden", "t", -1);                              // no pane behind it
    hidden.emit(OrchEvent::text_of("x"));
    OrchOp op;
    CHECK(!hidden.collect(op) && !hidden.shown());
    std::atomic<int> fired{0};
    hidden.set_cancel([&](const std::string&) { ++fired; });
    hidden.watchdog(now_s() + ORCH_IDLE_KILL + 1);
    CHECK(fired == 1);                                                // supervised although not shown
  }
  {
    OrchPanel utf("u", "t", 1);
    std::string emoji;
    for (int i = 0; i < 20000; ++i) emoji += "\xE2\x9C\x93";          // 3-byte characters across the write limit
    utf.emit(OrchEvent::text_of(emoji));
    OrchOp op;
    CHECK(utf.collect(op));
    CHECK(op.data.size() <= ORCH_WRITE_MAX && (op.data.size() % 3 == 0 || op.data.find("\xE2\x96\xB8") != std::string::npos));
    unsigned char first_next = 0;
    OrchOp op2;
    if (utf.collect(op2) && !op2.data.empty()) first_next = static_cast<unsigned char>(op2.data[0]);
    CHECK((first_next & 0xC0) != 0x80);                               // never splits a character
  }

  // ---- board over a fake transport
  {
    auto fake = std::make_unique<FakeTransport>();
    FakeTransport* t = fake.get();
    OrchBoard board(std::move(fake));
    auto a = board.open("agent #1", "task a");
    auto b = board.open("agent #2", "task b");
    CHECK(a->shown() && b->shown() && board.shown() == 2);
    for (int i = 0; i < 200; ++i) {
      a->emit(OrchEvent::text_of("tok" + std::to_string(i) + " "));
      if (i % 50 == 0) std::this_thread::sleep_for(std::chrono::milliseconds(5));
    }
    b->emit(OrchEvent::text_of("from b"));
    a->finish(OrchPanel::State::Done, "ok");
    std::this_thread::sleep_for(std::chrono::milliseconds(200));       // the held word is settled after a pause
    board.flush(2.0);
    std::string da = all_data(*t, a->remote_id());
    CHECK(da.find("tok0 ") != std::string::npos && da.find("tok199 ") != std::string::npos);
    CHECK(da.find("tok5 ") < da.find("tok6 "));                       // order kept
    CHECK(t->writes.size() < 120);                                    // coalesced: far fewer writes than events
    CHECK(all_data(*t, b->remote_id()).find("from b") != std::string::npos);
    bool done_op = false;
    for (const auto& w : t->writes) for (const auto& op : w) if (op.id == a->remote_id() && op.done) done_op = true;
    CHECK(done_op);
    // the user closes b's pane: the board learns it from the write reply and cancels the job
    std::atomic<int> fired{0};
    b->set_cancel([&](const std::string&) { ++fired; });
    t->to_close.push_back(b->remote_id());
    b->emit(OrchEvent::text_of("more"));
    for (int i = 0; i < 100 && fired == 0; ++i) std::this_thread::sleep_for(std::chrono::milliseconds(10));
    CHECK(fired == 1 && b->state() == OrchPanel::State::Cancelled);
    board.stop();
  }
  {
    auto fake = std::make_unique<FakeTransport>();
    fake->max_open = 1;
    OrchBoard board(std::move(fake), false);
    auto a = board.open("a", "t");
    auto b = board.open("b", "t");                                    // no space: still a usable panel
    CHECK(a->shown() && !b->shown() && !board.last_error().empty());
    b->emit(OrchEvent::text_of("works anyway"));
    b->finish(OrchPanel::State::Done, "ok");
    CHECK(b->finished());
  }
  {
    auto fake = std::make_unique<FakeTransport>();
    fake->fail_writes = true;
    OrchBoard board(std::move(fake), false);
    auto a = board.open("a", "t");
    for (int i = 0; i < 5; ++i) { a->emit(OrchEvent::text_of("x ")); board.flush(0.05); }
    CHECK(!board.transport_ok());                                     // given up after repeated failures
    CHECK(board.last_error().find("unreachable") != std::string::npos);
    CHECK(!board.open("b", "t")->shown());
  }
  {
    auto fake = std::make_unique<FakeTransport>();
    OrchBoard board(std::move(fake), false);
    std::vector<std::shared_ptr<OrchPanel>> keep;
    for (int i = 0; i < ORCH_PANELS_MAX + 2; ++i) keep.push_back(board.open("p" + std::to_string(i), "t"));
    CHECK(board.shown() == static_cast<size_t>(ORCH_PANELS_MAX));    // the ceiling
    CHECK(!keep.back()->shown());
  }

  // ---- the spec lists every constant of orch_policy.h and mplex_vpane.h
  {
    std::ifstream in("docs/orchestration.md");
    std::stringstream ss;
    ss << in.rdbuf();
    std::string doc = ss.str();
    CHECK(!doc.empty());
    for (const char* name : {"ORCH_PANELS_MAX", "ORCH_FLUSH_INTERVAL", "ORCH_MONITOR_INTERVAL", "ORCH_PENDING_MAX", "ORCH_WRITE_MAX",
                             "ORCH_LINGER", "ORCH_IDLE_WARN", "ORCH_IDLE_KILL", "ORCH_CANCEL_GRACE", "ORCH_TASK_SHOW_MAX",
                             "ORCH_TOOL_ARGS_MAX", "ORCH_TOOL_RESULT_LINES", "ORCH_TOOL_RESULT_COLS", "ORCH_LABEL_MAX",
                             "ORCH_CALL_TIMEOUT_MS", "MPLEX_VPANE_MAX", "MPLEX_VPANE_MIN_ROWS", "MPLEX_VPANE_MIN_COLS",
                             "MPLEX_VPANE_MAIN_PERCENT", "MPLEX_VPANE_OPS_MAX", "MPLEX_VPANE_WRITE_MAX", "MPLEX_VPANE_HEADER_MAX",
                             "MPLEX_VPANE_LEASE", "MPLEX_VPANE_ORPHAN_LINGER", "MPLEX_VPANE_LINGER_MAX"}) {
      if (doc.find(name) == std::string::npos) { std::printf("FAIL docs/orchestration.md lacks %s\n", name); ++failures; }
    }
    for (const char* cmd : {"vpane_open", "vpane_write", "vpane_close", "vpane_list"})
      if (doc.find(cmd) == std::string::npos) { std::printf("FAIL docs/orchestration.md lacks %s\n", cmd); ++failures; }
  }

  std::printf(failures ? "FAILED %d\n" : "orch_test OK\n", failures);
  return failures ? 1 : 0;
}