agent: scheduler, tui control client, e2e and scheduler differential tests
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5.5
@@ -6,6 +6,7 @@
|
||||
#include "consts.h"
|
||||
#include "data.h"
|
||||
#include "text.h"
|
||||
#include "tui_control.h"
|
||||
#include <algorithm>
|
||||
#include <deque>
|
||||
#include <filesystem>
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -302,12 +302,40 @@ class OpsPort {
|
||||
virtual std::tuple<std::string, std::string, std::string, int64_t> do_release(const std::string& part, const std::string& message,
|
||||
const std::string& actor) = 0;
|
||||
virtual std::string collect_sysinfo(const std::optional<std::vector<std::string>>& checks) = 0;
|
||||
virtual std::vector<std::string> 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<DueSchedule> 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();
|
||||
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
#include "scheduler.h"
|
||||
#include "../llm/helpers.h"
|
||||
#include "../llm/py.h"
|
||||
#include "../llm/term.h"
|
||||
#include "consts.h"
|
||||
#include <chrono>
|
||||
#include <cmath>
|
||||
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<int64_t>(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<double>(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<DueSchedule> 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<double>(row.every_sec)) + 1.0;
|
||||
e += steps * static_cast<double>(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<double>(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<double>(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
|
||||
@@ -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 <atomic>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
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<bool> stop_{false};
|
||||
std::thread worker_;
|
||||
};
|
||||
|
||||
} // namespace detz::agent
|
||||
@@ -39,7 +39,7 @@ std::optional<std::vector<std::string>> 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();
|
||||
|
||||
@@ -5,6 +5,7 @@
|
||||
#include "../llm/term.h"
|
||||
#include "consts.h"
|
||||
#include "text.h"
|
||||
#include "tui_control.h"
|
||||
#include "zipfile.h"
|
||||
#include <algorithm>
|
||||
#include <cmath>
|
||||
@@ -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<std::string> 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<std::string> valid = ports().ops->sysinfo_check_names();
|
||||
std::vector<std::string> 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);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
#include "tui_control.h"
|
||||
#include "../llm/proc.h"
|
||||
#include "../llm/py.h"
|
||||
#include "text.h"
|
||||
#include <cerrno>
|
||||
#include <cstring>
|
||||
#include <poll.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/un.h>
|
||||
#include <unistd.h>
|
||||
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<sockaddr*>(&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<size_t>(w);
|
||||
}
|
||||
std::string data;
|
||||
while (data.empty() || data.back() != '\n') {
|
||||
struct pollfd pfd{fd, POLLIN, 0};
|
||||
int pr = ::poll(&pfd, 1, static_cast<int>(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<size_t>(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
|
||||
@@ -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 <string>
|
||||
namespace detz::agent {
|
||||
|
||||
inline constexpr int GRR_CONTROL_WAIT_MAX = 60;
|
||||
// TAI_TUI_SOCK or <TAI_HOME or ~/.tai>/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
|
||||
@@ -658,6 +658,15 @@ double to_epoch(const DateTime& dt) {
|
||||
// naive: CPython's local_to_seconds with fold=0
|
||||
return static_cast<double>(naive_local_to_epoch(secs)) + frac;
|
||||
}
|
||||
DateTime from_epoch(double epoch, int offset_seconds, bool aware, int micro) {
|
||||
int64_t secs = static_cast<int64_t>(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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<Value()>& f) {
|
||||
}
|
||||
}
|
||||
|
||||
class FailTransport : public llm::HttpTransport {
|
||||
public:
|
||||
llm::HttpResult send(const llm::HttpRequest&) override { llm::HttpResult r; r.error = "<urlopen error fake>"; return r; }
|
||||
llm::HttpResult stream(const llm::HttpRequest&, const llm::LineSink&) override { llm::HttpResult r; r.error = "<urlopen error fake>"; return r; }
|
||||
};
|
||||
|
||||
std::vector<std::string> str_list(const Value& v) {
|
||||
std::vector<std::string> 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<double> 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<Value(const Value&)>;
|
||||
std::map<std::string, Fn> registry() {
|
||||
std::map<std::string, Fn> r;
|
||||
r["chat_candidates"] = chat_candidates;
|
||||
r["json_roundtrip"] = [](const Value& a) {
|
||||
return try_value([&]() { return S(Value::loads(a.at(0).str()).dumps()); });
|
||||
};
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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 <cstdio>
|
||||
#include <filesystem>
|
||||
#include <fstream>
|
||||
#include <iostream>
|
||||
#include <sstream>
|
||||
#include <unistd.h>
|
||||
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 = "<urlopen error fake>";
|
||||
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 = "<urlopen error scripted>"; 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<std::string, std::string>& 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<std::string>&, agent::LiveStreamPort*, const agent::AskFn&) override {
|
||||
throw agent::PortNotWired("shell argv");
|
||||
}
|
||||
std::unique_ptr<agent::LiveStreamPort> make_live_stream() override { return nullptr; }
|
||||
std::vector<std::string> 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<std::string>&, const std::optional<std::string>&, double) override {
|
||||
throw agent::PortNotWired("box");
|
||||
}
|
||||
std::unique_ptr<agent::LaneBoardPort> 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<std::string> skill_extra_files(const std::string&) override { return {}; }
|
||||
std::optional<Value> 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::string>&) -> std::optional<std::string> {
|
||||
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<int>(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;
|
||||
}
|
||||
@@ -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("<html><head><style>p{}</style><script>var x=1;</script></head><body><h1>Title & more</h1><p>Para text</p></body></html>", "text/html; charset=utf-8"),
|
||||
raw("<b>raw</b> 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", "<WORK>", json.dumps(py, sort_keys=True))
|
||||
cpp_text = re.sub(r"/tmp/agent-e2e-[A-Za-z0-9_]+/work", "<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())
|
||||
@@ -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 <algorithm>
|
||||
#include <map>
|
||||
#include <random>
|
||||
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<int64_t> old_size, new_size;
|
||||
std::optional<std::string> old_text, new_text;
|
||||
};
|
||||
std::vector<Event> events;
|
||||
std::vector<Audit> audits;
|
||||
std::map<std::string, std::string> file_records; // path -> content (kind file)
|
||||
std::map<std::string, Value> 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<std::pair<std::string, std::string>> 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<std::string> 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<std::string> 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<std::string> 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<std::string> 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<std::string> 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<std::string>& provider,
|
||||
const std::optional<std::string>& model, const std::optional<std::string>& bot,
|
||||
const std::vector<std::string>& 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<std::string>& 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<std::string>& texts, const std::string& profile) override {
|
||||
chunks_[scope(profile) + "|" + kind + "|" + id] = texts;
|
||||
}
|
||||
Value get_chunks_by_ids(const std::vector<int64_t>&, 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<int64_t>(py::len(r.content));
|
||||
int64_t start = std::max<int64_t>(0, std::min(offset, total));
|
||||
int64_t end = std::min(total, start + std::max<int64_t>(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<std::string>&, const std::vector<std::string>&, 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<std::string> 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<std::pair<std::string, int64_t>> 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<std::string>&, 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<std::string>&) 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<std::string>& old_text, const std::optional<std::string>& new_text,
|
||||
const std::vector<std::string>& tags, const std::optional<int64_t>& old_size,
|
||||
const std::optional<int64_t>& new_size, const std::string& profile) override {
|
||||
Capped o = capped_text(old_text), n = capped_text(new_text);
|
||||
std::optional<std::string> ot = o.text, nt = n.text;
|
||||
std::optional<int64_t> os = o.size, ns = n.size;
|
||||
if (old_text.has_value() && old_size.has_value() && *old_size > static_cast<int64_t>(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<int64_t>(new_text->size())) {
|
||||
nt = py::head(*new_text, 20000) + trunc_marker(*new_size);
|
||||
ns = new_size;
|
||||
}
|
||||
Audit a;
|
||||
a.id = static_cast<int64_t>(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<std::string>& path, const std::optional<std::string>& tag, int64_t limit,
|
||||
const std::string& profile) override {
|
||||
Value out = Value::array();
|
||||
int64_t lim = std::max<int64_t>(1, std::min<int64_t>(limit, 100));
|
||||
for (auto it = audits.rbegin(); it != audits.rend() && static_cast<int64_t>(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<int64_t>&, 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<std::string> 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<std::string>& tags,
|
||||
const std::string& profile) override {
|
||||
auto& v = secrets_[scope(profile)];
|
||||
if (!v.values.count(name)) { v.order.push_back(name); }
|
||||
std::vector<std::string> 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<std::string> 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<std::string> 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<std::string>& 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<Value> 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<std::string> mcp_enabled_profiles(const std::string&) override { return {}; }
|
||||
std::vector<std::string> 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<std::string> 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<std::string> tags; int64_t reads; };
|
||||
struct Sec { std::string value; Value meta; std::vector<std::string> tags; };
|
||||
struct Vault { std::vector<std::string> order; std::map<std::string, Sec> 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<std::string, Value> sessions_, turns_, bots_;
|
||||
std::map<std::string, std::string> systems_, bot_systems_;
|
||||
std::map<std::string, std::vector<Rec>> records_;
|
||||
std::map<std::string, std::vector<std::string>> chunks_;
|
||||
std::map<std::string, Vault> secrets_;
|
||||
std::vector<Value> schedules_;
|
||||
int64_t sched_id_{0};
|
||||
};
|
||||
|
||||
} // namespace detz::agent::testing
|
||||
@@ -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 <chrono>
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
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<Row> rows;
|
||||
std::vector<agent::DueSchedule> select_due(const std::string& now_iso) override {
|
||||
std::vector<agent::DueSchedule> 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<char>(std::cin)), std::istreambuf_iterator<char>());
|
||||
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;
|
||||
}
|
||||
@@ -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())
|
||||
Reference in New Issue
Block a user