From c63c89883335f213c64b8fd3612586e7bbd1bc81 Mon Sep 17 00:00:00 2001 From: Jorijn van der Graaf Date: Sat, 31 Oct 2026 08:15:03 +0100 Subject: [PATCH] feat(seatpool): the seat pool and the board several drivers share The seats of .env and the ledger, the probe, placement, settling a finished pass, the child's credentials, the OpenRouter seats, and the board every driver on the machine shares, in one shape and under one lock. A driver is alive while its lifetime lock is held; with no lock file natively its pid and start time decide, and in the sandbox it is taken for alive. :Http is the HTTP the driver needs, on FetchHTTP1: a status outside 2xx is HttpError, every other failure carries a fixed text. :Platform gains MakeDirs, with the last directory's own mode, and errno texts for its writes and locks. The tests cover the seat pool and several drivers at once, less the checks that drive the pool's Scheduler, Admission and Stacks. Co-Authored-By: Claude Opus 5.5 --- implementations/Crafter.Review-Http.cpp | 108 ++ implementations/Crafter.Review-Platform.cpp | 24 +- implementations/Crafter.Review-SeatPool.cpp | 1498 +++++++++++++++++++ interfaces/Crafter.Review-Http.cppm | 49 + interfaces/Crafter.Review-Platform.cppm | 6 + interfaces/Crafter.Review-SeatPool.cppm | 208 +++ interfaces/Crafter.Review.cppm | 2 + project.cpp | 12 +- tests/Drivers/main.cpp | 496 ++++++ tests/SeatPool/main.cpp | 167 +++ 10 files changed, 2562 insertions(+), 8 deletions(-) create mode 100644 implementations/Crafter.Review-Http.cpp create mode 100644 implementations/Crafter.Review-SeatPool.cpp create mode 100644 interfaces/Crafter.Review-Http.cppm create mode 100644 interfaces/Crafter.Review-SeatPool.cppm create mode 100644 tests/Drivers/main.cpp create mode 100644 tests/SeatPool/main.cpp 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; +}