Jorijn van der Graaf
61ee4602f4
Jorijn: "there is no need to do a global cap anymore, the cap should be 128 per available account". `--concurrency` no longer defaults to 128: the limits are each OAuth account's 128 (and the local login's), every driver's counted, an API seat's own REVIEW_GATES_API_CONCURRENCY, and free memory. `--concurrency N` still caps the machine when given; a value below 1 is a usage error. The dump's args.concurrency is null then. The board's driver settings read `128 per account`, which every driver's `seats` shows as that driver's width, and the `seats` header says a width is that driver's cap. The Scheduler test runs 300 passes on two accounts: 256 at once, 128 each. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1065 lines
55 KiB
C++
1065 lines
55 KiB
C++
// SPDX-License-Identifier: AGPL-3.0-only
|
|
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
|
|
|
|
module;
|
|
#include <signal.h>
|
|
|
|
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<double>();
|
|
if (value.is_boolean()) return value.get<bool>() ? 1.0 : 0.0;
|
|
return 0.0;
|
|
}
|
|
|
|
std::int64_t Integer(const Json& value) {
|
|
return static_cast<std::int64_t>(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<std::int64_t>(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<std::string> TotpCommand(const std::optional<std::string>& fixture) {
|
|
if (!fixture || fixture->empty()) return std::nullopt;
|
|
return std::format("{} totp {}", Platform::SelfExe().string(), *fixture);
|
|
}
|
|
|
|
std::vector<std::string> ExtraDirs(std::span<const std::string> trees) {
|
|
const std::string cwd = fs::current_path().string();
|
|
std::vector<std::string> 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<std::int32_t> 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<std::string> 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<std::string> AllowedTools(std::string_view, std::string_view, std::span<const std::string> trees, std::span<const std::string> extra,
|
|
const std::optional<std::string>& loginFixture, std::string_view environment) {
|
|
std::vector<std::string> 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<std::string> 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<std::string> totp = TotpCommand(loginFixture)) allow.push_back(std::format("Bash({}:*)", *totp));
|
|
std::vector<std::string> 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<Json> 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<Json>& 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<std::string>& prefer, const std::set<std::string>& held) const {
|
|
const std::lock_guard guard(lock);
|
|
std::vector<std::string> 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<std::string>& prefer, const std::set<std::string>& 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<std::string> Stacks::Free() const {
|
|
const std::lock_guard guard(lock);
|
|
return free;
|
|
}
|
|
|
|
// ---- Admission --------------------------------------------------------------------
|
|
Admission::Admission(CostOf costOf, NeedsStack needsStack, Stacks* stacks, std::optional<double> 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<double> {
|
|
const std::optional<std::int64_t> mb = Memory::Available();
|
|
return mb ? std::optional<double>(static_cast<double>(*mb)) : std::nullopt;
|
|
});
|
|
}
|
|
|
|
std::optional<std::string> 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<std::string>(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<double> 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<std::string> 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<std::string, std::string> Runner::ProcessEnv(const Json& job, const std::map<std::string, std::string>& base) const {
|
|
if (!EnvironmentOf(EnvOf(job)).db) return base;
|
|
const Json& s = Dyn::Get(job, "stack");
|
|
std::optional<std::string> url = options.dbUrl;
|
|
if (Dyn::Truthy(s)) {
|
|
const Json& own = Dyn::Get(s, "dbUrl");
|
|
url = Dyn::Truthy(own) ? std::optional<std::string>(Dyn::Str(own)) : std::nullopt;
|
|
}
|
|
std::map<std::string, std::string> out = base;
|
|
out[std::string(PgWire::BrokerEnv)] = options.broker && url && !url->empty() ? options.broker->Address(*url) : std::string();
|
|
return out;
|
|
}
|
|
|
|
std::optional<std::string> 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<std::string>(Dyn::Str(login));
|
|
}
|
|
|
|
std::optional<std::string> 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<std::string> Runner::Cmd(const Json& job, const fs::path& systemPath, std::string_view schema, std::span<const std::string> trees, std::int64_t maxTurns, bool persist,
|
|
const std::string& sessionId, bool resume) const {
|
|
std::vector<std::string> 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<std::string> 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<std::string> extra;
|
|
for (const Json& x : Dyn::Or(Dyn::Get(options.functional, "extraTools"), Dyn::EmptyList())) extra.push_back(Dyn::Str(x));
|
|
const std::vector<std::string> 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<Json>& 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<std::string, std::string>& Runner::Env() {
|
|
return env;
|
|
}
|
|
|
|
const fs::path& Runner::Scratch() const {
|
|
return options.scratch;
|
|
}
|
|
|
|
std::optional<std::int32_t> Runner::Wait(std::int64_t pid, std::int64_t timeoutMin, bool& timedOut) {
|
|
std::optional<std::int32_t> 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<std::int32_t>(Process::SupervisedReturnCode(*code)) : std::nullopt;
|
|
}
|
|
|
|
std::pair<Json, Json> Runner::Run(const Json& job, const std::string& prompt, const fs::path& systemPath, std::string_view schema, std::span<const std::string> 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<std::string> 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<std::string>())) || (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<std::string, std::string> 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<std::string> cmd = Cmd(job, systemPath, schema, trees, maxTurns, persist, sid, resume);
|
|
bool timedOut = false;
|
|
std::optional<std::int32_t> 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<double> 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<Json> structured = Stream::StructuredOutput(parsed.at("result"));
|
|
return {rec, structured ? *structured : Json()};
|
|
}
|
|
|
|
std::pair<Json, Json> Runner::RunOpenCode(const Json& job, const std::string& prompt, const fs::path& systemPath, std::span<const std::string> 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<std::string, std::string> 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<std::string> 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<std::int32_t> 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<std::string> Scheduler::Aborted() const {
|
|
const std::lock_guard guard(lock);
|
|
return abort;
|
|
}
|
|
|
|
const std::vector<Done>& 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<double>(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<Json*>& live, double now, const std::optional<std::string>& 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<std::string> k = o.cacheKey(job);
|
|
if (!k) return true;
|
|
const std::pair<std::string, std::string> 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<std::string> 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<std::string>& running, const std::vector<std::string>& queued) {
|
|
if (!o.progressPath) return;
|
|
try {
|
|
const std::optional<std::string> 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<Done>& Scheduler::Run(std::vector<Json> 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<Json, Json> result;
|
|
std::optional<std::string> error;
|
|
};
|
|
std::deque<Json> queue(std::make_move_iterator(jobs.begin()), std::make_move_iterator(jobs.end()));
|
|
std::list<Slot> running;
|
|
std::uint64_t finishes = 0;
|
|
const bool withSeats = o.seats && !o.seats->empty();
|
|
auto labelsOf = [](const auto& items, auto get) {
|
|
std::vector<std::string> 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<std::string> 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() && (!o.concurrency || static_cast<std::int64_t>(running.size()) < *o.concurrency) && !paused) {
|
|
std::vector<Json*> 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<Json*>& cands, const SeatPool::Others* others) -> std::optional<std::pair<Json*, Json*>> {
|
|
if (others && o.admission) o.admission->SetOthers(*others);
|
|
for (Json* seat : cands) {
|
|
const std::optional<std::string> sid = seat ? std::optional<std::string>(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<Json*> 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; a pool width, when
|
|
// given, is the machine's cap.
|
|
const SeatPool::Board::Placed placed = o.board->Place(*o.seats, o.now(), o.seatConfig, o.apiConfig,
|
|
[&](const std::vector<Json*>& 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.value_or(0), 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<double> resume = SeatPool::EarliestResume(*o.seats, o.now(), o.board ? std::optional<double>(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<Json, Json> result;
|
|
std::optional<std::string> 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<std::string>(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<std::list<Slot>::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<double>(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<std::string> k = o.cacheKey(j)) {
|
|
const std::pair<std::string, std::string> 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<std::int64_t>(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<double, 4> Backoff = {300, 900, 1800, 1800};
|
|
const double delay = Dyn::Truthy(reset) ? std::max(60.0, Number(reset) - Platform::Now() + 30) : Backoff[static_cast<std::size_t>(std::min<std::int64_t>(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<std::int64_t>(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<std::int64_t>(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<Json> 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;
|
|
}
|
|
}
|