From 554a5ddd4875b313af2629f1641c04bf84ee487f Mon Sep 17 00:00:00 2001 From: Jorijn van der Graaf Date: Sat, 31 Oct 2026 08:55:00 +0100 Subject: [PATCH] feat(pool): the pool - stacks, admission, the runner and the scheduler The allowlists and the instrument checks (mcp_connected, db_answered, exercised), the stack keys, Stacks and Admission, the Runner that runs one claude or opencode process per pass, and the Scheduler with its retries, rate-window pauses, the prompt-prefix warm-up, the seats and the board. One thread per pass; each worker runs on its own copy of the job. Every pass runs under the supervisor, so a timeout or an abort takes its whole tree down; a timed-out pass reports exit -9, a SIGKILLed child's code. The helpers the lenses run are this binary's `sql` and `totp`. :Process gains an Append stream for the OpenCode meta line. The tests cover the pool and its share of the commits-mode, seat-pool and several-drivers checks, a fake claude under the supervisor included. Co-Authored-By: Claude Opus 5.5 --- implementations/Crafter.Review-Pool.cpp | 1065 ++++++++++++++++++++ implementations/Crafter.Review-Process.cpp | 5 +- interfaces/Crafter.Review-Pool.cppm | 269 +++++ interfaces/Crafter.Review-Process.cppm | 3 + interfaces/Crafter.Review.cppm | 1 + project.cpp | 10 +- tests/Runner/main.cpp | 277 +++++ tests/Scheduler/main.cpp | 506 ++++++++++ 8 files changed, 2132 insertions(+), 4 deletions(-) create mode 100644 implementations/Crafter.Review-Pool.cpp create mode 100644 interfaces/Crafter.Review-Pool.cppm create mode 100644 tests/Runner/main.cpp create mode 100644 tests/Scheduler/main.cpp diff --git a/implementations/Crafter.Review-Pool.cpp b/implementations/Crafter.Review-Pool.cpp new file mode 100644 index 0000000..2140cf7 --- /dev/null +++ b/implementations/Crafter.Review-Pool.cpp @@ -0,0 +1,1065 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +module; +#include + +module Crafter.Review:Pool_impl; +import :Pool; +import :Json; +import :Text; +import :Crypto; +import :Platform; +import :Process; +import :Plan; +import :Tools; +import :Mcp; +import :Stream; +import :PgWire; +import :Seats; +import :SeatPool; +import :OpenCode; +import :Memory; +import std; + +namespace fs = std::filesystem; + +namespace Crafter::Review::Pool { + namespace { + double Number(const Json& value) { + if (value.is_number()) return value.get(); + if (value.is_boolean()) return value.get() ? 1.0 : 0.0; + return 0.0; + } + + std::int64_t Integer(const Json& value) { + return static_cast(Number(value)); + } + + std::string Clock(double epoch) { + return Platform::LocalTime("%H:%M:%S", epoch); + } + + // A number as the caller passed it: an integral value without a + // fraction. + std::string Plain(double value) { + if (value == std::floor(value) && std::abs(value) < 1e15) return std::to_string(static_cast(value)); + return Dyn::FloatRepr(value); + } + + const Plan::Environment& EnvironmentOf(std::string_view name) { + const Plan::Environment* e = Plan::FindEnvironment(name); + if (!e) throw Dyn::Error("KeyError", Dyn::Repr(name)); + return *e; + } + + // label.replace('/', '_').replace(' ', '_'): the file a pass's output + // and prompt go to. + std::string FileLabel(const Json& job) { + return Dyn::Replace(Dyn::Replace(Dyn::Str(Dyn::At(job, "label")), "/", "_"), " ", "_"); + } + + std::string ReadLenient(const fs::path& path) { + return Dyn::UniversalNewlines(Dyn::DecodeReplace(Platform::ReadFile(path).value_or(""))); + } + + void WriteText(const fs::path& path, std::string_view text) { + std::ofstream file(path, std::ios::binary | std::ios::trunc); + file << text; + } + } + + std::string DefaultBin() { + return Seats::FindClaudeBin(); + } + + std::string SqlCommand() { + return std::format("{} sql", Platform::SelfExe().string()); + } + + std::optional TotpCommand(const std::optional& fixture) { + if (!fixture || fixture->empty()) return std::nullopt; + return std::format("{} totp {}", Platform::SelfExe().string(), *fixture); + } + + std::vector ExtraDirs(std::span trees) { + const std::string cwd = fs::current_path().string(); + std::vector out; + for (const std::string& t : trees) { + const std::string a = Dyn::AbsPath(t); + if (!(std::format("{}/", a)).starts_with(std::format("{}/", cwd)) && a != cwd && !std::ranges::contains(out, a)) out.push_back(a); + } + return out; + } + + Verdict McpConnected(const Json& parsed, std::string_view kind, bool empirical) { + const Json& init = Dyn::Or(Dyn::Get(parsed, "init"), Dyn::EmptyDict()); + const Json& servers = Dyn::Or(Dyn::Get(init, "mcp_servers"), Dyn::EmptyList()); + const bool connected = std::ranges::any_of(servers, [](const Json& s) { return Dyn::Get(s, "name") == Tools::McpServer && Dyn::Get(s, "status") == "connected"; }); + const std::string prefix = std::format("mcp__{}__", Tools::McpServer); + const bool tools = std::ranges::any_of(Dyn::Or(Dyn::Get(init, "tools"), Dyn::EmptyList()), [&](const Json& t) { return Dyn::Str(t).starts_with(prefix); }); + if (!(connected && tools)) { + return {"error", std::format("the {} MCP server did not connect in this pass ({}) - no browser, so the pass's verdict is not functional evidence", Tools::McpServer, + Dyn::Truthy(servers) ? Dyn::ReprValue(servers) : std::string("no server in the init line"))}; + } + const Json none = {{"calls", 0}, {"errors", 0}}; + const Json& b = Dyn::Or(Dyn::Get(parsed, "browser"), none); + if (empirical && kind == "whole" && Dyn::At(b, "calls") == 0) return {"error", "the whole pass made no browser call - nothing was exercised, so its verdict is not functional evidence"}; + if (Dyn::Truthy(Dyn::At(b, "calls")) && Number(Dyn::At(b, "errors")) >= Number(Dyn::At(b, "calls"))) { + return {"error", std::format("every browser call of this pass failed ({} of {}) - the browser did not work, so the pass's verdict is not functional evidence", Dyn::Str(b.at("errors")), + Dyn::Str(b.at("calls")))}; + } + return {"ok", ""}; + } + + Verdict DbAnswered(const Json& parsed, std::string_view, bool) { + const Json none = {{"calls", 0}, {"answered", 0}}; + const Json& d = Dyn::Or(Dyn::Get(parsed, "db"), none); + if (Dyn::Truthy(Dyn::At(d, "calls")) && !Dyn::Truthy(Dyn::At(d, "answered"))) { + return {"error", std::format("none of this pass's {} SQL call(s) reached the database (no `{}` line in any result) - no database, so the pass's verdict is not evidence from it", + Dyn::Str(d.at("calls")), PgWire::Answered)}; + } + return {"ok", ""}; + } + + Verdict Exercised(const Json& parsed, std::string_view environment, std::string_view kind, bool empirical) { + const Plan::Environment* found = Plan::FindEnvironment(environment); + const Plan::Environment& e = found ? *found : *Plan::FindEnvironment("none"); + if (e.mcp == "playwright") { + Verdict v = McpConnected(parsed, kind, empirical); + if (v.first != "ok") return v; + } + if (e.db) { + Verdict v = DbAnswered(parsed, kind, empirical); + if (v.first != "ok") return v; + } + return {"ok", ""}; + } + + std::string StackKey(const Json& stack) { + const std::string u = Dyn::Lower(Dyn::RStrip(Dyn::Strip(Dyn::StrOr(Dyn::Get(stack, "appUrl"), "")), "/")); + if (u.empty()) return Dyn::Str(Dyn::At(stack, "id")); + const Dyn::SplitUrl p = Dyn::UrlSplit(u); + std::optional port; + try { + port = p.Port(); + } catch (const std::invalid_argument&) { + return u; + } + std::string portText; + if (port && *port != 0) { + portText = std::to_string(*port); + } else if (p.scheme == "http") { + portText = "80"; + } else if (p.scheme == "https") { + portText = "443"; + } else { + portText = "None"; + } + const std::optional hostname = p.Hostname(); + const std::string host = hostname && (*hostname == "localhost" || *hostname == "127.0.0.1" || *hostname == "::1" || *hostname == "0.0.0.0") ? "127.0.0.1" : hostname.value_or(""); + return std::format("{}://{}:{}{}", p.scheme, host, portText, p.path); + } + + std::vector AllowedTools(std::string_view, std::string_view, std::span trees, std::span extra, + const std::optional& loginFixture, std::string_view environment) { + std::vector allow = Tools::BaseAllow(); + const std::string cwd = fs::current_path().string(); + for (const std::string& t : trees) { + // Both spellings: lenses write absolute -C paths too (sabotage + // proof 2026-09-17: five denials on an absolute path). + std::vector spellings; + for (const std::string& spelled : {t, Dyn::AbsPath(t), Dyn::JoinPath(cwd, t)}) { + if (!std::ranges::contains(spellings, spelled)) spellings.push_back(spelled); + } + for (const std::string& spelled : spellings) { + for (const std::string_view s : Tools::GitReadOnly) allow.push_back(std::format("Bash(git -C {} {}:*)", spelled, s)); + } + } + const Plan::Environment& e = EnvironmentOf(environment); + if (e.mcp == "playwright") std::ranges::move(Tools::McpToolNames(), std::back_inserter(allow)); + for (const std::string_view tool : e.tools) allow.emplace_back(tool); + if (e.db) allow.push_back(std::format("Bash({}:*)", SqlCommand())); + if (e.app) { + if (const std::optional totp = TotpCommand(loginFixture)) allow.push_back(std::format("Bash({}:*)", *totp)); + std::vector extras; + for (const std::string& x : extra) { + if (!std::ranges::contains(allow, x)) extras.push_back(x); + } + std::ranges::move(extras, std::back_inserter(allow)); + } + // Read(//abs/path/**): the double slash marks an absolute path. + for (const std::string& d : ExtraDirs(trees)) allow.push_back(std::format("Read(/{}/**)", d)); + return allow; + } + + // ---- Stacks ----------------------------------------------------------------------- + Stacks::Stacks(std::vector stacks) : all(std::move(stacks)) { + for (const Json& s : all) { + const std::string id = Dyn::Str(Dyn::At(s, "id")); + keys[id] = StackKey(s); + free.push_back(id); + } + } + + std::size_t Stacks::Size() const { + return all.size(); + } + + const Json* Stacks::First() const { + return all.empty() ? nullptr : &all.front(); + } + + const std::vector& Stacks::All() const { + return all; + } + + const std::string& Stacks::KeyOf(const std::string& id) const { + return keys.at(id); + } + + bool Stacks::Can(const std::optional& prefer, const std::set& held) const { + const std::lock_guard guard(lock); + std::vector open; + for (const std::string& id : free) { + if (!held.contains(keys.at(id))) open.push_back(id); + } + return prefer ? std::ranges::contains(open, *prefer) : !open.empty(); + } + + Json Stacks::Take(const std::optional& prefer, const std::set& held) { + const std::lock_guard guard(lock); + std::string id; + if (prefer) { + id = *prefer; + } else { + const auto open = std::ranges::find_if(free, [&](const std::string& f) { return !held.contains(keys.at(f)); }); + if (open == free.end()) throw Dyn::Error("IndexError", "list index out of range"); + id = *open; + } + const auto at = std::ranges::find(free, id); + if (at == free.end()) throw Dyn::Error("ValueError", "list.remove(x): x not in list"); + free.erase(at); + return *std::ranges::find_if(all, [&](const Json& s) { return Dyn::Str(s.at("id")) == id; }); + } + + void Stacks::Give(const std::string& id) { + const std::lock_guard guard(lock); + if (keys.contains(id) && !std::ranges::contains(free, id)) free.push_back(id); + } + + std::vector Stacks::Free() const { + const std::lock_guard guard(lock); + return free; + } + + // ---- Admission -------------------------------------------------------------------- + Admission::Admission(CostOf costOf, NeedsStack needsStack, Stacks* stacks, std::optional budget, double reserve, Available available, Log log) + : costOf(std::move(costOf)), needsStack(std::move(needsStack)), stacks(stacks), budget(budget), reserve(reserve), log(log ? std::move(log) : Log([](const std::string& line) { std::println("{}", line); })) { + this->available = available ? std::move(available) : Available([]() -> std::optional { + const std::optional mb = Memory::Available(); + return mb ? std::optional(static_cast(*mb)) : std::nullopt; + }); + } + + std::optional Admission::Prefer(const Json& job) const { + if (!Dyn::Truthy(Dyn::Get(job, "resumeSession"))) return std::nullopt; + const Json& previous = Dyn::Get(job, "prevStack"); + return previous.is_null() ? std::nullopt : std::optional(Dyn::Str(previous)); + } + + bool Admission::Can(const Json& job) { + SeatPool::Others seen; + { + const std::lock_guard guard(lock); + seen = others; + } + if (needsStack(job) && !(stacks && stacks->Can(Prefer(job), seen.stacks))) return false; + if (!budget) return true; + double cost = 0; + double inUse = 0; + { + const std::lock_guard guard(lock); + if (inflight == 0 && seen.count == 0) return true; + cost = costOf(job); + inUse = held + seen.mb; + } + if (inUse + cost > *budget) return false; + const std::optional now = available(); + return !now || *now - cost >= reserve; + } + + void Admission::Take(Json& job) { + const double cost = costOf(job); + SeatPool::Others seen; + { + const std::lock_guard guard(lock); + seen = others; + if (inflight == 0 && seen.count == 0 && budget && cost > *budget && !warned) { + warned = true; + log(std::format("pool: {} costs ~{:.0f} MB and the memory budget is {:.0f} MB - launched alone, as nothing else runs", Dyn::Str(job.at("label")), cost, *budget)); + } + ++inflight; + held += cost; + } + job["_admitted"] = cost; + if (needsStack(job)) { + const Json s = stacks->Take(Prefer(job), seen.stacks); + const std::string id = Dyn::Str(s.at("id")); + job["stack"] = s; + job["prevStack"] = id; + // Its name on the board. + job["stackKey"] = stacks->KeyOf(id); + } + } + + void Admission::Give(Json& job) { + if (!job.contains("_admitted")) return; + const double cost = Number(job.at("_admitted")); + job.erase("_admitted"); + { + const std::lock_guard guard(lock); + --inflight; + held = std::max(0.0, held - cost); + } + job.erase("stackKey"); + if (job.contains("stack")) { + const Json s = job.at("stack"); + job.erase("stack"); + if (Dyn::Truthy(s) && stacks) stacks->Give(Dyn::Str(s.at("id"))); + } + } + + void Admission::SetOthers(SeatPool::Others value) { + const std::lock_guard guard(lock); + others = std::move(value); + } + + SeatPool::Others Admission::OthersNow() const { + const std::lock_guard guard(lock); + return others; + } + + std::int64_t Admission::Inflight() const { + const std::lock_guard guard(lock); + return inflight; + } + + double Admission::Held() const { + const std::lock_guard guard(lock); + return held; + } + + // ---- Runner ----------------------------------------------------------------------- + Runner::Runner(RunnerOptions given) : options(std::move(given)) { + if (options.bin.empty()) { + const std::optional configured = Platform::GetEnv("REVIEW_GATES_CLAUDE_BIN"); + options.bin = configured && !configured->empty() ? *configured : DefaultBin(); + } + if (options.opencodeBin.empty()) options.opencodeBin = OpenCode::DefaultBin(); + if (!options.log) options.log = [](const std::string& line) { std::println("{}", line); }; + for (const std::string_view d : {"prompts", "out", "err"}) fs::create_directories(options.scratch / d); + env = Platform::Environment(); + env["CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC"] = "1"; + env["DISABLE_AUTOUPDATER"] = "1"; + for (const std::string_view key : {std::string_view("MAX_THINKING_TOKENS"), std::string_view("CLAUDE_CODE_OAUTH_TOKEN"), std::string_view("ANTHROPIC_API_KEY"), + std::string_view("ANTHROPIC_AUTH_TOKEN"), std::string_view("ANTHROPIC_BASE_URL"), PgWire::UrlEnv, PgWire::BrokerEnv}) { + env.erase(std::string(key)); + } + } + + std::int32_t Runner::KillChildren() { + const std::lock_guard guard(lock); + std::int32_t n = 0; + for (const auto& [pid, live] : running) { + Process::Signal(pid, SIGTERM); + ++n; + } + return n; + } + + std::string Runner::EnvOf(const Json& job) const { + const auto found = options.environments.find(Dyn::Str(Dyn::At(job, "lensFile"))); + return found == options.environments.end() ? "none" : found->second; + } + + std::map Runner::ProcessEnv(const Json& job, const std::map& base) const { + if (!EnvironmentOf(EnvOf(job)).db) return base; + const Json& s = Dyn::Get(job, "stack"); + std::optional url = options.dbUrl; + if (Dyn::Truthy(s)) { + const Json& own = Dyn::Get(s, "dbUrl"); + url = Dyn::Truthy(own) ? std::optional(Dyn::Str(own)) : std::nullopt; + } + std::map out = base; + out[std::string(PgWire::BrokerEnv)] = options.broker && url && !url->empty() ? options.broker->Address(*url) : std::string(); + return out; + } + + std::optional Runner::LoginOf(const Json& job) const { + const Json& s = Dyn::Get(job, "stack"); + if (!Dyn::Truthy(s)) return options.loginFixture; + const Json& login = Dyn::Get(s, "login"); + return login.is_null() ? std::nullopt : std::optional(Dyn::Str(login)); + } + + std::optional Runner::McpConfig(const Json& job) const { + if (EnvironmentOf(EnvOf(job)).mcp != "playwright") return std::nullopt; + return Mcp::ConfigJson(Mcp::PlaywrightConfig(options.functional, Mcp::OutputDir(options.scratch, Dyn::Str(Dyn::At(job, "label"))).string())); + } + + std::vector Runner::Cmd(const Json& job, const fs::path& systemPath, std::string_view schema, std::span trees, std::int64_t maxTurns, bool persist, + const std::string& sessionId, bool resume) const { + std::vector c = {options.bin, "--print", "--model", Dyn::Str(Dyn::At(job, "model")), "--effort", options.effort}; + c.insert(c.end(), {resume ? std::string("--resume") : std::string("--session-id"), sessionId}); + if (!persist && !resume) c.emplace_back("--no-session-persistence"); + const std::optional mcp = McpConfig(job); + // --safe-mode drops every MCP server (measured 2026-10-15, CLI + // 2.1.300); a pass with the browser runs without it - + // --strict-mcp-config and --setting-sources '' still keep the + // consumer's own servers, hooks and settings out. + if (!mcp) c.emplace_back("--safe-mode"); + c.insert(c.end(), {"--strict-mcp-config", "--setting-sources", "", "--disable-slash-commands", "--append-system-prompt-file", systemPath.string(), "--exclude-dynamic-system-prompt-sections"}); + if (mcp) c.insert(c.end(), {"--mcp-config", *mcp}); + for (const std::string& d : ExtraDirs(trees)) c.insert(c.end(), {"--add-dir", d}); + std::vector extra; + for (const Json& x : Dyn::Or(Dyn::Get(options.functional, "extraTools"), Dyn::EmptyList())) extra.push_back(Dyn::Str(x)); + const std::vector allow = AllowedTools(Dyn::Str(job.at("lensFile")), Dyn::Str(Dyn::At(job, "kind")), trees, extra, LoginOf(job), EnvOf(job)); + c.insert(c.end(), {"--tools", "Read,Glob,Grep,Bash", "--allowedTools", Dyn::Join(allow, ","), "--disallowedTools", Dyn::Join(Tools::Deny(), ","), "--permission-prompts", "none", "--max-turns", + std::to_string(maxTurns), "--json-schema", std::string(schema), "--output-format", "stream-json", "--verbose"}); + return c; + } + + void Runner::SetSeats(const std::vector& seats) { + const std::lock_guard guard(lock); + seatsById.clear(); + for (const Json& s : seats) seatsById[Dyn::Str(s.at("id"))] = {{"id", s.at("id")}, {"kind", s.at("kind")}, {"secret", Dyn::Get(s, "secret")}}; + } + + const std::string& Runner::Bin() const { + return options.bin; + } + + std::map& Runner::Env() { + return env; + } + + const fs::path& Runner::Scratch() const { + return options.scratch; + } + + std::optional Runner::Wait(std::int64_t pid, std::int64_t timeoutMin, bool& timedOut) { + std::optional code = Process::Wait(pid, std::chrono::milliseconds(timeoutMin * 60 * 1000)); + if (!code) { + timedOut = true; + // The supervisor takes the tree down on SIGTERM and reports the + // kill; it gets ten seconds for that. + Process::Signal(pid, SIGTERM); + code = Process::Wait(pid, std::chrono::milliseconds(10000)); + if (!code) { + Process::Signal(pid, SIGKILL); + code = Process::Wait(pid); + } + } + { + const std::lock_guard guard(lock); + running.erase(pid); + } + return code ? std::optional(Process::SupervisedReturnCode(*code)) : std::nullopt; + } + + std::pair Runner::Run(const Json& job, const std::string& prompt, const fs::path& systemPath, std::string_view schema, std::span trees, std::int64_t maxTurns, + std::int64_t timeoutMin, bool persist) { + const std::string label = FileLabel(job); + const fs::path pp = options.scratch / "prompts" / std::format("{}.txt", label); + WriteText(pp, prompt); + const fs::path op = options.scratch / "out" / std::format("{}.jsonl", label); + const fs::path ep = options.scratch / "err" / std::format("{}.txt", label); + std::string sid = Uuid4(); + Json rec = {{"label", job.at("label")}, + {"sessionId", sid}, + {"startedAt", Platform::LocalTime("%Y-%m-%dT%H:%M:%S")}, + {"maxTurns", maxTurns}, + {"timeoutMin", timeoutMin}, + {"outFile", Dyn::RelPath(op.string(), options.repo.string())}, + {"promptBytes", Dyn::Length(prompt)}}; + if (options.dryRun) { + std::vector fail; + for (const std::string& x : Dyn::Split(Platform::GetEnv("REVIEW_GATES_SIMULATE_FAIL").value_or(""), ",")) { + if (!x.empty()) fail.push_back(x); + } + const Json& unit = Dyn::Get(job, "unit"); + if ((unit.is_string() && std::ranges::contains(fail, unit.get())) || (job.at("kind") == "whole" && std::ranges::contains(fail, std::string("whole")))) { + rec["status"] = "error"; + rec["reason"] = "simulated failure (REVIEW_GATES_SIMULATE_FAIL)"; + return {rec, nullptr}; + } + rec["status"] = "dry-run"; + rec["reason"] = ""; + return {rec, nullptr}; + } + Json seat; + if (Dyn::Truthy(Dyn::Get(job, "seat"))) { + const std::lock_guard guard(lock); + const auto found = seatsById.find(Dyn::Str(job.at("seat"))); + if (found != seatsById.end()) seat = found->second; + } + if (options.harness == "opencode") return RunOpenCode(job, prompt, systemPath, trees, timeoutMin, seat, std::move(rec), op, ep, sid); + std::map processEnv = ProcessEnv(job, Dyn::Truthy(seat) ? SeatPool::ChildEnv(seat, env, options.scratch) : env); + bool resume = false; + if (Dyn::Truthy(seat)) { + const std::string seatId = Dyn::Str(seat.at("id")); + const bool local = seat.at("kind") == "local"; + rec["seat"] = seatId; + persist = persist || !local; + const Json& resumeSession = Dyn::Get(job, "resumeSession"); + if (Dyn::Truthy(resumeSession) && !local) { + const std::string fromSeat = Dyn::StrOr(Dyn::Get(job, "resumeSeat"), ""); + const std::string session = Dyn::Str(resumeSession); + if (SeatPool::CarryTranscript(options.scratch, fromSeat, seatId, session, options.log)) { + sid = session; + resume = true; + WriteText(pp, SeatPool::ResumePrompt); + rec["sessionId"] = sid; + rec["resumedFrom"] = {{"seat", Dyn::Get(job, "resumeSeat")}, {"attempt", Integer(Dyn::Has(job, "attempt") ? job.at("attempt") : Json(1)) - 1}}; + rec["promptBytes"] = Dyn::Length(SeatPool::ResumePrompt); + } else { + options.log(std::format("{}: no transcript of {} on seat {} - the pass restarts on seat {}", Dyn::Str(job.at("label")), Dyn::Slice(session, 0, 8), Dyn::Str(Dyn::Get(job, "resumeSeat")), seatId)); + } + } + } + const std::vector cmd = Cmd(job, systemPath, schema, trees, maxTurns, persist, sid, resume); + bool timedOut = false; + std::optional code; + const std::string key = std::format("{}\n{}", Dyn::Str(job.at("label")), sid); + { + Process::Spec spec {.argv = Process::Supervised(cmd), .environment = processEnv, .cwd = options.repo}; + spec.input = {.kind = Process::Stream::Kind::File, .file = pp}; + spec.output = {.kind = Process::Stream::Kind::File, .file = op}; + spec.error = {.kind = Process::Stream::Kind::File, .file = ep}; + spec.deathSignal = SIGTERM; + const Process::Child child = Process::Start(spec); + { + const std::lock_guard guard(lock); + running[child.pid] = true; + } + // The supervisor's pid: the driver's measure leaves the supervisor + // itself out, so a peak is claude's tree alone, as the calibration + // records it. + if (options.sampler) options.sampler->Watch(key, child.pid); + code = Wait(child.pid, timeoutMin, timedOut); + const std::optional peak = options.sampler ? options.sampler->Done(key) : std::nullopt; + rec["memPeakMb"] = peak ? Json(*peak) : Json(); + } + const std::string out = ReadLenient(op); + const std::string err = ReadLenient(ep); + const Json parsed = Stream::ParseStream(out); + auto [status, reason] = Stream::Classify(parsed, code, err, timedOut); + const std::string environment = EnvOf(job); + if (status == "ok") std::tie(status, reason) = Exercised(parsed, environment, Dyn::Str(job.at("kind")), Plan::IsLens(Dyn::Str(job.at("lensFile"))) && Plan::LensOf(Dyn::Str(job.at("lensFile"))).empirical); + const Json& rate = parsed.at("rateLimit"); + rec["endedAt"] = Platform::LocalTime("%Y-%m-%dT%H:%M:%S"); + rec["exit"] = code ? Json(*code) : Json(); + rec["status"] = status; + rec["reason"] = reason; + rec["usage"] = parsed.at("usage"); + rec["firstTurn"] = parsed.at("firstTurn"); + rec["turns"] = parsed.at("turns"); + rec["toolCalls"] = parsed.at("toolCalls"); + rec["toolErrors"] = parsed.at("toolErrors"); + rec["costUsd"] = parsed.at("costUsd"); + rec["denials"] = parsed.at("denials"); + rec["rateLimit"] = Stream::Windows(rate); + rec["rateStatus"] = Dyn::Get(Dyn::Or(rate, Dyn::EmptyDict()), "status"); + rec["rateType"] = Dyn::Get(Dyn::Or(rate, Dyn::EmptyDict()), "rateLimitType"); + rec["mcp"] = McpConfig(job).has_value(); + rec["environment"] = environment; + rec["browser"] = Dyn::Get(parsed, "browser"); + rec["db"] = Dyn::Get(parsed, "db"); + if (status != "ok") return {rec, nullptr}; + const std::optional structured = Stream::StructuredOutput(parsed.at("result")); + return {rec, structured ? *structured : Json()}; + } + + std::pair Runner::RunOpenCode(const Json& job, const std::string& prompt, const fs::path& systemPath, std::span trees, std::int64_t timeoutMin, const Json& seat, + Json rec, const fs::path& out, const fs::path& err, const std::string& sessionId) { + // The same prompts, the allowlist as permission rules, the result + // through the StructuredOutput tool. No MCP server: a lens whose + // environment needs one never runs on this harness. + if (!Dyn::Truthy(seat) || seat.at("kind") != "openrouter") { + rec["status"] = "error"; + rec["reason"] = "the opencode harness runs on an OpenRouter seat only (--seats openrouter)"; + return {rec, nullptr}; + } + const fs::path& home = opencodeHome; + const std::string outText = out.string(); + const fs::path so = std::format("{}.so.json", outText.substr(0, outText.size() - std::string_view(".jsonl").size())); + const std::map processEnv = OpenCode::ChildEnv(OpenCode::BaseEnv(home, Dyn::Str(seat.at("secret")), env), systemPath, so, + AllowedTools(Dyn::Str(job.at("lensFile")), Dyn::Str(job.at("kind")), trees), home / "data" / sessionId); + const std::string variant = options.variant.value_or(""); + const std::vector cmd = OpenCode::Cmd(options.opencodeBin, Dyn::Str(job.at("model")), home / "view", Dyn::Str(job.at("label")), variant, prompt); + rec["seat"] = seat.at("id"); + rec["harness"] = "opencode"; + rec["variant"] = options.variant ? Json(*options.variant) : Json(); + rec["environment"] = EnvOf(job); + // The line salvage knows an OpenCode output by, with DumpSpaced's + // separators. + WriteText(out, std::format("{}\n", DumpSpaced({{"type", "review_gates_meta"}, {"model", job.at("model")}, {"harness", "opencode"}, {"variant", rec.at("variant")}}))); + bool timedOut = false; + std::optional code; + { + Process::Spec spec {.argv = Process::Supervised(cmd), .environment = processEnv, .cwd = home / "view"}; + spec.input = {.kind = Process::Stream::Kind::Null}; + spec.output = {.kind = Process::Stream::Kind::Append, .file = out}; + spec.error = {.kind = Process::Stream::Kind::File, .file = err}; + spec.deathSignal = SIGTERM; + const Process::Child child = Process::Start(spec); + { + const std::lock_guard guard(lock); + running[child.pid] = true; + } + code = Wait(child.pid, timeoutMin, timedOut); + } + const Json parsed = OpenCode::ParseEvents(ReadLenient(out), so); + const auto [status, reason] = OpenCode::Classify(parsed, code.value_or(-1), ReadLenient(err), timedOut); + rec["endedAt"] = Platform::LocalTime("%Y-%m-%dT%H:%M:%S"); + rec["exit"] = code ? Json(*code) : Json(); + rec["status"] = status; + rec["reason"] = reason; + rec["usage"] = parsed.at("usage"); + rec["firstTurn"] = parsed.at("firstTurn"); + rec["turns"] = parsed.at("turns"); + rec["toolCalls"] = parsed.at("toolCalls"); + rec["toolErrors"] = parsed.at("toolErrors"); + rec["costUsd"] = parsed.at("costUsd"); + rec["denials"] = Json::array(); + rec["rateLimit"] = Json::object(); + rec["rateStatus"] = nullptr; + rec["rateType"] = nullptr; + rec["opencodeSession"] = parsed.at("sessionId"); + rec["stop"] = parsed.at("stop"); + return {rec, status == "ok" ? parsed.at("structured") : Json()}; + } + + // ---- Scheduler -------------------------------------------------------------------- + Scheduler::Scheduler(SchedulerOptions options) : o(std::move(options)) { + if (!o.log) o.log = [](const std::string& line) { std::println("{}", line); }; + if (!o.now) o.now = [] { return Platform::Now(); }; + if (o.seats && !o.seats->empty()) { + if (!Dyn::Truthy(o.seatConfig)) o.seatConfig = SeatPool::PoolConfig(Platform::Environment()); + if (!Dyn::Truthy(o.apiConfig)) o.apiConfig = SeatPool::ApiConfig(Platform::Environment()); + if (o.runner) o.runner->SetSeats(*o.seats); + } + abort = o.abort; + } + + void Scheduler::Log(const std::string& line) const { + o.log(line); + } + + void Scheduler::Abort(const std::string& reason) { + { + const std::lock_guard guard(lock); + abort = reason; + } + wake.notify_all(); + } + + std::optional Scheduler::Aborted() const { + const std::lock_guard guard(lock); + return abort; + } + + const std::vector& Scheduler::DoneList() const { + return done; + } + + Json Scheduler::PauseUntil() const { + return pauseUntil; + } + + void Scheduler::RaisePause(double until) { + if (until > Number(pauseUntil)) pauseUntil = until; + } + + void Scheduler::Sleep(double seconds) { + if (seconds <= 0) return; + std::unique_lock guard(lock); + wake.wait_for(guard, std::chrono::duration(seconds), [&] { return abort.has_value(); }); + } + + bool Scheduler::Capped(const Json& job) { + return o.admission && !o.admission->Can(job); + } + + bool Scheduler::Launchable(Json& job, const std::vector& live, double now, const std::optional& seatId) { + // The API serves a prompt prefix from its cache only once the first + // response that carries it has begun, so a wave of passes launched + // together all WRITE the same prefix. One pass per cold prefix goes + // first; the others follow its first response, its end, or + // warmTimeout. With seats the prefix is warmed once per seat. + if (!o.cacheKey) return true; + const std::optional k = o.cacheKey(job); + if (!k) return true; + const std::pair key = {*k, seatId.value_or("")}; + const std::string where = seatId ? std::format(" on seat {}", *seatId) : std::string(); + const auto known = warm.find(key); + if (now - (known == warm.end() ? 0.0 : known->second) < o.warmTtl) return true; + const bool answered = o.warmProbe && std::ranges::any_of(live, [&](Json* r) { + if (r == &job) return false; + const std::optional rk = o.cacheKey(*r); + return rk && *rk == key.first && Dyn::StrOr(Dyn::Get(*r, "seat"), "") == key.second && o.warmProbe(*r); + }); + if (answered) { + warm[key] = now; + warming.erase(key); + Log(std::format("pool: prompt prefix cached{} (first response in) - the passes that share it follow", where)); + return true; + } + const auto w = warming.find(key); + if (w == warming.end()) { + warming[key] = {Dyn::Str(job.at("label")), now}; + Log(std::format("pool: {} goes first{} - its prompt prefix is not cached yet there", Dyn::Str(job.at("label")), where)); + return true; + } + if (now - w->second.second > o.warmTimeout) { + warm[key] = now; + const std::string first = w->second.first; + warming.erase(w); + Log(std::format("pool: no first response from {}{} after {} s - the passes that share its prefix follow anyway", first, where, Plain(o.warmTimeout))); + return true; + } + return false; + } + + void Scheduler::Progress(const std::vector& running, const std::vector& queued) { + if (!o.progressPath) return; + try { + const std::optional reason = Aborted(); + Json runningJson = Json::array(); + for (const std::string& label : running) runningJson.push_back(label); + Json queuedJson = Json::array(); + for (const std::string& label : queued) queuedJson.push_back(label); + const Json progress = {{"time", Platform::LocalTime("%H:%M:%S")}, {"running", runningJson}, {"queued", queuedJson}, + {"done", done.size()}, {"pauseUntil", pauseUntil}, {"abort", reason ? Json(*reason) : Json()}}; + Platform::WriteAtomic(*o.progressPath, DumpSpaced(progress), 0666); + } catch (const std::exception&) { + // A record, never a dependency. + } + } + + const std::vector& Scheduler::Run(std::vector jobs, const ExecJob& exec, const OnDone& onDone) { + struct Slot { + // The pool's copy; the worker runs on its own. + Json job; + std::thread thread; + bool finished = false; + std::uint64_t order = 0; + std::pair result; + std::optional error; + }; + std::deque queue(std::make_move_iterator(jobs.begin()), std::make_move_iterator(jobs.end())); + std::list running; + std::uint64_t finishes = 0; + const bool withSeats = o.seats && !o.seats->empty(); + auto labelsOf = [](const auto& items, auto get) { + std::vector labels; + for (const auto& item : items) labels.push_back(Dyn::Str(get(item).at("label"))); + return labels; + }; + auto runningLabels = [&] { return labelsOf(running, [](const Slot& s) -> const Json& { return s.job; }); }; + auto queuedLabels = [&] { return labelsOf(queue, [](const Json& j) -> const Json& { return j; }); }; + auto everySeat = [&] { + std::vector parts; + for (const Json& s : *o.seats) parts.push_back(std::format("{} {} ({})", Dyn::Str(s.at("id")), Dyn::Str(s.at("state")), Dyn::Slice(Dyn::StrOr(Dyn::Get(s, "error"), ""), 0, 60))); + return std::format("every seat is out: {}", Dyn::Join(parts, "; ")); + }; + for (;;) { + if (queue.empty() && running.empty()) break; + if (Aborted()) break; + const double now = Platform::Now(); + bool paused = now < Number(pauseUntil) && !queue.empty(); + if (paused && running.empty()) { + Log(std::format("pool: paused until {} (rate window), {} queued", Clock(Number(pauseUntil)), queue.size())); + Sleep(std::min(60.0, Number(pauseUntil) - now)); + continue; + } + while (!queue.empty() && static_cast(running.size()) < o.concurrency && !paused) { + std::vector live; + for (Slot& slot : running) live.push_back(&slot.job); + // Without a board every queued pass waiting for a stack or + // memory waits for a completion; with one the check needs the + // other drivers' passes as they are now, which only the fold + // in Place has. + if (!o.board && std::ranges::all_of(queue, [&](const Json& q) { return Capped(q); })) break; + Json pickedJob; + Json* pickedSeat = nullptr; + bool picked = false; + // The first queued pass the admission lets in that may launch + // on the best seat with room, popped and admitted. + auto choose = [&](const std::vector& cands, const SeatPool::Others* others) -> std::optional> { + if (others && o.admission) o.admission->SetOthers(*others); + for (Json* seat : cands) { + const std::optional sid = seat ? std::optional(Dyn::Str(seat->at("id"))) : std::nullopt; + for (auto it = queue.begin(); it != queue.end(); ++it) { + if (Capped(*it) || !Launchable(*it, live, Platform::Now(), sid)) continue; + pickedJob = std::move(*it); + queue.erase(it); + if (!pickedJob.contains("attempt")) pickedJob["attempt"] = 1; + if (o.admission) o.admission->Take(pickedJob); + pickedSeat = seat; + picked = true; + return std::pair(seat, &pickedJob); + } + } + return std::nullopt; + }; + std::vector cands = {nullptr}; + std::int64_t full = 0; + if (withSeats) { + if (o.board) { + // The board folded, the seats ranked, the admission + // checked against every driver's passes and the pick + // registered in one locked section; the pool width is + // the machine's cap. + const SeatPool::Board::Placed placed = o.board->Place(*o.seats, o.now(), o.seatConfig, o.apiConfig, + [&](const std::vector& c, const SeatPool::Others& others) { return choose(c, &others); }, o.concurrency); + cands = placed.ranked; + full = placed.full; + } else { + cands = SeatPool::Ranked(*o.seats, o.now(), o.seatConfig, o.apiConfig); + if (!cands.empty()) choose(cands, nullptr); + } + if (full) { + if (roomWait.is_null()) roomWait = pauseUntil; + RaisePause(o.now() + o.noRoomWait); + Log(std::format("pool: {} passes in flight on this machine, the cap is {} - re-checked in {:g} s, {} queued", full, o.concurrency, o.noRoomWait, queue.size())); + paused = true; + break; + } + if (cands.empty()) { + if (SeatPool::AllOut(*o.seats)) { + Abort(everySeat()); + Log(std::format("pool: {}", *Aborted())); + break; + } + // Wait for the earliest reset, never less than + // noRoomWait; with other drivers' passes in flight on a + // usable seat their completions free room. + std::int64_t held = 0; + for (const Json& s : *o.seats) { + if (s.at("state") != "out" && s.at("state") != "invalid") held += Integer(SeatPool::Foreign(s).at("count")); + } + const std::optional resume = SeatPool::EarliestResume(*o.seats, o.now(), o.board ? std::optional(o.noRoomWait) : std::nullopt); + if (roomWait.is_null()) roomWait = pauseUntil; + RaisePause(std::max(resume.value_or(0.0), o.now() + o.noRoomWait)); + Log(std::format("pool: no seat has room for a new launch - waiting until {}, {} queued{}", Clock(Number(pauseUntil)), queue.size(), + held ? std::format(" ({} passes of other drivers hold the room - re-checked in {:g} s)", held, o.noRoomWait) : std::string())); + paused = true; + break; + } + } else { + choose(cands, nullptr); + } + // Every queued pass waits: for the one warming its prefix, for + // a stack, for memory. + if (!picked) break; + if (pickedSeat) { + (*pickedSeat)["running"] = Integer(pickedSeat->at("running")) + 1; + pickedJob["seat"] = pickedSeat->at("id"); + } + Slot& slot = running.emplace_back(); + slot.job = std::move(pickedJob); + const std::int64_t attempt = Integer(slot.job.at("attempt")); + slot.thread = std::thread([this, &slot, &exec, &finishes, copy = slot.job, attempt] { + std::pair result; + std::optional error; + try { + result = exec(copy, attempt); + } catch (const std::exception& e) { + error = e.what(); + } + { + const std::lock_guard guard(lock); + slot.result = std::move(result); + slot.error = std::move(error); + slot.finished = true; + slot.order = ++finishes; + } + wake.notify_all(); + }); + heldOff = false; + } + if (Aborted()) break; + Progress(runningLabels(), queuedLabels()); + if (running.empty()) { + if (!queue.empty() && !paused) { + // Nothing of this driver runs and nothing could launch: + // what holds the room is other drivers' passes - their + // completions free it; re-checked, not spun on. + if (!heldOff) { + const SeatPool::Others others = o.admission ? o.admission->OthersNow() : SeatPool::Others(); + std::string what = "nothing in flight to wait for"; + if (others.count) { + what = std::format("{} pass{} of other drivers on this machine hold{} {:.0f} MB", others.count, others.count != 1 ? "es" : "", others.count == 1 ? "s" : "", others.mb); + if (!others.stacks.empty()) what += std::format(" and stack(s) {}", Dyn::Join(std::vector(others.stacks.begin(), others.stacks.end()), ", ")); + } + Log(std::format("pool: nothing can launch, {} queued - {}; re-checked every {:g} s", queue.size(), what, o.noRoomWait)); + heldOff = true; + } + Sleep(o.noRoomWait); + } + continue; + } + std::vector::iterator> finished; + { + std::unique_lock guard(lock); + auto anyFinished = [&] { return std::ranges::any_of(running, [](const Slot& s) { return s.finished; }); }; + wake.wait_for(guard, std::chrono::duration(warming.empty() ? 30.0 : o.warmPoll), [&] { return anyFinished() || abort.has_value(); }); + for (auto it = running.begin(); it != running.end(); ++it) { + if (it->finished) finished.push_back(it); + } + } + std::ranges::sort(finished, {}, [](const auto& it) { return it->order; }); + if (!finished.empty() && !roomWait.is_null()) { + pauseUntil = roomWait; + roomWait = nullptr; + Log("pool: a pass finished - seat room is re-checked"); + } + if (paused && finished.empty()) Log(std::format("pool: paused until {} (rate window), {} running, {} queued", Clock(Number(pauseUntil)), running.size(), queue.size())); + for (const auto& it : finished) { + it->thread.join(); + Json j = std::move(it->job); + Json rec; + Json raw; + if (it->error) { + rec = {{"label", j.at("label")}, {"status", "error"}, {"reason", std::format("driver exception: {}", *it->error)}}; + } else { + rec = std::move(it->result.first); + raw = std::move(it->result.second); + } + running.erase(it); + if (o.admission) o.admission->Give(j); + if (o.cacheKey) { + if (const std::optional k = o.cacheKey(j)) { + const std::pair key = {*k, Dyn::StrOr(Dyn::Get(j, "seat"), "")}; + if (o.warmProbe && o.warmProbe(j)) warm[key] = Platform::Now(); + if (const auto w = warming.find(key); w != warming.end() && w->second.first == Dyn::Str(j.at("label"))) warming.erase(w); + } + } + const std::int64_t attempt = Integer(j.at("attempt")); + rec["attempt"] = attempt; + const std::string st = Dyn::StrOr(Dyn::Get(rec, "status"), ""); + const std::string label = Dyn::Str(j.at("label")); + Json fh = Json::object(); + if (withSeats && Dyn::Truthy(Dyn::Get(j, "seat"))) { + Json* seat = nullptr; + for (Json& s : *o.seats) { + if (s.at("id") == j.at("seat")) { + seat = &s; + break; + } + } + if (seat) { + (*seat)["running"] = std::max(0, Integer(seat->at("running")) - 1); + if (!rec.contains("seat")) rec["seat"] = seat->at("id"); + const bool requeue = SeatPool::Settle(*seat, rec, o.now(), o.seatConfig, o.apiConfig, o.ledger, o.log); + // Off the board whether or not it is placed again. + if (o.board) o.board->Finish(*seat, label, o.now()); + if (requeue) { + if (attempt <= SeatPool::MaxAttempts && !SeatPool::AllOut(*o.seats)) { + Json next = j; + next["attempt"] = attempt + 1; + next["seat"] = nullptr; + if (Dyn::Truthy(Dyn::Get(rec, "sessionId")) && seat->at("kind") != "local" && !Dyn::Truthy(Dyn::Get(rec, "salvaged"))) { + next["resumeSession"] = rec.at("sessionId"); + next["resumeSeat"] = seat->at("id"); + } + queue.push_front(std::move(next)); + continue; + } + if (SeatPool::AllOut(*o.seats)) { + Abort(everySeat()); + Log(std::format("pool: {}", *Aborted())); + } + done.push_back({std::move(j), std::move(rec), std::move(raw)}); + continue; + } + } + if (st == "limit" || st == "credits" || st == "auth") { + done.push_back({std::move(j), std::move(rec), std::move(raw)}); + continue; + } + } else if (st == "credits") { + Abort(std::format("credit balance exhausted ({})", Dyn::Str(Dyn::Get(rec, "reason")))); + Log(std::format("pool: {}", *Aborted())); + done.push_back({std::move(j), std::move(rec), std::move(raw)}); + continue; + } else { + const Json& win = Dyn::Or(Dyn::Get(rec, "rateLimit"), Dyn::EmptyDict()); + fh = Dyn::Or(Dyn::Get(win, "five_hour"), Dyn::EmptyDict()); + const Json& utilization = Dyn::Get(fh, "utilization"); + if (!utilization.is_null() && Number(utilization) >= o.pauseAt && Dyn::Truthy(Dyn::Get(fh, "resetsAt"))) { + RaisePause(Number(fh.at("resetsAt")) + 30); + Log(std::format("pool: five_hour window at {}% - pausing new launches until reset", Dyn::Str(utilization))); + } + if (st == "limit" && Dyn::StrOr(Dyn::Get(rec, "rateType"), "").starts_with("seven_day")) { + Abort(std::format("seven-day window rejected ({})", Dyn::Str(Dyn::Get(rec, "reason")))); + Log(std::format("pool: {}", *Aborted())); + done.push_back({std::move(j), std::move(rec), std::move(raw)}); + continue; + } + if (st == "auth") { + Abort(std::format("authentication failure ({})", Dyn::Str(Dyn::Get(rec, "reason")))); + Log(std::format("pool: {}", *Aborted())); + done.push_back({std::move(j), std::move(rec), std::move(raw)}); + continue; + } + } + if (st == "limit" && attempt <= 4) { + const Json& reset = Dyn::Get(fh, "resetsAt"); + constexpr std::array Backoff = {300, 900, 1800, 1800}; + const double delay = Dyn::Truthy(reset) ? std::max(60.0, Number(reset) - Platform::Now() + 30) : Backoff[static_cast(std::min(attempt - 1, 3))]; + if (waitedMin + delay / 60 > o.maxWaitMin) { + Log(std::format("pool: {} limit - max wait exceeded, giving up", label)); + done.push_back({std::move(j), std::move(rec), std::move(raw)}); + continue; + } + waitedMin += delay / 60; + RaisePause(Platform::Now() + delay); + Log(std::format("pool: {} hit a limit - requeued after {} min (attempt {})", label, static_cast(delay / 60), attempt + 1)); + Json next = j; + next["attempt"] = attempt + 1; + queue.push_front(std::move(next)); + continue; + } + const std::string reason = Dyn::Has(rec, "reason") ? Dyn::Str(rec.at("reason")) : std::string(); + if (st == "error" && reason.contains("max_turns")) { + Log(std::format("pool: {} ran out of turns ({}) - not retried, unit unaddressed", label, Dyn::Str(Dyn::Get(j, "maxTurns")))); + } else if ((st == "error" || st == "timeout") && attempt <= 1 && !Aborted()) { + Json next = j; + next["attempt"] = attempt + 1; + if (st == "timeout") next["timeoutMin"] = static_cast(Number(Dyn::Has(j, "timeoutMin") ? j.at("timeoutMin") : Json(20)) * 1.5); + Log(std::format("pool: {} {} ({}) - one retry", label, st, Dyn::Slice(reason, 0, 80))); + queue.push_back(std::move(next)); + continue; + } + const Done& entry = done.emplace_back(std::move(j), std::move(rec), std::move(raw)); + std::vector follow; + try { + follow = onDone(entry.job, entry.rec, entry.raw); + } catch (const std::exception& e) { + Log(std::format("pool: on_done failed for {}: {}", label, e.what())); + follow.clear(); + } + std::ranges::move(follow, std::back_inserter(queue)); + } + Progress(runningLabels(), queuedLabels()); + } + // As ThreadPoolExecutor's exit: the passes in flight end first. + for (Slot& slot : running) { + if (slot.thread.joinable()) slot.thread.join(); + } + return done; + } +} diff --git a/implementations/Crafter.Review-Process.cpp b/implementations/Crafter.Review-Process.cpp index 219d7ed..04e6d1b 100644 --- a/implementations/Crafter.Review-Process.cpp +++ b/implementations/Crafter.Review-Process.cpp @@ -97,9 +97,12 @@ namespace Crafter::Review::Process { return -1; case Stream::Kind::Null: return open("/dev/null", (reading ? O_RDONLY : O_WRONLY) | O_CLOEXEC); + // Created 0666 less the umask, truncated. case Stream::Kind::File: return reading ? open(stream.file.c_str(), O_RDONLY | O_CLOEXEC) - : open(stream.file.c_str(), O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0644); + : open(stream.file.c_str(), O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0666); + case Stream::Kind::Append: + return open(stream.file.c_str(), O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC, 0666); case Stream::Kind::Pipe: { std::array ends {}; if (pipe2(ends.data(), O_CLOEXEC) != 0) return -2; diff --git a/interfaces/Crafter.Review-Pool.cppm b/interfaces/Crafter.Review-Pool.cppm new file mode 100644 index 0000000..405dc1a --- /dev/null +++ b/interfaces/Crafter.Review-Pool.cppm @@ -0,0 +1,269 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +export module Crafter.Review:Pool; +import std; +import :Json; +import :SeatPool; +import :Memory; +import :PgWire; + +// One `claude --print` process per pass, a pool with gating, retries, +// rate-window pauses and progress; with seats every process runs on the seat +// picked for it and a seat's limit pauses that seat, not the pool. What else +// holds a launch back is the pass's environment, never its lens (Admission): +// a pass that drives an application waits for a stack of its own (Stacks), +// and every pass waits until it fits in free memory at its environment's +// measured cost. A pass with the database alone shares the first stack: +// every SQL call is its own transaction, rolled back. +// +// Every pass runs under the supervisor (`crafter-review __supervise`), so a +// timeout, an abort or the driver's death takes its whole tree down - the +// MCP server and Chromium with it. +export namespace Crafter::Review::Pool { + using Log = std::function; + // (status, reason) + using Verdict = std::pair; + + // seats.find_claude_bin + std::string DefaultBin(); + // The exact Bash prefix a lens with the database under test runs SQL with: + // this binary's `sql`, spelled absolute (allowlisted as a prefix). + std::string SqlCommand(); + // The exact Bash line a lens with an application under test runs for a + // one-time code: this binary's `totp `; nullopt without one. + std::optional TotpCommand(const std::optional& fixture); + // Trees outside the working directory: the CLI must be told (--add-dir) + // and Read allowed there, because --setting-sources '' drops the + // project's additionalDirectories and a read there would prompt. + std::vector ExtraDirs(std::span trees); + // A pass with the browser that returned ok, fail closed: the Playwright + // server connected with its tools (2026-10-15: --safe-mode had dropped it + // and four units came back n/a), and the browser answered - an empirical + // lens's whole pass must have called it, and no pass may have had every + // browser call fail. + Verdict McpConnected(const Json& parsed, std::string_view kind = "block", bool empirical = true); + // A pass that ran the SQL helper and never got the database's answer had + // no database: its verdict is not evidence from one. + Verdict DbAnswered(const Json& parsed, std::string_view kind = "block", bool empirical = false); + // Every instrument the environment gives must show it was there. + Verdict Exercised(const Json& parsed, std::string_view environment, std::string_view kind = "block", bool empirical = false); + // A stack's name on the board, the same in every driver: its + // application's address, localhost and 127.0.0.1 one host, a default port + // spelled out; a stack without a URL is its id. + std::string StackKey(const Json& stack); + // The read-only tools, git -C on every tree in both spellings, and what + // the environment adds: the browser tools and curl, the SQL helper, and + // with an application the one-time-code command and the config's extras. + std::vector AllowedTools(std::string_view lensFile, std::string_view kind, std::span trees, std::span extra = {}, + const std::optional& loginFixture = std::nullopt, std::string_view environment = "none"); + + // The stacks a run's passes use (2026-10-22): the one the config names, + // or the slots brought up side by side. A pass whose environment drives an + // application holds a stack of its own from launch to end, least recently + // returned first; another driver's pass holds one too (`held`, the + // StackKey of every stack the other drivers on this machine hold). + class Stacks { + public: + explicit Stacks(std::vector stacks); + std::size_t Size() const; + // nullptr when there is none. + const Json* First() const; + const std::vector& All() const; + const std::string& KeyOf(const std::string& id) const; + // Is a stack free (prefer: the one a resumed session ran on - only + // that one will do)? + bool Can(const std::optional& prefer = std::nullopt, const std::set& held = {}) const; + Json Take(const std::optional& prefer = std::nullopt, const std::set& held = {}); + void Give(const std::string& id); + std::vector Free() const; + + private: + std::vector all; + std::map keys; + std::vector free; + mutable std::mutex lock; + }; + + // May the pool launch this pass now? No per-lens number (2026-10-22): an + // environment that drives an application needs a stack of its own, and the + // pass must fit the memory budget - free memory at the gate's start less + // the reserve, less what the passes in flight were admitted at - with the + // machine still holding its cost free beyond the reserve right now. With + // nothing in flight on the machine a pass is admitted whatever the memory + // says, and the log says so once. `others` is what the other drivers' + // passes hold, as the board's last fold left it. + class Admission { + public: + using CostOf = std::function; + using NeedsStack = std::function; + using Available = std::function()>; + Admission(CostOf costOf, NeedsStack needsStack, Stacks* stacks = nullptr, std::optional budget = std::nullopt, double reserve = 0, Available available = {}, Log log = {}); + bool Can(const Json& job); + // The pass's memory and its stack taken: _admitted, stack, prevStack + // and stackKey on the job. + void Take(Json& job); + void Give(Json& job); + void SetOthers(SeatPool::Others others); + SeatPool::Others OthersNow() const; + std::int64_t Inflight() const; + double Held() const; + + private: + CostOf costOf; + NeedsStack needsStack; + Stacks* stacks; + std::optional budget; + double reserve; + Available available; + Log log; + std::int64_t inflight = 0; + double held = 0; + bool warned = false; + SeatPool::Others others; + mutable std::mutex lock; + std::optional Prefer(const Json& job) const; + }; + + struct RunnerOptions { + std::filesystem::path repo; + std::filesystem::path scratch; + // "": REVIEW_GATES_CLAUDE_BIN, else DefaultBin(). + std::string bin; + std::string effort = "xhigh"; + Log log; + bool dryRun = false; + std::string harness = "claude"; + std::optional variant; + std::string opencodeBin; + // The config's functional section: the MCP server, the extra tools. + Json functional = Json::object(); + std::optional loginFixture; + // {lens file: environment} + std::map environments; + // The first stack's: a pass with the database alone runs there. + std::optional dbUrl; + Memory::Sampler* sampler = nullptr; + // The database URLs stay in the driver; a db pass gets the broker's + // address and a token. + PgWire::Broker* broker = nullptr; + }; + + class Runner { + public: + explicit Runner(RunnerOptions options); + // Every running pass stopped (its supervisor takes the tree down); how + // many. + std::int32_t KillChildren(); + std::string EnvOf(const Json& job) const; + // A pass's process environment: the seat's (base), plus - when its + // environment reaches the database - the broker's address and the + // token of the database of the stack it holds (none for a stack + // without one), else the run's first stack's. + std::map ProcessEnv(const Json& job, const std::map& base) const; + // The login fixture of the stack it holds, else the run's. + std::optional LoginOf(const Json& job) const; + // The --mcp-config of a pass, nullopt for an environment without the + // browser. + std::optional McpConfig(const Json& job) const; + std::vector Cmd(const Json& job, const std::filesystem::path& systemPath, std::string_view schema, std::span trees, std::int64_t maxTurns, bool persist, + const std::string& sessionId, bool resume = false) const; + // One pass: (record, the structured output or null). + std::pair Run(const Json& job, const std::string& prompt, const std::filesystem::path& systemPath, std::string_view schema, std::span trees, + std::int64_t maxTurns, std::int64_t timeoutMin, bool persist = false); + // The seats a job's `seat` names, as they were when the pool began: + // only what never changes (id, kind, secret) is read from them. + void SetSeats(const std::vector& seats); + const std::string& Bin() const; + // The environment every pass starts from. + std::map& Env(); + const std::filesystem::path& Scratch() const; + std::filesystem::path opencodeHome; + + private: + RunnerOptions options; + std::map env; + std::map seatsById; + std::map running; + mutable std::mutex lock; + std::pair RunOpenCode(const Json& job, const std::string& prompt, const std::filesystem::path& systemPath, std::span trees, std::int64_t timeoutMin, + const Json& seat, Json rec, const std::filesystem::path& out, const std::filesystem::path& err, const std::string& sessionId); + std::optional Wait(std::int64_t pid, std::int64_t timeoutMin, bool& timedOut); + }; + + struct SchedulerOptions { + Runner* runner = nullptr; + std::int64_t concurrency = 8; + Log log; + std::optional progressPath; + double pauseAt = 85; + double maxWaitMin = 360; + // The key of a pass's prompt prefix (nullopt: none), which the driver + // keeps on the job as `cacheKey`; and whether a running pass's first + // response is in. + std::function(Json& job)> cacheKey; + std::function warmProbe; + double warmTtl = 240; + double warmTimeout = 180; + double warmPoll = 5; + std::vector* seats = nullptr; + // null: from the environment. + Json seatConfig; + Json apiConfig; + SeatPool::Ledger* ledger = nullptr; + std::function now; + Admission* admission = nullptr; + SeatPool::Board* board = nullptr; + double noRoomWait = 30; + // A stop that came before this pool (a signal, `crafter-review stop`): + // the pool starts aborted. + std::optional abort; + }; + + struct Done { + Json job; + Json rec; + Json raw; + }; + + // Runs jobs on a pool; onDone(job, rec, raw) -> the follow-up jobs (the + // blocks after a passing whole pass). Retries: error/timeout once (x1.5 + // budget for a timeout); limit -> pause until resetsAt (or a backoff) and + // requeue (<= 4); auth or a seven_day rejection -> abort. With seats every + // launch is placed (with the board: placed, admitted and registered under + // its lock, the pool width the machine's cap), every finished process + // settled onto its seat and taken off the board. One thread per pass. + class Scheduler { + public: + using ExecJob = std::function(const Json& job, std::int64_t attempt)>; + using OnDone = std::function(const Json& job, const Json& rec, const Json& raw)>; + explicit Scheduler(SchedulerOptions options); + const std::vector& Run(std::vector jobs, const ExecJob& exec, const OnDone& onDone); + // From any thread: the pool stops launching and returns once what runs + // has ended. + void Abort(const std::string& reason); + std::optional Aborted() const; + const std::vector& DoneList() const; + // pause_until: 0 until a pause raised it. + Json PauseUntil() const; + + private: + SchedulerOptions o; + std::vector done; + std::optional abort; + Json pauseUntil = 0; + Json roomWait; + double waitedMin = 0; + bool heldOff = false; + std::map, double> warm; + std::map, std::pair> warming; + mutable std::mutex lock; + std::condition_variable wake; + bool Capped(const Json& job); + bool Launchable(Json& job, const std::vector& running, double now, const std::optional& seatId); + void RaisePause(double until); + void Sleep(double seconds); + void Progress(const std::vector& running, const std::vector& queued); + void Log(const std::string& line) const; + }; +} diff --git a/interfaces/Crafter.Review-Process.cppm b/interfaces/Crafter.Review-Process.cppm index 6ca5c5e..66e2fbd 100644 --- a/interfaces/Crafter.Review-Process.cppm +++ b/interfaces/Crafter.Review-Process.cppm @@ -32,6 +32,9 @@ export namespace Crafter::Review::Process { // Opened for reading (stdin) or created and truncated (stdout, // stderr). File, + // stdout or stderr after what the file already holds: a line the + // driver wrote before the child starts stays first. + Append, // A pipe whose other end Start() returns. Pipe, }; diff --git a/interfaces/Crafter.Review.cppm b/interfaces/Crafter.Review.cppm index 75bb961..5ffb78f 100644 --- a/interfaces/Crafter.Review.cppm +++ b/interfaces/Crafter.Review.cppm @@ -29,5 +29,6 @@ export import :Mcp; export import :Seats; export import :SeatPool; export import :OpenCode; +export import :Pool; export import :Freeze; export import :Driver; diff --git a/project.cpp b/project.cpp index be8527d..17a6db5 100644 --- a/project.cpp +++ b/project.cpp @@ -13,7 +13,7 @@ using namespace Crafter; namespace { // The umbrella, then the partitions in import order: each after the ones // it imports, which is also the order the module map in AGENTS.md uses. - constexpr std::array Interfaces = { + constexpr std::array Interfaces = { "interfaces/Crafter.Review", "interfaces/Crafter.Review-Unicode", "interfaces/Crafter.Review-Json", @@ -41,10 +41,11 @@ namespace { "interfaces/Crafter.Review-Seats", "interfaces/Crafter.Review-SeatPool", "interfaces/Crafter.Review-OpenCode", + "interfaces/Crafter.Review-Pool", "interfaces/Crafter.Review-Freeze", "interfaces/Crafter.Review-Driver", }; - constexpr std::array Implementations = { + constexpr std::array Implementations = { "implementations/Crafter.Review-Text", "implementations/Crafter.Review-Crypto", "implementations/Crafter.Review-Platform", @@ -69,6 +70,7 @@ namespace { "implementations/Crafter.Review-Seats", "implementations/Crafter.Review-SeatPool", "implementations/Crafter.Review-OpenCode", + "implementations/Crafter.Review-Pool", "implementations/Crafter.Review-Freeze", "implementations/Crafter.Review-Driver", }; @@ -80,7 +82,7 @@ namespace { // One per directory under tests/, each an ordinary main() that links the // library: exit 0 passes, nonzero fails, 77 skips. - constexpr std::array Tests = { + constexpr std::array Tests = { "Crypto", "Text", "Platform", @@ -99,6 +101,8 @@ namespace { "SeatPool", "Drivers", "OpenCode", + "Runner", + "Scheduler", }; // The version the binary reports: CI sets it from the tag, a checkout diff --git a/tests/Runner/main.cpp b/tests/Runner/main.cpp new file mode 100644 index 0000000..19637c2 --- /dev/null +++ b/tests/Runner/main.cpp @@ -0,0 +1,277 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// The pool's command lines: the allowlists and the deny list, the MCP +// config, the command line by environment, the process environment of a +// pass with the database, mcp_connected and exercised, and the stream +// accounting behind them. The helpers are this binary's `sql` and `totp`. + +#include +import std; +import Crafter.Review; +namespace fs = std::filesystem; +using namespace Crafter::Review; + +namespace { + std::int32_t Failures = 0; + const std::string Fn(Plan::FunctionalLens); + + void Check(std::string_view name, bool cond, std::string_view detail = {}) { + std::println("{}{}{}", cond ? " ok " : " FAIL ", name, !cond && !detail.empty() ? std::format(" ({})", detail) : std::string()); + if (!cond) ++Failures; + } + + bool Has(const std::vector& list, std::string_view item) { + return std::ranges::contains(list, item); + } + + std::ptrdiff_t Index(const std::vector& list, std::string_view item) { + const auto at = std::ranges::find(list, item); + return at == list.end() ? -1 : at - list.begin(); + } + + // c[c.index(flag) + 1] + std::string After(const std::vector& c, std::string_view flag) { + const std::ptrdiff_t at = Index(c, flag); + return at < 0 || at + 1 >= static_cast(c.size()) ? std::string() : c[static_cast(at + 1)]; + } + + Json Job(std::string_view label, std::string_view lensFile) { + return {{"label", label}, {"model", "m"}, {"lensFile", lensFile}, {"kind", "block"}}; + } + + Json With(Json value, const Json& changes) { + for (auto it = changes.begin(); it != changes.end(); ++it) value[it.key()] = it.value(); + return value; + } +} + +int main() { + std::println("commits mode (the pool's part): allowlists, the deny list, the MCP config, the command line, the process environment, mcp_connected, exercised"); + Platform::SetMode(Platform::Mode::Native); + const std::string cwd = fs::current_path().string(); + const std::vector here = {"."}; + const std::vector al = Pool::AllowedTools("repo-bug-prover.md", "block", here); + Check("static lens allowlist: git -C . and absolute, python3 -, no browser tool, no curl, no Read(//) rule for the cwd", + Has(al, "Bash(git -C . show:*)") && Has(al, std::format("Bash(git -C {} show:*)", cwd)) && Has(al, "Bash(python3 -:*)") && std::ranges::none_of(al, [](const std::string& x) { return x.starts_with("mcp__"); }) + && !Has(al, "Bash(curl:*)") && std::ranges::none_of(al, [](const std::string& x) { return x.starts_with("Read("); })); + const std::vector extras = {"Bash(psql:*)", "mcp__x__y"}; + const std::string fixture = "cypress/fixtures/created-user.json"; + const std::vector af = Pool::AllowedTools(Fn, "block", here, extras, fixture, "browser"); + const std::string totp = std::format("Bash({}:*)", Pool::TotpCommand(fixture).value_or("")); + Check("functional allowlist: every Playwright tool by name, curl, the TOTP command as a prefix, the extra tools", + std::ranges::all_of(Tools::McpToolNames(), [&](const std::string& t) { return Has(af, t); }) && Has(af, "Bash(curl:*)") && Has(af, totp) && Has(af, "Bash(psql:*)") && Has(af, "mcp__x__y") + && Pool::TotpCommand(fixture)->starts_with(std::format("{} totp ", Platform::SelfExe().string()))); + const std::vector psql = {"Bash(psql:*)"}; + const std::vector staticBrowser = Pool::AllowedTools("repo-bug-prover.md", "block", here, {}, std::nullopt, "browser"); + std::vector wanted = Tools::McpToolNames(); + wanted.emplace_back("Bash(curl:*)"); + const std::vector functionalNone = Pool::AllowedTools(Fn, "block", here, psql, std::string("u.json"), "none"); + Check("the environment carries the instrument, not the lens name: a static lens with browser gets the browser tools and curl; the functional lens with none gets neither, nor the extras or the TOTP command", + std::ranges::all_of(wanted, [&](const std::string& t) { return Has(staticBrowser, t); }) + && std::ranges::none_of(functionalNone, [](const std::string& x) { return x.starts_with("mcp__") || x == "Bash(curl:*)" || x == "Bash(psql:*)" || x.contains(" totp "); })); + Check("the browser allowlist keeps the order of the functional allowlist it replaces: the browser tools, curl, the TOTP command, the extras", + Index(af, "mcp__playwright__browser_navigate") < Index(af, "Bash(curl:*)") && Index(af, "Bash(curl:*)") < Index(af, totp) && Index(af, totp) < Index(af, "Bash(psql:*)") + && af == Pool::AllowedTools(Fn, "block", here, extras, fixture, "browser")); + const std::vector& deny = Tools::Deny(); + Check("deny list: run_code_unsafe, package managers, stacks, writes, a database client that commits", + Has(deny, "mcp__playwright__browser_run_code_unsafe") && Has(deny, "Bash(npm:*)") && Has(deny, "Bash(supabase:*)") && Has(deny, "Write") && Has(deny, "Bash(psql:*)")); + const std::string sq = std::format("Bash({}:*)", Pool::SqlCommand()); + const std::vector redis = {"Bash(redis-cli:*)"}; + const std::vector asb = Pool::AllowedTools("repo-bug-prover.md", "block", here, redis, std::string("u.json"), "supabase"); + Check("supabase allowlist: the SQL helper as its exact prefix (absolute, this binary's sql); no browser tool, no curl, no TOTP command, no extras", + Has(asb, sq) && Pool::SqlCommand() == std::format("{} sql", Platform::SelfExe().string()) + && std::ranges::none_of(asb, [](const std::string& x) { return x.starts_with("mcp__") || x == "Bash(curl:*)" || x == "Bash(redis-cli:*)" || x.contains(" totp "); })); + const std::vector abs = Pool::AllowedTools(Fn, "block", here, redis, std::string("u.json"), "browser+supabase"); + const std::string totpU = std::format("Bash({}:*)", Pool::TotpCommand("u.json").value_or("")); + Check("browser+supabase allowlist: the browser tools, curl, the SQL helper, the TOTP command, the extras - in that order; browser alone has no SQL helper", + Index(abs, "mcp__playwright__browser_navigate") < Index(abs, "Bash(curl:*)") && Index(abs, "Bash(curl:*)") < Index(abs, sq) && Index(abs, sq) < Index(abs, totpU) && Index(abs, totpU) < Index(abs, "Bash(redis-cli:*)") + && !Has(Pool::AllowedTools(Fn, "block", here, {}, std::string("u.json"), "browser"), sq) && !Has(Pool::AllowedTools(Fn, "block", here, {}, std::string("u.json"), "none"), sq)); + const std::vector outside = {"/home/x/repo"}; + const std::vector al2 = Pool::AllowedTools("repo-bug-prover.md", "block", outside); + Check("allowlist (outside tree): Read(//abs/**)", Has(al2, "Read(//home/x/repo/**)") && Has(al2, "Bash(git -C /home/x/repo show:*)")); + const std::vector trees = {".", "sub", "/home/x/repo", "/home/x/repo"}; + Check("extra_dirs: the cwd and paths inside it dropped, outside kept once", Pool::ExtraDirs(trees) == std::vector{"/home/x/repo"}); + + // The MCP config and the command line. + const Json npx = {{"mcp", {{"command", "npx"}, {"args", nullptr}}}}; + const Json mc = Mcp::PlaywrightConfig(npx, "/tmp/out"); + Check("mcp config: the pinned server, headless, isolated, the output dir appended", + mc.at("mcpServers").at("playwright").at("args") == Json::array({"-y", std::format("@playwright/mcp@{}", Mcp::PlaywrightMcpVersion), "--headless", "--isolated", "--browser", "chromium", "--output-dir", "/tmp/out"}) + && mc.at("mcpServers").at("playwright").at("command") == "npx"); + const Json own = {{"mcp", {{"command", "playwright-mcp"}, {"args", {"--headless", "--output-dir", "/x"}}}}}; + Check("a consumer's own command and args are kept; an output dir it names is not overridden", + Dyn::Equal(Mcp::PlaywrightConfig(own, "/tmp/out").at("mcpServers").at("playwright"), {{"command", "playwright-mcp"}, {"args", {"--headless", "--output-dir", "/x"}}})); + const fs::path td = fs::temp_directory_path() / std::format("crafter-review-runner-{}", Platform::PublishedPid()); + fs::remove_all(td); + fs::create_directories(td); + const Json functional = {{"mcp", {{"command", "npx"}, {"args", nullptr}}}, {"extraTools", Json::array()}}; + auto options = [&](std::map environments) { + return Pool::RunnerOptions {.repo = cwd, .scratch = td, .bin = "/bin/true", .effort = "xhigh", .log = [](const std::string&) {}, .dryRun = true, .functional = functional, + .environments = std::move(environments)}; + }; + { + const Pool::Runner runner(options({{Fn, "browser"}})); + const std::vector cf = runner.Cmd(Job("f", Fn), "/dev/null", "{}", here, 20, false, "sid"); + const std::vector cs = runner.Cmd(Job("s", "repo-bug-prover.md"), "/dev/null", "{}", here, 20, false, "sid"); + Check("cmd: the functional pass gets --mcp-config with its own output dir and the browser tools allowed; a static pass gets neither; both keep --tools Read,Glob,Grep,Bash and --strict-mcp-config", + Has(cf, "--mcp-config") && After(cf, "--mcp-config").contains((td / "playwright" / "f").string()) && After(cf, "--allowedTools").contains("mcp__playwright__browser_snapshot") + && !Has(cs, "--mcp-config") && !After(cs, "--allowedTools").contains("mcp__playwright") && After(cf, "--tools") == "Read,Glob,Grep,Bash" && Has(cs, "--strict-mcp-config") + && After(cf, "--disallowedTools").contains("mcp__playwright__browser_run_code_unsafe")); + Check("cmd: --safe-mode drops MCP servers, so the functional pass runs without it and the static pass keeps it", !Has(cf, "--safe-mode") && Has(cs, "--safe-mode")); + } + { + const Pool::Runner rb(options({{"repo-bug-prover.md", "browser"}})); + const std::vector cb = rb.Cmd(Job("b", "repo-bug-prover.md"), "/dev/null", "{}", here, 20, false, "sid"); + const std::vector cfn = rb.Cmd(Job("f", Fn), "/dev/null", "{}", here, 20, false, "sid"); + Check("cmd by environment: a static lens given the browser gets --mcp-config, the browser tools and no --safe-mode; a lens with no environment entry runs as none", + Has(cb, "--mcp-config") && !Has(cb, "--safe-mode") && After(cb, "--allowedTools").contains("mcp__playwright__browser_snapshot") && !Has(cfn, "--mcp-config") && Has(cfn, "--safe-mode") + && !After(cfn, "--allowedTools").contains("mcp__playwright")); + } + { + const std::string url = "postgresql://postgres:hunter2@127.0.0.1:54322/postgres"; + // An operator's shell values never reach a pass. + setenv(std::string(PgWire::UrlEnv).c_str(), "postgresql://leaked@127.0.0.1:1/x", 1); + setenv(std::string(PgWire::BrokerEnv).c_str(), "127.0.0.1:1#leaked", 1); + PgWire::Broker broker; + broker.Start(); + Pool::RunnerOptions withDb = options({{"repo-bug-prover.md", "supabase"}, {Fn, "browser+supabase"}}); + withDb.dbUrl = url; + withDb.broker = &broker; + Pool::Runner rd(withDb); + unsetenv(std::string(PgWire::UrlEnv).c_str()); + unsetenv(std::string(PgWire::BrokerEnv).c_str()); + const std::vector cd = rd.Cmd(Job("d", "repo-bug-prover.md"), "/dev/null", "{}", here, 20, false, "sid"); + const std::vector cbs = rd.Cmd(Job("e", Fn), "/dev/null", "{}", here, 20, false, "sid"); + Check("cmd: the supabase environment keeps --safe-mode (no MCP server) and gets the SQL helper allowed; browser+supabase gets --mcp-config, the helper and no --safe-mode; the URL is on no command line", + Has(cd, "--safe-mode") && !Has(cd, "--mcp-config") && After(cd, "--allowedTools").contains(sq) && Has(cbs, "--mcp-config") && !Has(cbs, "--safe-mode") && After(cbs, "--allowedTools").contains(sq) + && !(Dyn::Join(cd, " ") + Dyn::Join(cbs, " ")).contains("hunter2")); + const Json jd = {{"lensFile", "repo-bug-prover.md"}}; + const Json js = {{"lensFile", "repo-better-solution.md"}}; + const Json jf = {{"lensFile", Fn}}; + const std::string brokerEnv(PgWire::BrokerEnv); + const std::map ed = rd.ProcessEnv(jd, rd.Env()); + const std::map ef = rd.ProcessEnv(jf, rd.Env()); + const std::map es = rd.ProcessEnv(js, rd.Env()); + auto token = [&](const std::map& e) { + const auto found = e.find(brokerEnv); + const std::string value = found == e.end() ? std::string("#") : found->second; + return broker.UrlOf(value.substr(value.rfind('#') + 1)); + }; + auto shaped = [](std::string_view address) { + // ^127\.0\.0\.1:\d+#[0-9a-f]{32}$ + if (!address.starts_with("127.0.0.1:")) return false; + const std::size_t hash = address.find('#'); + if (hash == std::string_view::npos || hash == 10) return false; + const std::string_view port = address.substr(10, hash - 10); + const std::string_view tok = address.substr(hash + 1); + return std::ranges::all_of(port, [](char c) { return c >= '0' && c <= '9'; }) && tok.size() == 32 && std::ranges::all_of(tok, [](char c) { return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f'); }); + }; + bool leaked = false; + for (const auto* e : {&ed, &ef}) { + for (const auto& [k, v] : *e) leaked = leaked || v.contains("hunter2"); + } + Check("process_env: a db pass gets the broker's address and its database's token - never the URL or its password; a static pass gets neither; the operator's shell values are dropped", + shaped(ed.at(brokerEnv)) && token(ed) == url && token(ef) == url && !leaked && !ed.contains(std::string(PgWire::UrlEnv)) && !es.contains(brokerEnv) && !rd.Env().contains(std::string(PgWire::UrlEnv)) + && !rd.Env().contains(brokerEnv)); + const std::string ownDb = "postgresql://postgres:postgres@127.0.0.1:20222/postgres"; + Pool::RunnerOptions noBroker = options({{"repo-bug-prover.md", "supabase"}}); + noBroker.dbUrl = url; + Check("process_env: a pass that holds a stack gets that stack's database and no other - a stack without one gets none, never the first stack's (finding 1)", + token(rd.ProcessEnv({{"lensFile", Fn}, {"stack", {{"id", "slot 2"}, {"dbUrl", ownDb}}}}, rd.Env())) == ownDb + && rd.ProcessEnv({{"lensFile", Fn}, {"stack", {{"id", "slot 2"}, {"dbUrl", nullptr}}}}, rd.Env()).at(brokerEnv).empty() + && Pool::Runner(noBroker).ProcessEnv(jd, {}).at(brokerEnv).empty()); + broker.Close(); + } + + auto init = [](const Json& servers, const Json& tools, const Json& browser = Json()) { + return Json {{"init", {{"mcp_servers", servers}, {"tools", tools}}}, {"browser", browser.is_null() ? Json {{"calls", 2}, {"errors", 0}} : browser}}; + }; + const Json conn = Json::array({{{"name", "playwright"}, {"status", "connected"}}}); + const Json none = Json::array(); + const Json read = Json::array({"Read"}); + const Json pw = Json::array({"mcp__playwright__x"}); + const Json zero = {{"calls", 0}, {"errors", 0}}; + const Pool::Verdict ok = {"ok", ""}; + Check("mcp_connected: ok only with a connected playwright server AND its tools in the init line; else an error that says so", + Pool::McpConnected(init(conn, Json::array({"Read", "mcp__playwright__browser_snapshot"}))) == ok && Pool::McpConnected(init(none, read)).first == "error" + && Pool::McpConnected(init(none, read)).second.contains("did not connect") && Pool::McpConnected(init(Json::array({{{"name", "playwright"}, {"status", "failed"}}}), none)).first == "error" + && Pool::McpConnected({{"init", nullptr}}).first == "error"); + Check("mcp_connected: a whole pass with no browser call, or any pass whose every browser call failed, is an error; a block pass with no browser call is allowed; one answered call is enough", + Pool::McpConnected(init(conn, pw, zero), "whole").second.contains("no browser call") && Pool::McpConnected(init(conn, pw, zero), "block") == ok + && Pool::McpConnected(init(conn, pw, {{"calls", 3}, {"errors", 3}}), "block").second.contains("every browser call") + && Pool::McpConnected(init(conn, pw, {{"calls", 3}, {"errors", 3}}), "whole").second.contains("every browser call") && Pool::McpConnected(init(conn, pw, {{"calls", 3}, {"errors", 2}}), "whole") == ok); + Check("mcp_connected: a static lens given the browser (not empirical) may finish a whole pass without a browser call; a server that did not connect or answered every call with an error still fails it", + Pool::McpConnected(init(conn, pw, zero), "whole", false) == ok && Pool::McpConnected(init(conn, pw, {{"calls", 2}, {"errors", 2}}), "whole", false).first == "error" + && Pool::McpConnected(init(none, read), "block", false).first == "error"); + Check("exercised: 'browser' is mcp_connected with the lens's empirical flag, 'none' is ok as classified", + Pool::Exercised(init(none, read), "browser", "whole", true) == Pool::McpConnected(init(none, read), "whole", true) && Pool::Exercised(init(conn, pw, zero), "browser", "whole", true).first == "error" + && Pool::Exercised(init(none, read), "none", "whole", true) == ok); + std::vector names; + for (const Plan::Environment& e : Plan::Environments) names.emplace_back(e.name); + const Plan::Environment& browser = *Plan::FindEnvironment("browser"); + const Plan::Environment& noneEnv = *Plan::FindEnvironment("none"); + const Plan::Environment& supabase = *Plan::FindEnvironment("supabase"); + const Plan::Environment& both = *Plan::FindEnvironment("browser+supabase"); + auto tools = [](const Plan::Environment& e) { return std::vector(e.tools.begin(), e.tools.end()); }; + bool otherEmpirical = false; + for (const Plan::Lens& l : Plan::Lenses) otherEmpirical = otherEmpirical || (l.file != Fn && l.empirical); + Check("environments: exactly none, browser, supabase, browser+supabase - a new entry first meets the seven-item list in plan.ENVIRONMENTS (the environment in the reuse keys, _resumable and _salvage; its prompt texts; pool.exercised; validate/preflight; DENY; the opencode refusal; a firing proof)", + names == std::vector{"none", "browser", "supabase", "browser+supabase"} && browser.mcp == "playwright" && tools(browser) == std::vector{"Bash(curl:*)"} && browser.app && !browser.db + && browser.memMb == Plan::MemBrowser && noneEnv.mcp.empty() && tools(noneEnv).empty() && !noneEnv.app && !noneEnv.db && noneEnv.memMb == Plan::MemNone && supabase.mcp.empty() && tools(supabase).empty() + && !supabase.app && supabase.db && supabase.memMb == Plan::MemNone && both.mcp == browser.mcp && tools(both) == tools(browser) && both.app && both.db && both.memMb == browser.memMb + && Plan::LensOf(Fn).empirical && !otherEmpirical); + + // The stream accounting behind it: a navigate the server answered with an + // error, a snapshot it answered. + const std::vector ev = { + R"({"type":"system","subtype":"init","mcp_servers":[{"name":"playwright","status":"connected"}],"tools":["mcp__playwright__browser_navigate"]})", + R"({"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"t1","name":"mcp__playwright__browser_navigate","input":{}},{"type":"tool_use","id":"t2","name":"Bash","input":{}}]}})", + R"({"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t1","is_error":true,"content":"Chromium distribution chrome is not found"},{"type":"tool_result","tool_use_id":"t2","content":"ok"}]}})", + R"({"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"t3","name":"mcp__playwright__browser_snapshot","input":{}}]}})", + R"({"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t3","content":"- page"}]}})", + R"({"type":"result","subtype":"success","is_error":false,"structured_output":{"verdict":"SEND-READY","findings":[],"notes":[],"coverage":[]},"usage":{"input_tokens":1,"output_tokens":1},"num_turns":2})", + }; + const Json ps = Stream::ParseStream(Dyn::Join(ev, "\n")); + Check("stream: browser calls and their errors are counted by tool_use id (2 calls, 1 error), Bash errors are not browser errors", + Dyn::Equal(ps.at("browser"), {{"calls", 2}, {"errors", 1}}) && ps.at("toolErrors") == 1 && Pool::McpConnected(ps, "whole") == ok); + const std::vector ev2 = {ev[0], ev[1], ev[2], ev[5]}; + const Json ps2 = Stream::ParseStream(Dyn::Join(ev2, "\n")); + Check("...one failed navigate and nothing else -> every browser call failed -> error", Dyn::Equal(ps2.at("browser"), {{"calls", 1}, {"errors", 1}}) && Pool::McpConnected(ps2, "block").first == "error"); + // The helper as the Flatpak spells it: what the stream recognises is the + // program's name, whatever path a build runs it from. + const std::string sqlCommand = "/app/bin/crafter-review sql"; + auto sqlEv = [&](bool answered, const std::optional& cmd = std::nullopt, const Json& content = Json()) { + const Json result = content.is_null() ? Json(answered ? std::format("{} to postgres on 127.0.0.1:54322 as postgres - ONE transaction\n SELECT 1", PgWire::Answered) + : std::format("{} - cannot connect", PgWire::NoDatabase)) + : content; + return std::vector { + R"({"type":"system","subtype":"init","mcp_servers":[],"tools":["Bash"]})", + DumpSpaced({{"type", "assistant"}, + {"message", {{"role", "assistant"}, {"content", Json::array({{{"type", "tool_use"}, {"id", "s1"}, {"name", "Bash"}, {"input", {{"command", cmd.value_or(std::format("{} <<'SQL'\nselect 1;\nSQL", sqlCommand))}}}}})}}}}), + DumpSpaced({{"type", "user"}, {"message", {{"role", "user"}, {"content", Json::array({{{"type", "tool_result"}, {"tool_use_id", "s1"}, {"is_error", !answered}, {"content", result}}})}}}}), + ev[5], + }; + }; + const Json pa = Stream::ParseStream(Dyn::Join(sqlEv(true), "\n")); + const Json pn = Stream::ParseStream(Dyn::Join(sqlEv(false), "\n")); + Check("stream: a SQL helper call is counted by its command, and answered by the helper's connected line in its result", + Dyn::Equal(pa.at("db"), {{"calls", 1}, {"answered", 1}}) && Dyn::Equal(pn.at("db"), {{"calls", 1}, {"answered", 0}}) && Dyn::Equal(pa.at("browser"), zero)); + Check("stream: the helper inside a compound command counts; the marker in a list-of-blocks result counts; the marker mid-line does not; another python3 script is no SQL call", + Dyn::Equal(Stream::ParseStream(Dyn::Join(sqlEv(true, std::format("cd . && {} 'select 1'", sqlCommand)), "\n")).at("db"), {{"calls", 1}, {"answered", 1}}) + && Stream::ParseStream(Dyn::Join(sqlEv(true, std::nullopt, Json::array({{{"type", "text"}, {"text", std::format("{} to x", PgWire::Answered)}}})), "\n")).at("db").at("answered") == 1 + && Stream::ParseStream(Dyn::Join(sqlEv(true, std::nullopt, Json(std::format("echo says {}", PgWire::Answered))), "\n")).at("db").at("answered") == 0 + && Dyn::Equal(Stream::ParseStream(Dyn::Join(sqlEv(true, std::string("python3 - < noSql = {sqlEv(true)[0], ev[5]}; + Check("exercised: supabase is db_answered - a pass whose every SQL call missed the database is an error, a pass with no SQL call is ok (a lens may judge by the code)", + Pool::Exercised(pa, "supabase", "block") == ok && Pool::Exercised(pn, "supabase", "block").first == "error" && Pool::Exercised(pn, "supabase", "whole", true).second.contains("reached the database") + && Pool::Exercised(Stream::ParseStream(Dyn::Join(noSql, "\n")), "supabase", "whole", true) == ok && Pool::Exercised(pn, "none", "block") == ok); + const Json bothOk = With(pa, {{"init", {{"mcp_servers", conn}, {"tools", pw}}}, {"browser", {{"calls", 2}, {"errors", 0}}}}); + Check("exercised: browser+supabase needs both - the browser (connected, used by the empirical whole pass) and, when SQL ran, the database", + Pool::Exercised(bothOk, "browser+supabase", "whole", true) == ok && Pool::Exercised(With(pa, {{"init", {{"mcp_servers", Json::array()}}}}), "browser+supabase", "block").second.contains("did not connect") + && Pool::Exercised(With(bothOk, {{"db", {{"calls", 2}, {"answered", 0}}}}), "browser+supabase", "whole", true).second.contains("reached the database") + && Pool::Exercised(With(bothOk, {{"browser", zero}}), "browser+supabase", "whole", true).second.contains("no browser call")); + + fs::remove_all(td); + return Failures == 0 ? 0 : 1; +} diff --git a/tests/Scheduler/main.cpp b/tests/Scheduler/main.cpp new file mode 100644 index 0000000..d7f2fcc --- /dev/null +++ b/tests/Scheduler/main.cpp @@ -0,0 +1,506 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// The pool at work: re-placement after a seat's failure, the Runner on a +// fake claude (the seat, the MCP check, resume, the database broker), a +// no-room wait other drivers hold, the machine-wide cap, the stack keys, the +// admission against other drivers' memory and stacks, and the race for the +// last stack. +// +// The test binary is the supervisor every pass runs under (`__supervise`); +// another driver is a live `sleep 300`. + +#include +#include +#include +#include +import std; +import Crafter.Review; +namespace fs = std::filesystem; +using namespace Crafter::Review; +using namespace std::chrono_literals; + +namespace { + std::int32_t Failures = 0; + const std::string Fn(Plan::FunctionalLens); + + void Check(std::string_view name, bool cond, std::string_view detail = {}) { + std::println("{}{}{}", cond ? " ok " : " FAIL ", name, !cond && !detail.empty() ? std::format(" ({})", detail) : std::string()); + if (!cond) ++Failures; + } + + void Write(const fs::path& path, std::string_view text) { + fs::create_directories(path.parent_path()); + std::ofstream(path, std::ios::binary) << text; + } + + std::string Read(const fs::path& path) { + return Platform::ReadFile(path).value_or(""); + } + + Json With(Json value, const Json& changes) { + for (auto it = changes.begin(); it != changes.end(); ++it) value[it.key()] = it.value(); + return value; + } + + bool AnyContains(const std::vector& lines, std::string_view part) { + return std::ranges::any_of(lines, [&](const std::string& line) { return line.contains(part); }); + } + + // The log a pool writes into from its threads. + struct Logs { + std::mutex lock; + std::vector lines; + Pool::Log Sink() { + return [this](const std::string& line) { + const std::lock_guard guard(lock); + lines.push_back(line); + }; + } + std::vector Copy() { + const std::lock_guard guard(lock); + return lines; + } + void Clear() { + const std::lock_guard guard(lock); + lines.clear(); + } + }; + + struct Sleeper { + std::int64_t pid = Process::Start({.argv = {"sleep", "300"}}).pid; + std::string key = std::format("{}:{}", pid, Platform::ProcStart(pid).value_or(0)); + + void Kill() const { + Process::Signal(pid, SIGKILL); + Process::Wait(pid); + } + }; + + double Seconds(std::chrono::steady_clock::time_point since) { + return std::chrono::duration(std::chrono::steady_clock::now() - since).count(); + } + + const Pool::Scheduler::OnDone NoFollowUp = [](const Json&, const Json&, const Json&) { return std::vector {}; }; + + void TestPool() { + std::println("pool: admission (stacks, memory), a no-room wait ends on the next completion, one pass per cold prefix goes first"); + std::mutex lock; + std::map running = {{"F", 0}, {"S", 0}}; + std::map peak = {{"F", 0}, {"S", 0}}; + auto execJob = [&](const Json& job, std::int64_t) { + const std::string lens = job.at("lens").get(); + { + const std::lock_guard guard(lock); + peak[lens] = std::max(peak[lens], ++running[lens]); + } + std::this_thread::sleep_for(150ms); + { + const std::lock_guard guard(lock); + --running[lens]; + } + return std::pair({{"label", job.at("label")}, {"status", "ok"}}, nullptr); + }; + std::vector slots; + for (const std::int32_t n : {1, 2}) slots.push_back({{"id", std::format("slot {}", n)}, {"appUrl", std::format("http://localhost:{}", n)}, {"apiUrls", Json::array()}, {"dbUrl", nullptr}, {"login", nullptr}, {"slot", true}}); + Pool::Stacks stacks(slots); + auto plenty = []() -> std::optional { return 1e6; }; + Pool::Admission adm([](const Json&) { return 100.0; }, [](const Json& j) { return j.at("lens") == "F"; }, &stacks, 10'000.0, 0, plenty); + std::vector jobs; + for (std::int32_t i = 0; i < 6; ++i) jobs.push_back({{"label", std::format("f{}", i)}, {"lens", "F"}}); + for (std::int32_t i = 0; i < 3; ++i) jobs.push_back({{"label", std::format("s{}", i)}, {"lens", "S"}}); + Pool::Scheduler s({.concurrency = 8, .log = [](const std::string&) {}, .admission = &adm}); + s.Run(jobs, execJob, NoFollowUp); + std::vector free = stacks.Free(); + std::ranges::sort(free); + Check("admission: the passes that need a stack never outnumber the stacks (2), the rest fill the pool, everything finishes, every stack is back", + s.DoneList().size() == 9 && peak["F"] == 2 && peak["S"] == 3 && free == std::vector{"slot 1", "slot 2"} && adm.Inflight() == 0 && adm.Held() == 0, + std::format("F {} S {}", peak["F"], peak["S"])); + running = {{"F", 0}, {"S", 0}}; + peak = {{"F", 0}, {"S", 0}}; + Pool::Admission memory([](const Json&) { return 300.0; }, [](const Json&) { return false; }, nullptr, 1000.0, 0, plenty); + jobs.clear(); + for (std::int32_t i = 0; i < 7; ++i) jobs.push_back({{"label", std::format("s{}", i)}, {"lens", "S"}}); + Pool::Scheduler m({.concurrency = 8, .log = [](const std::string&) {}, .admission = &memory}); + m.Run(jobs, execJob, NoFollowUp); + Check("admission: memory - passes at 300 MB in a 1000 MB budget run three at once, and all of them finish", m.DoneList().size() == 7 && peak["S"] == 3, std::format("S {}", peak["S"])); + + Logs logs; + const double now = Platform::Now(); + std::vector seat = {{{"id", "3"}, {"kind", "oauth"}, {"secret", "x"}, {"state", "ok"}, {"util", 80.0}, {"binding", "five_hour"}, {"resetsAt", nullptr}, + {"windows", {{"five_hour", {{"utilization", 80.0}, {"resetsAt", now + 3600}}}, {"seven_day", {{"utilization", 30.0}, {"resetsAt", now + 86400}}}}}, + {"pauseUntil", 0.0}, {"running", 0}, {"procs", 0}, {"spentUsd", 0.0}, {"error", nullptr}, {"probedWith", nullptr}}}; + Json cfg = SeatPool::PoolConfig({}); + cfg["lensCostPct"] = 2; + cfg["maxUtil"] = 85; + Pool::Scheduler r({.concurrency = 8, .log = logs.Sink(), .seats = &seat, .seatConfig = cfg, .apiConfig = SeatPool::ApiConfig({})}); + const auto t0 = std::chrono::steady_clock::now(); + jobs.clear(); + for (std::int32_t i = 0; i < 3; ++i) jobs.push_back({{"label", std::format("r{}", i)}}); + r.Run(jobs, [](const Json& j, std::int64_t) { + std::this_thread::sleep_for(200ms); + return std::pair({{"label", j.at("label")}, {"status", "ok"}}, nullptr); + }, NoFollowUp); + const std::vector waited = logs.Copy(); + Check("a no-room wait ends on the next completion, not at the window reset", + r.DoneList().size() == 3 && Seconds(t0) < 5 && AnyContains(waited, "no seat has room") && AnyContains(waited, "seat room is re-checked") && r.PauseUntil() == 0, Dyn::Join(waited, " | ")); + + std::map started; + std::map answered; + logs.Clear(); + Pool::Scheduler w({.concurrency = 8, + .log = logs.Sink(), + .cacheKey = [](Json& j) -> std::optional { + const Json& k = Dyn::Get(j, "key"); + return k.is_null() ? std::nullopt : std::optional(k.get()); + }, + .warmProbe = [&](const Json& j) { + const std::lock_guard guard(lock); + return answered.contains(j.at("label").get()); + }, + .warmPoll = 0.02}); + jobs.clear(); + for (std::int32_t i = 0; i < 5; ++i) jobs.push_back({{"label", std::format("a{}", i)}, {"key", "A"}, {"respondAfter", i == 0 ? 0.3 : 0.05}}); + jobs.push_back({{"label", "n0"}, {"key", nullptr}}); + w.Run(jobs, [&](const Json& j, std::int64_t) { + const std::string label = j.at("label").get(); + { + const std::lock_guard guard(lock); + started[label] = Platform::Now(); + } + std::this_thread::sleep_for(std::chrono::duration(Dyn::Has(j, "respondAfter") ? j.at("respondAfter").get() : 0.05)); + { + const std::lock_guard guard(lock); + answered[label] = Platform::Now(); + } + std::this_thread::sleep_for(300ms); + return std::pair({{"label", j.at("label")}, {"status", "ok"}}, nullptr); + }, NoFollowUp); + const double first = answered.at("a0"); + bool followed = true; + for (std::int32_t i = 1; i < 5; ++i) { + const double at = started.at(std::format("a{}", i)); + followed = followed && first <= at && at < first + 0.3; + } + Check("warm-up: the first pass of a cold prefix starts alone, the others follow its first response, a pass without a key is never held", + w.DoneList().size() == 6 && followed && started.at("n0") < first && AnyContains(logs.Copy(), "a0 goes first")); + } + + void TestSeatPool(const fs::path& td) { + std::println("seat pool: re-placement on another seat, Runner.run on a fake claude"); + const Json acfg = SeatPool::ApiConfig({{"REVIEW_GATES_API_BUDGET_USD", "10"}, {"REVIEW_GATES_API_LENS_COST_USD", "1"}, {"REVIEW_GATES_API_CONCURRENCY", "2"}}); + const Json cfg = Seats::ConfigFromEnv({}); + Logs logs; + std::mutex lock; + std::vector> order; + std::vector pool = {SeatPool::NewSeat("api1", "api", "k"), SeatPool::NewSeat("1", "oauth", "t1")}; + for (Json& s : pool) s["state"] = "ok"; + SeatPool::Ledger ledger(td / "l2.json", acfg); + Pool::Scheduler sch({.concurrency = 4, .log = logs.Sink(), .seats = &pool, .seatConfig = cfg, .apiConfig = With(acfg, {{"budgetUsd", 100}}), .ledger = &ledger}); + sch.Run({{{"label", "j1"}}}, [&](const Json& job, std::int64_t) { + { + const std::lock_guard guard(lock); + order.emplace_back(job.at("label").get(), job.at("seat").get()); + } + if (job.at("seat") == "api1") return std::pair({{"label", job.at("label")}, {"status", "credits"}, {"reason", "Your credit balance is too low"}, {"costUsd", 0.0}, {"rateLimit", Json::object()}}, nullptr); + return std::pair({{"label", job.at("label")}, {"status", "ok"}, {"costUsd", 0.5}, {"rateLimit", {{"five_hour", {{"utilization", 30}, {"resetsAt", 9'999'999'999LL}}}}}}, nullptr); + }, NoFollowUp); + std::map statuses; + for (const Pool::Done& d : sch.DoneList()) statuses[d.rec.at("label").get()] = d.rec.at("status").get(); + Check("a process that fails on its seat is placed again on another; the first seat is out", + statuses == std::map{{"j1", "ok"}} && pool[0].at("state") == "out" && !order.empty() && order.front().second == "api1" && order.back().second == "1"); + + const fs::path fake = td / "claude"; + const fs::path argsf = td / "args.txt"; + Write(fake, "#!/bin/sh\nprintf \"%s\\n\" \"$@\" > \"$FAKE_ARGS\"\nprintf \"%s\" \"$REVIEW_GATES_DB\" > \"$FAKE_ARGS.db\"\ncat > /dev/null\n" + "if [ -n \"$FAKE_MCP\" ]; then echo '{\"type\":\"system\",\"subtype\":\"init\",\"mcp_servers\":[{\"name\":\"playwright\",\"status\":\"connected\"}],\"tools\":[\"Read\",\"mcp__playwright__browser_snapshot\"]}'; " + "echo '{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"t1\",\"name\":\"mcp__playwright__browser_snapshot\",\"input\":{}}]}}'; " + "echo '{\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"t1\",\"content\":\"- page\"}]}}'; " + "elif [ -n \"$FAKE_SQL\" ]; then echo '{\"type\":\"system\",\"subtype\":\"init\",\"mcp_servers\":[],\"tools\":[\"Read\",\"Bash\"]}'; printf \"%s\\n\" \"$FAKE_SQL\"; " + "else echo '{\"type\":\"system\",\"subtype\":\"init\",\"mcp_servers\":[],\"tools\":[\"Read\"]}'; fi\n" + "echo '{\"type\":\"result\",\"subtype\":\"success\",\"is_error\":false,\"result\":\"ok\",\"structured_output\":{\"verdict\":\"SEND-READY\",\"findings\":[],\"notes\":[],\"coverage\":[]}," + "\"usage\":{\"input_tokens\":1,\"output_tokens\":1},\"total_cost_usd\":0.01}'\n"); + fs::permissions(fake, fs::perms::owner_all | fs::perms::group_read | fs::perms::group_exec | fs::perms::others_read | fs::perms::others_exec); + setenv("FAKE_ARGS", argsf.c_str(), 1); + const fs::path rscratch = td / "rscratch"; + const std::string cwd = fs::current_path().string(); + Pool::Runner runner({.repo = cwd, .scratch = rscratch, .bin = fake.string(), .effort = "xhigh", .log = logs.Sink(), .dryRun = false, + .functional = {{"mcp", {{"command", "npx"}, {"args", nullptr}}}, {"extraTools", Json::array()}}, .environments = {{Fn, "browser"}}}); + runner.SetSeats({SeatPool::NewSeat("api1", "api", "k1"), SeatPool::NewSeat("api2", "api", "k2")}); + const fs::path sysf = td / "sys.md"; + Write(sysf, "x"); + const std::vector trees = {"."}; + auto args = [&] { return Dyn::Split(Read(argsf), "\n"); }; + auto has = [](const std::vector& list, std::string_view item) { return std::ranges::contains(list, item); }; + auto [rec, raw] = runner.Run({{"label", "f1"}, {"model", "claude-opus-5-5"}, {"lensFile", "repo-bug-prover.md"}, {"kind", "block"}, {"seat", "api1"}, {"attempt", 1}}, "the prompt", sysf, "{}", trees, 20, 1); + std::vector a = args(); + Check("Runner.run on a fake claude: ok, the seat on the record, persisted, no --mcp-config for a static pass, environment none", + rec.at("status") == "ok" && rec.at("seat") == "api1" && !has(a, "--no-session-persistence") && has(a, "--session-id") && !has(a, "--mcp-config") && rec.at("mcp") == false && rec.at("environment") == "none", + DumpSpaced(rec)); + std::tie(rec, raw) = runner.Run({{"label", "f2"}, {"model", "claude-opus-5-5"}, {"lensFile", Fn}, {"kind", "block"}, {"seat", "api2"}, {"attempt", 1}}, "the prompt", sysf, "{}", trees, 20, 1); + a = args(); + Check("...a functional pass whose init line shows no connected server is an error (no browser = no functional evidence), with --mcp-config passed, its playwright output dir made and environment browser", + rec.at("mcp") == true && rec.at("environment") == "browser" && rec.at("status") == "error" && rec.at("reason").get().contains("did not connect") && has(a, "--mcp-config") + && fs::is_directory(rscratch / "playwright" / "f2")); + // The runner captured its environment at construction; the fake reads + // this one. + runner.Env()["FAKE_MCP"] = "1"; + std::tie(rec, raw) = runner.Run({{"label", "f2"}, {"model", "claude-opus-5-5"}, {"lensFile", Fn}, {"kind", "block"}, {"seat", "api2"}, {"attempt", 1}}, "the prompt", sysf, "{}", trees, 20, 1); + runner.Env().erase("FAKE_MCP"); + Check("...and ok when the server connected", rec.at("status") == "ok" && !raw.is_null()); + const std::string sid1 = rec.at("sessionId").get(); + Write(rscratch / "seat-api2" / "projects" / "-home-repo" / (std::format("{}.jsonl", sid1)), "{}\n"); + std::tie(rec, raw) = runner.Run({{"label", "f2"}, {"model", "claude-opus-5-5"}, {"lensFile", Fn}, {"kind", "block"}, {"seat", "api1"}, {"attempt", 2}, {"resumeSession", sid1}, {"resumeSeat", "api2"}}, + "the prompt", sysf, "{}", trees, 20, 1); + a = args(); + const auto resumeAt = std::ranges::find(a, "--resume"); + Check("resumed pass: --resume , the transcript carried to the new seat, the resume prompt", + rec.at("sessionId") == sid1 && resumeAt != a.end() && resumeAt + 1 != a.end() && *(resumeAt + 1) == sid1 && fs::exists(rscratch / "seat-api1" / "projects" / "-home-repo" / (std::format("{}.jsonl", sid1))) + && Read(rscratch / "prompts" / "f2.txt") == SeatPool::ResumePrompt); + + // A pass with the database under test: the broker's address and a token + // in its process alone, never the URL; and no evidence when no SQL call + // reached the database. + const std::string durl = "postgresql://postgres:hunter2@127.0.0.1:54322/postgres"; + PgWire::Broker dbr; + dbr.Start(); + Pool::Runner rdb({.repo = cwd, .scratch = td / "dscratch", .bin = fake.string(), .effort = "xhigh", .log = logs.Sink(), .dryRun = false, + .environments = {{"repo-bug-prover.md", "supabase"}}, .dbUrl = durl, .broker = &dbr}); + rdb.SetSeats({SeatPool::NewSeat("api1", "api", "k1")}); + auto sqlLines = [](const std::string& content) { + return std::format("{}\n{}", DumpSpaced({{"type", "assistant"}, + {"message", {{"role", "assistant"}, {"content", Json::array({{{"type", "tool_use"}, {"id", "q1"}, {"name", "Bash"}, {"input", {{"command", "/app/bin/crafter-review sql <<'SQL'\nselect 1;\nSQL"}}}}})}}}}), + DumpSpaced({{"type", "user"}, {"message", {{"role", "user"}, {"content", Json::array({{{"type", "tool_result"}, {"tool_use_id", "q1"}, {"content", content}}})}}}})); + }; + rdb.Env()["FAKE_SQL"] = sqlLines(std::format("{} - cannot connect to 127.0.0.1:54322", PgWire::NoDatabase)); + std::tie(rec, raw) = rdb.Run({{"label", "d1"}, {"model", "claude-opus-5-5"}, {"lensFile", "repo-bug-prover.md"}, {"kind", "block"}, {"seat", "api1"}, {"attempt", 1}}, "the prompt", sysf, "{}", trees, 20, 1); + const std::string argsText = Read(argsf); + const std::string gotEnv = Read(std::format("{}.db", argsf.string())); + Check("Runner.run, supabase: the db pass's process gets the broker and its token (no URL, no password, on no command line); a pass whose every SQL call missed the database is an error with no result", + gotEnv == dbr.Address(durl) && dbr.UrlOf(gotEnv.substr(gotEnv.rfind('#') + 1)) == durl && !(gotEnv + argsText).contains("hunter2") && has(Dyn::Split(argsText, "\n"), "--safe-mode") && rec.at("status") == "error" + && rec.at("reason").get().contains("reached the database") && raw.is_null() && Dyn::Equal(rec.at("db"), {{"calls", 1}, {"answered", 0}}) && rec.at("environment") == "supabase", + DumpSpaced(rec)); + rdb.Env()["FAKE_SQL"] = sqlLines(std::format("{} to postgres on 127.0.0.1:54322 as postgres - ONE transaction\n SELECT 1", PgWire::Answered)); + std::tie(rec, raw) = rdb.Run({{"label", "d1"}, {"model", "claude-opus-5-5"}, {"lensFile", "repo-bug-prover.md"}, {"kind", "block"}, {"seat", "api1"}, {"attempt", 1}}, "the prompt", sysf, "{}", trees, 20, 1); + rdb.Env().erase("FAKE_SQL"); + Check("...and ok once the database answered", rec.at("status") == "ok" && !raw.is_null() && Dyn::Equal(rec.at("db"), {{"calls", 1}, {"answered", 1}})); + std::tie(rec, raw) = rdb.Run({{"label", "d2"}, {"model", "claude-opus-5-5"}, {"lensFile", "repo-better-solution.md"}, {"kind", "block"}, {"seat", "api1"}, {"attempt", 1}}, "the prompt", sysf, "{}", trees, 20, 1); + Check("...while a lens without the database gets no broker at all", Read(std::format("{}.db", argsf.string())).empty() && rec.at("status") == "ok" && rec.at("environment") == "none"); + dbr.Close(); + unsetenv("FAKE_ARGS"); + } + + void TestDrivers(const fs::path& td) { + std::println("several drivers at once: the pool against the board - a wait other drivers hold, the machine's cap, stacks and memory across drivers"); + const fs::path path = td / "seat-board.json"; + Logs logs; + const double now = Platform::Now(); + const Json cfg = SeatPool::PoolConfig({}); + const Json acfg = SeatPool::ApiConfig({{"REVIEW_GATES_API_CONCURRENCY", "8"}, {"REVIEW_GATES_API_LENS_COST_USD", "1"}}); + auto mk = [&](const std::string& id, const std::string& secret, std::int64_t fh, std::int64_t sd) { + Json s = SeatPool::NewSeat(id, "oauth", secret); + s["state"] = "ok"; + s["windows"] = {{"five_hour", {{"utilization", fh}, {"resetsAt", now + 3600}}}, {"seven_day", {{"utilization", sd}, {"resetsAt", now + 86400}}}}; + const auto [util, binding] = Seats::SeatUtil(s.at("windows")); + s["util"] = util; + s["binding"] = binding; + return s; + }; + std::deque jobs; + std::mutex jobsLock; + auto chooserFor = [&](const std::string& id, const std::string& label, const Json& extra = Json::object()) { + return SeatPool::Board::Chooser([&jobs, &jobsLock, id, label, extra](const std::vector& ranked, const SeatPool::Others&) -> std::optional> { + for (Json* s : ranked) { + if (s->at("id") == id) { + const std::lock_guard guard(jobsLock); + jobs.push_back(With({{"label", label}, {"timeoutMin", 20}}, extra)); + return std::pair(s, &jobs.back()); + } + } + return std::nullopt; + }); + }; + auto place = [&](SeatPool::Board& board, Json& seat, const Json& c, const SeatPool::Board::Chooser& chooser) { + std::vector one = {seat}; + board.Place(one, now, c, acfg, chooser); + seat = one.front(); + }; + SeatPool::Board A(path, {}, std::nullopt, logs.Sink()); + A.Join({{"run", "A"}}, now); + const Sleeper child2; + SeatPool::Board D(path, child2.key, std::nullopt, logs.Sink()); + D.Join({{"run", "D"}}, now); + + const Json cfg3 = With(cfg, {{"lensCostPct", 2}, {"maxUtil", 85}}); + std::vector seat3 = {mk("3", "t3", 80, 30)}; + Json sd3 = mk("3", "t3", 80, 30); + place(D, sd3, cfg3, chooserFor("3", "h0")); + // 80 + 4 held: 80 + 4 + 2 > 85, no room for A until D's passes finish. + place(D, sd3, cfg3, chooserFor("3", "h1")); + auto execJob = [&](const Json& job, std::int64_t) { + std::this_thread::sleep_for(100ms); + return std::pair({{"label", job.at("label")}, {"status", "ok"}, {"rateLimit", {{"five_hour", {{"utilization", 81}, {"resetsAt", now + 3600}}}}}}, nullptr); + }; + { + std::jthread freer([&] { + std::this_thread::sleep_for(500ms); + D.Finish(sd3, "h0", Platform::Now()); + D.Finish(sd3, "h1", Platform::Now()); + }); + logs.Clear(); + Pool::Scheduler sch({.concurrency = 8, .log = logs.Sink(), .seats = &seat3, .seatConfig = cfg3, .apiConfig = acfg, .board = &A, .noRoomWait = 0.2}); + const auto t0 = std::chrono::steady_clock::now(); + sch.Run({{{"label", "a0"}, {"timeoutMin", 20}}}, execJob, NoFollowUp); + const double dt = Seconds(t0); + const Json bd = Dyn::LoadJsonFile(path); + const Json& inflight = bd.at("seats").at(seat3[0].at("boardKey").get()).at("inflight"); + bool mine = false; + for (auto it = inflight.begin(); it != inflight.end(); ++it) mine = mine || it.key().starts_with(A.Me()); + const std::vector lines = logs.Copy(); + Check("a no-room wait held by another driver's passes ends when they finish, not at the window reset", + sch.DoneList().size() == 1 && dt < 5 && AnyContains(lines, "hold the room") && !mine, std::format("{:.1f}s {}", dt, Dyn::Join(lines, " | "))); + Check("...and the finished pass's reading is on the board", bd.at("seats").at(seat3[0].at("boardKey").get()).at("windows").at("five_hour").at("utilization") == 81); + } + place(D, sd3, cfg3, chooserFor("3", "w0")); + // Two of D's on the machine: a width of 2 holds A's pass until one + // finishes. + place(D, sd3, cfg3, chooserFor("3", "w1")); + { + std::jthread freer([&] { + std::this_thread::sleep_for(500ms); + D.Finish(sd3, "w0", Platform::Now()); + }); + logs.Clear(); + std::vector seat3b = {mk("3", "t3", 10, 30)}; + Pool::Scheduler sch({.concurrency = 2, .log = logs.Sink(), .seats = &seat3b, .seatConfig = cfg3, .apiConfig = acfg, .board = &A, .noRoomWait = 0.2}); + const auto t0 = std::chrono::steady_clock::now(); + sch.Run({{{"label", "a1"}, {"timeoutMin", 20}}}, execJob, NoFollowUp); + const double dt = Seconds(t0); + const std::vector lines = logs.Copy(); + Check("the machine-wide cap holds a launch while other drivers fill it, and releases it when one of theirs finishes", + sch.DoneList().size() == 1 && 0.4 < dt && dt < 5 && AnyContains(lines, "the cap is 2"), std::format("{:.1f}s {}", dt, Dyn::Join(lines, " | "))); + } + D.Finish(sd3, "w1", Platform::Now()); + + // Memory and stacks across drivers (2026-10-31). + Check("stack keys: one application whatever the spelling (localhost/127.0.0.1, case, trailing slash, the default port); another port another stack; no URL, its id", + Pool::StackKey({{"id", "a"}, {"appUrl", "http://localhost:5173/"}}) == Pool::StackKey({{"id", "b"}, {"appUrl", "HTTP://127.0.0.1:5173"}}) + && Pool::StackKey({{"id", "b"}, {"appUrl", "HTTP://127.0.0.1:5173"}}) == "http://127.0.0.1:5173" && Pool::StackKey({{"id", "c"}, {"appUrl", "http://localhost"}}) == "http://127.0.0.1:80" + && Pool::StackKey({{"id", "d"}, {"appUrl", "http://localhost:5273"}}) != Pool::StackKey({{"id", "a"}, {"appUrl", "http://localhost:5173"}}) && Pool::StackKey({{"id", "s1"}}) == "s1"); + // What the board's fold left in A.others when another driver had two + // passes in flight (test_drivers checks the fold itself). + SeatPool::Others held; + held.count = 2; + held.mb = 1050.0; + held.stacks = {"http://127.0.0.1:5173"}; + auto plenty = []() -> std::optional { return 1e6; }; + Pool::Stacks stx({{{"id", "slot 1"}, {"appUrl", "http://localhost:5173/"}}, {{"id", "slot 2"}, {"appUrl", "http://localhost:5273"}}}); + Pool::Admission adm([](const Json& j) { return j.at("mb").get(); }, [](const Json& j) { return Dyn::Truthy(Dyn::Get(j, "app")); }, &stx, 2000.0, 0, plenty, logs.Sink()); + adm.SetOthers(held); + Json j1 = {{"label", "x"}, {"mb", 840}, {"app", true}}; + const bool ok = adm.Can(j1); + adm.Take(j1); + Check("Admission: a stack another driver holds is not free here - the pass takes the other; their memory counts against the budget beside this driver's (1050 + 840 + 200 > 2000, + 100 fits)", + ok && j1.at("stack").at("id") == "slot 2" && j1.at("stackKey") == "http://127.0.0.1:5273" && !adm.Can({{"label", "y"}, {"mb", 840}, {"app", true}}) && !adm.Can({{"label", "z"}, {"mb", 200}}) + && adm.Can({{"label", "z"}, {"mb", 100}}), + DumpSpaced(Dyn::Get(j1, "stack"))); + adm.Give(j1); + logs.Clear(); + const Json big = {{"label", "b"}, {"mb", 5000}}; + const bool waits = !adm.Can(big); + adm.SetOthers(SeatPool::Others()); + Check("Admission: nothing of this driver's in flight but other drivers' passes on the machine - a pass over the budget waits for them instead of running alone; with the machine idle it runs alone", + waits && adm.Can(big) && !j1.contains("stackKey") && logs.Copy().empty()); + + // The pool: a stack another driver holds - the pass waits (re-checked, + // not spun on) and runs when theirs finishes. + Json sE = mk("1", "t1", 10, 50); + place(D, sE, cfg, chooserFor("1", "hold", {{"_admitted", 840.0}, {"stackKey", "http://127.0.0.1:5173"}})); + Pool::Stacks stw({{{"id", "slot 1"}, {"appUrl", "http://localhost:5173"}}}); + Pool::Admission admw([](const Json&) { return 100.0; }, [](const Json&) { return true; }, &stw, 10'000.0, 0, plenty, logs.Sink()); + double seenAt = 0; + Json seenEntry; + const std::string keyE = sE.at("boardKey").get(); + auto execBoard = [&](const Json& job, std::int64_t) { + seenAt = Platform::Now(); + seenEntry = Dyn::Get(Dyn::LoadJsonFile(path).at("seats").at(keyE).at("inflight"), std::format("{}/{}", A.Me(), job.at("label").get())); + std::this_thread::sleep_for(100ms); + return std::pair({{"label", job.at("label")}, {"status", "ok"}}, nullptr); + }; + { + std::jthread freer([&] { + std::this_thread::sleep_for(600ms); + D.Finish(sE, "hold", Platform::Now()); + }); + logs.Clear(); + std::vector seats = {mk("1", "t1", 10, 50)}; + Pool::Scheduler sch({.concurrency = 8, .log = logs.Sink(), .seats = &seats, .seatConfig = cfg, .apiConfig = acfg, .admission = &admw, .board = &A, .noRoomWait = 0.2}); + const double t0 = Platform::Now(); + sch.Run({{{"label", "w0"}, {"timeoutMin", 20}}}, execBoard, NoFollowUp); + const Json bd = Dyn::LoadJsonFile(path); + bool mine = false; + for (const Json& s : bd.at("seats")) { + for (auto it = Dyn::Or(Dyn::Get(s, "inflight"), Dyn::EmptyDict()).begin(); it != Dyn::Or(Dyn::Get(s, "inflight"), Dyn::EmptyDict()).end(); ++it) mine = mine || it.key().starts_with(A.Me()); + } + const std::vector lines = logs.Copy(); + const auto once = std::ranges::count_if(lines, [](const std::string& l) { return l.contains("nothing can launch"); }); + Check("the pool: a pass whose only stack another driver holds waits, re-checked every no_room_wait and logged once, and launches when theirs finishes - on the board with its memory and its stack", + sch.DoneList().size() == 1 && 0.5 < seenAt - t0 && seenAt - t0 < 5 && once == 1 && AnyContains(lines, "stack(s) http://127.0.0.1:5173") && Dyn::Equal(Dyn::Get(seenEntry, "mb"), 100.0) + && Dyn::Get(seenEntry, "stack") == "http://127.0.0.1:5173" && stw.Can() && !mine, + std::format("{:.2f}s {} {}", seenAt - t0, DumpSpaced(seenEntry), Dyn::Join(lines, " | "))); + } + + // Two drivers racing for the last stack: one gets it. + const Sleeper child3; + SeatPool::Board F(path, child3.key, std::nullopt, logs.Sink()); + F.Join({{"run", "F"}}, now); + std::mutex picksLock; + std::vector picks; + auto racer = [&](SeatPool::Board& board, const std::string& label) { + Pool::Stacks st({{{"id", "slot 1"}, {"appUrl", "http://localhost:5999"}}}); + Pool::Admission ad([](const Json&) { return 100.0; }, [](const Json&) { return true; }, &st, 10'000.0, 0, plenty); + Json job = {{"label", label}, {"timeoutMin", 20}}; + std::vector one = {mk("1", "t1", 10, 50)}; + const SeatPool::Board::Placed got = board.Place(one, now, cfg, acfg, [&](const std::vector& cands, const SeatPool::Others& others) -> std::optional> { + ad.SetOthers(others); + std::this_thread::sleep_for(50ms); + if (!ad.Can(job)) return std::nullopt; + ad.Take(job); + return std::pair(cands.front(), &job); + }); + const std::lock_guard guard(picksLock); + picks.push_back(got.picked.has_value()); + }; + { + const std::jthread ta(racer, std::ref(A), std::string("ra")); + const std::jthread tf(racer, std::ref(F), std::string("rf")); + } + Check("two drivers racing for the last stack on the machine: exactly one gets it", std::ranges::count(picks, true) == 1); + F.Leave(); + child3.Kill(); + A.Leave(); + child2.Kill(); + } +} + +int main(int argc, char** argv) { + const std::vector args(argv, argv + argc); + if (args.size() > 1 && args[1] == Process::SuperviseVerb) return Process::SuperviseMain(std::span(args).subspan(2)); + Platform::SetMode(Platform::Mode::Native); + const fs::path td = fs::temp_directory_path() / std::format("crafter-review-scheduler-{}", Platform::PublishedPid()); + fs::remove_all(td); + fs::create_directories(td); + // No check reaches the machine's own board: the drivers running on it are + // real. + setenv("REVIEW_GATES_BOARD", (td / "seat-board.json").c_str(), 1); + try { + TestPool(); + TestSeatPool(td); + TestDrivers(td); + } catch (const std::exception& error) { + std::println(std::cerr, "scheduler: {}", error.what()); + return 1; + } + fs::remove_all(td); + return Failures == 0 ? 0 : 1; +}