diff --git a/implementations/Crafter.Review-Http.cpp b/implementations/Crafter.Review-Http.cpp new file mode 100644 index 0000000..73247ac --- /dev/null +++ b/implementations/Crafter.Review-Http.cpp @@ -0,0 +1,108 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +module; +#include + +module Crafter.Review:Http_impl; +import :Http; +import :Json; +import :Text; +import Crafter.Network; +import std; + +namespace Crafter::Review::Http { + namespace { + std::string Wrapped(std::string_view reason) { + return std::format("", reason); + } + + // ClientTCP's resolve, socket() and connect failures, which happen + // while the request goes out, apart from a send's or a receive's. + bool WhileSending(const SocketErrorException& failure) { + const std::string_view text = failure.what(); + return failure.kind == SocketErrorKind::Resolver || text.starts_with("Could not connect") || text.starts_with("Could not open socket"); + } + + std::string SocketText(const SocketErrorException& failure) { + if (failure.kind == SocketErrorKind::Resolver) return std::format("[Errno {}] {}", failure.code, gai_strerror(failure.code)); + return Dyn::ErrnoText(failure.code); + } + + constexpr std::array, 62> Phrases = {{ + {100, "Continue"}, {101, "Switching Protocols"}, {102, "Processing"}, {103, "Early Hints"}, {200, "OK"}, {201, "Created"}, {202, "Accepted"}, + {203, "Non-Authoritative Information"}, {204, "No Content"}, {205, "Reset Content"}, {206, "Partial Content"}, {207, "Multi-Status"}, + {208, "Already Reported"}, {226, "IM Used"}, {300, "Multiple Choices"}, {301, "Moved Permanently"}, {302, "Found"}, {303, "See Other"}, + {304, "Not Modified"}, {305, "Use Proxy"}, {307, "Temporary Redirect"}, {308, "Permanent Redirect"}, {400, "Bad Request"}, {401, "Unauthorized"}, + {402, "Payment Required"}, {403, "Forbidden"}, {404, "Not Found"}, {405, "Method Not Allowed"}, {406, "Not Acceptable"}, + {407, "Proxy Authentication Required"}, {408, "Request Timeout"}, {409, "Conflict"}, {410, "Gone"}, {411, "Length Required"}, + {412, "Precondition Failed"}, {413, "Content Too Large"}, {414, "URI Too Long"}, {415, "Unsupported Media Type"}, {416, "Range Not Satisfiable"}, + {417, "Expectation Failed"}, {418, "I'm a Teapot"}, {421, "Misdirected Request"}, {422, "Unprocessable Content"}, {423, "Locked"}, + {424, "Failed Dependency"}, {425, "Too Early"}, {426, "Upgrade Required"}, {428, "Precondition Required"}, {429, "Too Many Requests"}, + {431, "Request Header Fields Too Large"}, {451, "Unavailable For Legal Reasons"}, {500, "Internal Server Error"}, {501, "Not Implemented"}, + {502, "Bad Gateway"}, {503, "Service Unavailable"}, {504, "Gateway Timeout"}, {505, "HTTP Version Not Supported"}, {506, "Variant Also Negotiates"}, + {507, "Insufficient Storage"}, {508, "Loop Detected"}, {510, "Not Extended"}, {511, "Network Authentication Required"}, + }}; + } + + HttpError::HttpError(std::int32_t code, const std::string& message) : std::runtime_error(std::format("HTTP Error {}: {}", code, message)), code(code) {} + + HttpError::HttpError(std::int32_t code) : HttpError(code, std::string(Reason(code))) {} + + std::int32_t HttpError::Code() const { + return code; + } + + std::string_view Reason(std::int32_t code) { + for (const auto& [value, phrase] : Phrases) { + if (value == code) return phrase; + } + return ""; + } + + Response UrlOpen(std::string_view url, const std::map& headers, double timeout) { + FetchOptionsHTTP1 options; + options.headers = {{"accept-encoding", "identity"}, {"connection", "close"}, {"user-agent", "crafter-review"}}; + for (const auto& [name, value] : headers) options.headers[Dyn::Lower(name)] = value; + // The timeout bounds each blocking step: the connect, the handshake, + // every read. + const std::chrono::milliseconds budget(static_cast(timeout * 1000)); + options.connectTimeout = budget; + options.handshakeTimeout = budget; + options.timeout = budget; + FetchResultHTTP1 result; + try { + result = FetchHTTP1(url, options); + } catch (const SocketTimeoutException&) { + throw Error(Wrapped("timed out")); + } catch (const SocketErrorException& failure) { + throw Error(WhileSending(failure) ? Wrapped(SocketText(failure)) : SocketText(failure)); + } catch (const SocketClosedException&) { + throw Error("Remote end closed connection without response"); + } catch (const TLSException& failure) { + throw Error(Wrapped(failure.what())); + } catch (const std::invalid_argument& failure) { + throw Error(failure.what()); + } catch (const std::runtime_error& failure) { + const std::string_view text = failure.what(); + if (text.starts_with("timed out")) throw Error("timed out"); + if (text.starts_with("connection closed")) throw Error("Remote end closed connection without response"); + throw Error(failure.what()); + } catch (const std::exception& failure) { + throw Error(failure.what()); + } + const std::optional status = Dyn::ParseInt(result.response.status); + const auto code = static_cast(status.value_or(0)); + if (result.limited) { + throw HttpError(code, std::format("The HTTP server returned a redirect error that would lead to an infinite loop.\nThe last 30x error message was:\n{}", Reason(code))); + } + if (code < 200 || code >= 300) throw HttpError(code); + return {code, std::move(result.response.body)}; + } + + Json JsonBody(const Response& response) { + if (const std::optional error = Dyn::Utf8DecodeError(response.body)) throw Dyn::Error("UnicodeDecodeError", *error); + if (const std::optional error = Dyn::JsonDecodeError(response.body)) throw Dyn::Error("JSONDecodeError", *error); + return ParseLoose(response.body); + } +} diff --git a/implementations/Crafter.Review-Platform.cpp b/implementations/Crafter.Review-Platform.cpp index 4b051bd..7e681a0 100644 --- a/implementations/Crafter.Review-Platform.cpp +++ b/implementations/Crafter.Review-Platform.cpp @@ -169,7 +169,7 @@ namespace Crafter::Review::Platform { if (path.has_parent_path()) fs::create_directories(path.parent_path()); const fs::path temp = std::format("{}.{}.tmp", path.string(), PublishedPid()); const std::int32_t fd = open(temp.c_str(), O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, static_cast(mode)); - if (fd < 0) throw std::runtime_error(std::format("{}: {}", temp.string(), Dyn::ErrnoText(errno))); + if (fd < 0) throw std::runtime_error(std::format("{}: {}", Dyn::ErrnoText(errno), Dyn::Repr(temp.string()))); std::size_t written = 0; while (written < content.size()) { const ssize_t step = write(fd, content.data() + written, content.size() - written); @@ -178,7 +178,7 @@ namespace Crafter::Review::Platform { const std::int32_t failure = errno; close(fd); unlink(temp.c_str()); - throw std::runtime_error(std::format("{}: {}", temp.string(), Dyn::ErrnoText(failure))); + throw std::runtime_error(Dyn::ErrnoText(failure)); } written += static_cast(step); } @@ -186,7 +186,21 @@ namespace Crafter::Review::Platform { if (rename(temp.c_str(), path.c_str()) != 0) { const std::int32_t failure = errno; unlink(temp.c_str()); - throw std::runtime_error(std::format("{}: {}", path.string(), Dyn::ErrnoText(failure))); + throw std::runtime_error(std::format("{}: {} -> {}", Dyn::ErrnoText(failure), Dyn::Repr(temp.string()), Dyn::Repr(path.string()))); + } + } + + void MakeDirs(const fs::path& path, std::uint32_t mode) { + const fs::path name = path.filename().empty() && path.has_parent_path() ? path.parent_path() : path; + const fs::path head = name.parent_path(); + std::error_code error; + if (!head.empty() && head != name && !fs::exists(head, error)) { + MakeDirs(head, 0777); + if (name.filename() == ".") return; + } + if (mkdir(name.c_str(), static_cast(mode)) != 0) { + const std::int32_t failure = errno; + if (!fs::is_directory(name, error)) throw std::runtime_error(std::format("{}: {}", Dyn::ErrnoText(failure), Dyn::Repr(name.string()))); } } @@ -195,7 +209,7 @@ namespace Crafter::Review::Platform { if (lockPath.has_parent_path()) fs::create_directories(lockPath.parent_path(), error); fd = open(lockPath.c_str(), O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC, 0666); if (fd < 0) { - problem = Dyn::ErrnoText(errno); + problem = std::format("{}: {}", Dyn::ErrnoText(errno), Dyn::Repr(lockPath.string())); return; } while (flock(fd, LOCK_EX) != 0) { @@ -230,7 +244,7 @@ namespace Crafter::Review::Platform { } Json data; try { - if (const std::optional text = ReadFile(path)) data = Json::parse(*text); + if (const std::optional text = ReadFile(path)) data = ParseLoose(*text); } catch (const std::exception&) { data = Json(); } diff --git a/implementations/Crafter.Review-SeatPool.cpp b/implementations/Crafter.Review-SeatPool.cpp new file mode 100644 index 0000000..8ec8979 --- /dev/null +++ b/implementations/Crafter.Review-SeatPool.cpp @@ -0,0 +1,1498 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +module; +#include +#include +#include + +module Crafter.Review:SeatPool_impl; +import :SeatPool; +import :Json; +import :Text; +import :Crypto; +import :Platform; +import :Http; +import :Seats; +import std; + +namespace fs = std::filesystem; + +namespace Crafter::Review::SeatPool { + const std::string_view ResumePrompt = + "The previous attempt of this pass was interrupted by a limit on the account it ran on and continues here on " + "another account with the same transcript. Do not repeat work already done: continue from where it stopped, " + "finish the pass, and return the structured output exactly as the instructions require.\n"; + + namespace { + using Log = std::function; + + // A number as the files and the probes hold it, as a float. + double AsDouble(const Json& value) { + if (value.is_number_float()) return value.get(); + if (value.is_boolean()) return value.get() ? 1.0 : 0.0; + if (value.is_number_unsigned()) return static_cast(value.get()); + if (value.is_number_integer()) return static_cast(value.get()); + if (value.is_string()) { + if (const std::optional parsed = Dyn::ParseFloat(value.get_ref())) return *parsed; + throw Dyn::Error("ValueError", Dyn::FloatError(value.get_ref())); + } + throw Dyn::Error("TypeError", std::format("float() argument must be a string or a real number, not '{}'", Dyn::TypeName(value))); + } + + // The value as a float, 0 when it is falsy. + double Num(const Json& value) { + return Dyn::Truthy(value) ? AsDouble(value) : 0.0; + } + + // A float truncated toward zero, saturating at the ends of int64. + std::int64_t Int(double value) { + if (std::isinf(value)) throw Dyn::Error("OverflowError", "cannot convert float infinity to integer"); + if (std::isnan(value)) throw Dyn::Error("ValueError", "cannot convert float NaN to integer"); + if (value >= 9.2e18) return std::numeric_limits::max(); + if (value <= -9.2e18) return std::numeric_limits::min(); + return static_cast(value); + } + + // `now or time.time()` + double NowOr(std::optional now) { + return now && *now != 0 ? *now : Platform::Now(); + } + + // f'{value:g}' + std::string G(double value) { + return std::format("{:g}", value); + } + + std::string Clock(double epoch) { + return Platform::LocalTime("%H:%M:%S", epoch); + } + + Json ZeroForeign() { + return {{"count", 0}, {"cost5h", 0.0}, {"costUsd", 0.0}}; + } + + std::string Kind(const Json& seat) { + return Dyn::Str(Dyn::At(seat, "kind")); + } + + std::string Id(const Json& seat) { + return Dyn::Str(Dyn::At(seat, "id")); + } + + std::int64_t Running(const Json& seat) { + return Int(AsDouble(Dyn::At(seat, "running"))); + } + + const std::string& EnvOf(const std::map& env, const std::string& key) { + static const std::string none; + const auto found = env.find(key); + return found == env.end() ? none : found->second; + } + + bool IsOpenRouter(const Json& seat) { + return Kind(seat) == "openrouter"; + } + + double Budget(const Json& s, const Json& a) { + return AsDouble(Dyn::At(a, IsOpenRouter(s) ? "orBudgetUsd" : "budgetUsd")); + } + + const Json& LensCostValue(const Json& s, const Json& a) { + return Dyn::At(a, IsOpenRouter(s) ? "orLensCostUsd" : "lensCostUsd"); + } + + double LensCost(const Json& s, const Json& a) { + return AsDouble(LensCostValue(s, a)); + } + + std::int64_t Concurrency(const Json& s, const Json& a) { + return Int(AsDouble(Dyn::At(a, IsOpenRouter(s) ? "orConcurrency" : "concurrency"))); + } + + // The windows a seat's utilisation is read from: all but the Fable + // one, or all when that leaves none. + void Reutil(Json& s) { + Json own = Json::object(); + const Json& windows = s.at("windows"); + for (auto it = windows.begin(); it != windows.end(); ++it) { + if (!Seats::IsFableWindow(it.key())) own[it.key()] = it.value(); + } + const auto [util, binding] = Seats::SeatUtil(own.empty() ? windows : own); + s["util"] = util; + s["binding"] = binding; + } + + bool LongOk(const Json& s) { + const Json& windows = Dyn::At(s, "windows"); + for (auto it = windows.begin(); it != windows.end(); ++it) { + if (it.key() == "five_hour" || Seats::IsFableWindow(it.key())) continue; + if (Num(Dyn::Get(it.value(), "utilization")) >= Seats::LongWindowCutoff) return false; + } + return true; + } + + bool ApiRoom(const Json& s, const Json& a) { + return AsDouble(s.at("spentUsd")) + static_cast(Running(s) + 1) * LensCost(s, a) + AsDouble(Foreign(s).at("costUsd")) <= Budget(s, a); + } + + const Json& FiveHour(const Json& s) { + return Dyn::Or(Dyn::Get(Dyn::At(s, "windows"), "five_hour"), Dyn::EmptyDict()); + } + + // ik.split('/', 1)[0]: the driver an in-flight entry belongs to. + std::string Owner(std::string_view inflightKey) { + return std::string(inflightKey.substr(0, inflightKey.find('/'))); + } + + // key.split(':')[0] + std::string PidOf(std::string_view key) { + return std::string(key.substr(0, key.find(':'))); + } + + // Board._key + std::string KeyOf(const Json& seat) { + const Json& key = Dyn::Get(seat, "boardKey"); + return Dyn::Truthy(key) ? Dyn::Str(key) : SeatKey(Kind(seat), Dyn::StrOr(Dyn::Get(seat, "secret"), "")); + } + + // Per-token list prices when the catalogue cannot be fetched + // (OpenRouter, 2026-10-12). + const Json& BuiltInPrices() { + static const Json prices = {{"xiaomi/mimo-v2.6-pro", {{"prompt", 0.435e-6}, {"completion", 0.87e-6}, {"cacheRead", 0.0036e-6}, {"cacheWrite", nullptr}}}}; + return prices; + } + + // model -> prices, filled by FetchOpenRouterPrices. + std::map Prices; + std::mutex PricesLock; + + void WriteAll(std::int32_t fd, std::string_view text) { + std::size_t written = 0; + while (written < text.size()) { + const ssize_t step = write(fd, text.data() + written, text.size() - written); + if (step < 0) { + if (errno == EINTR) continue; + const std::int32_t failure = errno; + close(fd); + throw std::runtime_error(Dyn::ErrnoText(failure)); + } + written += static_cast(step); + } + close(fd); + } + + // shutil.copyfile: the bytes only, the destination created as open() + // creates it. + void CopyContents(const fs::path& from, const fs::path& to) { + const std::optional bytes = Platform::ReadFile(from); + if (!bytes) throw std::runtime_error(std::format("{}: {}", Dyn::ErrnoText(ENOENT), Dyn::Repr(from.string()))); + const std::int32_t fd = open(to.c_str(), O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0666); + if (fd < 0) throw std::runtime_error(std::format("{}: {}", Dyn::ErrnoText(errno), Dyn::Repr(to.string()))); + WriteAll(fd, *bytes); + } + } + + bool IsPaid(const Json& seat) { + return std::ranges::contains(Paid, Kind(seat)); + } + + Json ApiConfig(const std::map& env) { + auto num = [&](const std::string& key, double fallback) { + const std::optional v = Dyn::ParseFloat(EnvOf(env, key)); + return v && *v >= 0 ? *v : fallback; + }; + return { + {"budgetUsd", num("REVIEW_GATES_API_BUDGET_USD", 200.0)}, + {"lensCostUsd", num("REVIEW_GATES_API_LENS_COST_USD", 1.0)}, + {"concurrency", Int(num("REVIEW_GATES_API_CONCURRENCY", 8))}, + {"cycleDay", std::max(1, std::min(28, Int(num("REVIEW_GATES_API_CYCLE_DAY", 1))))}, + {"order", EnvOf(env, "REVIEW_GATES_SEAT_ORDER") == "oauth-first" ? "oauth-first" : "api-first"}, + // OpenRouter seats: a prepaid key, cheap models - a smaller reserve + // per running process. + {"orBudgetUsd", num("REVIEW_GATES_OPENROUTER_BUDGET_USD", 100.0)}, + {"orLensCostUsd", num("REVIEW_GATES_OPENROUTER_LENS_COST_USD", 0.1)}, + {"orConcurrency", Int(num("REVIEW_GATES_OPENROUTER_CONCURRENCY", 16))}, + }; + } + + Json Foreign(const Json& seat) { + const Json& foreign = Dyn::Get(seat, "foreign"); + return Dyn::Truthy(foreign) ? foreign : ZeroForeign(); + } + + fs::path LedgerPath(const std::optional& scratchRoot) { + fs::path root; + if (scratchRoot && !scratchRoot->empty()) { + root = *scratchRoot; + } else if (const std::optional configured = Platform::GetEnv("REVIEW_GATES_SCRATCH"); configured && !configured->empty()) { + root = *configured; + } else { + root = fs::path(Platform::Home()) / ".cache" / "review-gates"; + } + return root / "seat-ledger.json"; + } + + std::string CycleStart(double now, std::int32_t day) { + using namespace std::chrono; + const std::string today = Platform::LocalTime("%Y-%m-%d", now); + const year_month_day date(year(std::stoi(today.substr(0, 4))), month(static_cast(std::stoi(today.substr(5, 2)))), + std::chrono::day(static_cast(std::stoi(today.substr(8, 2))))); + auto iso = [](const year_month_day& d) { + return std::format("{:04d}-{:02d}-{:02d}", static_cast(d.year()), static_cast(d.month()), static_cast(d.day())); + }; + if (static_cast(static_cast(date.day())) >= day) { + return iso(year_month_day(date.year(), date.month(), std::chrono::day(static_cast(day)))); + } + const year_month_day previous(sys_days(year_month_day(date.year(), date.month(), std::chrono::day(1))) - days(1)); + const auto last = static_cast(static_cast(previous.day())); + return iso(year_month_day(previous.year(), previous.month(), std::chrono::day(static_cast(std::min(day, last))))); + } + + // ---- the ledger ------------------------------------------------------------------- + Ledger::Ledger(fs::path path, Json apiConfig, Json data) : path(std::move(path)), apiConfig(std::move(apiConfig)), data(Dyn::Truthy(data) ? std::move(data) : Json::object()) {} + + Ledger Ledger::Load(const fs::path& path, const Json& apiConfig) { + Json data; + try { + data = Dyn::LoadJsonFile(path); + } catch (const std::exception&) { + data = Json::object(); + } + return Ledger(path, apiConfig, data.is_object() ? data : Json::object()); + } + + Json& Ledger::Entry(const std::string& seatId, std::optional now) { + const std::string cycle = CycleStart(NowOr(now), static_cast(Int(AsDouble(Dyn::At(apiConfig, "cycleDay"))))); + const Json& e = Dyn::Get(data, seatId); + // Only a newer cycle starts over: an entry of a later one (another + // driver's cycle day) is not wiped. + if (!Dyn::Truthy(e) || cycle > Dyn::StrOr(Dyn::Get(e, "cycle"), "")) { + data[seatId] = {{"cycle", cycle}, {"spentUsd", 0.0}, {"procs", 0}, {"outSince", nullptr}, {"outReason", nullptr}}; + } + return data[seatId]; + } + + Json Ledger::Update(const std::string& seatId, const std::function& change, std::optional now) { + // The file is the record: re-read under its lock, this process only + // ever adds to it (2026-10-15: each driver saved the copy it had loaded + // and the last writer won - api4 reached the credit error with $157.54 + // recorded of the $199.66 its processes had cost). + Json entry; + Platform::LockedUpdate(path, [&](Json& file) { + // The file wins where both have a seat. + Json merged = data; + for (auto it = file.begin(); it != file.end(); ++it) merged[it.key()] = it.value(); + data = std::move(merged); + Json& e = Entry(seatId, now); + change(e); + entry = e; + file = data; + return true; + }, data, 0600, {}); + return entry; + } + + Json Ledger::Add(const std::string& seatId, double usd, std::optional now) { + return Update(seatId, [&](Json& e) { + e["spentUsd"] = Dyn::Round(AsDouble(Dyn::At(e, "spentUsd")) + usd, 4); + e["procs"] = Dyn::Add(Dyn::At(e, "procs"), Json(1)); + }, now); + } + + Json Ledger::MarkOut(const std::string& seatId, std::string_view reason, std::optional now) { + return Update(seatId, [&](Json& e) { + e["outSince"] = Platform::LocalTime("%Y-%m-%dT%H:%M:%S", NowOr(now)); + e["outReason"] = Dyn::Slice(reason, 0, 160); + }, now); + } + + void Ledger::Save() { + try { + Platform::WriteAtomic(path, DumpFile(data), 0600); + } catch (const std::exception&) { + // A record, never a dependency. + } + } + + const Json& Ledger::Data() const { + return data; + } + + // ---- the seats --------------------------------------------------------------------- + Json NewSeat(const std::string& id, const std::string& kind, const std::string& secret) { + return { + {"id", id}, + {"kind", kind}, + {"secret", secret}, + {"state", "unknown"}, + {"windows", Json::object()}, + {"util", nullptr}, + {"binding", ""}, + {"resetsAt", nullptr}, + {"pauseUntil", 0.0}, + {"running", 0}, + {"procs", 0}, + {"spentUsd", 0.0}, + {"error", nullptr}, + {"probedWith", nullptr}, + {"foreign", ZeroForeign()}, + // The board's view of the seat. + {"boardKey", SeatKey(kind, secret)}, + {"stateFrom", nullptr}, + }; + } + + std::vector LoadSeats(const std::map& env, const std::optional& envFile, std::string_view specText, Ledger* ledger, const Json& apiConfig) { + std::map merged = env; + if (envFile && !envFile->empty()) { + for (const auto& [key, value] : Seats::SeatTokensFromEnvFile(Dyn::ReadText(*envFile))) merged[key] = value; + } + std::vector> pool; + for (const auto& [n, key] : Seats::ApiKeysFromEnv(merged)) pool.emplace_back(std::format("api{}", n), NewSeat(std::format("api{}", n), "api", key)); + for (const auto& [n, token] : Seats::SeatTokensFromEnv(merged)) pool.emplace_back(std::to_string(n), NewSeat(std::to_string(n), "oauth", token)); + for (const auto& [n, key] : Seats::OpenRouterKeysFromEnv(merged)) pool.emplace_back(std::format("or{}", n), NewSeat(std::format("or{}", n), "openrouter", key)); + pool.emplace_back("local", NewSeat("local", "local", "")); + auto find = [&](std::string_view id) -> const Json* { + for (const auto& [key, seat] : pool) { + if (key == id) return &seat; + } + return nullptr; + }; + const std::string spec = Dyn::Strip(specText.empty() ? std::string_view("auto") : specText); + std::vector ids; + if (spec == "auto") { + // Never an OpenRouter seat by default. + for (const auto& [key, seat] : pool) { + if (Kind(seat) == "api" || Kind(seat) == "oauth") ids.push_back(key); + } + if (ids.empty()) ids.emplace_back("local"); + } else if (spec == "local" || spec == "api" || spec == "oauth" || spec == "openrouter") { + for (const auto& [key, seat] : pool) { + if (Kind(seat) == spec) ids.push_back(key); + } + } else { + std::vector bad; + for (const std::string& part : Dyn::Split(spec, ",")) { + const std::string id = Dyn::Strip(part); + if (id.empty()) continue; + ids.push_back(id); + if (!find(id)) bad.push_back(id); + } + if (!bad.empty()) { + std::vector configured; + for (const auto& [key, seat] : pool) { + if (key != "local") configured.push_back(key); + } + throw std::invalid_argument(std::format("--seats names seat(s) not configured: {} (configured: {}, local)", Dyn::Join(bad, ", "), Dyn::Join(configured, ", "))); + } + } + std::vector seats; + for (const std::string& id : ids) seats.push_back(*find(id)); + if (std::ranges::any_of(seats, IsOpenRouter) && !std::ranges::all_of(seats, IsOpenRouter)) { + throw std::invalid_argument("--seats: an OpenRouter seat cannot share a pool with Anthropic seats or the local login - a model name means a different model per endpoint"); + } + if (ledger) { + const double now = Platform::Now(); + for (Json& s : seats) { + if (!IsPaid(s)) continue; + const Json& e = ledger->Entry(Id(s), now); + s["spentUsd"] = AsDouble(Dyn::At(e, "spentUsd")); + if (Dyn::Truthy(Dyn::Get(e, "outSince"))) { + s["state"] = "out"; + s["error"] = std::format("ledger: {} since {}", Dyn::StrOr(Dyn::Get(e, "outReason"), "out"), Dyn::Slice(Dyn::Str(e.at("outSince")), 0, 16)); + } else if (Dyn::Truthy(apiConfig) && AsDouble(s.at("spentUsd")) >= Budget(s, apiConfig)) { + s["state"] = "out"; + s["error"] = std::format("ledger: ${:.2f} spent this cycle, budget ${}", AsDouble(s.at("spentUsd")), G(Budget(s, apiConfig))); + } + } + } + return seats; + } + + void ApplyProbe(Json& s, const Json& r) { + s["state"] = Dyn::Has(r, "state") ? r.at("state") : Json("unknown"); + s["windows"] = Dyn::Or(Dyn::Get(r, "windows"), Dyn::EmptyDict()); + s["util"] = Dyn::Get(r, "util"); + s["binding"] = Dyn::Truthy(Dyn::Get(r, "binding")) ? r.at("binding") : Json(""); + s["resetsAt"] = Dyn::Get(r, "resetsAt"); + s["error"] = Dyn::Get(r, "error"); + s["probedWith"] = Dyn::Get(r, "probedWith"); + if (Kind(s) == "oauth" && Dyn::Truthy(s.at("windows"))) Reutil(s); + if (s.at("state") == "limited" && Dyn::Truthy(s.at("resetsAt"))) s["pauseUntil"] = AsDouble(s.at("resetsAt")) + 30; + } + + std::string Describe(const Json& s) { + if (IsPaid(s)) { + std::string state = Dyn::Str(s.at("state")); + if (Dyn::Truthy(Dyn::Get(s, "error")) && s.at("state") != "ok") state += std::format(" ({})", Dyn::Str(s.at("error"))); + const Json& remaining = Dyn::Get(s, "remainingUsd"); + return std::format("{}, ${:.2f} spent this cycle{}", state, AsDouble(s.at("spentUsd")), remaining.is_null() ? std::string() : std::format(", ${:.2f} left on the key", AsDouble(remaining))); + } + if (!Dyn::Truthy(s.at("windows"))) { + return Dyn::Str(s.at("state")) + (Dyn::Truthy(Dyn::Get(s, "error")) ? std::format(" ({})", Dyn::Str(s.at("error"))) : std::string()); + } + std::vector parts; + const Json& windows = s.at("windows"); + for (auto it = windows.begin(); it != windows.end(); ++it) { + const Json& u = Dyn::Get(it.value(), "utilization"); + if (u.is_null()) continue; + std::string label = it.key(); + if (it.key() == "five_hour") { + label = "5h"; + } else if (it.key() == "seven_day") { + label = "7d"; + } else if (it.key() == Seats::FableWindow || Seats::IsFableWindow(it.key())) { + label = "fable"; + } + parts.push_back(std::format("{} {}%", label, Seats::Fmt(u))); + } + const Json& binding = s.at("binding"); + return std::format("{}: {}{}", Dyn::Str(s.at("state")), Dyn::Join(parts, ", "), Dyn::Truthy(binding) ? std::format(" (binding {})", Dyn::Str(binding)) : std::string()); + } + + void Probe(std::vector& seats, const std::string& claudeBin, double timeout, const std::vector& models, const Log& log, const ProbeFunction& probe) { + const ProbeFunction run = probe ? probe : ProbeFunction([](std::int32_t seat, const std::string& token, const std::string& model, double t, const std::string& bin, bool local, bool api) { + return Seats::ProbeSeat(seat, token, {.model = model, .timeoutS = t, .claudeBin = bin, .local = local, .api = api}); + }); + // OAuth seats probe on the lens model when that is Fable (its window + // is seen only there), else on Haiku; API seats and the local login + // on Haiku. + std::optional fable; + for (const std::string& m : models) { + if (Dyn::Lower(m).contains("fable")) { + fable = m; + break; + } + } + if (std::ranges::any_of(seats, IsOpenRouter)) { + const Json prices = FetchOpenRouterPrices(models, log); + for (auto it = prices.begin(); it != prices.end(); ++it) { + const Json& p = it.value(); + if (!Dyn::Truthy(p)) { + log(std::format("openrouter: NO PRICE for {} - its spend is not counted", it.key())); + continue; + } + log(std::format("openrouter: {} priced ${:.3f}/${:.3f} per M in/out, cache read {}", it.key(), AsDouble(p.at("prompt")) * 1e6, AsDouble(p.at("completion")) * 1e6, + Dyn::Get(p, "cacheRead").is_null() ? std::string("n/a") : std::format("${:.4f}", AsDouble(p.at("cacheRead")) * 1e6))); + } + } + const std::string lensModel = fable.value_or(std::string(Seats::ProbeModel)); + for (Json& s : seats) { + if (s.at("state") == "out") continue; + const std::string kind = Kind(s); + const std::string id = Id(s); + const std::string secret = Dyn::StrOr(Dyn::Get(s, "secret"), ""); + if (kind == "openrouter") { + const Json r = ProbeOpenRouter(secret, timeout); + ApplyProbe(s, r); + s["remainingUsd"] = Dyn::Get(r, "remaining"); + log(std::format("seat {} ({}): {}", id, kind, Describe(s))); + continue; + } + const std::int32_t n = kind == "api" ? static_cast(Dyn::ParseInt(Dyn::Slice(id, 3)).value_or(0)) : (kind == "local" ? 0 : static_cast(Dyn::ParseInt(id).value_or(0))); + Json r; + if (kind == "api") { + r = run(n, secret, std::string(Seats::ProbeModel), timeout, claudeBin, false, true); + } else if (kind == "local") { + r = run(n, "", lensModel, timeout, claudeBin, true, false); + } else { + r = run(n, secret, lensModel, timeout, claudeBin, false, false); + if (fable && Seats::NeedsHaiku(r)) r = Seats::MergeFableProbe(run(n, secret, std::string(Seats::ProbeModel), timeout, claudeBin, false, false), r); + } + ApplyProbe(s, r); + log(std::format("seat {} ({}): {}", id, kind, Describe(s))); + } + } + + // ---- placement ------------------------------------------------------------------------- + Json PoolConfig(const std::map& env) { + Json config = Seats::ConfigFromEnv(env); + if (Dyn::Strip(EnvOf(env, "REVIEW_GATES_LENS_COST_PCT")).empty()) config["lensCostPct"] = 0.25; + return config; + } + + double OauthProjected(const Json& s, const Json& config) { + // Where the seat's 5-hour window will be with its running processes - + // this driver's at its projection, the other drivers' at theirs (the + // board). Unknown = at the ceiling, ranked last. + const Json& five = Dyn::Get(FiveHour(s), "utilization"); + const double base = five.is_null() ? AsDouble(config.at("maxUtil")) : AsDouble(five); + return base + static_cast(Running(s)) * AsDouble(config.at("lensCostPct")) + AsDouble(Foreign(s).at("cost5h")); + } + + std::vector ExpireWindows(Json& s, double now) { + // 2026-10-13: S153 sat in a 30-second pause loop for 30 minutes after + // the 20:20 resets - nothing was running, so no rate event ever + // replaced the 83-89 % readings. + std::vector done; + if (!Dyn::Truthy(Dyn::Get(s, "windows"))) return done; + for (auto it = s["windows"].begin(); it != s["windows"].end(); ++it) { + Json& w = it.value(); + if (Dyn::Truthy(Dyn::Get(w, "resetsAt")) && AsDouble(w.at("resetsAt")) + 30 <= now && Num(Dyn::Get(w, "utilization")) > 0) { + w["utilization"] = 0; + w["resetsAt"] = 0; + done.push_back(it.key()); + } + } + return done; + } + + bool Eligible(Json& s, double now, const Json& config, const Json& apiConfig) { + if (s.at("state") == "out" || s.at("state") == "invalid" || now < AsDouble(s.at("pauseUntil"))) return false; + if (IsPaid(s)) { + // The cap counts every driver's processes. + return Running(s) + Int(AsDouble(Foreign(s).at("count"))) < Concurrency(s, apiConfig) && ApiRoom(s, apiConfig); + } + ExpireWindows(s, now); + // Its pause is over (checked above): back in placement, its windows + // decide. + if (s.at("state") == "limited") s["state"] = "ok"; + // No 5-hour reading (no probe, or it failed): usable, ranked at the + // ceiling - after every known seat. + if (Dyn::Get(FiveHour(s), "utilization").is_null()) return true; + return OauthProjected(s, config) + AsDouble(config.at("lensCostPct")) <= AsDouble(config.at("maxUtil")) && LongOk(s); + } + + std::vector Ranked(std::vector& seats, double now, const Json& config, const Json& apiConfig) { + std::vector api; + std::vector oauth; + for (Json& s : seats) { + if (!Eligible(s, now, config, apiConfig)) continue; + (IsPaid(s) ? api : oauth).push_back(&s); + } + auto spend = [&](const Json* s) { + return std::pair(AsDouble(s->at("spentUsd")) + static_cast(Running(*s)) * LensCost(*s, apiConfig) + AsDouble(Foreign(*s).at("costUsd")), Id(*s)); + }; + std::ranges::stable_sort(api, [&](const Json* a, const Json* b) { return spend(a) < spend(b); }); + if (config.at("policy") == "pack") { + std::ranges::stable_sort(oauth, [](const Json* a, const Json* b) { return std::pair(Kind(*a) != "oauth", Id(*a)) < std::pair(Kind(*b) != "oauth", Id(*b)); }); + } else { + auto projected = [&](const Json* s) { return std::pair(Seats::Quantize(OauthProjected(*s, config)), Id(*s)); }; + std::ranges::stable_sort(oauth, [&](const Json* a, const Json* b) { return projected(a) < projected(b); }); + } + const bool apiFirst = apiConfig.at("order") == "api-first"; + std::vector out = apiFirst ? api : oauth; + const std::vector& rest = apiFirst ? oauth : api; + out.insert(out.end(), rest.begin(), rest.end()); + return out; + } + + Json* Pick(std::vector& seats, double now, const Json& config, const Json& apiConfig) { + const std::vector ranked = Ranked(seats, now, config, apiConfig); + return ranked.empty() ? nullptr : ranked.front(); + } + + // ---- resuming a pass on another seat ------------------------------------------------ + fs::path SeatDir(const fs::path& scratch, std::string_view seatId) { + return scratch / std::format("seat-{}", seatId); + } + + std::optional CarryTranscript(const fs::path& scratch, std::string_view fromSeat, std::string_view toSeat, std::string_view sessionId, const Log& log) { + // The CLI keeps a session's transcript under + // /projects//.jsonl, and + // `claude --resume ` continues it; every process runs in the + // repository, so the slug is the same on the other seat. + if (fromSeat.empty() || toSeat.empty() || sessionId.empty() || fromSeat == toSeat) return std::nullopt; + // glob(/projects/*/.jsonl), whose * skips dot names. + std::optional source; + std::error_code error; + for (fs::directory_iterator it(SeatDir(scratch, fromSeat) / "projects", error), end; !error && it != end; it.increment(error)) { + if (it->path().filename().string().starts_with('.')) continue; + const fs::path candidate = it->path() / std::format("{}.jsonl", sessionId); + std::error_code missing; + if (fs::exists(fs::symlink_status(candidate, missing))) { + source = candidate; + break; + } + } + if (!source) return std::nullopt; + const fs::path destination = SeatDir(scratch, toSeat) / "projects" / source->parent_path().filename() / source->filename(); + Platform::MakeDirs(destination.parent_path(), 0700); + CopyContents(*source, destination); + if (log) { + log(std::format("seat {}: transcript of {} carried over from seat {} ({} bytes) - the pass resumes", toSeat, Dyn::Slice(sessionId, 0, 8), fromSeat, fs::file_size(destination))); + } + return destination; + } + + bool AllOut(const std::vector& seats) { + return std::ranges::all_of(seats, [](const Json& s) { return s.at("state") == "out" || s.at("state") == "invalid"; }); + } + + std::optional EarliestResume(const std::vector& seats, double now, std::optional recheck) { + std::vector times; + for (const Json& s : seats) { + if (s.at("state") == "out" || s.at("state") == "invalid") continue; + const Json& five = FiveHour(s); + if (AsDouble(s.at("pauseUntil")) > now) { + times.push_back(AsDouble(s.at("pauseUntil"))); + } else if (Kind(s) != "api" && Dyn::Truthy(Dyn::Get(five, "resetsAt")) && AsDouble(five.at("resetsAt")) + 30 > now) { + times.push_back(AsDouble(five.at("resetsAt")) + 30); + } + if (recheck && AsDouble(Foreign(s).at("count")) > 0) times.push_back(now + *recheck); + } + if (times.empty()) return std::nullopt; + return std::ranges::min(times); + } + + double RetryAfter(std::string_view text, double fallback) { + // re.search(r'retry[- ]after[^0-9]{0,12}(\d+)', text, re.I): the class + // run is greedy, so the digits are the last \d that it leaves. + const std::u32string t = Dyn::Decode(text); + auto letters = [&](std::size_t at, std::string_view word) { + for (std::size_t i = 0; i < word.size(); ++i) { + if (Dyn::IgnoreCaseLetter(t[at + i]) != word[i]) return false; + } + return true; + }; + for (std::size_t at = 0; at + 11 <= t.size(); ++at) { + if (!letters(at, "retry") || (t[at + 5] != U'-' && t[at + 5] != U' ') || !letters(at + 6, "after")) continue; + const std::size_t gap = at + 11; + std::size_t run = 0; + while (run < 12 && gap + run < t.size() && !(t[gap + run] >= U'0' && t[gap + run] <= U'9')) ++run; + for (std::size_t k = run + 1; k-- > 0;) { + std::size_t end = gap + k; + std::int64_t seconds = 0; + while (end < t.size() && Dyn::IsDecimal(t[end])) { + seconds = std::min(1'000'000, seconds * 10 + Dyn::DecimalValue(t[end])); + ++end; + } + if (end > gap + k) return static_cast(std::min(900, std::max(30, seconds + 5))); + } + } + return fallback; + } + + bool Settle(Json& s, Json& record, double now, const Json& config, const Json& apiConfig, Ledger* ledger, const Log& log) { + s["procs"] = Dyn::Add(s.at("procs"), Json(1)); + const Json& rate = Dyn::Or(Dyn::Get(record, "rateLimit"), Dyn::EmptyDict()); + if (Kind(s) != "api" && Dyn::Truthy(rate)) { + Json windows = Json::object(); + for (auto it = rate.begin(); it != rate.end(); ++it) { + const Json& u = Dyn::Get(it.value(), "utilization"); + const Json& r = Dyn::Get(it.value(), "resetsAt"); + windows[it.key()] = {{"utilization", Dyn::Truthy(u) ? u : Json(0)}, {"resetsAt", Dyn::Truthy(r) ? r : Json(0)}}; + } + s["windows"] = std::move(windows); + Reutil(s); + const Json& five = FiveHour(s); + if (Num(Dyn::Get(five, "utilization")) >= AsDouble(config.at("maxUtil")) && Dyn::Truthy(Dyn::Get(five, "resetsAt"))) { + s["pauseUntil"] = std::max(AsDouble(s.at("pauseUntil")), AsDouble(five.at("resetsAt")) + 30); + if (s.at("state") == "ok") log(std::format("seat {}: five_hour window at {}% - no new launches until its reset", Id(s), Seats::Fmt(five.at("utilization")))); + } + } + if (IsPaid(s)) { + if (IsOpenRouter(s)) { + // The harness's own figure is wrong for a model it does not + // know (the CLI) or is the catalogue's guess (OpenCode): the + // seat's spend is tokens x OpenRouter's list prices, and that + // is what the dump sums. + record["costUsdHarness"] = Dyn::Get(record, "costUsd"); + record["costUsd"] = OpenRouterCost(Dyn::StrOr(Dyn::Get(record, "model"), ""), Dyn::Or(Dyn::Get(record, "usage"), Dyn::EmptyDict())); + } + const double usd = Num(Dyn::Get(record, "costUsd")); + s["spentUsd"] = AsDouble(s.at("spentUsd")) + usd; + if (ledger) ledger->Add(Id(s), usd, now); + const double budget = Budget(s, apiConfig); + const double spent = AsDouble(s.at("spentUsd")); + if (spent >= budget && s.at("state") != "out" && s.at("state") != "invalid") { + s["state"] = "out"; + s["error"] = std::format("budget: ${:.2f} spent this cycle, budget ${}", spent, G(budget)); + log(std::format("seat {}: {} - out for the rest of the cycle", Id(s), Dyn::Str(s.at("error")))); + } else if (spent >= 0.8 * budget && spent - usd < 0.8 * budget) { + log(std::format("seat {}: ${:.2f} of ${} spent this cycle", Id(s), spent, G(budget))); + } + } + const Json& status = Dyn::Get(record, "status"); + const std::string reason = Dyn::StrOr(Dyn::Get(record, "reason"), ""); + if (status == "credits") { + s["state"] = "out"; + s["error"] = std::format("credit balance exhausted ({})", Dyn::Slice(reason, 0, 80)); + if (ledger && IsPaid(s)) ledger->MarkOut(Id(s), "credit balance too low", now); + log(std::format("seat {}: {} - out, the process is placed again", Id(s), Dyn::Str(s.at("error")))); + return true; + } + if (status == "auth") { + s["state"] = "invalid"; + s["error"] = std::format("authentication failure ({})", Dyn::Slice(reason, 0, 80)); + log(std::format("seat {}: {} - out, the process is placed again", Id(s), Dyn::Str(s.at("error")))); + return true; + } + if (status == "limit") { + if (IsPaid(s)) { + s["pauseUntil"] = std::max(AsDouble(s.at("pauseUntil")), now + RetryAfter(reason, ApiBackoffS)); + log(std::format("seat {}: rate limited - paused {} s, the process is placed again", Id(s), Int(AsDouble(s.at("pauseUntil")) - now))); + return true; + } + const std::string rateType = Dyn::StrOr(Dyn::Get(record, "rateType"), ""); + const Json& binding = s.at("binding"); + if (rateType.starts_with("seven_day") || (Dyn::StrOr(binding, "").starts_with("seven_day") && Num(Dyn::Get(s, "util")) >= 100)) { + s["state"] = "out"; + s["error"] = std::format("seven-day window rejected ({})", rateType.empty() ? Dyn::Str(binding) : rateType); + log(std::format("seat {}: {} - out for the run, the process is placed again", Id(s), Dyn::Str(s.at("error")))); + return true; + } + const Json& five = FiveHour(s); + s["pauseUntil"] = std::max(AsDouble(s.at("pauseUntil")), Dyn::Truthy(Dyn::Get(five, "resetsAt")) ? AsDouble(five.at("resetsAt")) + 30 : now + OauthFallbackPauseS); + s["state"] = "limited"; + log(std::format("seat {}: limit ({}) - paused until {}, the process is placed again", Id(s), Dyn::Slice(reason, 0, 60), Clock(AsDouble(s.at("pauseUntil"))))); + return true; + } + if (s.at("state") == "limited" && now >= AsDouble(s.at("pauseUntil"))) s["state"] = "ok"; + return false; + } + + // ---- the child's credentials ----------------------------------------------------------- + fs::path WriteCredentials(const fs::path& configDir, std::string_view token) { + // A setup-token is long-lived and never refreshed by the CLI: the + // expiry only has to be far away. + Platform::MakeDirs(configDir, 0700); + const Json body = {{"claudeAiOauth", {{"accessToken", std::string(token)}, {"expiresAt", Int((Platform::Now() + 365 * 86400) * 1000)}, {"scopes", Json::array({"user:inference"})}}}}; + const fs::path path = configDir / ".credentials.json"; + // One temp name per writer: the pool's threads launch a fresh seat's + // first passes in the same instant, each sees no credentials file yet + // and each writes one - with a pid-only name one rename found the temp + // file gone (S115, S116, S119 on 2026-10-10). The contents are + // identical, so the last rename to win is fine. + const fs::path temp = std::format("{}.{}.{}.tmp", path.string(), Platform::PublishedPid(), Uuid4().substr(0, 8)); + const std::int32_t fd = open(temp.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC, 0600); + if (fd < 0) throw std::runtime_error(std::format("{}: {}", Dyn::ErrnoText(errno), Dyn::Repr(temp.string()))); + WriteAll(fd, std::format("{}\n", DumpSpaced(body))); + fs::rename(temp, path); + return path; + } + + std::map ChildEnv(const Json& seat, const std::map& base, const fs::path& scratch) { + // `local`: the driver's environment as it is (the machine's own + // login). OAuth: the seat's config dir with its credentials file, no + // token variable. API: the key in ANTHROPIC_API_KEY plus a config dir + // of its own that approves that key, so no login on the machine is + // picked up. + std::map env = base; + // No Anthropic seat ever inherits another endpoint or credential from + // the driver's environment. + for (const std::string_view key : {"ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_AUTH_TOKEN", "ANTHROPIC_BASE_URL"}) env.erase(std::string(key)); + const std::string kind = Kind(seat); + if (kind == "local") return env; + const fs::path dir = scratch / std::format("seat-{}", Id(seat)); + Platform::MakeDirs(dir, 0700); + env["CLAUDE_CONFIG_DIR"] = dir.string(); + const std::string secret = Dyn::StrOr(Dyn::Get(seat, "secret"), ""); + if (kind == "oauth") { + if (!fs::exists(dir / ".credentials.json")) WriteCredentials(dir, secret); + } else if (kind == "openrouter") { + // The CLI against OpenRouter's Anthropic-format endpoint + // (/api/v1/messages), the key as a bearer token. Verified + // 2026-10-12 (CLI 2.1.296, xiaomi/mimo-v2.6-pro): tools, thinking, + // cache reads and --json-schema all work; `--effort` does not + // reach the model, and the CLI's total_cost_usd is wrong. + env["ANTHROPIC_BASE_URL"] = std::string(OpenRouterBaseUrl); + env["ANTHROPIC_AUTH_TOKEN"] = secret; + env.erase("CLAUDE_CODE_ENTRYPOINT"); + // hasCompletedOnboarding, so --print asks nothing. + Seats::WriteApiKeyApproval(dir, secret); + } else { + env["ANTHROPIC_API_KEY"] = secret; + // The launcher's CLAUDE_CODE_ENTRYPOINT=claude-vscode is what broke + // S111's API seats (2026-10-08, CLI 2.1.294): with it the key is + // refused ("Not logged in") and, once approved, the subscriber + // credits answer "Credit balance is too low". Bisected over every + // CLAUDE_*/ANTHROPIC_* variable: only this one matters. + env.erase("CLAUDE_CODE_ENTRYPOINT"); + Seats::WriteApiKeyApproval(dir, secret); + } + return env; + } + + // ---- OpenRouter seats ------------------------------------------------------------------ + std::string ModelKey(std::string_view model) { + return model.starts_with("openrouter/") ? std::string(model.substr(11)) : std::string(model); + } + + Json FetchOpenRouterPrices(const std::vector& models, const Log& log, double timeout) { + std::set want; + for (const std::string& m : models) { + if (!m.empty()) want.insert(ModelKey(m)); + } + Json data = Json::array(); + try { + const Json catalogue = Http::JsonBody(Http::UrlOpen(OpenRouterModelsUrl, {}, timeout)); + data = Dyn::Or(Dyn::Get(catalogue, "data"), Dyn::EmptyList()); + } catch (const std::exception& error) { + if (log) log(std::format("openrouter: price catalogue not fetched ({}) - list prices from the built-in table", Dyn::Slice(error.what(), 0, 80))); + } + const std::lock_guard guard(PricesLock); + for (const Json& m : data) { + const Json& id = Dyn::Get(m, "id"); + if (!id.is_string() || !want.contains(id.get())) continue; + const Json& p = Dyn::Or(Dyn::Get(m, "pricing"), Dyn::EmptyDict()); + const Json& cacheRead = Dyn::Get(p, "input_cache_read"); + const Json& cacheWrite = Dyn::Get(p, "input_cache_write"); + Prices[id.get()] = { + {"prompt", Num(Dyn::Get(p, "prompt"))}, + {"completion", Num(Dyn::Get(p, "completion"))}, + {"cacheRead", cacheRead.is_null() ? Json() : Json(AsDouble(cacheRead))}, + {"cacheWrite", cacheWrite.is_null() ? Json() : Json(AsDouble(cacheWrite))}, + }; + } + Json out = Json::object(); + for (const std::string& key : want) { + if (!Prices.contains(key) && BuiltInPrices().contains(key)) Prices[key] = BuiltInPrices().at(key); + out[key] = Prices.contains(key) ? Prices.at(key) : Json(); + } + return out; + } + + double OpenRouterCost(std::string_view model, const Json& usage) { + // Uncached input and cache writes at the prompt price (a cache write + // costs nothing extra on OpenRouter unless the catalogue says so), + // cache reads and output at theirs. 0 for a model without a price. + Json p; + { + const std::lock_guard guard(PricesLock); + const auto found = Prices.find(ModelKey(model)); + if (found != Prices.end()) p = found->second; + } + if (!Dyn::Truthy(p)) p = Dyn::Get(BuiltInPrices(), ModelKey(model)); + if (!Dyn::Truthy(p)) return 0.0; + const double prompt = AsDouble(p.at("prompt")); + const double cacheRead = Dyn::Get(p, "cacheRead").is_null() ? prompt : AsDouble(p.at("cacheRead")); + const double cacheWrite = Dyn::Get(p, "cacheWrite").is_null() ? prompt : AsDouble(p.at("cacheWrite")); + const Json& u = Dyn::Or(usage, Dyn::EmptyDict()); + double total = Num(Dyn::Get(u, "input")) * prompt; + total = total + Num(Dyn::Get(u, "cacheCreation")) * cacheWrite; + total = total + Num(Dyn::Get(u, "cacheRead")) * cacheRead; + total = total + Num(Dyn::Get(u, "output")) * AsDouble(p.at("completion")); + return Dyn::Round(total, 6); + } + + Json ProbeOpenRouter(const std::string& secret, double timeout) { + Json d; + try { + const Json body = Http::JsonBody(Http::UrlOpen(OpenRouterKeyUrl, {{"Authorization", std::format("Bearer {}", secret)}}, timeout)); + d = Dyn::Or(Dyn::Get(body, "data"), Dyn::EmptyDict()); + } catch (const Http::HttpError& error) { + return {{"state", error.Code() == 401 || error.Code() == 403 ? "invalid" : "unknown"}, {"error", std::format("HTTP {} from {}", error.Code(), OpenRouterKeyUrl)}}; + } catch (const std::exception& error) { + return {{"state", "unknown"}, {"error", std::format("probe failed: {}", Dyn::Slice(error.what(), 0, 100))}}; + } + const Json& remaining = Dyn::Get(d, "limit_remaining"); + if (!remaining.is_null() && AsDouble(remaining) <= 0) { + return {{"state", "out"}, {"error", std::format("key limit reached (${:.2f} of ${} used)", Num(Dyn::Get(d, "usage")), G(Num(Dyn::Get(d, "limit"))))}, {"remaining", remaining}}; + } + return {{"state", "ok"}, {"remaining", remaining}, {"usage", Dyn::Get(d, "usage")}, {"limit", Dyn::Get(d, "limit")}}; + } + + // ---- the board ------------------------------------------------------------------------- + fs::path BoardPath() { + if (const std::optional configured = Platform::GetEnv("REVIEW_GATES_BOARD"); configured && !configured->empty()) return *configured; + return fs::path(Platform::Home()) / ".cache" / "review-gates" / BoardFile; + } + + std::string SeatKey(std::string_view kind, std::string_view secret) { + if (kind == "local" || secret.empty()) return "local"; + return std::format("{}:{}", kind, Sha256Hex(secret).substr(0, 12)); + } + + std::string MyKey() { + if (const std::optional self = Platform::Self()) return self->Key(); + return std::format("{}:0", Platform::PublishedPid()); + } + + bool DriverAlive(std::string_view key, const fs::path& boardDir, bool reap) { + const std::vector parts = Dyn::Split(key, ":", 1); + const std::optional pid = parts.size() == 2 ? Dyn::ParseInt(parts[0]) : std::nullopt; + const std::optional start = parts.size() == 2 ? Dyn::ParseInt(parts[1]) : std::nullopt; + if (!pid || !start) return false; + const fs::path locks = boardDir / "drivers"; + if (const std::optional held = Platform::ProbeLifetimeLock(locks, *pid, *start)) { + if (!*held && reap) { + // Its driver died; nobody can take that name again. + std::error_code error; + fs::remove(Platform::LifetimeLockFile(locks, *pid, *start), error); + } + return *held; + } + // No lock file: a driver that takes none. In the sandbox a pid is + // another namespace's and says nothing. + if (Platform::InSandbox()) return true; + const std::optional running = Platform::ProcStart(*pid); + // No /proc at all: the deadline sweep is the backstop. + if (!running) return !fs::is_directory(Platform::Paths().proc); + return *running == *start; + } + + bool ClaimScratch(const fs::path& path) { + // 2026-10-18: two test runs launched seconds apart on one hash shared + // `-run-2` and its out/ files - the outputs rule alone saw an + // empty out/ in the first seconds. + fs::create_directories(path); + const fs::path marker = path / ".driver"; + const std::string me = MyKey(); + std::int32_t fd = open(marker.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC, 0644); + if (fd < 0) { + if (errno != EEXIST) throw std::runtime_error(std::format("{}: {}", Dyn::ErrnoText(errno), Dyn::Repr(marker.string()))); + const std::vector holder = Dyn::Split(Dyn::Strip(Platform::ReadFile(marker).value_or("")), ":", 1); + const std::optional pid = holder.size() == 2 ? Dyn::ParseInt(holder[0]) : std::nullopt; + const std::optional start = holder.size() == 2 ? Dyn::ParseInt(holder[1]) : std::nullopt; + if (pid && start && *pid != 0 && *pid != Platform::PublishedPid() && DriverAlive(std::format("{}:{}", *pid, *start), BoardPath().parent_path())) return false; + fd = open(marker.c_str(), O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0644); + if (fd < 0) throw std::runtime_error(std::format("{}: {}", Dyn::ErrnoText(errno), Dyn::Repr(marker.string()))); + } + WriteAll(fd, me); + return true; + } + + std::vector MergeWindows(Json& into, const Json& other, double now) { + // A window whose reset has passed is ignored on either side; of the + // rest the newer resetsAt wins, and within one reset epoch the higher + // utilization: utilization never falls inside an epoch (2026-10-18: + // 1948 rate_limit_events of three runs, 942 rises, 0 falls), so the + // merge is exact without a timestamp. + std::vector changed; + const Json& readings = Dyn::Or(other, Dyn::EmptyDict()); + for (auto it = readings.begin(); it != readings.end(); ++it) { + const Json& w = it.value(); + if (!w.is_object() || Dyn::Get(w, "utilization").is_null()) continue; + const double r = Num(Dyn::Get(w, "resetsAt")); + if (r + 30 <= now) continue; + const Json& found = Dyn::Get(into, it.key()); + const Json* current = found.is_object() ? &found : nullptr; + const double cr = current ? Num(Dyn::Get(*current, "resetsAt")) : 0.0; + if (!current || Dyn::Get(*current, "utilization").is_null() || cr + 30 <= now || r > cr || (r == cr && AsDouble(w.at("utilization")) > AsDouble(current->at("utilization")))) { + into[it.key()] = {{"utilization", w.at("utilization")}, {"resetsAt", Dyn::Get(w, "resetsAt")}}; + changed.push_back(it.key()); + } + } + return changed; + } + + Board::Board(fs::path path, std::string me, std::optional boot, Log log) + : path(std::move(path)), me(me.empty() ? MyKey() : std::move(me)), boot(boot ? *boot : Platform::BootId()), log(log ? std::move(log) : Log([](const std::string&) {})) { + data = {{"version", 1}, {"drivers", Json::object()}, {"seats", Json::object()}}; + } + + void Board::Warn(const std::string& message) { + if (warned) return; + warned = true; + log(std::format("board: {} - placement continues on this driver's own view", message)); + } + + Json Board::Update(const std::function& change) { + data = Platform::LockedUpdate(path, [&](Json& d) { + if (!Dyn::Has(d, "version")) d["version"] = 1; + if (!Dyn::Has(d, "drivers")) d["drivers"] = Json::object(); + if (!Dyn::Has(d, "seats")) d["seats"] = Json::object(); + return change(d); + }, data, 0600, [this](const std::string& message) { Warn(message); }); + return data; + } + + bool Board::Alive(std::string_view key, const Json& drivers, bool reap) const { + if (key == me) return true; + const std::vector parts = Dyn::Split(key, ":", 1); + if (parts.size() != 2 || !Dyn::ParseInt(parts[0]) || !Dyn::ParseInt(parts[1])) return false; + const Json& d = Dyn::Or(Dyn::Get(drivers, key), Dyn::EmptyDict()); + if (!boot.empty() && Dyn::Truthy(Dyn::Get(d, "boot")) && d.at("boot") != boot) return false; + return DriverAlive(key, path.parent_path(), reap); + } + + std::pair> Board::Sweep(Json& d, double now, bool reap) const { + std::vector lines; + bool changed = false; + Json& drivers = d["drivers"]; + std::vector dead; + for (auto it = drivers.begin(); it != drivers.end(); ++it) { + if (!Alive(it.key(), drivers, reap)) dead.push_back(it.key()); + } + for (const std::string& key : dead) { + drivers.erase(key); + changed = true; + } + std::vector gone; + Json& seats = d["seats"]; + for (auto it = seats.begin(); it != seats.end(); ++it) { + Json& s = it.value(); + if (!Dyn::Get(s, "inflight").is_object()) s["inflight"] = Json::object(); + Json& inflight = s["inflight"]; + // {owner: entries}, in the order they were met. + std::vector> deadOwners; + std::vector> lateOwners; + auto count = [](std::vector>& list, const std::string& owner) { + for (auto& [o, n] : list) { + if (o == owner) { + ++n; + return; + } + } + list.emplace_back(owner, 1); + }; + std::vector drop; + for (auto e = inflight.begin(); e != inflight.end(); ++e) { + const std::string owner = Owner(e.key()); + if (owner != me && !drivers.contains(owner)) { + count(deadOwners, owner); + drop.push_back(e.key()); + } else if (Num(Dyn::Get(e.value(), "deadline")) <= now) { + count(lateOwners, owner); + drop.push_back(e.key()); + } + } + for (const std::string& key : drop) inflight.erase(key); + const std::string name = Dyn::StrOr(Dyn::Get(s, "id"), it.key()); + for (const auto& [owner, n] : deadOwners) lines.push_back(std::format("swept pid {} (dead) with {} in-flight entr{} on seat {}", PidOf(owner), n, n == 1 ? "y" : "ies", name)); + for (const auto& [owner, n] : lateOwners) lines.push_back(std::format("swept {} in-flight entr{} of pid {} past their deadline on seat {}", n, n == 1 ? "y" : "ies", PidOf(owner), name)); + const Json& state = Dyn::Get(s, "state"); + if (Dyn::Truthy(state)) { + const Json& until = Dyn::Get(state, "until"); + const Json& by = Dyn::Get(state, "by"); + if (!until.is_null() ? AsDouble(until) <= now : (by != me && !(by.is_string() && drivers.contains(by.get())))) { + s["state"] = nullptr; + changed = true; + } + } + bool liveWindow = false; + for (const Json& w : Dyn::Or(Dyn::Get(s, "windows"), Dyn::EmptyDict())) liveWindow = liveWindow || (w.is_object() && Num(Dyn::Get(w, "resetsAt")) + 30 > now); + if (inflight.empty() && !Dyn::Truthy(Dyn::Get(s, "state")) && !liveWindow && Num(Dyn::Get(s, "pauseUntil")) <= now) gone.push_back(it.key()); + } + for (const std::string& key : gone) { + seats.erase(key); + changed = true; + } + return {changed || !lines.empty(), lines}; + } + + Json& Board::Record(Json& d, const Json& seat) const { + const std::string key = KeyOf(seat); + Json& seats = d["seats"]; + if (!seats.contains(key) || seats[key].is_null()) { + seats[key] = {{"id", seat.at("id")}, {"kind", seat.at("kind")}, {"windows", Json::object()}, {"pauseUntil", 0.0}, {"state", nullptr}, {"inflight", Json::object()}}; + } + return seats[key]; + } + + bool Board::StateHolds(const Json& state, const Json& d, double now) { + const Json& until = Dyn::Get(state, "until"); + if (!until.is_null()) return AsDouble(until) > now; + const Json& by = Dyn::Get(state, "by"); + return by.is_string() && d.at("drivers").contains(by.get()); + } + + Others Board::OthersOf(const Json& d) const { + // Memory and stacks are the machine's, not a seat's: every other + // driver's pass counts, whatever account it runs on. + Others o; + for (const Json& b : d.at("seats")) { + const Json& inflight = Dyn::Or(Dyn::Get(b, "inflight"), Dyn::EmptyDict()); + for (auto it = inflight.begin(); it != inflight.end(); ++it) { + if (Owner(it.key()) == me) continue; + ++o.count; + if (Dyn::Get(it.value(), "mb").is_null()) ++o.unmeasured; + o.mb += Num(Dyn::Get(it.value(), "mb")); + if (Dyn::Truthy(Dyn::Get(it.value(), "stack"))) o.stacks.insert(Dyn::Str(it.value().at("stack"))); + } + } + o.mb = Dyn::Round(o.mb, 1); + return o; + } + + std::pair> Board::FoldInto(std::vector& seats, double now, Json& d) { + auto [changed, lines] = Sweep(d, now); + others = OthersOf(d); + for (Json& s : seats) { + Json f = ZeroForeign(); + f["drivers"] = Json::object(); + const Json& b = Dyn::Get(d.at("seats"), KeyOf(s)); + if (Dyn::Truthy(b)) { + const Json& inflight = Dyn::Or(Dyn::Get(b, "inflight"), Dyn::EmptyDict()); + std::int64_t count = 0; + double cost5h = 0.0; + double costUsd = 0.0; + for (auto it = inflight.begin(); it != inflight.end(); ++it) { + const std::string owner = Owner(it.key()); + if (owner == me) continue; + ++count; + cost5h += Num(Dyn::Get(it.value(), "cost5h")); + costUsd += Num(Dyn::Get(it.value(), "costUsd")); + f["drivers"][owner] = Dyn::Has(f["drivers"], owner) ? f["drivers"][owner].get() + 1 : 1; + } + f["count"] = count; + f["cost5h"] = cost5h; + f["costUsd"] = costUsd; + if (!IsPaid(s) && !MergeWindows(s["windows"], Dyn::Get(b, "windows"), now).empty()) Reutil(s); + const double pause = Num(Dyn::Get(b, "pauseUntil")); + if (pause > Num(Dyn::Get(s, "pauseUntil"))) { + s["pauseUntil"] = pause; + if (pause > now) lines.push_back(std::format("seat {}: paused until {} - another driver's limit", Id(s), Clock(pause))); + } + const Json& st = Dyn::Get(b, "state"); + if (Dyn::Truthy(st) && std::ranges::contains(StateShared, Kind(s)) && s.at("state") != "out" && s.at("state") != "invalid" && StateHolds(st, d, now)) { + s["state"] = Dyn::At(st, "state"); + s["stateFrom"] = Dyn::Get(st, "by"); + const Json& error = Dyn::Get(st, "error"); + s["error"] = std::format("{} (seen by pid {})", Dyn::Truthy(error) ? Dyn::Str(error) : Dyn::Str(st.at("state")), PidOf(Dyn::StrOr(Dyn::Get(st, "by"), "?"))); + lines.push_back(std::format("seat {}: {} - {}", Id(s), Dyn::Str(s.at("state")), Dyn::Str(s.at("error")))); + } + } + f["cost5h"] = Dyn::Round(AsDouble(f.at("cost5h")), 4); + f["costUsd"] = Dyn::Round(AsDouble(f.at("costUsd")), 4); + s["foreign"] = std::move(f); + } + return {changed, lines}; + } + + bool Board::PublishSeat(Json& d, const Json& s, double now, bool probed) const { + // Readings merged, pause raised, an out/invalid verdict of our own + // written (OAuth and local seats; a paid seat's credit-out is the + // ledger's, its budget-out this driver's alone); a seat our own probe + // read ok clears a verdict another driver left. + Json& b = Record(d, s); + bool changed = !Dyn::Equal(Dyn::Get(b, "id"), s.at("id")); + b["id"] = s.at("id"); + if (!IsPaid(s)) { + if (!Dyn::Has(b, "windows")) b["windows"] = Json::object(); + if (!MergeWindows(b["windows"], Dyn::Get(s, "windows"), now).empty()) changed = true; + } + if (Num(Dyn::Get(s, "pauseUntil")) > Num(Dyn::Get(b, "pauseUntil"))) { + b["pauseUntil"] = AsDouble(s.at("pauseUntil")); + changed = true; + } + if (std::ranges::contains(StateShared, Kind(s))) { + const Json& state = s.at("state"); + const Json& held = Dyn::Get(b, "state"); + if ((state == "out" || state == "invalid") && !Dyn::Truthy(Dyn::Get(s, "stateFrom")) && (!Dyn::Truthy(held) || Dyn::Get(held, "by") == me)) { + Json until; + if (state == "out") { + // A seven-day rejection stands until that window resets. + const Json& seven = Dyn::Or(Dyn::Get(Dyn::Or(Dyn::Get(s, "windows"), Dyn::EmptyDict()), "seven_day"), Dyn::EmptyDict()); + if (Dyn::Truthy(Dyn::Get(seven, "resetsAt"))) until = AsDouble(seven.at("resetsAt")) + 30; + } + const Json record = {{"state", state}, {"error", Dyn::Slice(Dyn::StrOr(Dyn::Get(s, "error"), ""), 0, 160)}, {"until", until}, {"by", me}}; + if (!Dyn::Equal(held, record)) { + b["state"] = record; + changed = true; + } + } else if (probed && state == "ok" && Dyn::Truthy(held)) { + b["state"] = nullptr; + changed = true; + } + } + return changed; + } + + void Board::Join(const Json& info, double now) { + Update([&](Json& d) { + Sweep(d, now); + Json entry = info; + entry["since"] = now; + entry["boot"] = boot; + d["drivers"][me] = std::move(entry); + return true; + }); + } + + void Board::Leave() { + try { + Update([&](Json& d) { + bool changed = d["drivers"].erase(me) > 0; + for (auto it = d["seats"].begin(); it != d["seats"].end(); ++it) { + if (!Dyn::Get(it.value(), "inflight").is_object()) continue; + Json& inflight = it.value()["inflight"]; + std::vector mine; + for (auto e = inflight.begin(); e != inflight.end(); ++e) { + if (e.key().starts_with(std::format("{}/", me))) mine.push_back(e.key()); + } + for (const std::string& key : mine) { + inflight.erase(key); + changed = true; + } + } + return changed; + }); + } catch (const std::exception& error) { + Warn(std::format("leave failed ({})", error.what())); + } + } + + void Board::Publish(const std::vector& seats, double now, bool probed) { + Update([&](Json& d) { + bool changed = Sweep(d, now).first; + for (const Json& s : seats) changed = PublishSeat(d, s, now, probed) || changed; + return changed; + }); + } + + std::vector Board::Fold(std::vector& seats, double now) { + std::vector lines; + Update([&](Json& d) { + auto [changed, folded] = FoldInto(seats, now, d); + lines = std::move(folded); + return changed; + }); + for (const std::string& line : lines) log(std::format("board: {}", line)); + return lines; + } + + std::int64_t Board::TotalInflight(const Json& d) { + std::int64_t total = 0; + for (const Json& s : Dyn::Or(Dyn::Get(d, "seats"), Dyn::EmptyDict())) total += static_cast(Dyn::Or(Dyn::Get(s, "inflight"), Dyn::EmptyDict()).size()); + return total; + } + + std::pair Board::MemoryBase(std::int64_t free, double now) { + // MemAvailable as it was with no pass in flight on the machine. With + // other drivers' passes in flight the recorded reading stands: those + // passes are in the budget at what they were admitted at, and the + // reading of now has their use taken off already - it would count + // them twice. With no record, or a pass in flight that records no + // memory, `free` - the safe side. + std::optional> result; + Update([&](Json& d) { + const bool changed = Sweep(d, now).first; + std::vector entries; + for (const Json& s : d.at("seats")) { + for (const Json& e : Dyn::Or(Dyn::Get(s, "inflight"), Dyn::EmptyDict())) entries.push_back(&e); + } + const Json memory = Dyn::Get(d, "memory").is_object() ? d.at("memory") : Json::object(); + if (entries.empty()) { + d["memory"] = {{"freeMb", free}, {"at", now}, {"by", me}}; + result = {static_cast(free), "no pass in flight on this machine"}; + return true; + } + std::vector held; + for (const Json* e : entries) held.push_back(Num(Dyn::Get(*e, "mb"))); + const double mb = Dyn::Sum(held); + const std::size_t n = entries.size(); + const std::string plural = n != 1 ? "es" : ""; + if (!Dyn::Get(memory, "freeMb").is_null() && std::ranges::all_of(entries, [](const Json* e) { return !Dyn::Get(*e, "mb").is_null(); })) { + result = {AsDouble(memory.at("freeMb")), std::format("the board's reading of {} with no pass in flight; {} pass{} of other drivers in flight hold{} {:.0f} MB of it", + Clock(Num(Dyn::Get(memory, "at"))), n, plural, n == 1 ? "s" : "", mb)}; + } else { + const std::string why = Dyn::Get(memory, "freeMb").is_null() + ? std::string("no reading on the board") + : std::format("{} of them record no memory", std::ranges::count_if(entries, [](const Json* e) { return Dyn::Get(*e, "mb").is_null(); })); + result = {static_cast(free), std::format("read now with {} pass{} of other drivers in flight ({}): their {:.0f} MB counted again, on the safe side", n, plural, why, mb)}; + } + return changed; + }); + return result.value_or(std::pair{static_cast(free), "read now"}); + } + + Board::Placed Board::Place(std::vector& seats, double now, const Json& config, const Json& apiConfig, const Chooser& chooser, std::optional cap) { + // Two drivers polling in the same instant cannot both take the last + // slot, the last stack or the last of the memory. `cap` is the pool + // width, enforced across every driver on the machine (Jorijn, + // 2026-10-18). + Placed placed; + std::vector lines; + Update([&](Json& d) { + auto [changed, folded] = FoldInto(seats, now, d); + lines = std::move(folded); + placed.ranked = Ranked(seats, now, config, apiConfig); + const std::int64_t total = TotalInflight(d); + if (cap && total >= *cap) { + placed.full = total; + return changed; + } + placed.picked = placed.ranked.empty() ? std::nullopt : chooser(placed.ranked, others); + if (placed.picked) { + const Json& seat = *placed.picked->first; + const Json& job = *placed.picked->second; + const bool paid = IsPaid(seat); + const Json& timeout = Dyn::Get(job, "timeoutMin"); + Json entry = { + {"since", now}, + {"deadline", now + 60 * (Dyn::Truthy(timeout) ? AsDouble(timeout) : 20.0) * 1.5 + InflightGraceS}, + {"cost5h", paid ? Json(0) : config.at("lensCostPct")}, + {"costUsd", paid ? LensCostValue(seat, apiConfig) : Json(0)}, + }; + if (!Dyn::Get(job, "_admitted").is_null()) entry["mb"] = Dyn::Round(AsDouble(job.at("_admitted")), 1); + if (Dyn::Truthy(Dyn::Get(job, "stackKey"))) entry["stack"] = job.at("stackKey"); + Record(d, seat)["inflight"][std::format("{}/{}", me, Dyn::Str(Dyn::At(job, "label")))] = std::move(entry); + changed = true; + } + return changed; + }); + for (const std::string& line : lines) log(std::format("board: {}", line)); + return placed; + } + + void Board::Finish(const Json& seat, std::string_view label, double now) { + // Off the board whether or not it is placed again, and the seat's + // settled state onto it. + Update([&](Json& d) { + bool changed = Sweep(d, now).first; + const std::string key = KeyOf(seat); + if (Dyn::Truthy(Dyn::Get(d.at("seats"), key))) { + Json& b = d["seats"][key]; + if (Dyn::Get(b, "inflight").is_object() && b["inflight"].erase(std::format("{}/{}", me, label)) > 0) changed = true; + } + return PublishSeat(d, seat, now, false) || changed; + }); + } + + Json Board::View(double now) const { + Json d; + try { + d = Dyn::LoadJsonFile(path); + } catch (const std::exception&) { + d = Json::object(); + } + if (!d.is_object()) d = Json::object(); + if (!Dyn::Has(d, "drivers")) d["drivers"] = Json::object(); + if (!Dyn::Has(d, "seats")) d["seats"] = Json::object(); + // Nothing written, not even a dead driver's lock file taken away. + Sweep(d, now, false); + Json drivers = Json::object(); + for (auto it = d.at("drivers").begin(); it != d.at("drivers").end(); ++it) { + Json driver = it.value(); + driver["inflight"] = 0; + driver["mb"] = 0.0; + driver["stacks"] = Json::array(); + drivers[it.key()] = std::move(driver); + } + Json seats = Json::object(); + for (auto it = d.at("seats").begin(); it != d.at("seats").end(); ++it) { + const Json& s = it.value(); + Json by = Json::object(); + std::int64_t count = 0; + double cost5h = 0.0; + double costUsd = 0.0; + Json oldest; + const Json& inflight = Dyn::Or(Dyn::Get(s, "inflight"), Dyn::EmptyDict()); + for (auto e = inflight.begin(); e != inflight.end(); ++e) { + const std::string owner = Owner(e.key()); + by[owner] = Dyn::Has(by, owner) ? by[owner].get() + 1 : 1; + ++count; + cost5h += Num(Dyn::Get(e.value(), "cost5h")); + costUsd += Num(Dyn::Get(e.value(), "costUsd")); + const double since = Num(Dyn::Get(e.value(), "since")); + if (oldest.is_null() || since < oldest.get()) oldest = since; + if (drivers.contains(owner)) { + Json& driver = drivers[owner]; + driver["inflight"] = driver.at("inflight").get() + 1; + driver["mb"] = Dyn::Round(AsDouble(driver.at("mb")) + Num(Dyn::Get(e.value(), "mb")), 1); + if (Dyn::Truthy(Dyn::Get(e.value(), "stack"))) driver["stacks"].push_back(e.value().at("stack")); + } + } + const Json& pause = Dyn::Get(s, "pauseUntil"); + seats[it.key()] = { + {"id", Dyn::Get(s, "id")}, + {"kind", Dyn::Get(s, "kind")}, + {"inflight", count}, + {"cost5h", Dyn::Round(cost5h, 4)}, + {"costUsd", Dyn::Round(costUsd, 4)}, + {"by", by}, + {"oldest", oldest}, + {"windows", Dyn::Or(Dyn::Get(s, "windows"), Dyn::EmptyDict())}, + {"pauseUntil", Dyn::Truthy(pause) ? pause : Json(0)}, + {"state", Dyn::Get(s, "state")}, + }; + } + return {{"path", path.string()}, {"drivers", drivers}, {"seats", seats}, {"memory", Dyn::Get(d, "memory")}}; + } + + std::vector Board::OthersLines(const std::vector& seats, double, const Json& config, const Json& apiConfig, std::optional cap) const { + Json drivers = Json::object(); + const Json& all = Dyn::Or(Dyn::Get(data, "drivers"), Dyn::EmptyDict()); + for (auto it = all.begin(); it != all.end(); ++it) { + if (it.key() != me) drivers[it.key()] = it.value(); + } + const std::string width = cap ? std::format("; {} in flight on this machine, the cap is {}", TotalInflight(data), *cap) : std::string(); + if (drivers.empty()) return {std::format("board: {}; no other driver running{}", path.string(), width)}; + std::vector each; + for (auto it = drivers.begin(); it != drivers.end(); ++it) { + std::int64_t n = 0; + for (const Json& s : seats) { + const Json f = Foreign(s); + const Json& by = Dyn::Or(Dyn::Get(f, "drivers"), Dyn::EmptyDict()); + if (Dyn::Has(by, it.key())) n += Int(AsDouble(by.at(it.key()))); + } + const Json& d = it.value(); + each.push_back(std::format("pid {} ({}, {}) since {}, {} in flight on our seats", PidOf(it.key()), Dyn::Has(d, "run") ? Dyn::Str(d.at("run")) : std::string("?"), + Dyn::Has(d, "effort") ? Dyn::Str(d.at("effort")) : std::string("?"), Platform::LocalTime("%H:%M", Num(Dyn::Get(d, "since"))), n)); + } + std::string held; + if (others.count) { + held = std::format("; theirs hold {:.0f} MB of memory", others.mb); + if (!others.stacks.empty()) { + held += std::format(" and {} stack(s) ({})", others.stacks.size(), Dyn::Join(std::vector(others.stacks.begin(), others.stacks.end()), ", ")); + } + } + std::vector lines = {std::format("board: {}; {} other driver(s) running: {}{}{}", path.string(), drivers.size(), Dyn::Join(each, "; "), width, held)}; + for (const Json& s : seats) { + const Json f = Foreign(s); + if (!Dyn::Truthy(f.at("count"))) continue; + std::vector who; + const Json& by = Dyn::Or(Dyn::Get(f, "drivers"), Dyn::EmptyDict()); + for (auto it = by.begin(); it != by.end(); ++it) who.push_back(std::format("{} from pid {}", Dyn::Str(it.value()), PidOf(it.key()))); + const std::int64_t count = Int(AsDouble(f.at("count"))); + if (IsPaid(s)) { + const std::int64_t c = Concurrency(s, apiConfig); + lines.push_back(std::format(" seat {}: {} in flight ({}) - {} of {} slots free, ${:.2f} reserved", Id(s), count, Dyn::Join(who, ", "), std::max(0, c - count - Running(s)), c, + AsDouble(f.at("costUsd")))); + } else { + lines.push_back(std::format(" seat {}: {} passes in flight ({}) - projected 5h {:.1f} %", Id(s), count, Dyn::Join(who, ", "), OauthProjected(s, config))); + } + } + return lines; + } + + const Others& Board::OthersNow() const { + return others; + } + + const std::string& Board::Me() const { + return me; + } + + const fs::path& Board::Path() const { + return path; + } + + const Json& Board::Data() const { + return data; + } + + // ---- reporting ----------------------------------------------------------------------------- + Json Summary(const std::vector& seats) { + Json out = Json::array(); + for (const Json& s : seats) { + Json copy = s; + copy.erase("secret"); + out.push_back(std::move(copy)); + } + return out; + } + + std::string Table(const std::vector& seats, std::optional nowGiven) { + const double now = NowOr(nowGiven); + std::vector> rows = {{"seat", "kind", "state", "5h", "7d", "fable", "running", "foreign", "done", "spent", "note"}}; + for (const Json& s : seats) { + const Json& windows = s.at("windows"); + auto w = [&](std::string_view k) { + const Json& u = Dyn::Get(Dyn::Or(Dyn::Get(windows, k), Dyn::EmptyDict()), "utilization"); + return u.is_null() ? std::string("—") : std::format("{}%", Seats::Fmt(u)); + }; + std::string fable = "—"; + for (auto it = windows.begin(); it != windows.end(); ++it) { + if (Seats::IsFableWindow(it.key()) && !Dyn::Get(it.value(), "utilization").is_null()) { + fable = std::format("{}%", Seats::Fmt(it.value().at("utilization"))); + break; + } + } + std::string note; + if (s.at("state") != "ok") { + note = Dyn::Slice(Dyn::StrOr(s.at("error"), ""), 0, 60); + } else if (AsDouble(s.at("pauseUntil")) > now) { + note = std::format("paused until {}", Platform::LocalTime("%H:%M", AsDouble(s.at("pauseUntil")))); + } + const Json f = Foreign(s); + rows.push_back({Id(s), Kind(s), Dyn::Str(s.at("state")), w("five_hour"), w("seven_day"), fable, Dyn::Str(s.at("running")), + Dyn::Truthy(f.at("count")) ? Dyn::Str(f.at("count")) : std::string("—"), Dyn::Str(s.at("procs")), + IsPaid(s) ? std::format("${:.2f}", AsDouble(s.at("spentUsd"))) : std::string("—"), note}); + } + return Seats::RenderTable(rows); + } +} diff --git a/interfaces/Crafter.Review-Http.cppm b/interfaces/Crafter.Review-Http.cppm new file mode 100644 index 0000000..7206686 --- /dev/null +++ b/interfaces/Crafter.Review-Http.cppm @@ -0,0 +1,49 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +export module Crafter.Review:Http; +import std; +import :Json; + +// The HTTP the driver needs: app_ready's preflight, the OpenRouter catalogue +// and key probe. On Crafter.Network's FetchHTTP1, which follows redirects; on +// top of it a status outside 2xx thrown as HttpError, and every other failure +// as Error with a fixed text, because those texts reach the run log and +// app_ready's detail. Proxies are not consulted (AGENTS.md). +export namespace Crafter::Review::Http { + struct Response { + std::int32_t status = 0; + std::string body; + }; + + // A status outside 2xx. what(): "HTTP Error 404: Not Found", with the + // standard reason phrase rather than the server's. + class HttpError : public std::runtime_error { + public: + HttpError(std::int32_t code, const std::string& message); + explicit HttpError(std::int32_t code); + std::int32_t Code() const; + + private: + std::int32_t code; + }; + + // Every other failure. What fails while the request goes out reads + // "" or ""; what fails while the response comes in reads as it is: + // "timed out". + class Error : public std::runtime_error { + public: + using std::runtime_error::runtime_error; + }; + + // A GET of `url`, read whole. Header names in any case; Accept-Encoding: + // identity and Connection: close always go out, and a User-Agent unless + // one is given. + Response UrlOpen(std::string_view url, const std::map& headers, double timeout); + // The body as JSON: Dyn::Error JSONDecodeError (or UnicodeDecodeError) + // with the scanner's message for a body that is not. + Json JsonBody(const Response& response); + // http.HTTPStatus(code).phrase, "" for a code it does not know. + std::string_view Reason(std::int32_t code); +} diff --git a/interfaces/Crafter.Review-Platform.cppm b/interfaces/Crafter.Review-Platform.cppm index c76f38d..49315a6 100644 --- a/interfaces/Crafter.Review-Platform.cppm +++ b/interfaces/Crafter.Review-Platform.cppm @@ -74,6 +74,12 @@ export namespace Crafter::Review::Platform { // Directories are created as needed. void WriteAtomic(const std::filesystem::path& path, std::string_view content, std::uint32_t mode = 0644); + // `path` and its missing parents: the parents made as mkdir makes them, + // the last directory with `mode`, both less the umask; a directory that is + // there keeps its mode. Throws with the errno text ("[Errno 13] Permission + // denied: '/x'"). + void MakeDirs(const std::filesystem::path& path, std::uint32_t mode = 0777); + // An exclusive flock(2) on `lockPath`, held while the object lives. flock, // not fcntl or OFD locks: every driver on the machine takes flock on the // same side files, and on Linux the two kinds do not exclude each other. A diff --git a/interfaces/Crafter.Review-SeatPool.cppm b/interfaces/Crafter.Review-SeatPool.cppm new file mode 100644 index 0000000..daace60 --- /dev/null +++ b/interfaces/Crafter.Review-SeatPool.cppm @@ -0,0 +1,208 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +export module Crafter.Review:SeatPool; +import std; +import :Json; + +// The seat pool: which credential each lens process runs on. OAuth seats +// (CLAUDE_CODE_OAUTH_TOKEN1..6), API-credit seats (ANTHROPIC_API_KEY1..6, +// real money, a ledger and a budget), OpenRouter seats +// (OPENROUTER_KEY[1..6], never in the default pool), and `local`, this +// machine's login. Placement: API seats first (their credit expires monthly), +// then OAuth seats by projected utilisation; a process that fails on a seat's +// limit, credit or auth is placed again; the pool aborts only when every seat +// is out. Secrets never reach a log, a dump or a child's environment except +// the one the CLI reads. +// +// Several drivers at once (2026-10-31): every driver on this machine +// registers on the board (~/.cache/review-gates/seat-board.json) what it has +// in flight on each seat, and folds the board into its seats and admission +// before every launch; the pick is registered under the same lock. Every +// driver on the machine shares the file, its shape and its lock. A seat is a +// JSON object: the dump records it (without the secret). +// +// Liveness (README, AGENTS.md): a driver is alive while its lifetime lock +// (/drivers/-.lock) is held; without one, natively +// the pid and start time decide, and in the Flatpak the driver is taken for +// alive, because pids of another namespace say nothing here. +export namespace Crafter::Review::SeatPool { + // A job is re-placed this often after a limit, credit or auth failure. + constexpr std::int32_t MaxAttempts = 4; + // An API seat's pause after a 429. + constexpr double ApiBackoffS = 90; + constexpr double OauthFallbackPauseS = 300; + // The kinds whose processes cost money the ledger counts. + constexpr std::array Paid = {"api", "openrouter"}; + // The kinds whose out/invalid verdict crosses drivers on the board. + constexpr std::array StateShared = {"oauth", "local"}; + constexpr std::string_view BoardFile = "seat-board.json"; + // An in-flight entry outlives its pass's time budget x1.5 by this much + // before the sweep takes it. + constexpr double InflightGraceS = 300; + constexpr std::string_view OpenRouterBaseUrl = "https://openrouter.ai/api"; + constexpr std::string_view OpenRouterModelsUrl = "https://openrouter.ai/api/v1/models"; + constexpr std::string_view OpenRouterKeyUrl = "https://openrouter.ai/api/v1/key"; + extern const std::string_view ResumePrompt; + + bool IsPaid(const Json& seat); + // The API-seat knobs. + Json ApiConfig(const std::map& env); + // The other drivers' load on a seat as the last board fold left it. + Json Foreign(const Json& seat); + std::filesystem::path LedgerPath(const std::optional& scratchRoot = std::nullopt); + // The most recent date with that day of the month, as YYYY-MM-DD. + std::string CycleStart(double now, std::int32_t day); + + // What the driver spent on each API seat this cycle, and whether the key + // answered the credit error; one file for every driver, each write a delta + // applied under its lock. + class Ledger { + public: + Ledger(std::filesystem::path path, Json apiConfig, Json data = Json::object()); + static Ledger Load(const std::filesystem::path& path, const Json& apiConfig); + Json& Entry(const std::string& seatId, std::optional now = std::nullopt); + Json Add(const std::string& seatId, double usd, std::optional now = std::nullopt); + Json MarkOut(const std::string& seatId, std::string_view reason, std::optional now = std::nullopt); + void Save(); + const Json& Data() const; + + private: + std::filesystem::path path; + Json apiConfig; + Json data; + Json Update(const std::string& seatId, const std::function& change, std::optional now); + }; + + Json NewSeat(const std::string& id, const std::string& kind, const std::string& secret); + // The seats a run may use. spec: auto (every Anthropic seat of the env + // file and the environment, else local), local | api | oauth | openrouter, + // or a list such as api1,1,2,local. std::invalid_argument with a + // ValueError message for a seat that is not configured, or OpenRouter + // mixed with the rest. + std::vector LoadSeats(const std::map& env, const std::optional& envFile, std::string_view spec, + Ledger* ledger = nullptr, const Json& apiConfig = nullptr); + using ProbeFunction = std::function; + // One probe per seat before the run; a seat already out is skipped. + void Probe(std::vector& seats, const std::string& claudeBin, double timeout, const std::vector& models, + const std::function& log, const ProbeFunction& probe = {}); + void ApplyProbe(Json& seat, const Json& result); + std::string Describe(const Json& seat); + + // The OAuth placement knobs, with the pool's own per-process default (0.25: + // the 2026-10-15 load test moved a 20x seat's window 0.08 points a pass). + Json PoolConfig(const std::map& env); + double OauthProjected(const Json& seat, const Json& config); + // A window whose reset has passed counts as empty until the next reading. + std::vector ExpireWindows(Json& seat, double now); + bool Eligible(Json& seat, double now, const Json& config, const Json& apiConfig); + // Every seat with room for a launch now, best first (pointers into seats). + std::vector Ranked(std::vector& seats, double now, const Json& config, const Json& apiConfig); + Json* Pick(std::vector& seats, double now, const Json& config, const Json& apiConfig); + + std::filesystem::path SeatDir(const std::filesystem::path& scratch, std::string_view seatId); + // A pass interrupted by its seat's limit continues on another seat once + // its transcript is copied into that seat's config dir. + std::optional CarryTranscript(const std::filesystem::path& scratch, std::string_view fromSeat, std::string_view toSeat, + std::string_view sessionId, const std::function& log = {}); + bool AllOut(const std::vector& seats); + std::optional EarliestResume(const std::vector& seats, double now, std::optional recheck = std::nullopt); + // A finished process onto its seat; true when the job must be placed again + // (the seat failed it, not the lens). + bool Settle(Json& seat, Json& record, double now, const Json& config, const Json& apiConfig, Ledger* ledger, const std::function& log); + double RetryAfter(std::string_view text, double fallback); + + std::filesystem::path WriteCredentials(const std::filesystem::path& configDir, std::string_view token); + // The environment of a lens process on this seat. + std::map ChildEnv(const Json& seat, const std::map& base, const std::filesystem::path& scratch); + + std::string ModelKey(std::string_view model); + // {catalogue id: prices or null} of the run's models. + Json FetchOpenRouterPrices(const std::vector& models, const std::function& log = {}, double timeout = 20); + double OpenRouterCost(std::string_view model, const Json& usage); + Json ProbeOpenRouter(const std::string& secret, double timeout = 15); + + // ---- several drivers at once: the board -------------------------------------- + std::filesystem::path BoardPath(); + // `:<12 hex of the secret's sha256>`, `local` for the login. + std::string SeatKey(std::string_view kind, std::string_view secret); + // This driver as the board names it: `:`, the sandbox + // instance's in the Flatpak (Platform::Self). + std::string MyKey(); + // Is the driver `key` alive? Its lifetime lock decides when it has one + // (a free one is taken away when `reap`); else natively its pid and start + // time, and in the sandbox yes. + bool DriverAlive(std::string_view key, const std::filesystem::path& boardDir, bool reap = true); + // One live driver per scratch dir: a `.driver` marker, taken over when its + // driver is dead. + bool ClaimScratch(const std::filesystem::path& path); + // Fold `other`'s readings into `into`, in place; the windows that changed. + std::vector MergeWindows(Json& into, const Json& other, double now); + + // The other drivers' passes in flight on this machine. + struct Others { + std::int64_t count = 0; + double mb = 0; + std::set stacks; + std::int64_t unmeasured = 0; + }; + + class Board { + public: + explicit Board(std::filesystem::path path, std::string me = {}, std::optional boot = std::nullopt, + std::function log = {}); + // Is the driver `key` on the board alive: this driver; else not when + // it joined under another boot; else DriverAlive. + bool Alive(std::string_view key, const Json& drivers, bool reap = true) const; + // Drop what no longer stands: dead drivers with their in-flight + // entries, entries past their deadline, verdicts past their expiry + // or whose driver is gone, seat records with nothing live. (changed, + // the lines that say what went). + std::pair> Sweep(Json& data, double now, bool reap = true) const; + void Join(const Json& info, double now); + // Everything of this driver off the board; idempotent, never throws. + void Leave(); + void Publish(const std::vector& seats, double now, bool probed = false); + std::vector Fold(std::vector& seats, double now); + static std::int64_t TotalInflight(const Json& data); + // The memory a gate's budget starts from, and how it was had. + std::pair MemoryBase(std::int64_t free, double now); + using Chooser = std::function>(const std::vector& ranked, const Others& others)>; + struct Placed { + std::vector ranked; + std::optional> picked; + std::int64_t full = 0; + }; + // One locked section: the board folded, the seats ranked, the + // chooser's pick registered before the lock drops. + Placed Place(std::vector& seats, double now, const Json& config, const Json& apiConfig, const Chooser& chooser, std::optional cap = std::nullopt); + void Finish(const Json& seat, std::string_view label, double now); + // The board as it stands, swept in memory, nothing written. + Json View(double now) const; + std::vector OthersLines(const std::vector& seats, double now, const Json& config, const Json& apiConfig, std::optional cap = std::nullopt) const; + const Others& OthersNow() const; + const std::string& Me() const; + const std::filesystem::path& Path() const; + const Json& Data() const; + + private: + std::filesystem::path path; + std::string me; + std::string boot; + std::function log; + Json data; + Others others; + bool warned = false; + void Warn(const std::string& message); + Json Update(const std::function& change); + Json& Record(Json& data, const Json& seat) const; + static bool StateHolds(const Json& state, const Json& data, double now); + Others OthersOf(const Json& data) const; + std::pair> FoldInto(std::vector& seats, double now, Json& data); + bool PublishSeat(Json& data, const Json& seat, double now, bool probed) const; + }; + + // The seats as the dump records them: no secrets. + Json Summary(const std::vector& seats); + std::string Table(const std::vector& seats, std::optional now = std::nullopt); +} diff --git a/interfaces/Crafter.Review.cppm b/interfaces/Crafter.Review.cppm index 2996a36..d9031ba 100644 --- a/interfaces/Crafter.Review.cppm +++ b/interfaces/Crafter.Review.cppm @@ -9,6 +9,7 @@ export import :Text; export import :Crypto; export import :Platform; export import :Process; +export import :Http; export import :Core; export import :Schema; export import :Tools; @@ -26,5 +27,6 @@ export import :Config; export import :Memory; export import :Mcp; export import :Seats; +export import :SeatPool; export import :Freeze; export import :Driver; diff --git a/project.cpp b/project.cpp index 541fe0a..26905cd 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", @@ -21,6 +21,7 @@ namespace { "interfaces/Crafter.Review-Crypto", "interfaces/Crafter.Review-Platform", "interfaces/Crafter.Review-Process", + "interfaces/Crafter.Review-Http", "interfaces/Crafter.Review-Core", "interfaces/Crafter.Review-Schema", "interfaces/Crafter.Review-Tools", @@ -38,14 +39,16 @@ namespace { "interfaces/Crafter.Review-Memory", "interfaces/Crafter.Review-Mcp", "interfaces/Crafter.Review-Seats", + "interfaces/Crafter.Review-SeatPool", "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", "implementations/Crafter.Review-Process", + "implementations/Crafter.Review-Http", "implementations/Crafter.Review-Core", "implementations/Crafter.Review-Schema", "implementations/Crafter.Review-Tools", @@ -63,6 +66,7 @@ namespace { "implementations/Crafter.Review-Memory", "implementations/Crafter.Review-Mcp", "implementations/Crafter.Review-Seats", + "implementations/Crafter.Review-SeatPool", "implementations/Crafter.Review-Freeze", "implementations/Crafter.Review-Driver", }; @@ -74,7 +78,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", @@ -90,6 +94,8 @@ namespace { "Freeze", "PgWire", "Seats", + "SeatPool", + "Drivers", }; // The version the binary reports: CI sets it from the tag, a checkout diff --git a/tests/Drivers/main.cpp b/tests/Drivers/main.cpp new file mode 100644 index 0000000..a49ebdc --- /dev/null +++ b/tests/Drivers/main.cpp @@ -0,0 +1,496 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// Several drivers at once: the board (seats, memory, the cap), one full run +// per hash, one driver per scratch dir, the merged unit-state, the locked +// ledger, the shared memory record. The checks that run the Scheduler, +// Admission, Stacks, the stack keys and the run's state store come with the +// pool and the driver. +// +// Another driver is a live process of its own (`sleep 300`), so its entries +// stand until it dies; two drivers recording into one memory file are two +// runs of this binary (`--record`). + +#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; + + 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 Require(bool cond, std::string_view what) { + if (!cond) throw std::runtime_error(std::string(what)); + } + + Json With(Json value, const Json& changes) { + for (auto it = changes.begin(); it != changes.end(); ++it) value[it.key()] = it.value(); + return value; + } + + void Write(const fs::path& path, std::string_view text) { + fs::create_directories(path.parent_path()); + std::ofstream(path, std::ios::binary) << text; + } + + // The board file, as JSON. + Json Read(const fs::path& path) { + return Dyn::LoadJsonFile(path); + } + + std::set Keys(const Json& object) { + std::set keys; + for (auto it = object.begin(); it != object.end(); ++it) keys.insert(it.key()); + return keys; + } + + bool AnyContains(const std::vector& lines, std::string_view part) { + return std::ranges::any_of(lines, [&](const std::string& line) { return line.contains(part); }); + } + + Json& At(std::vector& seats, std::string_view id) { + for (Json& s : seats) { + if (s.at("id") == id) return s; + } + throw std::out_of_range(std::string(id)); + } + + // A driver of its own: a live process whose pid and start time name it. + 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); + } + }; + + std::int32_t Run() { + std::println("several drivers at once: the board (seats, memory, stacks, the cap), one full run per hash, one driver per scratch dir, the merged unit-state, the locked ledger"); + Platform::SetMode(Platform::Mode::Native); + const fs::path td = fs::temp_directory_path() / std::format("crafter-review-drivers-{}", Platform::PublishedPid()); + fs::remove_all(td); + fs::create_directories(td); + const fs::path path = td / "seat-board.json"; + // The lifetime locks ClaimScratch looks for are beside the board. + setenv("REVIEW_GATES_BOARD", path.c_str(), 1); + std::mutex logLock; + std::vector logs; + auto log = [&](const std::string& line) { + const std::lock_guard guard(logLock); + logs.push_back(line); + }; + const double now = Platform::Now(); + const std::int64_t start = Platform::ProcStart(getpid()).value_or(0); + const std::string mine = std::format("{}:{}", getpid(), start); + SeatPool::Board A(path, {}, std::nullopt, log); + Check("identity: pid + start time from /proc, alive; a nonexistent pid, a recycled pid (another start time) and another boot are not", + A.Me() == mine && A.Alive(A.Me(), Json::object()) && !A.Alive("2147483647:1", Json::object()) && !A.Alive(std::format("{}:{}", getpid(), start + 1), Json::object()) + && !SeatPool::Board(path, "x:y", "bootA").Alive(mine, {{mine, {{"boot", "bootB"}}}})); + Check("seat keys: one token under two .env numbers is one account, the key carries no secret, the login is local, the kinds differ", + SeatPool::NewSeat("1", "oauth", "tok-one").at("boardKey") == SeatPool::NewSeat("5", "oauth", "tok-one").at("boardKey") && !SeatPool::SeatKey("oauth", "tok-one").contains("tok-one") + && SeatPool::SeatKey("local", "") == "local" && SeatPool::SeatKey("api", "k") != SeatPool::SeatKey("oauth", "k")); + 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& kind, const std::string& secret, std::optional fh = std::nullopt, std::optional sd = std::nullopt, + const std::string& state = "ok") { + Json s = SeatPool::NewSeat(id, kind, secret); + s["state"] = state; + if (fh) { + s["windows"] = {{"five_hour", {{"utilization", *fh}, {"resetsAt", now + 3600}}}, {"seven_day", {{"utilization", sd.value_or(0)}, {"resetsAt", now + 86400}}}}; + const auto [util, binding] = Seats::SeatUtil(s.at("windows")); + s["util"] = util; + s["binding"] = binding; + } + return s; + }; + // The jobs the choosers hand out: they must outlive the place that + // registers them. + 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; + }); + }; + // A call on [seat]: the seat comes back changed. + auto fold1 = [&](SeatPool::Board& board, Json& seat, double at) { + std::vector one = {seat}; + std::vector lines = board.Fold(one, at); + seat = one.front(); + return lines; + }; + auto place1 = [&](SeatPool::Board& board, Json& seat, const Json& c, const SeatPool::Board::Chooser& chooser) { + std::vector one = {seat}; + const bool picked = board.Place(one, now, c, acfg, chooser).picked.has_value(); + seat = one.front(); + return picked; + }; + + const Sleeper child; + SeatPool::Board B(path, child.key, std::nullopt, log); + A.Join({{"run", "A"}}, now); + B.Join({{"run", "B"}}, now); + std::vector sa = {mk("1", "oauth", "t1", 10, 50), mk("2", "oauth", "t2", 10, 50), mk("api1", "api", "k1")}; + std::vector sb = {mk("1", "oauth", "t1", 10, 50), mk("2", "oauth", "t2", 10, 50), mk("api1", "api", "k1")}; + for (std::int32_t i = 0; i < 3; ++i) { + const SeatPool::Board::Placed placed = B.Place(sb, now, cfg, acfg, chooserFor("1", std::format("b{}", i))); + Require(placed.picked && placed.picked->first->at("id") == "1", "B places on seat 1"); + } + for (std::int32_t i = 0; i < 2; ++i) { + const SeatPool::Board::Placed placed = B.Place(sb, now, cfg, acfg, chooserFor("api1", std::format("ba{}", i))); + Require(placed.picked && placed.picked->first->at("id") == "api1", "B places on api1"); + } + // The pool width is the machine's cap: B's five passes count against + // A's. + SeatPool::Board::Placed placed = A.Place(sa, now, cfg, acfg, chooserFor("2", "cap0"), 5); + Check("the cap counts every driver: five in flight on the machine hold a sixth at cap 5, the seats still ranked", !placed.picked && placed.full == 5 && !placed.ranked.empty()); + placed = A.Place(sa, now, cfg, acfg, chooserFor("2", "cap0"), 6); + Check("...and a sixth goes at cap 6; it counts from then on", placed.picked && placed.full == 0 && SeatPool::Board::TotalInflight(Read(path)) == 6); + A.Finish(At(sa, "2"), "cap0", now); + A.Fold(sa, now); + const Json f1 = At(sa, "1").at("foreign"); + const Json fa = At(sa, "api1").at("foreign"); + Check("three passes of B on seat 1 and two on api1 fold into A as foreign, at B's own per-pass figures", + f1.at("count") == 3 && Dyn::Equal(f1.at("cost5h"), 0.75) && Dyn::Equal(f1.at("costUsd"), 0) && fa.at("count") == 2 && Dyn::Equal(fa.at("costUsd"), 2.0) + && Dyn::Equal(f1.at("drivers"), {{B.Me(), 3}}), + std::format("{} {}", DumpSpaced(f1), DumpSpaced(fa))); + Json s84 = mk("x", "oauth", "tx", 84, 50); + s84["foreign"] = {{"count", 3}, {"cost5h", 1.5}, {"costUsd", 0}}; + const bool over = !SeatPool::Eligible(s84, now, cfg, acfg); + Json s84Bare = With(s84, {{"foreign", {{"count", 0}, {"cost5h", 0.0}, {"costUsd", 0.0}}}}); + Check("the 5-hour projection counts the foreign passes: 84 + 1.5 + 0.25 is over 85, without them under", over && SeatPool::Eligible(s84Bare, now, cfg, acfg)); + Json a8 = mk("api9", "api", "k9"); + a8["foreign"] = {{"count", 6}, {"cost5h", 0}, {"costUsd", 6.0}}; + a8["running"] = 2; + Json a8One = With(a8, {{"running", 1}}); + Check("an API key's 8 slots count every driver: 6 foreign + 2 running is full, 6 + 1 is not", !SeatPool::Eligible(a8, now, cfg, acfg) && SeatPool::Eligible(a8One, now, cfg, acfg)); + Check("...and its budget the other drivers' reserve: $6 foreign + $1 running + $1 for the next is over a $7 budget, inside $8", + !SeatPool::Eligible(a8One, now, cfg, With(acfg, {{"budgetUsd", 7}})) && SeatPool::Eligible(a8One, now, cfg, With(acfg, {{"budgetUsd", 8}}))); + std::vector es = {mk("e1", "oauth", "te1", 20, 10), mk("e2", "oauth", "te2", 20, 10)}; + es[0]["foreign"] = {{"count", 10}, {"cost5h", 5.0}, {"costUsd", 0}}; + std::vector order; + for (const Json* s : SeatPool::Ranked(es, now, cfg, With(acfg, {{"budgetUsd", 0}}))) order.push_back(s->at("id").get()); + Check("ranked: equal readings, the seat with foreign load ranks second", order == std::vector{"e2", "e1"}, Dyn::Join(order, ",")); + // Readings merge by max per reset epoch, newer epoch wins, expired + // ignored. + Json w = {{"five_hour", {{"utilization", 40}, {"resetsAt", now + 3600}}}}; + const std::vector ch = SeatPool::MergeWindows(w, {{"five_hour", {{"utilization", 44}, {"resetsAt", now + 3600}}}}, now); + const bool higher = ch == std::vector{"five_hour"} && w["five_hour"]["utilization"] == 44; + Check("windows merge: the same epoch takes the higher utilization, a lower one changes nothing", + higher && SeatPool::MergeWindows(w, {{"five_hour", {{"utilization", 30}, {"resetsAt", now + 3600}}}}, now).empty() && w["five_hour"]["utilization"] == 44); + SeatPool::MergeWindows(w, {{"five_hour", {{"utilization", 3}, {"resetsAt", now + 7200}}}}, now); + Check("...a newer epoch replaces an older one, however low", Dyn::Equal(w["five_hour"], {{"utilization", 3}, {"resetsAt", now + 7200}})); + Json expired = {{"five_hour", {{"utilization", 0}, {"resetsAt", 0}}}}; + Check("...a reading whose reset has passed is ignored on either side", + SeatPool::MergeWindows(w, {{"five_hour", {{"utilization", 90}, {"resetsAt", now - 100}}}}, now).empty() && w["five_hour"]["utilization"] == 3 + && SeatPool::MergeWindows(expired, {{"five_hour", {{"utilization", 50}, {"resetsAt", now + 100}}}}, now) == std::vector{"five_hour"}); + Json sx = mk("1", "oauth", "t1"); + sx["windows"] = {{"five_hour", {{"utilization", 0}, {"resetsAt", 0}}}}; + A.Publish({sx}, now, true); + Check("...and a locally expired window (expire_windows) is never published", + !Dyn::Has(Dyn::Or(Dyn::Get(Read(path).at("seats").at(sx.at("boardKey").get()), "windows"), Dyn::EmptyDict()), "five_hour")); + At(sb, "1")["windows"]["five_hour"]["utilization"] = 40; + B.Publish({At(sb, "1")}, now); + fold1(A, At(sa, "1"), now); + Check("a higher reading of the same epoch on the board reaches the other driver's seat, util and binding recomputed", + At(sa, "1")["windows"]["five_hour"]["utilization"] == 40 && At(sa, "1")["util"] == 50 && At(sa, "1")["binding"] == "seven_day"); + const std::optional before = Platform::ReadFile(path); + SeatPool::Board C(path, A.Me(), std::nullopt, log); + C.Publish({At(sa, "1")}, now); + fold1(C, At(sa, "1"), now); + Check("publish then fold of what the board already holds writes nothing (the merge is idempotent)", Platform::ReadFile(path) == before); + // Verdicts: an OAuth seat's out stands until its window resets, + // invalid while its driver lives; never a paid seat's. + Json weekly = {{"status", "limit"}, + {"reason", "weekly"}, + {"rateType", "seven_day"}, + {"rateLimit", {{"five_hour", {{"utilization", 12}, {"resetsAt", now + 3600}}}, {"seven_day", {{"utilization", 100}, {"resetsAt", now + 500}}}}}}; + const bool requeued = SeatPool::Settle(At(sb, "2"), weekly, now, cfg, acfg, nullptr, log); + B.Finish(At(sb, "2"), "none", now); + A.Fold(sa, now); + const std::string key2 = At(sa, "2").at("boardKey").get(); + Check("an OAuth seat another driver found out (seven-day rejected) is out here, until that window resets", + requeued && At(sa, "2")["state"] == "out" && Dyn::Str(At(sa, "2")["error"]).contains("seen by pid") && Read(path)["seats"][key2]["state"]["until"].get() == now + 530, + Dyn::Str(At(sa, "2")["error"])); + Json sa2 = mk("2", "oauth", "t2", 10, 50); + fold1(A, sa2, now + 600); + const Json record2 = Dyn::Get(Read(path).at("seats"), key2); + Check("...and not once that reset has passed (the record is swept)", sa2["state"] == "ok" && Dyn::Get(Dyn::Or(record2, Dyn::EmptyDict()), "state").is_null()); + sa2 = mk("2", "oauth", "t2", 10, 50, "invalid"); + sa2["error"] = "mine"; + fold1(A, sa2, now); + Check("a local out/invalid is not overwritten by the board's", sa2["state"] == "invalid" && sa2["error"] == "mine"); + sa2 = mk("2", "oauth", "t2", 10, 50); + A.Publish({sa2}, now, true); + Check("a seat our own probe reads ok clears the verdict another driver left", Read(path)["seats"][key2]["state"].is_null()); + At(sb, "1")["state"] = "invalid"; + At(sb, "1")["error"] = "authentication failure (401)"; + B.Publish({At(sb, "1")}, now); + Json sa1 = mk("1", "oauth", "t1", 10, 50); + fold1(A, sa1, now); + Check("an invalid token seen by another driver is adopted while that driver lives", sa1["state"] == "invalid" && Dyn::Str(sa1["error"]).contains("authentication")); + const std::string kApi = At(sa, "api1").at("boardKey").get(); + Json bd = Read(path); + if (!bd["seats"].contains(kApi)) bd["seats"][kApi] = {{"id", "api1"}, {"kind", "api"}, {"windows", Json::object()}, {"pauseUntil", 0}, {"state", nullptr}, {"inflight", Json::object()}}; + bd["seats"][kApi]["state"] = {{"state", "out"}, {"error", "budget"}, {"until", now + 9999}, {"by", B.Me()}}; + bd["seats"][kApi]["pauseUntil"] = now + 90; + Write(path, DumpSpaced(bd)); + Json saApi = mk("api1", "api", "k1"); + fold1(A, saApi, now); + Check("a paid seat's verdict never crosses drivers (the ledger carries a credit-out, a budget-out is each driver's own) - its pause does, until the same moment", + saApi["state"] == "ok" && saApi["pauseUntil"].get() == now + 90 && !SeatPool::Eligible(saApi, now, cfg, acfg) && SeatPool::Eligible(saApi, now + 91, cfg, acfg)); + child.Kill(); + sa1 = mk("1", "oauth", "t1", 10, 50); + const std::vector lines = fold1(A, sa1, now); + Check("a driver that died is swept with its in-flight entries and its invalid verdict, and the sweep says so", + sa1["state"] == "ok" && sa1["foreign"]["count"] == 0 && !Read(path).at("drivers").contains(B.Me()) && AnyContains(lines, "(dead)"), Dyn::Join(lines, "; ")); + SeatPool::Board Z(path, std::format("{}:{}", getpid(), start + 1), std::nullopt, log); + Z.Join({{"run", "Z"}}, now); + Json z0 = mk("1", "oauth", "t1", 10, 50); + place1(Z, z0, cfg, chooserFor("1", "z0")); + Json sz = mk("1", "oauth", "t1", 10, 50); + fold1(A, sz, now); + Check("a driver record with a live pid but another start time is a recycled pid: swept", sz["foreign"]["count"] == 0 && !Read(path).at("drivers").contains(Z.Me())); + Json data = {{"drivers", {{A.Me(), Json::object()}}}, + {"seats", + {{"k", + {{"id", "1"}, + {"kind", "oauth"}, + {"windows", Json::object()}, + {"pauseUntil", 0}, + {"state", nullptr}, + {"inflight", {{std::format("{}/late", A.Me()), {{"since", now - 9999}, {"deadline", now - 1}}}, {std::format("{}/live", A.Me()), {{"since", now}, {"deadline", now + 100}}}}}}}}}}; + const auto [swept, sweptLines] = A.Sweep(data, now); + A.Finish(mk("1", "oauth", "t1", 10, 50), "late", now); + Check("an in-flight entry past its deadline is swept, a live one kept; finishing a pass the sweep already took is harmless", + swept && Keys(data["seats"]["k"]["inflight"]) == std::set{std::format("{}/live", A.Me())} && AnyContains(sweptLines, "past their deadline"), Dyn::Join(sweptLines, "; ")); + // A second live driver: leave, the atomic place, its own projection. + const Sleeper child2; + SeatPool::Board D(path, child2.key, std::nullopt, log); + D.Join({{"run", "D"}}, now); + Json d0 = mk("1", "oauth", "t1", 10, 50); + place1(D, d0, cfg, chooserFor("1", "d0")); + D.Leave(); + D.Leave(); + Json sl = mk("1", "oauth", "t1", 10, 50); + fold1(A, sl, now); + Check("leave takes the driver and its in-flight entries off the board; a second leave is harmless", sl["foreign"]["count"] == 0 && !Read(path).at("drivers").contains(D.Me())); + D.Join({{"run", "D"}}, now); + const Json cfg2 = With(cfg, {{"lensCostPct", 2}, {"maxUtil", 85}}); + // An account of its own: its finish publishes 82. + Json sa82 = mk("4", "oauth", "t4", 82, 10); + Json sd82 = mk("4", "oauth", "t4", 82, 10); + std::mutex resultsLock; + std::vector> results; + auto race = [&](SeatPool::Board& board, Json& seat) { + std::vector one = {seat}; + Json job = {{"label", "race"}, {"timeoutMin", 20}}; + const SeatPool::Board::Placed got = board.Place(one, now, cfg2, acfg, [&job](const std::vector& ranked, const SeatPool::Others&) -> std::optional> { + std::this_thread::sleep_for(50ms); + if (ranked.empty()) return std::nullopt; + return std::pair(ranked.front(), &job); + }); + seat = one.front(); + const std::lock_guard guard(resultsLock); + results.emplace_back(!got.ranked.empty(), got.picked.has_value()); + }; + { + const std::jthread ta(race, std::ref(A), std::ref(sa82)); + const std::jthread tb(race, std::ref(D), std::ref(sd82)); + } + const auto launched = std::ranges::count_if(results, [](const auto& r) { return r.second; }); + const auto roomless = std::ranges::count_if(results, [](const auto& r) { return !r.first; }); + Check("place is one locked section: two drivers racing for a seat with room for one pass get exactly one launch", + launched == 1 && roomless == 1 && Read(path)["seats"][sa82.at("boardKey").get()]["inflight"].size() == 1, std::format("{} launched, {} without room", launched, roomless)); + A.Finish(sa82, "race", now); + D.Finish(sd82, "race", now); + Json overCeiling = mk("o", "oauth", "to", 90, 10); + overCeiling["windows"]["five_hour"]["resetsAt"] = now + 600; + overCeiling["foreign"] = {{"count", 1}, {"cost5h", 0.5}, {"costUsd", 0}}; + const Json overBare = With(overCeiling, {{"foreign", {{"count", 0}, {"cost5h", 0.0}, {"costUsd", 0.0}}}}); + Check("earliest_resume: with another driver's pass in flight the wait is one re-check, else the window reset", + SeatPool::EarliestResume({overCeiling}, now, 30) == now + 30 && SeatPool::EarliestResume({overCeiling}, now) == now + 630 && SeatPool::EarliestResume({overBare}, now, 30) == now + 630); + D.Leave(); + D.Join({{"run", "D"}}, now); + Json sd1 = mk("1", "oauth", "t1", 10, 50); + place1(D, sd1, cfg2, chooserFor("1", "m0")); + place1(D, sd1, cfg2, chooserFor("1", "m1")); + Json sm = mk("1", "oauth", "t1", 10, 50); + fold1(A, sm, now); + // The board's reading of the account (40, B's) folded in before the + // projection. + const Json base = sm["windows"]["five_hour"]["utilization"]; + Check("a driver's passes fold at its own projection (2 % each) into another driver's (0.25 %)", + Dyn::Equal(sm["foreign"], {{"count", 2}, {"cost5h", 4.0}, {"costUsd", 0}, {"drivers", {{D.Me(), 2}}}}) && base == 40 && SeatPool::OauthProjected(sm, cfg) == base.get() + 4.0, + std::format("{} {}", DumpSpaced(sm["foreign"]), base.dump())); + const std::optional unviewed = Platform::ReadFile(path); + Json v = A.View(now); + const std::string km = sm.at("boardKey").get(); + Check("view: the drivers and, per account, the passes in flight by driver - nothing written", + v["drivers"].contains(A.Me()) && v["drivers"].contains(D.Me()) && v["seats"][km]["inflight"] == 2 && Dyn::Equal(v["seats"][km]["by"], {{D.Me(), 2}}) && v["drivers"][D.Me()]["inflight"] == 2 + && Platform::ReadFile(path) == unviewed); + D.Leave(); + Check("no secret on the board", !DumpSpaced(Read(path).at("seats")).contains("t1") && !Platform::ReadFile(path).value_or("").contains("k1")); + // Memory across drivers (2026-10-31): what the board's passes were + // admitted at and the stacks they hold. + D.Join({{"run", "D"}}, now); + Json sE = mk("1", "oauth", "t1", 10, 50); + place1(D, sE, cfg, chooserFor("1", "e0", {{"_admitted", 840.0}, {"stackKey", "http://127.0.0.1:5173"}})); + place1(D, sE, cfg, chooserFor("1", "e1", {{"_admitted", 210.0}})); + // Another account: memory and stacks are the machine's, not a seat's. + Json sA = mk("5", "oauth", "t5", 10, 50); + fold1(A, sA, now); + Json entry = Read(path)["seats"][sE.at("boardKey").get()]["inflight"][std::format("{}/e0", D.Me())]; + const SeatPool::Others& others = A.OthersNow(); + Check("a pass on the board carries the memory it was admitted at and the stack it holds; another driver's fold sees every pass on the machine, whatever account", + Dyn::Equal(entry["mb"], 840.0) && entry["stack"] == "http://127.0.0.1:5173" && others.count == 2 && others.mb == 1050.0 && others.stacks == std::set{"http://127.0.0.1:5173"} + && others.unmeasured == 0 && sA["foreign"]["count"] == 0, + std::format("{} passes, {} MB", others.count, others.mb)); + // The memory base the gates budget from. + D.Leave(); + const auto r0 = A.MemoryBase(50'000, now); + D.Join({{"run", "D"}}, now); + place1(D, sE, cfg, chooserFor("1", "f0", {{"_admitted", 840.0}})); + const auto r1 = A.MemoryBase(30'000, now); + // A pass that records no memory (a driver that does not measure it). + place1(D, sE, cfg, chooserFor("1", "f1")); + const auto r2 = A.MemoryBase(30'000, now); + Check("memory base: with no pass in flight on the machine the reading of now, recorded; with other drivers' measured passes in flight the board's reading from before them; with one that records none, the reading of now", + r0 == std::pair{50'000, "no pass in flight on this machine"} && r1.first == 50'000 && r1.second.contains("the board's reading") && r1.second.contains("840 MB") + && r2.first == 30'000 && r2.second.contains("record no memory"), + std::format("{} | {} | {}", r0.second, r1.second, r2.second)); + D.Leave(); + A.Leave(); + child2.Kill(); + + // One live driver per scratch dir. + const fs::path sd = td / "h8-run-2"; + const Sleeper child4; + fs::create_directories(sd); + Write(sd / ".driver", child4.key); + const bool held = SeatPool::ClaimScratch(sd); + child4.Kill(); + const bool takenOver = SeatPool::ClaimScratch(sd); + Check("a scratch dir a live driver claimed is refused; a dead claim is taken over; a fresh dir is claimed with our pid", + !held && takenOver && Platform::ReadFile(sd / ".driver") == mine && SeatPool::ClaimScratch(td / "h8-run-3") && SeatPool::ClaimScratch(sd)); + + // One full run per hash. + const fs::path gd = td / "gate-state"; + fs::create_directories(gd); + auto f = std::make_unique(gd, "commits", "abcd1234"); + try { + const State::RunLock second(gd, "commits", "abcd1234"); + Check("run lock: a second full run on the hash is refused", false); + } catch (const State::RunLocked& e) { + Check("run lock: a second full run on the hash is refused, the holder not yet identified", std::string_view(e.what()).contains("not identified"), e.what()); + } + f->WriteInfo({{"pid", 1}, {"since", "s"}, {"run_n", 1}, {"scratch", "x"}}); + f->WriteInfo({{"pid", 1}, {"since", "s"}, {"run_n", 2}, {"scratch", "x"}}); + try { + const State::RunLock second(gd, "commits", "abcd1234"); + Check("run lock: ...named once it wrote itself", false); + } catch (const State::RunLocked& e) { + const std::optional holder = State::LockHolder(gd, "commits", "abcd1234"); + Check("run lock: ...named once it wrote itself (one JSON document after two writes)", + std::string_view(e.what()).contains("pid 1") && std::string_view(e.what()).contains("run 2") && holder && holder->at("run_n") == 2, e.what()); + } + auto takes = [&](std::string_view hash8) { + try { + const State::RunLock taken(gd, "commits", hash8); + return true; + } catch (const State::RunLocked&) { + return false; + } + }; + Check("run lock: another hash is free; nobody holds an untaken one", takes("ffff0000") && !State::LockHolder(gd, "commits", "00000000")); + f.reset(); + Check("run lock: released on close (the kernel releases it on death the same way)", !State::LockHolder(gd, "commits", "abcd1234") && takes("abcd1234")); + + // unit-state merge-on-write + const fs::path sp = td / "unit-state.json"; + const Json a1 = {{"unit", "X"}, {"hash", "H1"}}; + const Json b1 = {{"unit", "Y"}, {"hash", "H1"}}; + const Json read1 = {{"hash", "H1"}, {"round", 1}, {"run", "r1"}}; + State::SaveState(sp, {{"ka", a1}}, {}, read1); + Json m = State::SaveState(sp, {{"kb", b1}}, {}, read1); + Check("unit-state merge: two drivers' stores both survive, each with its read", + Keys(m["entries"]) == std::set{"ka", "kb"} && Dyn::Equal(m["entries"]["ka"]["reads"], Json::array({read1})) && Keys(State::LoadState(sp)["entries"]) == std::set{"ka", "kb"}); + const std::vector dropKa = {"ka"}; + m = State::SaveState(sp, {{"ka", a1}}, dropKa); + Check("...a drop wins over a set in the same store", Keys(m["entries"]) == std::set{"kb"}); + const Json read2 = {{"hash", "H2"}, {"round", 1}, {"run", "r2"}}; + State::SaveState(sp, {{"kb", b1}}, {}, read2); + m = State::SaveState(sp, {{"kb", b1}}, {}, read2); + std::vector reads; + for (const Json& x : m["entries"]["kb"]["reads"]) reads.push_back(std::format("{}/{}", x.at("hash").get(), x.at("round").dump())); + Check("...a read of another hash is a second read, counted once however often it is stored", reads == std::vector{"H1/1", "H2/1"}, Dyn::Join(reads, " ")); + Check("...a whole pass carries no count", !State::SaveState(sp, {{"kw", {{"kind", "whole"}}}}, {}, {{"hash", "H"}, {"round", 1}, {"run", "r"}})["entries"]["kw"].contains("reads")); + Write(sp, "garbage"); + Check("...garbage on disk is the empty state", Dyn::Equal(State::SaveState(sp, {{"k", {{"unit", "Z"}}}})["entries"], {{"k", {{"unit", "Z"}}}})); + const Json old = {{"version", 1}, {"entries", {{"q", {{"unit", "Q"}}}}}, {"personaPass", nullptr}}; + State::SaveWhole(sp, old); + Check("...sets=None is the whole-file write as before", Dyn::Equal(Dyn::LoadJsonFile(sp), old)); + + // The ledger: two drivers on one file, every write a locked delta. + const fs::path lp = td / "seat-ledger.json"; + SeatPool::Ledger la = SeatPool::Ledger::Load(lp, acfg); + SeatPool::Ledger lb = SeatPool::Ledger::Load(lp, acfg); + la.Add("api9", 2.0, now); + lb.Add("api9", 3.0, now); + la.Add("api9", 2.0, now); + lb.MarkOut("api9", "credit balance too low", now); + const Json last = Dyn::LoadJsonFile(lp).at("api9"); + Check("two drivers on one ledger file: every add and the out mark survive, the last writer no longer wins", + Dyn::Equal(last.at("spentUsd"), 7.0) && last.at("procs") == 3 && Dyn::Truthy(last.at("outSince")) && Dyn::Equal(la.Entry("api9", now).at("spentUsd"), 7.0), DumpSpaced(last)); + SeatPool::Ledger lc(lp, acfg, {{"api8", {{"cycle", "2099-01-01"}, {"spentUsd", 5.0}, {"procs", 1}, {"outSince", nullptr}, {"outReason", nullptr}}}}); + Check("...an entry of a later cycle (another driver's cycle day) is not wiped", Dyn::Equal(lc.Entry("api8", now).at("spentUsd"), 5.0)); + + // The memory record: every driver's peaks in one file. + const fs::path mp = td / "memory.json"; + std::vector recorders; + for (const std::string_view environment : {"none", "browser"}) { + recorders.push_back(Process::Start({.argv = {Platform::SelfExe().string(), "--record", mp.string(), td.string(), std::string(environment)}}).pid); + } + for (const std::int64_t pid : recorders) Process::Wait(pid); + const Memory::Calibration mc(mp, td, {}); + Check("two drivers recording peaks into one memory file at once: every peak of both is kept", mc.Peaks("none").size() == 20 && mc.Peaks("browser").size() == 20, + std::format("{} {}", mc.Peaks("none").size(), mc.Peaks("browser").size())); + + fs::remove_all(td); + return Failures == 0 ? 0 : 1; + } +} + +int main(int argc, char** argv) { + // A second driver recording its peaks: twenty passes on one environment. + if (argc == 5 && std::string_view(argv[1]) == "--record") { + Memory::Calibration calibration(argv[2], argv[3], {}); + for (std::int32_t i = 0; i < 20; ++i) calibration.Record(argv[4], 100.0 + i); + return 0; + } + try { + return Run(); + } catch (const std::exception& error) { + std::println(std::cerr, "drivers: {}", error.what()); + return 1; + } +} diff --git a/tests/SeatPool/main.cpp b/tests/SeatPool/main.cpp new file mode 100644 index 0000000..dd21e2e --- /dev/null +++ b/tests/SeatPool/main.cpp @@ -0,0 +1,167 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +// The seat pool: seats from .env, placement, settling, credentials, the +// ledger; the checks that drive the Scheduler and the Runner are in the +// Scheduler test. Then reference vectors for the retry-after scan, the cycle +// start, the knobs, the OpenRouter cost and the description. + +#include +#include +import std; +import Crafter.Review; +namespace fs = std::filesystem; +using namespace Crafter::Review; + +namespace { + std::int32_t Failures = 0; + + 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; + } + + Json With(Json value, const Json& changes) { + for (auto it = changes.begin(); it != changes.end(); ++it) value[it.key()] = it.value(); + return value; + } + + 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 Ids(const std::vector& seats) { + std::vector ids; + for (const Json& s : seats) ids.push_back(s.at("id").get()); + return Dyn::Join(ids, ","); + } +} + +int main() { + std::println("seat pool: seats from .env, placement, settling, credentials, the ledger, re-placement"); + // The dates below are UTC's; the process zone decides what a timestamp's + // date is, as it does for dt.date.fromtimestamp. + setenv("TZ", "UTC", 1); + tzset(); + Platform::SetMode(Platform::Mode::Native); + const fs::path td = fs::temp_directory_path() / std::format("crafter-review-seatpool-{}", Platform::PublishedPid()); + fs::remove_all(td); + fs::create_directories(td); + setenv("REVIEW_GATES_BOARD", (td / "seat-board.json").c_str(), 1); + const fs::path envFile = td / ".env"; + Write(envFile, "CLAUDE_CODE_OAUTH_TOKEN1=\"tok-one\"\nCLAUDE_CODE_OAUTH_TOKEN2=tok-two\nANTHROPIC_API_KEY1=sk-ant-api-one\nANTHROPIC_API_KEY=sk-ant-bare-never-a-seat\nMAILOSAUR=x\n"); + 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({}); + SeatPool::Ledger ledger(td / "ledger.json", acfg); + std::vector seats = SeatPool::LoadSeats({}, envFile, "", &ledger, acfg); + Check(".env: api seats, oauth seats, never the bare ANTHROPIC_API_KEY, local only when nothing is configured", + Ids(seats) == "api1,1,2" && Ids(SeatPool::LoadSeats({}, std::nullopt, "")) == "local", Ids(seats)); + const std::string summary = DumpSpaced(SeatPool::Summary(seats)); + Check("no secret in the summary or the table", !(summary + SeatPool::Table(seats)).contains("tok-one") && !summary.contains("sk-ant")); + Check("the pool's default projection is 0.25% of the 5-hour window per process (measured 0.08, 2026-10-15); the env overrides", + Dyn::Equal(SeatPool::PoolConfig({}).at("lensCostPct"), 0.25) && Dyn::Equal(SeatPool::PoolConfig({{"REVIEW_GATES_LENS_COST_PCT", "5"}}).at("lensCostPct"), 5)); + + const double now = std::floor(Platform::Now()); + Json& api1 = seats[0]; + Json& s1 = seats[1]; + Json& s2 = seats[2]; + api1["state"] = "ok"; + s1 = With(s1, {{"state", "ok"}, {"util", 60}, {"windows", {{"five_hour", {{"utilization", 60}, {"resetsAt", now + 600}}}, {"seven_day", {{"utilization", 10}, {"resetsAt", now + 86400}}}}}}); + s2 = With(s2, {{"state", "ok"}, {"util", 50}, {"windows", {{"five_hour", {{"utilization", 20}, {"resetsAt", now + 900}}}, {"seven_day", {{"utilization", 50}, {"resetsAt", now + 86400}}}}}}); + auto pick = [&](const Json& apiConfig) { + const Json* picked = SeatPool::Pick(seats, now, cfg, apiConfig); + return picked ? picked->at("id").get() : std::string("None"); + }; + const std::string first = pick(acfg); + api1["running"] = 2; + Check("api seat first; at its cap the oauth seat with the most 5-hour headroom", first == "api1" && pick(acfg) == "2"); + api1["running"] = 0; + api1["spentUsd"] = 9.5; + Check("api seat without budget room: skipped; oauth-first order", pick(acfg) == "2" && pick(With(acfg, {{"order", "oauth-first"}, {"budgetUsd", 100}})) == "2"); + + std::vector logs; + auto log = [&](const std::string& line) { logs.push_back(line); }; + s1 = With(s1, {{"state", "ok"}, {"util", 20}, {"running", 1}, {"windows", {{"five_hour", {{"utilization", 20}, {"resetsAt", now + 600}}}}}}); + Json limit = {{"status", "limit"}, {"reason", "rate limit"}, {"rateType", "five_hour"}, {"rateLimit", {{"five_hour", {{"utilization", 100}, {"resetsAt", now + 600}}}}}}; + bool requeued = SeatPool::Settle(s1, limit, now, cfg, acfg, &ledger, log); + Check("5h limit on an oauth seat: limited, paused until its reset + 30 s, requeued", + requeued && s1.at("state") == "limited" && s1.at("pauseUntil").get() == now + 630 && !SeatPool::Eligible(s1, now, cfg, acfg)); + Json spend = {{"status", "ok"}, {"costUsd", 2.5}, {"rateLimit", Json::object()}}; + requeued = SeatPool::Settle(api1, spend, now, cfg, acfg, &ledger, log); + Check("api seat: spend accrues in the seat and the ledger, and the budget takes the seat out", + !requeued && api1.at("spentUsd").get() == 12.0 && ledger.Entry("api1").at("spentUsd").get() == 2.5 && api1.at("state") == "out"); + Json api2 = SeatPool::NewSeat("api2", "api", "k"); + api2["state"] = "ok"; + Json credits = {{"status", "credits"}, {"reason", "Your credit balance is too low"}, {"costUsd", 0}, {"rateLimit", Json::object()}}; + requeued = SeatPool::Settle(api2, credits, now, cfg, acfg, &ledger, log); + Check("credit error: out, recorded in the ledger, requeued", requeued && api2.at("state") == "out" && Dyn::Truthy(ledger.Entry("api2").at("outSince"))); + + const fs::path scratch = td / "scratch"; + fs::create_directories(scratch); + const std::map baseEnv = {{"PATH", "/bin"}, {"ANTHROPIC_API_KEY", "leak"}, {"CLAUDE_CODE_OAUTH_TOKEN", "leak2"}, {"HOME", td.string()}}; + const std::map e1 = SeatPool::ChildEnv(SeatPool::NewSeat("1", "oauth", "tok-one"), baseEnv, scratch); + const fs::path cred = scratch / "seat-1" / ".credentials.json"; + Check("oauth seat: its own config dir, the token in .credentials.json (0600), no token variable", + e1.at("CLAUDE_CONFIG_DIR") == (scratch / "seat-1").string() && Dyn::LoadJsonFile(cred).at("claudeAiOauth").at("accessToken") == "tok-one" + && (fs::status(cred).permissions() & fs::perms::all) == (fs::perms::owner_read | fs::perms::owner_write) && !e1.contains("CLAUDE_CODE_OAUTH_TOKEN") && !e1.contains("ANTHROPIC_API_KEY")); + std::map launcher = baseEnv; + launcher["CLAUDE_CODE_ENTRYPOINT"] = "claude-vscode"; + const std::map e2 = SeatPool::ChildEnv(SeatPool::NewSeat("api1", "api", "sk-ant-api-one"), launcher, scratch); + Check("api seat: the key in ANTHROPIC_API_KEY, the launcher's CLAUDE_CODE_ENTRYPOINT dropped, the key approved in its own .claude.json", + e2.at("ANTHROPIC_API_KEY") == "sk-ant-api-one" && !e2.contains("CLAUDE_CODE_ENTRYPOINT") + && Dyn::Equal(Dyn::LoadJsonFile(scratch / "seat-api1" / ".claude.json").at("customApiKeyResponses").at("approved"), Json::array({"sk-ant-api-one"}))); + + // ---- reference vectors ------------------------------------------------------------------ + const std::vector> retry = { + {"Retry-After: 30", 35}, + {"retry after 2000 seconds", 900}, + {"RETRY-AFTER:\u0663", 30}, + {"please retry after a moment", 90}, + {std::format("retry-after:{}45", std::string(14, ' ')), 90}, + {std::format("retry-after:{}45", std::string(12, ' ')), 90}, + {"x retry_after 5 retry after 7", 30}, + {"retry-after: \u0663\u0660 x", 30}, + {"Retry after 99999999999999999999999", 900}, + {"", 90}, + {"retry\u2028after 5", 90}, + {"RETRY AFTER\n\n12", 30}, + }; + std::vector missed; + for (const auto& [text, want] : retry) { + if (SeatPool::RetryAfter(text, 90) != want) missed.push_back(std::format("{} -> {}", Dyn::Repr(text), SeatPool::RetryAfter(text, 90))); + } + Check("retry-after: the regex's greedy run, Unicode digits, the 30..900 clamp, the fallback", missed.empty(), Dyn::Join(missed, "; ")); + const std::vector> cycles = { + {1788775200.0, 1, "2026-09-01"}, {1788775200.0, 7, "2026-09-07"}, {1788775200.0, 8, "2026-08-08"}, {1788775200.0, 28, "2026-08-28"}, + {1772323200.0, 28, "2026-02-28"}, {1772323200.0, 15, "2026-02-15"}, {1767225599.0, 5, "2025-12-05"}, {1709164800.0, 28, "2024-02-28"}, + }; + missed.clear(); + for (const auto& [at, day, want] : cycles) { + if (SeatPool::CycleStart(at, day) != want) missed.push_back(std::format("{} {} -> {}", at, day, SeatPool::CycleStart(at, day))); + } + Check("cycle start: this month's billing day once reached, else last month's, clamped to its length", missed.empty(), Dyn::Join(missed, "; ")); + const std::string knobs = DumpSpaced(SeatPool::ApiConfig({{"REVIEW_GATES_API_BUDGET_USD", " 1_0 "}, + {"REVIEW_GATES_API_CONCURRENCY", "3.9"}, + {"REVIEW_GATES_API_CYCLE_DAY", "40"}, + {"REVIEW_GATES_SEAT_ORDER", "oauth-first"}, + {"REVIEW_GATES_OPENROUTER_BUDGET_USD", "-1"}, + {"REVIEW_GATES_API_LENS_COST_USD", "nan"}})); + Check("api config: float() parsing, int() truncation, the cycle day clamped, a negative or NaN value the default", + knobs == R"({"budgetUsd": 10.0, "lensCostUsd": 1.0, "concurrency": 3, "cycleDay": 28, "order": "oauth-first", "orBudgetUsd": 100.0, "orLensCostUsd": 0.1, "orConcurrency": 16})", knobs); + const double cost = SeatPool::OpenRouterCost("openrouter/xiaomi/mimo-v2.6-pro", {{"input", 1234}, {"cacheCreation", 10}, {"cacheRead", 99999}, {"output", 777}}); + Check("openrouter cost: tokens x the built-in list prices, an unknown model free", cost == 0.001577 && SeatPool::OpenRouterCost("nope", {{"input", 1}}) == 0.0, Dyn::FloatRepr(cost)); + Json described = SeatPool::NewSeat("1", "oauth", "t"); + described["state"] = "ok"; + described["windows"] = {{"five_hour", {{"utilization", 61.5}, {"resetsAt", 1}}}, {"seven_day", {{"utilization", 18}, {"resetsAt", 2}}}, {"seven_day_overage_included", {{"utilization", 22}, {"resetsAt", 0}}}, + {"seven_day_fable", {{"utilization", 3}}}, {"other", {{"utilization", nullptr}}}}; + described["binding"] = "five_hour"; + Json paid = SeatPool::NewSeat("api1", "api", "k"); + paid = With(paid, {{"state", "out"}, {"error", "budget"}, {"spentUsd", 12.345}, {"remainingUsd", 3}}); + Check("describe: an OAuth seat's windows by label, a paid seat's spend and what is left on the key", + SeatPool::Describe(described) == "ok: 5h 61.5%, 7d 18%, fable 22%, fable 3% (binding five_hour)" && SeatPool::Describe(paid) == "out (budget), $12.35 spent this cycle, $3.00 left on the key", + std::format("{} | {}", SeatPool::Describe(described), SeatPool::Describe(paid))); + + fs::remove_all(td); + return Failures == 0 ? 0 : 1; +}