diff --git a/taicpp/src/agent/agent.cpp b/taicpp/src/agent/agent.cpp index 7aadc34..d1c207c 100644 --- a/taicpp/src/agent/agent.cpp +++ b/taicpp/src/agent/agent.cpp @@ -6,6 +6,7 @@ #include "consts.h" #include "data.h" #include "text.h" +#include "tui_control.h" #include #include #include @@ -53,7 +54,7 @@ bool tool_env_ok(const std::string& name) { return ports().plexer != nullptr && ports().plexer->detect() != nullptr; } if (name == "tui_screen" || name == "tui_keys" || name == "tui_type" || name == "tui_click" || name == "tui_goto" || name == "tui_result") { - return ports().tui != nullptr && ports().tui->control_exists(); + return ports().tui != nullptr ? ports().tui->control_exists() : path_exists(grr_control_path()); } return true; } diff --git a/taicpp/src/agent/ports.h b/taicpp/src/agent/ports.h index 6de35e8..07ec663 100644 --- a/taicpp/src/agent/ports.h +++ b/taicpp/src/agent/ports.h @@ -302,12 +302,40 @@ class OpsPort { virtual std::tuple do_release(const std::string& part, const std::string& message, const std::string& actor) = 0; virtual std::string collect_sysinfo(const std::optional>& checks) = 0; - virtual std::vector sysinfo_check_names() = 0; virtual std::string edge_synthesize(const std::string& text, const std::string& voice) = 0; // audio bytes; throws llm::BackendError or std::runtime_error virtual std::string play_audio(const std::string& path) = 0; virtual std::string listen_audio(int seconds, const std::string& audio_dir) = 0; }; +// --------------------------------------------------------------------------- +// Scheduler storage: the five statements scheduler_tick runs on the schedules table. +// --------------------------------------------------------------------------- +struct DueSchedule { + int64_t id{0}; + std::string prompt, profile; + int64_t every_sec{0}; + std::string next_run; + int64_t timeout{0}; + std::string last_status, updated; +}; +struct SealFailure : public std::runtime_error { using std::runtime_error::runtime_error; }; // SealError +class SchedulerDbPort { + public: + virtual ~SchedulerDbPort() = default; + // SELECT ... WHERE status = 'pending' AND next_run <= ? ORDER BY next_run + virtual std::vector select_due(const std::string& now_iso) = 0; + // UPDATE schedules SET next_run = ?, updated = ?, last_status = 'running' WHERE id = ? AND status = 'pending' AND next_run = ?; true when rowcount == 1 + virtual bool claim_every(int64_t id, const std::string& new_next, const std::string& stamp, const std::string& expected_next) = 0; + // UPDATE schedules SET status = 'done', updated = ? WHERE id = ? AND status = 'pending'; true when rowcount == 1 + virtual bool claim_once(int64_t id, const std::string& stamp) = 0; + // Seal.unlock for sealed prompts (Seal.PREFIX), identity otherwise; throws SealFailure. + virtual std::string unlock_prompt(const std::string& prompt) = 0; + virtual void release_every(int64_t id, const std::string& next_run, const std::string& stamp, const std::string& last_status) = 0; + virtual void release_once(int64_t id, const std::string& stamp) = 0; // status back to 'pending' + // UPDATE schedules SET last_status = ?, last_result = ?[:2000], updated = now WHERE id = ? + virtual void note_result(int64_t id, const std::string& status, const std::string& result) = 0; +}; + // Process-wide port registry; unset ports are null and the code that needs // them returns/throws the "port not wired" error instead of pretending. struct Ports { @@ -319,6 +347,7 @@ struct Ports { PlexerPort* plexer{nullptr}; TuiControlPort* tui{nullptr}; OpsPort* ops{nullptr}; + SchedulerDbPort* scheduler_db{nullptr}; }; Ports& ports(); diff --git a/taicpp/src/agent/scheduler.cpp b/taicpp/src/agent/scheduler.cpp new file mode 100644 index 0000000..1d2b8ac --- /dev/null +++ b/taicpp/src/agent/scheduler.cpp @@ -0,0 +1,126 @@ +#include "scheduler.h" +#include "../llm/helpers.h" +#include "../llm/py.h" +#include "../llm/term.h" +#include "consts.h" +#include +#include +namespace detz::agent { + +void note_schedule_result(SchedulerDbPort& db, int64_t row_id, const std::string& status, const std::string& result) { + try { + db.note_result(row_id, status, py::head(result, 2000)); + } catch (const std::exception&) { // sqlite3.Error + } +} + +void watch_schedule(SchedulerDbPort& db, int64_t row_id, int agent_id, double timeout, const std::string& profile) { + auto [status, text] = poll_agent(agent_id, timeout + SCHEDULER_LEASE_MARGIN); + note_schedule_result(db, row_id, status, text); + json::Value meta = json::Value::object(); + meta.set("id", row_id); + meta.set("status", status); + meta.set("result_chars", static_cast(py::len(text))); + llm::log_event("INFO", "sched", "finished", "schedule #" + std::to_string(row_id) + " finished: " + status, profile, "", meta); +} + +bool schedule_running(const std::string& last_status, const std::string& updated, int64_t timeout) { + if (last_status != "running") { return false; } + double age; + try { + age = py::now() - py::to_epoch(py::fromisoformat(updated)); + } catch (const std::invalid_argument&) { + return false; + } + return age < static_cast(timeout ? timeout : 600) + SCHEDULER_LEASE_MARGIN; +} + +int scheduler_tick(AgentConfig& config, SchedulerDbPort& db, const Runner& runner) { + int fired = 0, skipped_cap = 0, skipped_overlap = 0; + std::string now_iso = py::utc_now_iso(); + std::vector rows = db.select_due(now_iso); + for (const auto& row : rows) { + if (fired >= SCHEDULER_MAX_FIRES) { break; } + if (row.every_sec && schedule_running(row.last_status, row.updated, row.timeout)) { ++skipped_overlap; continue; } + if (running_agents() >= AGENTS_MAX_PARALLEL) { ++skipped_cap; continue; } + std::string stamp = py::utc_now_iso(); + bool done; + if (row.every_sec) { + py::DateTime claimed; + try { + claimed = py::fromisoformat(row.next_run); + } catch (const std::invalid_argument&) { + continue; + } + double e = py::to_epoch(claimed); + double now = py::now(); + // while claimed_next <= now: claimed_next += every_sec + if (e <= now) { + double steps = std::floor((now - e) / static_cast(row.every_sec)) + 1.0; + e += steps * static_cast(row.every_sec); + } + py::DateTime next = py::from_epoch(e, claimed.aware ? claimed.offset_seconds : 0, claimed.aware, claimed.micro); + done = db.claim_every(row.id, py::isoformat(next), stamp, row.next_run); + } else { + done = db.claim_once(row.id, stamp); + } + if (!done) { continue; } + std::string prompt = row.prompt; + try { + prompt = db.unlock_prompt(row.prompt); + } catch (const SealFailure& exc) { + note_schedule_result(db, row.id, "error: " + llm::short_error(exc)); + continue; + } + int agent_id; + try { + agent_id = spawn_agent(prompt, row.profile, static_cast(row.timeout), config, 0, runner); + } catch (const AgentsFull&) { + if (row.every_sec) { db.release_every(row.id, row.next_run, stamp, row.last_status); } + else { db.release_once(row.id, stamp); } + ++skipped_cap; + continue; + } + SchedulerDbPort* dbp = &db; + int64_t row_id = row.id; + double timeout = static_cast(row.timeout); + std::string profile = row.profile; + std::thread([dbp, row_id, agent_id, timeout, profile]() { watch_schedule(*dbp, row_id, agent_id, timeout, profile); }).detach(); + ++fired; + json::Value meta = json::Value::object(); + meta.set("id", row.id); + meta.set("profile", row.profile); + meta.set("every_sec", row.every_sec); + llm::log_event("INFO", "sched", "fired", "schedule #" + std::to_string(row.id) + " fired for " + row.profile, row.profile, "", meta); + } + if (skipped_cap || skipped_overlap) { + llm::print_out("scheduler: skipped " + std::to_string(skipped_cap) + " at parallel cap, " + std::to_string(skipped_overlap) + " overlapping"); + } + return fired; +} + +Scheduler::Scheduler(AgentConfig& config, SchedulerDbPort& db) : config_(config), db_(db) {} +Scheduler::~Scheduler() { stop(); } + +void Scheduler::start() { + stop_ = false; + worker_ = std::thread([this]() { + while (!stop_.load()) { + try { + scheduler_tick(config_, db_); + } catch (const std::exception& exc) { + llm::print_out("scheduler error: " + llm::short_error(exc)); + } + for (int i = 0; i < SCHEDULER_INTERVAL * 10 && !stop_.load(); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + } + }); +} + +void Scheduler::stop() { + stop_ = true; + if (worker_.joinable()) { worker_.join(); } +} + +} // namespace detz::agent diff --git a/taicpp/src/agent/scheduler.h b/taicpp/src/agent/scheduler.h new file mode 100644 index 0000000..d02ce5a --- /dev/null +++ b/taicpp/src/agent/scheduler.h @@ -0,0 +1,35 @@ +#pragma once +// Scheduler loop of tai.py (20691-20800): lease check, tick, watcher and the +// background loop. Storage goes through SchedulerDbPort. +#include "agent_config.h" +#include "ports.h" +#include "registry.h" +#include +#include +#include +#include +namespace detz::agent { + +void note_schedule_result(SchedulerDbPort& db, int64_t row_id, const std::string& status, const std::string& result = ""); +// Blocks up to timeout + SCHEDULER_LEASE_MARGIN for the agent, then records its outcome. +void watch_schedule(SchedulerDbPort& db, int64_t row_id, int agent_id, double timeout, const std::string& profile = ""); +// A running claim younger than timeout plus margin may still live. +bool schedule_running(const std::string& last_status, const std::string& updated, int64_t timeout); +// Fires due schedules (at most SCHEDULER_MAX_FIRES); returns how many fired. +int scheduler_tick(AgentConfig& config, SchedulerDbPort& db, const Runner& runner = nullptr); + +class Scheduler { + public: + Scheduler(AgentConfig& config, SchedulerDbPort& db); + ~Scheduler(); + // start_scheduler(): a daemon thread ticking every SCHEDULER_INTERVAL seconds. + void start(); + void stop(); // the stop_flag of tai.py + private: + AgentConfig& config_; + SchedulerDbPort& db_; + std::atomic stop_{false}; + std::thread worker_; +}; + +} // namespace detz::agent diff --git a/taicpp/src/agent/tools_memory.cpp b/taicpp/src/agent/tools_memory.cpp index 9bdd788..8ee04c9 100644 --- a/taicpp/src/agent/tools_memory.cpp +++ b/taicpp/src/agent/tools_memory.cpp @@ -39,7 +39,7 @@ std::optional> Tools::record_tag_args(const json::Value std::string Tools::run_remember(const json::Value& args) { std::string instruction = py::strip(py::str_or_empty(args.get("instruction"))); if (instruction.empty()) { return "error: empty instruction"; } - const std::string& current = app.system_message; + const std::string current = app.system_message; // a value: update_system replaces the member json::Value prompt = json::Value::array(); { json::Value sys = json::Value::object(); diff --git a/taicpp/src/agent/tools_ops.cpp b/taicpp/src/agent/tools_ops.cpp index 090fb2a..071be16 100644 --- a/taicpp/src/agent/tools_ops.cpp +++ b/taicpp/src/agent/tools_ops.cpp @@ -5,6 +5,7 @@ #include "../llm/term.h" #include "consts.h" #include "text.h" +#include "tui_control.h" #include "zipfile.h" #include #include @@ -202,8 +203,7 @@ std::string Tools::run_export_terminal_log(const json::Value& args) { // ------------------------------------------------------------------ plexer std::string Tools::run_plexer_send(const json::Value& args) { - if (ports().plexer == nullptr) { throw PortNotWired("plexer"); } - auto controller = ports().plexer->detect(); + auto controller = ports().plexer != nullptr ? ports().plexer->detect() : nullptr; // no port: no session if (!controller) { return kNoPlexer; } std::string text = py::str_or_empty(args.get("text")); bool enter = args.get("enter").truthy(); @@ -216,8 +216,7 @@ std::string Tools::run_plexer_send(const json::Value& args) { } std::string Tools::run_plexer_read(const json::Value& args) { - if (ports().plexer == nullptr) { throw PortNotWired("plexer"); } - auto controller = ports().plexer->detect(); + auto controller = ports().plexer != nullptr ? ports().plexer->detect() : nullptr; // no port: no session if (!controller) { return kNoPlexer; } int lines = 200; { @@ -234,8 +233,7 @@ std::string Tools::run_plexer_read(const json::Value& args) { } std::string Tools::run_plexer_type(const json::Value& args) { - if (ports().plexer == nullptr) { throw PortNotWired("plexer"); } - auto controller = ports().plexer->detect(); + auto controller = ports().plexer != nullptr ? ports().plexer->detect() : nullptr; // no port: no session if (!controller) { return kNoPlexer; } std::string text = py::str_or_empty(args.get("text")); if (text.empty()) { return "error: empty text"; } @@ -257,7 +255,6 @@ std::string Tools::run_plexer_type(const json::Value& args) { } std::string Tools::run_tui(const std::string& op, const json::Value& args) { - if (ports().tui == nullptr) { throw PortNotWired("tui"); } json::Value payload = json::Value::object(); payload.set("op", op); for (const auto& kv : args.members()) { @@ -271,12 +268,11 @@ std::string Tools::run_tui(const std::string& op, const json::Value& args) { else if (t.is_array()) { throw std::runtime_error("can only concatenate list (not \"int\") to list"); } else { throw std::runtime_error("unsupported operand type(s) for +: '" + py_type_name(t) + "' and 'int'"); } } - return ports().tui->control_call(payload).dumps(); + return (ports().tui != nullptr ? ports().tui->control_call(payload) : grr_control_call(payload)).dumps(); } std::string Tools::run_plexer_status(const json::Value&) { - if (ports().plexer == nullptr) { throw PortNotWired("plexer"); } - auto controller = ports().plexer->detect(); + auto controller = ports().plexer != nullptr ? ports().plexer->detect() : nullptr; // no port: no session if (!controller) { return "no plexer session active (neither tai's native wrapper nor tmux)"; } try { return controller->status().dumps(); @@ -460,11 +456,13 @@ std::string Tools::run_ask_user(const json::Value& args) { } std::string Tools::run_sysinfo(const json::Value& args) { - if (ports().ops == nullptr) { throw PortNotWired("ops"); } + static const std::vector valid = {"os", "python", "venv", "root", "container", "binaries", "cpu", "disk"}; // SYSINFO_CHECKS const json::Value& names = args.get("checks"); - if (names.is_null()) { return ports().ops->collect_sysinfo(std::nullopt); } + if (names.is_null()) { + if (ports().ops == nullptr) { throw PortNotWired("ops"); } + return ports().ops->collect_sysinfo(std::nullopt); + } if (!names.is_array() || names.size() == 0) { return "error: checks must be a non-empty list"; } - std::vector valid = ports().ops->sysinfo_check_names(); std::vector unknown, picked; for (const auto& item : names.items()) { std::string s = sv(item); @@ -474,6 +472,7 @@ std::string Tools::run_sysinfo(const json::Value& args) { if (!unknown.empty()) { return "error: unknown checks: " + py::join(unknown, ", ") + " (valid: " + py::join(valid, ", ") + ")"; } + if (ports().ops == nullptr) { throw PortNotWired("ops"); } return ports().ops->collect_sysinfo(picked); } diff --git a/taicpp/src/agent/tui_control.cpp b/taicpp/src/agent/tui_control.cpp new file mode 100644 index 0000000..ab66d28 --- /dev/null +++ b/taicpp/src/agent/tui_control.cpp @@ -0,0 +1,87 @@ +#include "tui_control.h" +#include "../llm/proc.h" +#include "../llm/py.h" +#include "text.h" +#include +#include +#include +#include +#include +#include +namespace detz::agent { + +std::string grr_control_path() { + std::string home = llm::env_get("TAI_HOME"); + if (home.empty()) { home = expanduser("~") + "/.tai"; } + std::string sock = llm::env_get("TAI_TUI_SOCK"); + return !sock.empty() ? sock : home + "/tui.sock"; +} + +json::Value grr_control_call(const json::Value& payload, const std::string& path_in, double timeout) { + std::string path = path_in.empty() ? grr_control_path() : path_in; + auto failure = [&](const std::string& what) { + json::Value o = json::Value::object(); + o.set("state", "error"); + o.set("error", "no running grr TUI answers on " + path + " (" + what + ")"); + return o; + }; + int fd = ::socket(AF_UNIX, SOCK_STREAM, 0); + if (fd < 0) { return failure(std::string("[Errno ") + std::to_string(errno) + "] " + std::strerror(errno)); } + struct sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + if (path.size() >= sizeof addr.sun_path) { + ::close(fd); + return failure("AF_UNIX path too long"); + } + std::memcpy(addr.sun_path, path.c_str(), path.size() + 1); + if (::connect(fd, reinterpret_cast(&addr), sizeof addr) != 0) { + int err = errno; + ::close(fd); + return failure("[Errno " + std::to_string(err) + "] " + std::strerror(err)); + } + std::string line = payload.dumps() + "\n"; + size_t off = 0; + while (off < line.size()) { + ssize_t w = ::send(fd, line.data() + off, line.size() - off, MSG_NOSIGNAL); + if (w < 0) { + if (errno == EINTR) { continue; } + int err = errno; + ::close(fd); + return failure("[Errno " + std::to_string(err) + "] " + std::strerror(err)); + } + off += static_cast(w); + } + std::string data; + while (data.empty() || data.back() != '\n') { + struct pollfd pfd{fd, POLLIN, 0}; + int pr = ::poll(&pfd, 1, static_cast(timeout * 1000)); + if (pr == 0) { ::close(fd); return failure("timed out"); } + if (pr < 0) { + if (errno == EINTR) { continue; } + int err = errno; + ::close(fd); + return failure("[Errno " + std::to_string(err) + "] " + std::strerror(err)); + } + char buf[65536]; + ssize_t c = ::recv(fd, buf, sizeof buf, 0); + if (c < 0) { + if (errno == EINTR) { continue; } + int err = errno; + ::close(fd); + return failure("[Errno " + std::to_string(err) + "] " + std::strerror(err)); + } + if (c == 0) { break; } + data.append(buf, static_cast(c)); + } + ::close(fd); + try { + return json::Value::loads(py::utf8_replace(data)); + } catch (const json::ParseError&) { + json::Value o = json::Value::object(); + o.set("state", "error"); + o.set("error", "unreadable reply"); + return o; + } +} + +} // namespace detz::agent diff --git a/taicpp/src/agent/tui_control.h b/taicpp/src/agent/tui_control.h new file mode 100644 index 0000000..bbf3339 --- /dev/null +++ b/taicpp/src/agent/tui_control.h @@ -0,0 +1,16 @@ +#pragma once +// Client side of the grr TUI control socket (tai.py 35479-35726): path lookup +// and one JSON request/reply over the unix socket. The server side belongs to +// the TUI module; this is only what the tui_* tools and the tool gate need. +#include "../llm/json.h" +#include +namespace detz::agent { + +inline constexpr int GRR_CONTROL_WAIT_MAX = 60; +// TAI_TUI_SOCK or /tui.sock +std::string grr_control_path(); +// grr_control_call(payload): the reply as an object; failures are +// {"state": "error", "error": "..."} exactly like tai.py. +json::Value grr_control_call(const json::Value& payload, const std::string& path = "", double timeout = GRR_CONTROL_WAIT_MAX + 10); + +} // namespace detz::agent diff --git a/taicpp/src/llm/py.cpp b/taicpp/src/llm/py.cpp index 5a3272b..e89a825 100644 --- a/taicpp/src/llm/py.cpp +++ b/taicpp/src/llm/py.cpp @@ -658,6 +658,15 @@ double to_epoch(const DateTime& dt) { // naive: CPython's local_to_seconds with fold=0 return static_cast(naive_local_to_epoch(secs)) + frac; } +DateTime from_epoch(double epoch, int offset_seconds, bool aware, int micro) { + int64_t secs = static_cast(std::floor(epoch)) + offset_seconds; + DateTime d; + civil_from_epoch(secs, d.year, d.month, d.day, d.hour, d.minute, d.second); + d.micro = micro; + d.aware = aware; + d.offset_seconds = aware ? offset_seconds : 0; + return d; +} std::string isoformat(const DateTime& dt) { char buf[96]; int n = std::snprintf(buf, sizeof buf, "%04d-%02d-%02dT%02d:%02d:%02d", dt.year, dt.month, dt.day, dt.hour, dt.minute, dt.second); diff --git a/taicpp/src/llm/py.h b/taicpp/src/llm/py.h index d152b04..0b9ef80 100644 --- a/taicpp/src/llm/py.h +++ b/taicpp/src/llm/py.h @@ -72,6 +72,8 @@ std::string utc_now_iso(); // datetime.now(timezone.utc).isofo // _fromisoformat; throws std::invalid_argument("invalid datetime...") on failure. DateTime fromisoformat(const std::string& text); double to_epoch(const DateTime& dt); // naive means local time +// A DateTime at the given instant expressed with a fixed UTC offset (datetime + timedelta keeps the zone). +DateTime from_epoch(double epoch, int offset_seconds, bool aware, int micro); std::string isoformat(const DateTime& dt); // moment.astimezone(utc).isoformat() std::string to_utc_iso(const DateTime& dt); diff --git a/taicpp/tests/agent_diff.cpp b/taicpp/tests/agent_diff.cpp index 49c5db9..9e5e311 100644 --- a/taicpp/tests/agent_diff.cpp +++ b/taicpp/tests/agent_diff.cpp @@ -4,7 +4,9 @@ #include "../src/agent/agent.h" #include "../src/agent/text.h" #include "../src/agent/data.h" +#include "../src/llm/chat_client.h" #include "../src/llm/cli_backends.h" +#include "../src/llm/http_transport.h" #include "../src/llm/helpers.h" #include "../src/llm/model_health.h" #include "../src/llm/muse.h" @@ -49,9 +51,84 @@ Value try_value(const std::function& f) { } } +class FailTransport : public llm::HttpTransport { + public: + llm::HttpResult send(const llm::HttpRequest&) override { llm::HttpResult r; r.error = ""; return r; } + llm::HttpResult stream(const llm::HttpRequest&, const llm::LineSink&) override { llm::HttpResult r; r.error = ""; return r; } +}; + +std::vector str_list(const Value& v) { + std::vector out; + for (const auto& x : v.items()) { out.push_back(x.str()); } + return out; +} + +Value candidate_view(const llm::Candidate& c) { + Value a = Value::array(); + a.push(c.provider); a.push(c.chat_url); a.push(c.models_url); a.push(c.key); a.push(c.model); a.push(c.profile); + a.push(c.runner); + return a; +} + +Value chat_candidates(const Value& a) { + static FailTransport fail; + llm::set_transport(&fail); + const Value& c = a.at(0); + llm::LlmConfig cfg; + cfg.home = c.get("home").str(); + cfg.model = c.get("model").str(); + cfg.explicit_model = c.get("explicit_model").str(); + for (const auto& b : c.get("backends").items()) { + llm::BackendSpec s; + s.label = b.get("label").str(); s.base = b.get("base").str(); s.key = b.get("key").str(); + s.model = b.get("model").str(); s.profile = b.get("profile").str(); + cfg.backends.push_back(s); + } + cfg.openrouter_key = c.get("openrouter_key").str(); + cfg.openrouter_models = str_list(c.get("openrouter_models")); + cfg.opencode_enabled = c.get("opencode_enabled").truthy(); + cfg.opencode_models = str_list(c.get("opencode_models")); + cfg.devplace_key = c.get("devplace_key").str(); + cfg.devplace_base = c.get("devplace_base").str(); + cfg.devplace_models = str_list(c.get("devplace_models")); + cfg.openrouter_enabled = c.get("openrouter_enabled").truthy(); + cfg.pollinations_enabled = c.get("pollinations_enabled").truthy(); + cfg.pollinations_models = str_list(c.get("pollinations_models")); + cfg.zen_enabled = c.get("zen_enabled").truthy(); + cfg.zen_models = str_list(c.get("zen_models")); + cfg.ollama_enabled = c.get("ollama_enabled").truthy(); + cfg.ollama_models = str_list(c.get("ollama_models")); + llm::ChatClient chat(cfg); + for (const auto& r : a.at(1).items()) { + std::optional lat, tps; + if (!r.at(3).is_null()) { lat = r.at(3).as_double(); } + if (!r.at(4).is_null()) { tps = r.at(4).as_double(); } + chat.health.record(r.at(0).str(), r.at(1).str(), r.at(2).as_bool(), lat, tps); + } + if (a.at(2).get("provider").is_string()) { chat.provider_pin = a.at(2).get("provider").str(); } + if (a.at(2).get("model").is_string()) { chat.model_pin = a.at(2).get("model").str(); } + Value out = Value::object(); + Value ordered = Value::array(); + for (const auto& cand : chat.ordered_candidates()) { ordered.push(candidate_view(cand)); } + out.set("ordered", ordered); + Value all = Value::array(); + for (const auto& cand : chat.all_candidates()) { all.push(candidate_view(cand)); } + out.set("all", all); + Value backs = Value::array(); + for (const auto& b : chat.backends()) { + Value row = Value::array(); + row.push(b.provider); row.push(b.chat_url); row.push(b.models_url); row.push(b.key); row.push(b.model); + backs.push(row); + } + out.set("backends", backs); + out.set("pin_note", chat.pin_note()); + return out; +} + using Fn = std::function; std::map registry() { std::map r; + r["chat_candidates"] = chat_candidates; r["json_roundtrip"] = [](const Value& a) { return try_value([&]() { return S(Value::loads(a.at(0).str()).dumps()); }); }; diff --git a/taicpp/tests/agent_diff.py b/taicpp/tests/agent_diff.py index 4f2d88b..e09dff8 100644 --- a/taicpp/tests/agent_diff.py +++ b/taicpp/tests/agent_diff.py @@ -30,9 +30,41 @@ for key in list(os.environ): del os.environ[key] os.environ["TAI_HOME"] = HOME os.environ["TZ"] = "Europe/Amsterdam" +os.environ["PATH"] = "/usr/bin:/bin" +for _name in ("MUSE_BIN", "OPENCODE_BIN", "GROK_BIN", "CLAUDE_BIN", "CODEX_BIN", "GEMINI_BIN"): + os.environ.pop(_name, None) sys.path.insert(0, ROOT) sys.dont_write_bytecode = True import tai # noqa: E402 +import types # noqa: E402 +import urllib.error # noqa: E402 +import urllib.request # noqa: E402 + + +def _fail_urlopen(*args, **kwargs): + raise urllib.error.URLError("fake") + + +def py_chat_candidates(cfg, records, pins): + original = urllib.request.urlopen + urllib.request.urlopen = _fail_urlopen + try: + home = tempfile.mkdtemp(prefix="agent-diff-chat-") + cfg = dict(cfg, home=home) + config = types.SimpleNamespace(**cfg) + chat = tai.ChatClient(config) + for provider, model, ok, lat, tps in records: + chat.health.record(provider, model, ok, lat, tps) + chat.provider_pin = pins.get("provider") + chat.model_pin = pins.get("model") + + def view(c): + return [c["provider"], c["chat_url"], c["models_url"], c["key"], c["model"], c["profile"], c.get("runner", "")] + + return {"ordered": [view(c) for c in chat.ordered_candidates()], "all": [view(c) for c in chat.all_candidates()], + "backends": [list(b) for b in chat.backends()], "pin_note": chat.pin_note()} + finally: + urllib.request.urlopen = original def dumps_sorted(value): @@ -79,6 +111,7 @@ def schedule_now(epoch): EXPECT = { + "chat_candidates": lambda cfg, records, pins: py_chat_candidates(cfg, records, pins), "json_roundtrip": guard(lambda text: json.dumps(json.loads(text))), "json_compact_sorted": guard(lambda text: json.dumps(json.loads(text), sort_keys=True, separators=(",", ":"))), "schemas": lambda: json.dumps(tai.TOOL_SCHEMAS), @@ -214,12 +247,40 @@ def fuzz_messages(rng): return messages +PROVIDERS = ["lab", "devplace", "openrouter", "pollinations", "zen", "ollama", "muse", "opencode", "other"] +MODELS = ["m1", "m2", "molodetz", "free", "muse-spark-x", "a:free", "b-free", "openai-fast", "deepseek/deepseek-v4.1-flash", "z"] + + +def fuzz_config(rng): + pick = lambda n: rng.sample(MODELS, rng.randint(0, n)) # noqa: E731 + backends = [] + for i in range(rng.randint(0, 3)): + backends.append({"label": rng.choice(["lab", "devplace", "Acme", "other"]), "base": "http://h%d/v1" % i, "key": rng.choice(["", "k"]), + "model": ",".join(pick(2)), "profile": rng.choice(["", "opencode"])}) + return {"model": rng.choice(["muse", "m1"]), "explicit_model": rng.choice(["", "m2", "xx"]), "backends": backends, + "openrouter_key": rng.choice(["", "key"]), "openrouter_models": pick(3), "retry_seconds": 0, "opencode_enabled": rng.random() < 0.5, + "opencode_models": pick(2), "devplace_key": rng.choice(["", "dk"]), "devplace_base": "https://devplace.net/openai/v1", + "devplace_models": pick(2), "openrouter_enabled": rng.random() < 0.7, "pollinations_enabled": rng.random() < 0.7, + "pollinations_models": pick(2), "zen_enabled": rng.random() < 0.4, "zen_models": pick(3), "ollama_enabled": rng.random() < 0.6, + "ollama_models": pick(2)} + + +def fuzz_records(rng): + out = [] + for _ in range(rng.randint(0, 12)): + out.append([rng.choice(PROVIDERS), rng.choice(MODELS), rng.random() < 0.6, rng.choice([None, 100.0, 5000.5]), rng.choice([None, 10.0, 80.0])]) + return out + + def build_cases(rng, fuzz): cases = [] def add(fn, *args): cases.append({"fn": fn, "args": list(args)}) + for _ in range(max(40, fuzz // 3)): + pins = rng.choice([{}, {}, {"provider": rng.choice(PROVIDERS)}, {"model": rng.choice(MODELS)}, {"provider": rng.choice(PROVIDERS), "model": rng.choice(MODELS)}]) + add("chat_candidates", fuzz_config(rng), fuzz_records(rng), pins) add("schemas") add("tool_catalog") add("data_texts") diff --git a/taicpp/tests/agent_e2e.cpp b/taicpp/tests/agent_e2e.cpp new file mode 100644 index 0000000..f181d23 --- /dev/null +++ b/taicpp/tests/agent_e2e.cpp @@ -0,0 +1,272 @@ +// End-to-end driver: runs one scenario (JSON file) through the C++ Agent with a +// scripted HTTP transport, an in-memory store and a popen shell, then prints +// the observable outcome as JSON. tests/agent_e2e.py runs the same scenario +// through tai.py and compares. +#include "../src/agent/agent.h" +#include "../src/llm/http_transport.h" +#include "../src/llm/py.h" +#include "agent_memstore.h" +#include +#include +#include +#include +#include +#include +using namespace detz; +using json::Value; +namespace fs = std::filesystem; + +namespace { + +struct ScriptExhausted {}; // deliberately not a std::exception: must not be retried + +class ScriptedTransport : public llm::HttpTransport { + public: + ScriptedTransport(Value script, Value single) : script_(std::move(script)), single_(std::move(single)) {} + Value requests = Value::array(); + llm::HttpResult send(const llm::HttpRequest& req) override { + if (req.url.size() >= 7 && req.url.compare(req.url.size() - 7, 7, "/models") == 0) { + llm::HttpResult r; + r.error = ""; + return r; + } + record(req); + const Value& item = next(single_, single_pos_); + return reply(item, nullptr); + } + llm::HttpResult stream(const llm::HttpRequest& req, const llm::LineSink& sink) override { + record(req); + const Value& item = next(script_, script_pos_); + return reply(item, &sink); + } + + private: + void record(const llm::HttpRequest& req) { + Value r = Value::object(); + r.set("method", req.method); + r.set("url", req.url); + Value body; + if (req.body.has_value()) { + try { body = Value::loads(*req.body); } catch (const json::ParseError&) { body = Value(*req.body); } + } + r.set("body", body); + requests.push(r); + } + const Value& next(const Value& queue, size_t& pos) { + if (pos >= queue.size()) { throw ScriptExhausted{}; } + return queue.at(pos++); + } + llm::HttpResult reply(const Value& item, const llm::LineSink* sink) { + llm::HttpResult r; + std::string kind = item.get("kind").str(); + if (kind == "transport_error") { r.error = ""; return r; } + if (kind == "status") { + r.status = item.get("status").as_int(); + r.body = item.get("body").str(); + return r; + } + r.status = 200; + if (item.get("content_type").is_string()) { r.content_type = item.get("content_type").str(); } + if (kind == "json") { + r.body = item.get("body").dumps(); + return r; + } + if (kind == "raw") { + r.body = item.get("body").str(); + return r; + } + // stream: SSE lines + if (sink == nullptr) { throw std::runtime_error("stream script item consumed by a non-stream call"); } + for (const auto& ev : item.get("events").items()) { + if (!(*sink)("data: " + ev.dumps() + "\n")) { return r; } + if (!(*sink)("\n")) { return r; } + } + (*sink)("data: [DONE]\n"); + return r; + } + Value script_, single_; + size_t script_pos_{0}, single_pos_{0}; +}; + +class PopenShell : public agent::ShellPort { + public: + agent::RunResult run_live_shell(const std::string& command, const std::string& cwd, const std::map& extra_env, + agent::LiveStreamPort*, const agent::AskFn&) override { + std::string env; + for (const auto& kv : extra_env) { env += "export " + kv.first + "='" + kv.second + "'; "; } + std::string full = "cd '" + cwd + "' && " + env + command + " 2>&1"; + FILE* pipe = popen(("/bin/sh -c '" + escape(full) + "' < /dev/null").c_str(), "r"); + agent::RunResult r; + if (pipe == nullptr) { r.code = 127; return r; } + char buf[4096]; + size_t n; + while ((n = fread(buf, 1, sizeof buf, pipe)) > 0) { r.output.append(buf, n); } + int status = pclose(pipe); + r.code = WIFEXITED(status) ? WEXITSTATUS(status) : -1; + r.output = py::utf8_replace(r.output); + return r; + } + agent::RunResult run_live_argv(const std::vector&, agent::LiveStreamPort*, const agent::AskFn&) override { + throw agent::PortNotWired("shell argv"); + } + std::unique_ptr make_live_stream() override { return nullptr; } + std::vector shell_target_paths(const std::string&, const std::string&) override { return {}; } + std::string ensure_box() override { throw llm::BackendError("no container engine (test)"); } + llm::ProcResult box_exec(const std::string&, const std::vector&, const std::optional&, double) override { + throw agent::PortNotWired("box"); + } + std::unique_ptr make_lane_board(const std::string&) override { return nullptr; } + + private: + static std::string escape(const std::string& s) { + std::string out; + for (char c : s) { + if (c == '\'') { out += "'\\''"; } else { out += c; } + } + return out; + } +}; + +class FakeSkills : public agent::SkillsPort { + public: + explicit FakeSkills(std::string catalog) : catalog_(std::move(catalog)) {} + Value discover_skills(const std::string&, const std::string&) override { return Value::object(); } + std::string skill_catalog(const Value&) override { return catalog_; } + std::string skill_provenance_note(const Value&) override { return ""; } + std::vector skill_extra_files(const std::string&) override { return {}; } + std::optional skill_blueprint(const std::string&) override { return std::nullopt; } + std::string skill_scope_dir(const std::string&, const std::string&, const std::string&, const std::string&) override { return ""; } + std::string default_skill_scope(const std::string&) override { return "project"; } + Value parse_skill_file(const std::string&) override { return Value(); } + bool is_external_skill_root(const std::string&) override { return false; } + void save_skill_install_meta(const std::string&, const std::string&, const Value&) override {} + private: + std::string catalog_; +}; + +std::string slurp(const std::string& path) { + std::ifstream in(path, std::ios::binary); + std::stringstream ss; + ss << in.rdbuf(); + return ss.str(); +} +} // namespace + +int main(int argc, char** argv) { + if (argc < 3) { std::fprintf(stderr, "usage: agent_e2e scenario.json workdir\n"); return 2; } + Value sc = Value::loads(slurp(argv[1])); + std::string workdir = argv[2]; + if (chdir(workdir.c_str()) != 0) { std::perror("chdir"); return 2; } + std::string home = workdir + "/.home"; + fs::create_directories(home); + + agent::AgentConfig cfg; + cfg.home = home; + cfg.audio_dir = home + "/audio"; + cfg.profile = sc.get("profile").is_string() ? sc.get("profile").str() : "default"; + cfg.auto_approve = sc.get("auto_approve").truthy(); + cfg.retry_seconds = sc.get("retry_seconds").truthy() ? sc.get("retry_seconds").as_double() : 0.0; + cfg.opencode_enabled = false; + cfg.openrouter_enabled = false; + cfg.pollinations_enabled = false; + cfg.zen_enabled = false; + llm::BackendSpec b; + b.label = "lab"; b.base = "http://fake.invalid/v1"; b.key = "k"; b.model = "m1"; b.profile = ""; + cfg.backends.push_back(b); + + ScriptedTransport transport(sc.get("script"), sc.get("single")); + llm::set_transport(&transport); + agent::testing::MemStore store; + PopenShell shell; + FakeSkills skills(sc.get("skill_catalog").str_or("")); + agent::ports().shell = &shell; + agent::ports().skills = &skills; + llm::ChatClient::sleeper() = [](double) {}; + + if (sc.get("presets").is_object()) { + // pre-existing vault state: systems and secrets + for (const auto& kv : sc.get("presets").get("systems").members()) { store.save_system(kv.first, kv.second.str()); } + for (const auto& kv : sc.get("presets").get("secrets").members()) { + store.save_secret(kv.first, kv.second.get("value").str(), Value::object(), {}, "default"); + } + for (const auto& kv : sc.get("presets").get("bots").members()) { + store.save_bot_system("default", kv.first, kv.second.get("system").str()); + Value bots = store.load_bots("default"); + Value e = Value::object(); + e.set("nicknames", kv.second.get("nicknames")); + e.set("created", "t"); + e.set("updated", "t"); + bots.set(kv.first, e); + store.save_bots("default", bots); + } + } + + agent::Agent ag(cfg, &store, sc.get("persist").is_bool() ? sc.get("persist").as_bool() : true, true, 0, false, sc.get("yolo").truthy()); + if (sc.get("deadline_expired").truthy()) { ag.deadline = 1.0; } + Value answers = sc.get("answers"); + size_t answer_pos = 0; + if (answers.is_array()) { + ag.prompter = [&](const std::string&, const std::string&, const std::vector&) -> std::optional { + if (answer_pos >= answers.size()) { return std::nullopt; } + const Value& a = answers.at(answer_pos++); + if (a.is_null()) { return std::nullopt; } + return a.str(); + }; + } + Value runner_script = sc.get("runner_script"); + size_t runner_pos = 0; + if (runner_script.is_array()) { + ag.runner_override = [&](const std::string&, const std::string&, double) -> std::string { + if (runner_pos >= runner_script.size()) { return "runner exhausted"; } + return runner_script.at(runner_pos++).str(); + }; + } + Value out = Value::object(); + try { + Value turns = sc.get("prompts").is_array() ? sc.get("prompts") : Value::array(); + if (turns.size() == 0) { turns.push(sc.get("prompt")); } + Value results = Value::array(); + for (const auto& p : turns.items()) { + results.push(ag.run_turn(p.str(), true, sc.get("max_steps").truthy() ? static_cast(sc.get("max_steps").as_int()) : agent::MAX_STEPS)); + } + out.set("results", results); + if (sc.get("verify").truthy()) { + auto [text, ok] = ag.verify_loop(sc.get("prompt").str(), results.at(results.size() - 1).str()); + Value v = Value::object(); + v.set("text", text); + v.set("ok", ok); + out.set("verify", v); + } + } catch (const ScriptExhausted&) { + out.set("exception", "script exhausted"); + } catch (const std::exception& e) { + out.set("exception", std::string(e.what())); + } + out.set("messages", ag.messages); + out.set("requests", transport.requests); + Value events = Value::array(); + for (const auto& e : store.events) { + Value row = Value::array(); + row.push(e.role); row.push(e.kind); row.push(e.text); + events.push(row); + } + out.set("events", events); + Value audits = Value::array(); + for (const auto& a : store.audits) { + Value row = Value::object(); + row.set("actor", a.actor); row.set("action", a.action); row.set("path", a.path); row.set("message", a.message); + row.set("tags", a.tags); + row.set("old_size", a.old_size ? Value(*a.old_size) : Value()); + row.set("new_size", a.new_size ? Value(*a.new_size) : Value()); + row.set("old", a.old_text ? Value(*a.old_text) : Value()); + row.set("new", a.new_text ? Value(*a.new_text) : Value()); + audits.push(row); + } + out.set("audits", audits); + out.set("tool_calls", ag.tools.calls); + out.set("yolo", ag.yolo); + out.set("timed_out", ag.timed_out); + out.set("system", ag.system_message); + std::cout << out.dumps() << "\n"; + return 0; +} diff --git a/taicpp/tests/agent_e2e.py b/taicpp/tests/agent_e2e.py new file mode 100644 index 0000000..4eba3b4 --- /dev/null +++ b/taicpp/tests/agent_e2e.py @@ -0,0 +1,487 @@ +#!/usr/bin/env python3 +"""End-to-end differential test: the same scripted conversation runs through +tai.py (real Store, real Agent, patched urlopen) and through the C++ Agent +(tests/agent_e2e.cpp: scripted transport, in-memory store). Messages, HTTP +requests, store events, audit rows and tool results must be identical. + + python3 tests/agent_e2e.py [--bin build/agent_e2e] [--filter NAME] [-v] +""" +import argparse +import http.client +import io +import json +import os +import re +import shutil +import subprocess +import sys +import tempfile +import types +import urllib.error +import urllib.request + +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) +sys.path.insert(0, ROOT) +sys.dont_write_bytecode = True + +SAFE_PATH = "/usr/bin:/bin" +TZ = "Europe/Amsterdam" + + +def clean_env(home): + for key in list(os.environ): + if key.startswith("TAI_") or key in ("MUSE_BIN", "OPENCODE_BIN", "GROK_BIN", "CLAUDE_BIN", "CODEX_BIN", "GEMINI_BIN", + "OPENROUTER_API_KEY", "DEVPLACE_API_KEY", "TMUX", "TMUX_PANE", "HISTFILE"): + del os.environ[key] + os.environ["PATH"] = SAFE_PATH + os.environ["TZ"] = TZ + os.environ["TAI_HOME"] = home + os.environ["TAI_BACKENDS"] = "lab|http://fake.invalid/v1|k|m1" + os.environ["TAI_OPENROUTER"] = "0" + os.environ["TAI_POLLINATIONS"] = "0" + os.environ["TAI_OPENCODE"] = "0" + os.environ["HOME"] = home + + +import tai # noqa: E402 + + +# ---------------------------------------------------------------- script builders +def say(text): + return {"kind": "stream", "events": [{"choices": [{"delta": {"content": text}}]}]} + + +def calls(*items, text=None): + events = [] + if text: + events.append({"choices": [{"delta": {"content": text}}]}) + for idx, (name, args, cid) in enumerate(items): + argtext = args if isinstance(args, str) else json.dumps(args) + half = len(argtext) // 2 + events.append({"choices": [{"delta": {"tool_calls": [{"index": idx, "id": cid, "function": {"name": name, "arguments": argtext[:half]}}]}}]}) + events.append({"choices": [{"delta": {"tool_calls": [{"index": idx, "function": {"arguments": argtext[half:]}}]}}]}) + return {"kind": "stream", "events": events} + + +def single(text): + return {"kind": "json", "body": {"choices": [{"message": {"content": text}}]}} + + +def raw(body, content_type="text/plain"): + return {"kind": "raw", "body": body, "content_type": content_type} + + +def status(code, body=""): + return {"kind": "status", "status": code, "body": body} + + +def tc(name, args, cid): + return (name, args, cid) + + +# ---------------------------------------------------------------- python side +class ScriptExhausted(BaseException): + """Not an Exception: the retry loops of ChatClient must not swallow it.""" + + +class FakeResponse(object): + def __init__(self, body=b"", lines=None, status=200, content_type=""): + self._body = body + self._lines = lines + self.status = status + self.headers = {"Content-Type": content_type} if content_type else {} + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self, n=-1): + data = self._body if n is None or n < 0 else self._body[:n] + return data + + def __iter__(self): + return iter(self._lines or []) + + def close(self): + pass + + +def python_run(sc, workdir): + home = os.path.join(workdir, ".home") + clean_env(home) + if sc.get("retry_seconds"): + os.environ["TAI_RETRY_SECONDS"] = str(sc["retry_seconds"]) + os.makedirs(home, exist_ok=True) + os.chdir(workdir) + tai.AGENTS.clear() + tai.AGENTS_NEXT[0] = 1 + requests = [] + queues = {"stream": list(sc.get("script", [])), "single": list(sc.get("single", []))} + + def fake_urlopen(request, timeout=None): + url = request.full_url + if url.endswith("/models"): + raise urllib.error.URLError("fake") + body = None + if request.data is not None: + try: + body = json.loads(request.data) + except ValueError: + body = request.data.decode("utf-8", "replace") + requests.append({"method": request.get_method(), "url": url, "body": body}) + streaming = isinstance(body, dict) and bool(body.get("stream")) + queue = queues["stream" if streaming else "single"] + if not queue: + raise ScriptExhausted() + item = queue.pop(0) + kind = item["kind"] + if kind == "transport_error": + raise urllib.error.URLError("scripted") + if kind == "status": + reason = http.client.responses.get(item["status"], "") + raise urllib.error.HTTPError(url, item["status"], reason, {}, io.BytesIO(item.get("body", "").encode("utf-8"))) + content_type = item.get("content_type", "") + if kind == "json": + return FakeResponse(body=json.dumps(item["body"]).encode("utf-8"), content_type=content_type) + if kind == "raw": + return FakeResponse(body=item["body"].encode("utf-8"), content_type=content_type) + lines = [] + for event in item["events"]: + lines.append(("data: %s\n" % json.dumps(event)).encode("utf-8")) + lines.append(b"\n") + lines.append(b"data: [DONE]\n") + return FakeResponse(lines=lines, content_type=content_type) + + original = urllib.request.urlopen + original_sleep = tai.time.sleep + urllib.request.urlopen = fake_urlopen + tai.time.sleep = lambda seconds: None + try: + args = types.SimpleNamespace(yes=bool(sc.get("auto_approve")), profile=sc.get("profile", "default"), no_opencode=True) + config = tai.Config(args) + seal = tai.Seal(config.home, tai.resolve_passphrase()[0]) + store = tai.Store(config, seal) + presets = sc.get("presets") or {} + for name, text in (presets.get("systems") or {}).items(): + store.save_system(name, text) + for name, item in (presets.get("secrets") or {}).items(): + store.save_secret(name, item["value"], {}, [], "default") + for name, item in (presets.get("bots") or {}).items(): + store.save_bot_system("default", name, item["system"]) + bots = store.load_bots("default") + bots[name] = {"nicknames": item["nicknames"], "created": "t", "updated": "t"} + store.save_bots("default", bots) + events = [] + original_log = store.log_event + + def wrapped(profile, role, kind, text, tags=()): + events.append([role, kind, text]) + return original_log(profile, role, kind, text, tags) + + store.log_event = wrapped + agent = tai.Agent(config, store, persist=sc.get("persist", True), quiet=True, yolo=bool(sc.get("yolo"))) + if sc.get("deadline_expired"): + agent.deadline = 1.0 + answers = sc.get("answers") + if isinstance(answers, list): + queue = list(answers) + agent.prompter = lambda kind, text, options: queue.pop(0) if queue else None + runner_script = sc.get("runner_script") + if isinstance(runner_script, list): + rq = list(runner_script) + agent.runner_override = lambda task, profile, timeout: rq.pop(0) if rq else "runner exhausted" + out = {} + try: + prompts = sc.get("prompts") or [sc.get("prompt")] + results = [] + for prompt in prompts: + results.append(agent.run_turn(prompt, capture=True, max_steps=sc.get("max_steps", tai.MAX_STEPS))) + out["results"] = results + if sc.get("verify"): + text, ok = agent.verify_loop(sc["prompt"], results[-1]) + out["verify"] = {"text": text, "ok": ok} + except ScriptExhausted: + out["exception"] = "script exhausted" + except Exception as exc: # noqa: BLE001 + out["exception"] = str(exc) + out["messages"] = agent.messages + out["requests"] = requests + out["events"] = events + audits = [] + for row in reversed(store.audit_history(None, None, 100)): + full = store.audit_get(row["id"]) + audits.append({"actor": full["actor"], "action": full["action"], "path": full["path"], "message": full["message"], + "tags": full["tags"], "old_size": full["old_size"], "new_size": full["new_size"], + "old": full["old"], "new": full["new"]}) + out["audits"] = audits + out["tool_calls"] = agent.tools.calls + out["yolo"] = agent.yolo + out["timed_out"] = agent.timed_out + out["system"] = agent.system_message + store.close() + return out + finally: + urllib.request.urlopen = original + tai.time.sleep = original_sleep + + +def cpp_run(sc, workdir, binary): + path = os.path.join(workdir, "..", "scenario.json") + with open(path, "w") as handle: + json.dump(sc, handle) + env = dict(os.environ) + env.update({"PATH": SAFE_PATH, "TZ": TZ, "HOME": workdir}) + done = subprocess.run([binary, path, workdir], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, cwd=workdir) + if done.returncode != 0: + raise RuntimeError("binary failed: %s %s" % (done.returncode, done.stderr.decode("utf-8", "replace")[-1500:])) + return json.loads(done.stdout.decode("utf-8")) + + +def normalize(value): + text = json.dumps(value, sort_keys=True) + text = re.sub(r"mem:[0-9a-f]{16}", "mem:ID", text) + return json.loads(text) + + +# ---------------------------------------------------------------- scenarios +def build_scenarios(): + big = "line of text number %d\n" * 1 + scenarios = [] + + def add(name, prompt, script, **kw): + sc = {"name": name, "prompt": prompt, "script": script, "skill_catalog": tai.skill_catalog({})} + sc.update(kw) + scenarios.append(sc) + + add("plain", "hi there", [say("Hello there")], auto_approve=True) + add("read_file", "read a.txt", [calls(tc("read_file", {"path": "a.txt"}, "c1")), say("done reading")], + files={"a.txt": "alpha\nbeta\n"}, auto_approve=True) + add("read_missing_and_big", "read", [calls(tc("read_file", {"path": "missing.txt"}, "c1"), tc("read_file", {"path": "big.txt"}, "c2")), say("ok")], + files={"big.txt": "x" * 200001}, auto_approve=True) + add("write_flow", "write", [ + calls(tc("write_file", {"path": "b.txt", "content": "new"}, "c1")), + calls(tc("read_file", {"path": "b.txt"}, "c2")), + calls(tc("write_file", {"path": "b.txt", "content": "second\nversion"}, "c3"), + tc("write_file", {"path": "sub/dir/c.txt", "content": "nested"}, "c4"), + tc("write_file", {"path": "", "content": "x"}, "c5")), + calls(tc("edit_file", {"path": "b.txt", "find": "second", "replace": "third"}, "c6"), + tc("edit_file", {"path": "b.txt", "find": "zzz", "replace": "q"}, "c7"), + tc("edit_file", {"path": "b.txt", "find": "", "replace": "q"}, "c8"), + tc("edit_file", {"path": "nope.txt", "find": "a", "replace": "b"}, "c9")), + say("finished")], files={"b.txt": "old"}, auto_approve=True) + add("edit_not_unique", "edit", [ + calls(tc("edit_file", {"path": "c.txt", "find": "x", "replace": "y"}, "c1")), + calls(tc("read_file", {"path": "c.txt"}, "c2")), + calls(tc("edit_file", {"path": "c.txt", "find": "x", "replace": "y"}, "c3"), + tc("edit_file", {"path": "c.txt", "find": "x x x", "replace": "é"}, "c4")), + say("end")], files={"c.txt": "x x x"}, auto_approve=True) + add("delete_restore", "delete and restore", [ + calls(tc("delete_file", {"path": "d.txt"}, "c1")), + calls(tc("read_file", {"path": "d.txt"}, "c2")), + calls(tc("delete_file", {"path": "d.txt"}, "c3")), + calls(tc("audit", {}, "c4"), tc("restore", {"id": 1}, "c5"), tc("restore", {"id": 99}, "c6"), tc("restore", {"id": "x"}, "c7")), + calls(tc("read_file", {"path": "d.txt"}, "c8")), + say("restored")], files={"d.txt": "keep me"}, auto_approve=True) + add("shell_basic", "run things", [ + calls(tc("shell", {"command": "echo hello"}, "c1"), tc("shell", {"command": "ls nonexistent_dir_xyz"}, "c2"), + tc("shell", {"command": "echo curl http://example.com"}, "c3"), tc("shell", {"command": " "}, "c4"), + tc("shell", {"command": "true"}, "c5"), tc("shell", {"command": "printf 'a\\nb\\n'; exit 3"}, "c6")), + calls(tc("shell", {"command": "echo x", "secrets": "oops"}, "c7"), tc("shell", {"command": "echo x", "secrets": ["nope"]}, "c8"), + tc("shell", {"command": "pwd", "workdir": "sub"}, "c9")), + say("done")], files={"sub/keep.txt": "k"}, auto_approve=True) + add("shell_long_output", "spill", [ + calls(tc("shell", {"command": "yes line | head -n 3000"}, "c1")), say("done")], auto_approve=True) + add("unknown_and_bad_args", "odd calls", [ + calls(tc("nonexistent_tool", {}, "c1"), tc("read_file", "this is not json", "c2"), tc("read_file", "[1, 2]", "c3"), + tc("list_secrets", "[1]", "c4"), tc("bad name!", {}, "c5")), + say("ok")], auto_approve=True) + add("loop_stop", "loop", [calls(tc("read_file", {"path": "a.txt"}, "c%d" % i)) for i in range(1, 8)] + [say("never")], + files={"a.txt": "same"}, auto_approve=True) + add("secrets", "secrets", [ + calls(tc("store_secret", {"name": "api_key", "value": "supersecret123", "username": "me", "host": "h", "port": 22, "notes": "n", "tags": ["Prod"]}, "c1")), + calls(tc("list_secrets", {}, "c2"), tc("store_secret", {"name": "bad name", "value": "v"}, "c3"), tc("store_secret", {"name": "x", "value": ""}, "c4"), + tc("store_secret", {"name": "y", "value": "valid1", "port": "abc"}, "c5"), tc("store_secret", {"name": "y", "value": "valid1", "port": 70000}, "c6"), + tc("store_secret", {"name": "y", "value": "valid1", "expires": "bogus"}, "c7"), tc("store_secret", {"name": "y", "value": "valid1", "tags": "x"}, "c8")), + calls(tc("shell", {"command": "echo $TAI_SECRET_API_KEY", "secrets": ["api_key"]}, "c9")), + calls(tc("delete_secret", {"name": "api_key"}, "c10"), tc("delete_secret", {"name": "missing"}, "c11"), tc("list_secrets", {}, "c12")), + say("fin")], auto_approve=True) + add("schedules", "plan", [ + calls(tc("schedule", {"prompt": "do it", "every": 120, "name": "nightly", "tags": ["Ops"]}, "c1"), + tc("schedule", {"prompt": "x", "at": "2026-12-24T10:00"}, "c2"), + tc("schedule", {"prompt": "x"}, "c3"), tc("schedule", {"prompt": "x", "every": 5}, "c4"), + tc("schedule", {"prompt": "x", "at": "garbage"}, "c5"), tc("schedule", {"prompt": "", "every": 100}, "c6"), + tc("schedule", {"prompt": "x", "every": "abc"}, "c7"), tc("schedule", {"prompt": "x", "every": 100, "profile": "other"}, "c8")), + calls(tc("schedules", {}, "c9"), tc("unschedule", {"id": 1}, "c10"), tc("unschedule", {"id": 77}, "c11"), tc("schedules", {}, "c12")), + say("planned")], auto_approve=True) + add("bots_and_mentions", "make a bot", [ + calls(tc("create_bot", {"name": "helper-bot", "description": "I help.", "rules": "be brief", "nicknames": ["hb"]}, "c1"), + tc("create_bot", {"name": "main", "description": "x"}, "c2"), tc("create_bot", {"name": "Bad Name", "description": "x"}, "c3"), + tc("create_bot", {"name": "empty"}, "c4"), tc("create_bot", {"name": "helper-bot", "description": "dup"}, "c5")), + say("bot created"), + say("hello from the bot"), + say("plain third answer")], + prompts=["make a bot", "@hb say hello", "@nobody hi", "@ "], auto_approve=True) + add("compaction", "read all", [calls(tc("read_file", {"path": "f%d.txt" % i}, "c%d" % i)) for i in range(1, 14)] + [say("all read")], + files={"f%d.txt" % i: ("%d " % i) * 95000 for i in range(1, 14)}, single=[single("SUMMARY OF HISTORY")], auto_approve=True) + add("backend_fail", "will fail", [status(500, "upstream oops")], retry_seconds=1, auto_approve=True) + add("tools_400_fallback", "fallback", [status(400, "bad tools"), say("worked without tools")], auto_approve=True) + add("all_permanent_fail", "perm", [status(401, "no key")], auto_approve=True) + add("ask_user", "ask", [ + calls(tc("ask_user", {"question": "Pick one", "options": ["A", "B", {"label": "C"}, "", 5]}, "c1"), + tc("ask_user", {"question": "Pick again", "options": ["A", "B"]}, "c2"), + tc("ask_user", {"question": "Free?"}, "c3"), tc("ask_user", {"question": ""}, "c4"), tc("ask_user", {"question": "q", "options": "no"}, "c5"), + tc("ask_user", {"question": "closed"}, "c6")), + say("asked")], answers=["2", "b", " hello world ", None], auto_approve=True) + add("approval_deny", "dangerous", [ + calls(tc("shell", {"command": "rm -rf somewhere"}, "c1"), tc("shell", {"command": "ls"}, "c2")), say("stopped")], + answers=["Deny", "Deny"], files={"somewhere/x": "1"}) + add("yolo_always", "yolo", [calls(tc("read_file", {"path": "a.txt"}, "c1")), say("ok")], yolo=True, files={"a.txt": "a"}) + add("verify_ok", "do the thing", [say("I did it")], runner_script=['{"success": true, "failures": [], "commentary": "good"}'], verify=True, auto_approve=True) + add("verify_retry", "do the thing", [say("first try"), say("second try")], + runner_script=['{"success": false, "failures": [{"item": "x", "reason": "r", "proposal": "p"}]}', '{"success": true, "failures": []}'], + verify=True, auto_approve=True) + add("remember", "remember my name", [ + calls(tc("remember", {"instruction": "The user is called Retoor and likes concise answers."}, "c1"), tc("remember", {"instruction": ""}, "c2")), + say("remembered")], single=[single("You are tai. The user is called Retoor and likes concise answers. Be professional.")], + auto_approve=True) + add("remember_fail", "remember", [calls(tc("remember", {"instruction": "x"}, "c1")), say("sorry")], single=[single("short")], auto_approve=True) + add("record_flow", "records", [ + calls(tc("record_save", {"content": "alpha beta gamma", "title": "T1", "kind": "note", "tags": ["Zeta"]}, "c1"), + tc("record_save", {"content": "", "title": "e"}, "c2"), tc("record_save", {"content": "x", "kind": "weird"}, "c3"), + tc("record_save", {"content": "x", "tags": "no"}, "c4")), + say("saved")], auto_approve=True) + add("web_fetch", "fetch", [ + calls(tc("web_fetch", {"url": "ftp://x"}, "c1"), tc("web_fetch", {"url": "http://example.com/a", "method": "TRACE"}, "c2"), + tc("web_fetch", {"url": "http://example.com/a"}, "c3"), tc("web_fetch", {"url": "http://example.com/b", "raw": True}, "c4"), + tc("web_fetch", {"url": "http://example.com/c", "headers": {"X-A": "b"}, "method": "post", "body": "{}"}, "c5"), + tc("web_fetch", {"url": "http://example.com/d", "headers": {"bad name": "b"}}, "c6"), + tc("web_fetch", {"url": "http://example.com/e", "timeout": "abc"}, "c7"), tc("web_fetch", {"url": "http://example.com/f"}, "c8"), + tc("web_fetch", {"url": "http://example.com/g"}, "c9")), + say("fetched")], + single=[raw("

Title & more

Para text

", "text/html; charset=utf-8"), + raw("raw text", "text/plain"), raw("posted", "application/json"), + status(404, "Not here"), status(500, ""), raw("", "text/html")], auto_approve=True) + add("web_search", "search", [ + calls(tc("web_search", {"query": "rust news", "images": True}, "c1"), tc("web_search", {"query": ""}, "c2"), tc("web_search", {"query": "x"}, "c3")), + say("searched")], + single=[{"kind": "json", "body": {"results": [{"title": "T", "url": "http://u", "description": "d" * 500, "content": "c" * 3000}, {"title": None, "url": "u2"}]}}, + {"kind": "json", "body": {"results": []}}], auto_approve=True) + add("poll_fork", "fork", [ + calls(tc("fork", {"task": "inspect the repo"}, "c1"), tc("fork", {"task": ""}, "c2"), tc("fork", {"task": "x", "profile": "zzz"}, "c3")), + calls(tc("poll", {"id": 1, "wait": 5}, "c4"), tc("poll", {"id": 99}, "c5"), tc("poll", {"id": "abc"}, "c6")), + say("collected")], runner_script=["worker result text"], auto_approve=True) + add("terminals", "terminals", [ + calls(tc("list_terminals", {}, "c1"), tc("plexer_send", {"text": "x"}, "c2"), tc("plexer_status", {}, "c3"), tc("plexer_read", {}, "c4")), + say("looked")], auto_approve=True) + add("persist_off", "worker style", [calls(tc("ask_user", {"question": "q"}, "c1"), tc("read_file", {"path": "a.txt"}, "c2")), say("end")], + persist=False, files={"a.txt": "z"}, auto_approve=True) + add("max_steps_stall", "stall", [calls(tc("shell", {"command": "echo %d" % i}, "c%d" % i)) for i in range(1, 5)] + [say("end")], + max_steps=2, auto_approve=True) + add("presets_profile", "with presets", [say("ok")], presets={"systems": {"default": "You are a custom bot. " + tai.VAULT_MEMORY_NOTE + " end"}}, + auto_approve=True) + def ev(delta): + return {"choices": [{"delta": delta}]} + + add("stream_quirks", "weird stream", [ + {"kind": "stream", "events": [ev({"reasoning_content": "thinking about it"}), ev({"content": "part one "}), ev({"content": "part two"}), + ev({"tool_calls": [{"index": 0, "id": "q1", "function": {"name": "read_file", "arguments": "{\"path\": "}}]}), + ev({"tool_calls": [{"index": 5, "function": {"arguments": "\"a.txt\"}"}}]}), + {"choices": []}, {"nochoices": True}, + ev({"tool_calls": [{"index": 1, "id": "q2", "function": {"name": "shell<|channel|>commentary", "arguments": "{\"command\": \"echo\"}"}}]}), + ev({"tool_calls": [{"index": 2, "id": "q3", "function": {"name": "bad name!", "arguments": "{}"}}]}), + ev({"tool_calls": [{"index": 3, "id": "q4", "function": {"name": "list_secrets", "arguments": "oops not json"}}]})]}, + {"kind": "stream", "events": [ev({"reasoning": "only reasoning, no content"})]}], + files={"a.txt": "A"}, auto_approve=True) + add("reasoning_only", "think", [{"kind": "stream", "events": [ev({"reasoning": "deep thought"})]}], auto_approve=True) + add("empty_reply", "nothing", [{"kind": "stream", "events": []}], auto_approve=True) + add("store_errors", "probe store tools", [ + calls(tc("record_read", {"id": "mem:0000000000000000"}, "c1"), tc("record_delete", {"id": ""}, "c2"), tc("record_delete", {"id": "mem:none"}, "c3"), + tc("graph_query", {"node": "bogus"}, "c4"), tc("graph_query", {"node": "mem:0123456789abcdef"}, "c5"), + tc("graph_link", {"src": "", "dst": "x"}, "c6"), tc("tags", {}, "c7"), tc("tags", {"limit": "x"}, "c8"), + tc("search", {"query": ""}, "c9"), tc("search", {"query": "x", "kinds": ["bogus"]}, "c10"), tc("search", {"query": "x", "limit": "q"}, "c11"), + tc("record_search", {"kind": "weird"}, "c12"), tc("audit", {"limit": "z"}, "c13"), tc("release", {"part": "huge"}, "c14"), + tc("release", {"part": "patch"}, "c15"), tc("install", {"action": "bogus"}, "c16"), tc("sysinfo", {"checks": []}, "c17"), + tc("mcp_call", {}, "c18"), tc("mcp_enable", {"server": "nope"}, "c19"), tc("mcp_list", {}, "c20"), tc("mcp_tools", {"server": "nope"}, "c21"), + tc("recall", {"query": ""}, "c22"), tc("recall", {"query": "x", "tags": "no"}, "c23"), tc("load_skill", {"name": "nope"}, "c24")), + say("probed")], auto_approve=True) + add("deadline", "late", [say("never reached")], deadline_expired=True, auto_approve=True) + add("mentions", "pins", [say("one"), say("two"), say("three"), say("four"), say("five")], + prompts=["@lab hello", "@m1 hello", "@a.txt what is this", "@auto hi", "@lab"], files={"a.txt": "A"}, auto_approve=True) + add("persist_resume", "session restore", [say("first"), say("second")], prompts=["one", "two"], auto_approve=True) + add("tool_error_text", "bad things", [ + calls(tc("describe_image", {"source": ""}, "c1"), tc("speak", {"text": ""}, "c2"), tc("create_skill", {"name": "Bad Name", "brief": "x"}, "c3"), + tc("create_skill", {"name": "ok-name", "brief": ""}, "c4"), tc("install_skill", {"source": ""}, "c5"), + tc("plexer_type", {"text": ""}, "c6"), tc("plexer_show", {"spec": "no"}, "c7"), tc("tui_screen", {}, "c8"), + tc("export_terminal_log", {}, "c9"), tc("get_current_terminal_content", {"lines": "x", "target": "nonexistent-pane"}, "c10"), + tc("tail_terminals", {}, "c11"), tc("web_fetch", {"url": "http://x", "headers": {"A": 1}}, "c12"), + tc("web_fetch", {"url": "http://x", "body": 5}, "c13"), tc("write_file", {"path": "/proc/forbidden/x", "content": "y"}, "c14")), + say("done")], auto_approve=True) + add("legacy_note", "legacy", [say("ok")], + presets={"systems": {"default": "Old " + tai.LEGACY_PASSWORD_NOTE + " tail"}}, auto_approve=True) + return scenarios + + +def run_all(args): + scenarios = build_scenarios() + failures = 0 + for sc in scenarios: + if args.filter and args.filter not in sc["name"]: + continue + results = [] + for who in ("python", "cpp"): + base = tempfile.mkdtemp(prefix="agent-e2e-") + workdir = os.path.join(base, "work") + os.makedirs(workdir) + for rel, content in (sc.get("files") or {}).items(): + path = os.path.join(workdir, rel) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + handle.write(content) + try: + if who == "python": + results.append(normalize(python_run(sc, workdir))) + else: + results.append(normalize(cpp_run(sc, workdir, args.bin))) + except Exception as exc: # noqa: BLE001 + results.append({"harness_error": "%s: %s" % (type(exc).__name__, exc)}) + finally: + os.chdir(ROOT) + shutil.rmtree(base, ignore_errors=True) + py, cpp = results + # workdir path differs per run: neutralise it + py_text = re.sub(r"/tmp/agent-e2e-[A-Za-z0-9_]+/work", "", json.dumps(py, sort_keys=True)) + cpp_text = re.sub(r"/tmp/agent-e2e-[A-Za-z0-9_]+/work", "", json.dumps(cpp, sort_keys=True)) + if py_text == cpp_text: + sys.stdout.write("ok %s\n" % sc["name"]) + continue + failures += 1 + sys.stdout.write("FAIL %s\n" % sc["name"]) + pyd, cppd = json.loads(py_text), json.loads(cpp_text) + for key in sorted(set(pyd) | set(cppd)): + if pyd.get(key) != cppd.get(key): + a = json.dumps(pyd.get(key), sort_keys=True) + b = json.dumps(cppd.get(key), sort_keys=True) + if args.verbose: + sys.stdout.write(" [%s]\n python: %s\n cpp: %s\n" % (key, a[:6000], b[:6000])) + else: + # first differing position + pos = next((i for i in range(min(len(a), len(b))) if a[i] != b[i]), min(len(a), len(b))) + sys.stdout.write(" [%s] differs at char %d\n python: ...%s\n cpp: ...%s\n" % (key, pos, a[max(0, pos - 150):pos + 250], b[max(0, pos - 150):pos + 250])) + sys.stdout.write("%d failing scenario(s)\n" % failures) + return 1 if failures else 0 + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--bin", default=os.path.join(HERE, "..", "build", "agent_e2e")) + parser.add_argument("--filter", default="") + parser.add_argument("-v", "--verbose", action="store_true") + args = parser.parse_args() + return run_all(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/taicpp/tests/agent_memstore.h b/taicpp/tests/agent_memstore.h new file mode 100644 index 0000000..e8375f9 --- /dev/null +++ b/taicpp/tests/agent_memstore.h @@ -0,0 +1,353 @@ +#pragma once +// In-memory StorePort used by the agent tests. It implements the Store +// behaviour the covered scenarios exercise (sessions, systems, bots, secrets, +// redaction, schedules, audit, file records, events, turn state, records); +// FTS-backed searches are not emulated and throw. +#include "../src/agent/ports.h" +#include "../src/agent/text.h" +#include "../src/llm/helpers.h" +#include "../src/llm/py.h" +#include +#include +#include +namespace detz::agent::testing { + +class MemStore : public StorePort { + public: + struct Event { std::string profile, role, kind, text; }; + struct Audit { + int64_t id; std::string profile, ts, actor, action, path, message, tags; + std::optional old_size, new_size; + std::optional old_text, new_text; + }; + std::vector events; + std::vector audits; + std::map file_records; // path -> content (kind file) + std::map dir_defaults_log; + + std::string profile() const override { return profile_; } + void set_profile(const std::string& name) override { profile_ = name; } + bool seal_enabled() const override { return seal_; } + bool fts_ok() const override { return false; } + void set_seal_enabled(bool v) { seal_ = v; } + + std::string redact(const std::string& text, const std::string& profile) override { + std::string out = text; + auto& vault = secrets_[scope(profile)]; + std::vector> items; + for (const auto& n : vault.order) { + const auto& value = vault.values.at(n).value; + if (py::len(value) >= 4) { items.emplace_back(n, value); } + } + std::stable_sort(items.begin(), items.end(), [](const auto& a, const auto& b) { return py::len(a.second) > py::len(b.second); }); + for (const auto& it : items) { + if (out.find(it.second) != std::string::npos) { out = py::replace(out, it.second, "[redacted:" + it.first + "]"); } + } + return out; + } + void log_event(const std::string& profile, const std::string& role, const std::string& kind, const std::string& text) override { + events.push_back({scope(profile), role, kind, text}); + } + + Value load_session(const std::string& name, const std::string& bot, const std::string&) override { + Value out = Value::array(); + auto it = sessions_.find(name + "|" + bot); + if (it == sessions_.end()) { return out; } + for (const auto& m : it->second.items()) { + const Value& r = m.get("role"); + if (m.is_object() && r.is_string() && (r.str() == "user" || r.str() == "assistant" || r.str() == "tool")) { out.push(m); } + } + return out; + } + void save_session(const std::string& name, const Value& messages, const std::string& bot, const std::string&) override { + Value kept = Value::array(); + for (const auto& m : messages.items()) { + if (!(m.get("role").is_string() && m.get("role").str() == "system")) { kept.push(m); } + } + sessions_[name + "|" + bot] = Value::loads(kept.dumps()); + } + std::optional load_system(const std::string& name) override { + auto it = systems_.find(name); + if (it == systems_.end()) { return std::nullopt; } + return it->second; + } + void save_system(const std::string& name, const std::string& text) override { systems_[name] = text; } + std::optional load_bot_system(const std::string& profile, const std::string& bot) override { + if (bot == "main") { return load_system(profile); } + auto it = bot_systems_.find(profile + "|" + bot); + if (it == bot_systems_.end()) { return std::nullopt; } + return it->second; + } + void save_bot_system(const std::string& profile, const std::string& bot, const std::string& text) override { + bot_systems_[profile + "|" + bot] = text; + } + Value load_bots(const std::string& profile) override { + auto it = bots_.find(profile); + return it == bots_.end() ? Value::object() : Value::loads(it->second.dumps()); + } + void save_bots(const std::string& profile, const Value& bots) override { bots_[profile] = Value::loads(bots.dumps()); } + Value list_bots(const std::string& profile) override { + Value bots = load_bots(profile); + Value found = Value::array(); + Value main = Value::object(); + main.set("name", "main"); + main.set("nicknames", Value::array()); + found.push(main); + std::vector names; + for (const auto& kv : bots.members()) { names.push_back(kv.first); } + std::sort(names.begin(), names.end()); + for (const auto& n : names) { + Value b = Value::object(); + b.set("name", n); + const Value& nick = bots.get(n).get("nicknames"); + b.set("nicknames", nick.is_array() ? nick : Value::array()); + found.push(b); + } + return found; + } + std::optional resolve_bot(const std::string& profile, const std::string& mention) override { + std::string want = py::lower(py::strip(mention)); + if (want.empty() || want == "main") { return "main"; } + Value bots = load_bots(profile); + if (bots.has(want)) { return want; } + std::vector names; + for (const auto& kv : bots.members()) { names.push_back(kv.first); } + std::sort(names.begin(), names.end()); + for (const auto& n : names) { + for (const auto& nick : bots.get(n).get("nicknames").items()) { if (nick.is_string() && nick.str() == want) { return n; } } + } + return std::nullopt; + } + Value load_turn_state(const std::string& profile, const std::string& bot, const std::string&) override { + auto it = turns_.find(profile + "|" + bot); + return it == turns_.end() ? Value::object() : it->second; + } + void save_turn_state(const std::string& profile, const std::string& bot, const Value& state, const std::string&) override { + turns_[profile + "|" + bot] = Value::loads(state.dumps()); + } + void save_dir_defaults(const std::string& directory, const std::optional& provider, + const std::optional& model, const std::optional& bot, + const std::vector& clear) override { + Value o = Value::object(); + o.set("provider", provider ? Value(*provider) : Value()); + o.set("model", model ? Value(*model) : Value()); + o.set("bot", bot ? Value(*bot) : Value()); + Value c = Value::array(); + for (const auto& x : clear) { c.push(x); } + o.set("clear", c); + dir_defaults_log[directory] = o; + } + + std::string add_record(const std::string& kind, const std::string& title, const std::string& content, + const std::vector& tags, const std::string& profile) override { + std::string id = "mem:" + py::uuid4_hex().substr(0, 16); + Rec r{id, kind, title, content, {"record"}, 0}; + for (const auto& t : normalize_tags(tags)) { if (std::find(r.tags.begin(), r.tags.end(), t) == r.tags.end()) { r.tags.push_back(t); } } + records_[scope(profile)].push_back(r); + return id; + } + void add_chunks(const std::string& kind, const std::string& id, const std::vector& texts, const std::string& profile) override { + chunks_[scope(profile) + "|" + kind + "|" + id] = texts; + } + Value get_chunks_by_ids(const std::vector&, const std::string&) override { return Value::object(); } + Value read_record(const std::string& id, int64_t offset, int64_t limit, const std::string& profile) override { + for (auto& r : records_[scope(profile)]) { + if (r.id != id) { continue; } + int64_t total = static_cast(py::len(r.content)); + int64_t start = std::max(0, std::min(offset, total)); + int64_t end = std::min(total, start + std::max(1, limit)); + r.reads++; + Value o = Value::object(); + o.set("id", id); + o.set("slice", py::slice(r.content, start, end)); + o.set("start", start); o.set("end", end); o.set("total", total); + Value t = Value::array(); + for (const auto& x : r.tags) { t.push(x); } + o.set("tags", t); + return o; + } + return Value(); + } + Value search_records(const std::string&, const std::optional&, const std::vector&, int64_t, const std::string&) override { + throw std::runtime_error("MemStore: search_records unsupported"); + } + bool delete_record(const std::string& id, const std::string& profile) override { + auto& v = records_[scope(profile)]; + for (auto it = v.begin(); it != v.end(); ++it) { + if (it->id == id) { v.erase(it); return true; } + } + return false; + } + std::vector item_tags(const std::string& item, const std::string& profile) override { + for (const auto& r : records_[scope(profile)]) { if (r.id == item) { return r.tags; } } + return {}; + } + std::vector> tag_counts(const std::string&, int64_t, const std::string&) override { return {}; } + std::string add_edge(const std::string&, const std::string&, const Value&, const std::string&) override { + throw std::runtime_error("MemStore: add_edge unsupported"); + } + Value edges_for(const std::string&, const std::string&) override { return Value::array(); } + bool has_node(const std::string&, const std::string&, const std::string&) override { return false; } + std::string node_title(const std::string& node, const std::string&) override { return node; } + Value traverse(const std::string&, int64_t, int64_t, const std::string&) override { return Value::array(); } + Value fts_search(const std::string&, const std::vector&, const std::string&, int64_t) override { return Value::array(); } + Value events_like(const std::string&, const std::string&, int64_t) override { return Value::array(); } + Value search_events(const std::string&, const std::string&, const std::vector&) override { return Value::array(); } + Value audit_search(const std::string&, const std::string&, int64_t) override { return Value::array(); } + + int64_t audit_event(const std::string& actor, const std::string& action, const std::string& path, const std::string& message, + const std::optional& old_text, const std::optional& new_text, + const std::vector& tags, const std::optional& old_size, + const std::optional& new_size, const std::string& profile) override { + Capped o = capped_text(old_text), n = capped_text(new_text); + std::optional ot = o.text, nt = n.text; + std::optional os = o.size, ns = n.size; + if (old_text.has_value() && old_size.has_value() && *old_size > static_cast(old_text->size())) { + ot = py::head(*old_text, 20000) + trunc_marker(*old_size); + os = old_size; + } + if (new_text.has_value() && new_size.has_value() && *new_size > static_cast(new_text->size())) { + nt = py::head(*new_text, 20000) + trunc_marker(*new_size); + ns = new_size; + } + Audit a; + a.id = static_cast(audits.size()) + 1; + a.profile = scope(profile); + a.ts = py::utc_now_iso(); + a.actor = actor; a.action = action; a.path = path; + a.message = py::head(message, 2000); + a.tags = py::join(normalize_tags(tags), " "); + a.old_size = os; a.new_size = ns; a.old_text = ot; a.new_text = nt; + audits.push_back(a); + return a.id; + } + Value audit_history(const std::optional& path, const std::optional& tag, int64_t limit, + const std::string& profile) override { + Value out = Value::array(); + int64_t lim = std::max(1, std::min(limit, 100)); + for (auto it = audits.rbegin(); it != audits.rend() && static_cast(out.size()) < lim; ++it) { + if (it->profile != scope(profile)) { continue; } + if (path.has_value() && it->path != *path) { continue; } + if (tag.has_value() && it->tags.find(*tag) == std::string::npos) { continue; } + out.push(audit_row(*it, false)); + } + return out; + } + Value audit_get(int64_t row_id, const std::string& profile) override { + for (const auto& a : audits) { if (a.id == row_id && a.profile == scope(profile)) { return audit_row(a, true); } } + return Value(); + } + void upsert_file_record(const std::string& path, const std::string& content, const std::optional&, const std::string&) override { + file_records[path] = content; + } + void delete_file_record(const std::string& path, const std::string&) override { file_records.erase(path); } + + std::optional load_secret(const std::string& name, const std::string& profile) override { + auto& v = secrets_[scope(profile)]; + auto it = v.values.find(name); + if (it == v.values.end()) { return std::nullopt; } + return it->second.value; + } + void save_secret(const std::string& name, const std::string& value, const Value& meta, const std::vector& tags, + const std::string& profile) override { + auto& v = secrets_[scope(profile)]; + if (!v.values.count(name)) { v.order.push_back(name); } + std::vector all = {"secret"}; + for (const auto& t : normalize_tags(tags)) { if (std::find(all.begin(), all.end(), t) == all.end()) { all.push_back(t); } } + v.values[name] = {value, meta, all}; + } + Value list_secret_infos(const std::string& profile) override { + auto& v = secrets_[scope(profile)]; + std::vector names = v.order; + std::sort(names.begin(), names.end()); + Value out = Value::array(); + for (const auto& n : names) { + Value o = Value::object(); + o.set("name", n); + o.set("meta", v.values.at(n).meta); + Value t = Value::array(); + std::vector sorted_tags = v.values.at(n).tags; + std::sort(sorted_tags.begin(), sorted_tags.end()); + for (const auto& x : sorted_tags) { t.push(x); } + o.set("tags", t); + o.set("updated", ""); + out.push(o); + } + return out; + } + void delete_secret(const std::string& name, const std::string& profile) override { + auto& v = secrets_[scope(profile)]; + v.values.erase(name); + v.order.erase(std::remove(v.order.begin(), v.order.end(), name), v.order.end()); + } + + int64_t add_schedule(const std::string& name, const std::string& prompt, const std::string& profile, int64_t every_sec, + const std::string& next_run, int64_t timeout, const std::vector& tags) override { + Value o = Value::object(); + int64_t id = ++sched_id_; + o.set("id", id); o.set("name", name); o.set("prompt", prompt); o.set("profile", profile); + o.set("every", every_sec); o.set("next_run", next_run); o.set("timeout", timeout); + o.set("status", "pending"); o.set("last", ""); o.set("result", ""); + Value t = Value::array(); + t.push("schedule"); + for (const auto& x : tags) { t.push(x); } + o.set("tags", t); + schedules_.push_back(o); + return id; + } + void remove_schedule(int64_t row_id, const std::string& profile) override { + for (auto it = schedules_.begin(); it != schedules_.end(); ++it) { + if (it->get("id").as_int() == row_id && it->get("profile").str() == scope(profile)) { schedules_.erase(it); return; } + } + } + Value list_schedules(const std::string& profile) override { + Value out = Value::array(); + std::vector rows; + for (const auto& s : schedules_) { if (s.get("profile").str() == scope(profile)) { rows.push_back(s); } } + std::stable_sort(rows.begin(), rows.end(), [](const Value& a, const Value& b) { return a.get("next_run").str() < b.get("next_run").str(); }); + for (const auto& r : rows) { out.push(r); } + return out; + } + + Value mcp_list_servers() override { return Value::array(); } + Value mcp_get_server(const std::string&) override { return Value(); } + void mcp_save_server(const Value&) override {} + void mcp_remove_server(const std::string&) override {} + void mcp_set_enabled(const std::string&, const std::string&, bool) override {} + std::vector mcp_enabled_profiles(const std::string&) override { return {}; } + std::vector mcp_server_profiles(const std::string&) override { return {}; } + void mcp_store_tools(const std::string&, const Value&) override {} + Value mcp_cached_tools(const std::string&) override { return Value::array(); } + std::vector mcp_tool_pins(const std::string&) override { return {}; } + Value sql_rows(const std::string&, const Value&) override { throw std::runtime_error("MemStore: sql unsupported"); } + + private: + struct Rec { std::string id, kind, title, content; std::vector tags; int64_t reads; }; + struct Sec { std::string value; Value meta; std::vector tags; }; + struct Vault { std::vector order; std::map values; }; + std::string scope(const std::string& p) const { return p.empty() ? profile_ : p; } + Value audit_row(const Audit& a, bool full) const { + Value o = Value::object(); + o.set("id", a.id); o.set("ts", a.ts); o.set("actor", a.actor); o.set("action", a.action); o.set("path", a.path); + o.set("message", a.message); + o.set("old_size", a.old_size ? Value(*a.old_size) : Value()); + o.set("new_size", a.new_size ? Value(*a.new_size) : Value()); + if (full) { + o.set("old", a.old_text ? Value(*a.old_text) : Value()); + o.set("new", a.new_text ? Value(*a.new_text) : Value()); + } + o.set("tags", a.tags); + return o; + } + std::string profile_{"default"}; + bool seal_{true}; + std::map sessions_, turns_, bots_; + std::map systems_, bot_systems_; + std::map> records_; + std::map> chunks_; + std::map secrets_; + std::vector schedules_; + int64_t sched_id_{0}; +}; + +} // namespace detz::agent::testing diff --git a/taicpp/tests/agent_sched.cpp b/taicpp/tests/agent_sched.cpp new file mode 100644 index 0000000..81c9cd6 --- /dev/null +++ b/taicpp/tests/agent_sched.cpp @@ -0,0 +1,91 @@ +// Scheduler driver: reads {"rows": [...]} from stdin, runs scheduler_tick over +// an in-memory schedules table with a stub runner and prints the final table. +#include "../src/agent/scheduler.h" +#include "../src/llm/py.h" +#include +#include +#include +using namespace detz; +using json::Value; + +namespace { +struct Row { + int64_t id; std::string name, prompt, profile; int64_t every_sec; std::string next_run; int64_t timeout; + std::string status, last_status, last_result, updated; +}; + +class MemSchedulerDb : public agent::SchedulerDbPort { + public: + std::vector rows; + std::vector select_due(const std::string& now_iso) override { + std::vector out; + for (const auto& r : rows) { + if (r.status == "pending" && r.next_run <= now_iso) { + out.push_back({r.id, r.prompt, r.profile, r.every_sec, r.next_run, r.timeout, r.last_status, r.updated}); + } + } + std::stable_sort(out.begin(), out.end(), [](const auto& a, const auto& b) { return a.next_run < b.next_run; }); + return out; + } + bool claim_every(int64_t id, const std::string& new_next, const std::string& stamp, const std::string& expected) override { + for (auto& r : rows) { + if (r.id == id && r.status == "pending" && r.next_run == expected) { + r.next_run = new_next; r.updated = stamp; r.last_status = "running"; + return true; + } + } + return false; + } + bool claim_once(int64_t id, const std::string& stamp) override { + for (auto& r : rows) { + if (r.id == id && r.status == "pending") { r.status = "done"; r.updated = stamp; return true; } + } + return false; + } + std::string unlock_prompt(const std::string& prompt) override { + if (prompt.rfind("SEALED:", 0) == 0) { + if (prompt == "SEALED:bad") { throw agent::SealFailure("cannot unlock"); } + return prompt.substr(7); + } + return prompt; + } + void release_every(int64_t id, const std::string& next_run, const std::string& stamp, const std::string& last_status) override { + for (auto& r : rows) { if (r.id == id) { r.next_run = next_run; r.updated = stamp; r.last_status = last_status; } } + } + void release_once(int64_t id, const std::string& stamp) override { + for (auto& r : rows) { if (r.id == id) { r.status = "pending"; r.updated = stamp; } } + } + void note_result(int64_t id, const std::string& status, const std::string& result) override { + for (auto& r : rows) { if (r.id == id) { r.last_status = status; r.last_result = result; r.updated = py::utc_now_iso(); } } + } +}; +} // namespace + +int main() { + std::string text((std::istreambuf_iterator(std::cin)), std::istreambuf_iterator()); + Value in = Value::loads(text); + MemSchedulerDb db; + for (const auto& r : in.get("rows").items()) { + db.rows.push_back({r.get("id").as_int(), r.get("name").str(), r.get("prompt").str(), r.get("profile").str(), r.get("every_sec").as_int(), + r.get("next_run").str(), r.get("timeout").as_int(), "pending", r.get("last_status").str(), "", r.get("updated").str()}); + } + agent::AgentConfig cfg; + int fired = 0; + agent::Runner runner = [](const std::string& task, const std::string&, double) { return "ran: " + task; }; + fired = agent::scheduler_tick(cfg, db, runner); + // let the watcher threads record their outcome + for (int i = 0; i < 100 && agent::running_agents() > 0; ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(20)); } + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + Value out = Value::object(); + out.set("fired", fired); + Value rows = Value::array(); + for (const auto& r : db.rows) { + Value o = Value::object(); + o.set("id", r.id); o.set("status", r.status); o.set("next_run", r.next_run); + o.set("last_status", r.last_status); o.set("last_result", r.last_result); + rows.push(o); + } + out.set("rows", rows); + std::cout << out.dumps() << "\n"; + return 0; +} diff --git a/taicpp/tests/agent_sched.py b/taicpp/tests/agent_sched.py new file mode 100644 index 0000000..8d3bdc5 --- /dev/null +++ b/taicpp/tests/agent_sched.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python3 +"""Differential test of scheduler_tick: tai.py over a real sqlite vault versus +the C++ scheduler over an in-memory table, same rows, stub runner.""" +import json +import os +import subprocess +import sys +import tempfile +import time +import types +from datetime import datetime, timedelta, timezone + +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.abspath(os.path.join(HERE, "..", "..")) +sys.path.insert(0, ROOT) +sys.dont_write_bytecode = True +import tai # noqa: E402 + + +def main(): + binary = os.path.join(HERE, "..", "build", "agent_sched") + home = tempfile.mkdtemp(prefix="agent-sched-") + for key in list(os.environ): + if key.startswith("TAI_"): + del os.environ[key] + os.environ["TAI_HOME"] = home + config = tai.Config(types.SimpleNamespace(yes=True, profile="default", no_opencode=True)) + seal = tai.Seal(config.home, tai.resolve_passphrase()[0]) + store = tai.Store(config, seal) + now = datetime.now(timezone.utc) + past = lambda **kw: (now - timedelta(**kw)).isoformat() # noqa: E731 + rows = [ + {"id": 1, "name": "once", "prompt": "do once", "profile": "default", "every_sec": 0, "next_run": past(minutes=5), "timeout": 60, "last_status": "", "updated": past(hours=1)}, + {"id": 2, "name": "hourly", "prompt": "do hourly", "profile": "default", "every_sec": 3600, "next_run": "2026-01-01T00:00:00+00:00", "timeout": 60, "last_status": "", "updated": past(hours=1)}, + {"id": 3, "name": "busy", "prompt": "still running", "profile": "default", "every_sec": 600, "next_run": past(minutes=1), "timeout": 600, "last_status": "running", "updated": past(seconds=30)}, + {"id": 4, "name": "stale", "prompt": "lease expired", "profile": "default", "every_sec": 600, "next_run": past(seconds=1500), "timeout": 60, "last_status": "running", "updated": past(hours=3)}, + {"id": 5, "name": "future", "prompt": "later", "profile": "default", "every_sec": 0, "next_run": (now + timedelta(hours=2)).isoformat(), "timeout": 60, "last_status": "", "updated": past(hours=1)}, + {"id": 6, "name": "badtime", "prompt": "bad", "profile": "default", "every_sec": 60, "next_run": "2020-01-01 nonsense", "timeout": 60, "last_status": "", "updated": past(hours=1)}, + {"id": 7, "name": "micro", "prompt": "micro", "profile": "default", "every_sec": 90, "next_run": "2026-03-01T10:00:00.250000+02:00", "timeout": 60, "last_status": "", "updated": past(hours=1)}, + ] + db = sqlite3_connect(config.db_path) + for r in rows: + db.execute("INSERT INTO schedules (id, name, prompt, profile, every_sec, next_run, timeout, status, last_status, last_result, created, updated) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, '', ?, ?)", + (r["id"], r["name"], r["prompt"], r["profile"], r["every_sec"], r["next_run"], r["timeout"], r["last_status"], r["updated"], r["updated"])) + db.commit() + db.close() + fired = tai.scheduler_tick(config, seal, runner=lambda task, profile, timeout: "ran: " + task) + deadline = time.time() + 5 + while tai.running_agents() and time.time() < deadline: + time.sleep(0.02) + time.sleep(0.4) + db = sqlite3_connect(config.db_path) + got = db.execute("SELECT id, status, next_run, last_status, last_result FROM schedules ORDER BY id").fetchall() + db.close() + expected = {"fired": fired, "rows": [{"id": r[0], "status": r[1], "next_run": r[2], "last_status": r[3], "last_result": r[4]} for r in got]} + done = subprocess.run([binary], input=json.dumps({"rows": rows}).encode(), stdout=subprocess.PIPE, stderr=subprocess.PIPE) + if done.returncode != 0: + print("binary failed", done.stderr.decode()[-500:]) + return 2 + actual = json.loads(done.stdout) + ok = json.dumps(expected, sort_keys=True) == json.dumps(actual, sort_keys=True) + if not ok: + for e, a in zip(expected["rows"], actual["rows"]): + if e != a: + print("row differs\n python:", e, "\n cpp: ", a) + print("fired python=%s cpp=%s" % (expected["fired"], actual["fired"])) + print("scheduler: %s (fired %d)" % ("ok" if ok else "MISMATCH", expected["fired"])) + return 0 if ok else 1 + + +def sqlite3_connect(path): + import sqlite3 + return sqlite3.connect(path, timeout=30) + + +if __name__ == "__main__": + sys.exit(main())