// SPDX-License-Identifier: AGPL-3.0-only // SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® module; #include module Crafter.Review:Pool_impl; import :Pool; import :Json; import :Text; import :Crypto; import :Platform; import :Process; import :Plan; import :Tools; import :Mcp; import :Stream; import :PgWire; import :Seats; import :SeatPool; import :OpenCode; import :Memory; import std; namespace fs = std::filesystem; namespace Crafter::Review::Pool { namespace { double Number(const Json& value) { if (value.is_number()) return value.get(); if (value.is_boolean()) return value.get() ? 1.0 : 0.0; return 0.0; } std::int64_t Integer(const Json& value) { return static_cast(Number(value)); } std::string Clock(double epoch) { return Platform::LocalTime("%H:%M:%S", epoch); } // A number as the caller passed it: an integral value without a // fraction. std::string Plain(double value) { if (value == std::floor(value) && std::abs(value) < 1e15) return std::to_string(static_cast(value)); return Dyn::FloatRepr(value); } const Plan::Environment& EnvironmentOf(std::string_view name) { const Plan::Environment* e = Plan::FindEnvironment(name); if (!e) throw Dyn::Error("KeyError", Dyn::Repr(name)); return *e; } // label.replace('/', '_').replace(' ', '_'): the file a pass's output // and prompt go to. std::string FileLabel(const Json& job) { return Dyn::Replace(Dyn::Replace(Dyn::Str(Dyn::At(job, "label")), "/", "_"), " ", "_"); } std::string ReadLenient(const fs::path& path) { return Dyn::UniversalNewlines(Dyn::DecodeReplace(Platform::ReadFile(path).value_or(""))); } void WriteText(const fs::path& path, std::string_view text) { std::ofstream file(path, std::ios::binary | std::ios::trunc); file << text; } } std::string DefaultBin() { return Seats::FindClaudeBin(); } std::string SqlCommand() { return std::format("{} sql", Platform::SelfExe().string()); } std::optional TotpCommand(const std::optional& fixture) { if (!fixture || fixture->empty()) return std::nullopt; return std::format("{} totp {}", Platform::SelfExe().string(), *fixture); } std::vector ExtraDirs(std::span trees) { const std::string cwd = fs::current_path().string(); std::vector out; for (const std::string& t : trees) { const std::string a = Dyn::AbsPath(t); if (!(std::format("{}/", a)).starts_with(std::format("{}/", cwd)) && a != cwd && !std::ranges::contains(out, a)) out.push_back(a); } return out; } Verdict McpConnected(const Json& parsed, std::string_view kind, bool empirical) { const Json& init = Dyn::Or(Dyn::Get(parsed, "init"), Dyn::EmptyDict()); const Json& servers = Dyn::Or(Dyn::Get(init, "mcp_servers"), Dyn::EmptyList()); const bool connected = std::ranges::any_of(servers, [](const Json& s) { return Dyn::Get(s, "name") == Tools::McpServer && Dyn::Get(s, "status") == "connected"; }); const std::string prefix = std::format("mcp__{}__", Tools::McpServer); const bool tools = std::ranges::any_of(Dyn::Or(Dyn::Get(init, "tools"), Dyn::EmptyList()), [&](const Json& t) { return Dyn::Str(t).starts_with(prefix); }); if (!(connected && tools)) { return {"error", std::format("the {} MCP server did not connect in this pass ({}) - no browser, so the pass's verdict is not functional evidence", Tools::McpServer, Dyn::Truthy(servers) ? Dyn::ReprValue(servers) : std::string("no server in the init line"))}; } const Json none = {{"calls", 0}, {"errors", 0}}; const Json& b = Dyn::Or(Dyn::Get(parsed, "browser"), none); if (empirical && kind == "whole" && Dyn::At(b, "calls") == 0) return {"error", "the whole pass made no browser call - nothing was exercised, so its verdict is not functional evidence"}; if (Dyn::Truthy(Dyn::At(b, "calls")) && Number(Dyn::At(b, "errors")) >= Number(Dyn::At(b, "calls"))) { return {"error", std::format("every browser call of this pass failed ({} of {}) - the browser did not work, so the pass's verdict is not functional evidence", Dyn::Str(b.at("errors")), Dyn::Str(b.at("calls")))}; } return {"ok", ""}; } Verdict DbAnswered(const Json& parsed, std::string_view, bool) { const Json none = {{"calls", 0}, {"answered", 0}}; const Json& d = Dyn::Or(Dyn::Get(parsed, "db"), none); if (Dyn::Truthy(Dyn::At(d, "calls")) && !Dyn::Truthy(Dyn::At(d, "answered"))) { return {"error", std::format("none of this pass's {} SQL call(s) reached the database (no `{}` line in any result) - no database, so the pass's verdict is not evidence from it", Dyn::Str(d.at("calls")), PgWire::Answered)}; } return {"ok", ""}; } Verdict Exercised(const Json& parsed, std::string_view environment, std::string_view kind, bool empirical) { const Plan::Environment* found = Plan::FindEnvironment(environment); const Plan::Environment& e = found ? *found : *Plan::FindEnvironment("none"); if (e.mcp == "playwright") { Verdict v = McpConnected(parsed, kind, empirical); if (v.first != "ok") return v; } if (e.db) { Verdict v = DbAnswered(parsed, kind, empirical); if (v.first != "ok") return v; } return {"ok", ""}; } std::string StackKey(const Json& stack) { const std::string u = Dyn::Lower(Dyn::RStrip(Dyn::Strip(Dyn::StrOr(Dyn::Get(stack, "appUrl"), "")), "/")); if (u.empty()) return Dyn::Str(Dyn::At(stack, "id")); const Dyn::SplitUrl p = Dyn::UrlSplit(u); std::optional port; try { port = p.Port(); } catch (const std::invalid_argument&) { return u; } std::string portText; if (port && *port != 0) { portText = std::to_string(*port); } else if (p.scheme == "http") { portText = "80"; } else if (p.scheme == "https") { portText = "443"; } else { portText = "None"; } const std::optional hostname = p.Hostname(); const std::string host = hostname && (*hostname == "localhost" || *hostname == "127.0.0.1" || *hostname == "::1" || *hostname == "0.0.0.0") ? "127.0.0.1" : hostname.value_or(""); return std::format("{}://{}:{}{}", p.scheme, host, portText, p.path); } std::vector AllowedTools(std::string_view, std::string_view, std::span trees, std::span extra, const std::optional& loginFixture, std::string_view environment) { std::vector allow = Tools::BaseAllow(); const std::string cwd = fs::current_path().string(); for (const std::string& t : trees) { // Both spellings: lenses write absolute -C paths too (sabotage // proof 2026-09-17: five denials on an absolute path). std::vector spellings; for (const std::string& spelled : {t, Dyn::AbsPath(t), Dyn::JoinPath(cwd, t)}) { if (!std::ranges::contains(spellings, spelled)) spellings.push_back(spelled); } for (const std::string& spelled : spellings) { for (const std::string_view s : Tools::GitReadOnly) allow.push_back(std::format("Bash(git -C {} {}:*)", spelled, s)); } } const Plan::Environment& e = EnvironmentOf(environment); if (e.mcp == "playwright") std::ranges::move(Tools::McpToolNames(), std::back_inserter(allow)); for (const std::string_view tool : e.tools) allow.emplace_back(tool); if (e.db) allow.push_back(std::format("Bash({}:*)", SqlCommand())); if (e.app) { if (const std::optional totp = TotpCommand(loginFixture)) allow.push_back(std::format("Bash({}:*)", *totp)); std::vector extras; for (const std::string& x : extra) { if (!std::ranges::contains(allow, x)) extras.push_back(x); } std::ranges::move(extras, std::back_inserter(allow)); } // Read(//abs/path/**): the double slash marks an absolute path. for (const std::string& d : ExtraDirs(trees)) allow.push_back(std::format("Read(/{}/**)", d)); return allow; } // ---- Stacks ----------------------------------------------------------------------- Stacks::Stacks(std::vector stacks) : all(std::move(stacks)) { for (const Json& s : all) { const std::string id = Dyn::Str(Dyn::At(s, "id")); keys[id] = StackKey(s); free.push_back(id); } } std::size_t Stacks::Size() const { return all.size(); } const Json* Stacks::First() const { return all.empty() ? nullptr : &all.front(); } const std::vector& Stacks::All() const { return all; } const std::string& Stacks::KeyOf(const std::string& id) const { return keys.at(id); } bool Stacks::Can(const std::optional& prefer, const std::set& held) const { const std::lock_guard guard(lock); std::vector open; for (const std::string& id : free) { if (!held.contains(keys.at(id))) open.push_back(id); } return prefer ? std::ranges::contains(open, *prefer) : !open.empty(); } Json Stacks::Take(const std::optional& prefer, const std::set& held) { const std::lock_guard guard(lock); std::string id; if (prefer) { id = *prefer; } else { const auto open = std::ranges::find_if(free, [&](const std::string& f) { return !held.contains(keys.at(f)); }); if (open == free.end()) throw Dyn::Error("IndexError", "list index out of range"); id = *open; } const auto at = std::ranges::find(free, id); if (at == free.end()) throw Dyn::Error("ValueError", "list.remove(x): x not in list"); free.erase(at); return *std::ranges::find_if(all, [&](const Json& s) { return Dyn::Str(s.at("id")) == id; }); } void Stacks::Give(const std::string& id) { const std::lock_guard guard(lock); if (keys.contains(id) && !std::ranges::contains(free, id)) free.push_back(id); } std::vector Stacks::Free() const { const std::lock_guard guard(lock); return free; } // ---- Admission -------------------------------------------------------------------- Admission::Admission(CostOf costOf, NeedsStack needsStack, Stacks* stacks, std::optional budget, double reserve, Available available, Log log) : costOf(std::move(costOf)), needsStack(std::move(needsStack)), stacks(stacks), budget(budget), reserve(reserve), log(log ? std::move(log) : Log([](const std::string& line) { std::println("{}", line); })) { this->available = available ? std::move(available) : Available([]() -> std::optional { const std::optional mb = Memory::Available(); return mb ? std::optional(static_cast(*mb)) : std::nullopt; }); } std::optional Admission::Prefer(const Json& job) const { if (!Dyn::Truthy(Dyn::Get(job, "resumeSession"))) return std::nullopt; const Json& previous = Dyn::Get(job, "prevStack"); return previous.is_null() ? std::nullopt : std::optional(Dyn::Str(previous)); } bool Admission::Can(const Json& job) { SeatPool::Others seen; { const std::lock_guard guard(lock); seen = others; } if (needsStack(job) && !(stacks && stacks->Can(Prefer(job), seen.stacks))) return false; if (!budget) return true; double cost = 0; double inUse = 0; { const std::lock_guard guard(lock); if (inflight == 0 && seen.count == 0) return true; cost = costOf(job); inUse = held + seen.mb; } if (inUse + cost > *budget) return false; const std::optional now = available(); return !now || *now - cost >= reserve; } void Admission::Take(Json& job) { const double cost = costOf(job); SeatPool::Others seen; { const std::lock_guard guard(lock); seen = others; if (inflight == 0 && seen.count == 0 && budget && cost > *budget && !warned) { warned = true; log(std::format("pool: {} costs ~{:.0f} MB and the memory budget is {:.0f} MB - launched alone, as nothing else runs", Dyn::Str(job.at("label")), cost, *budget)); } ++inflight; held += cost; } job["_admitted"] = cost; if (needsStack(job)) { const Json s = stacks->Take(Prefer(job), seen.stacks); const std::string id = Dyn::Str(s.at("id")); job["stack"] = s; job["prevStack"] = id; // Its name on the board. job["stackKey"] = stacks->KeyOf(id); } } void Admission::Give(Json& job) { if (!job.contains("_admitted")) return; const double cost = Number(job.at("_admitted")); job.erase("_admitted"); { const std::lock_guard guard(lock); --inflight; held = std::max(0.0, held - cost); } job.erase("stackKey"); if (job.contains("stack")) { const Json s = job.at("stack"); job.erase("stack"); if (Dyn::Truthy(s) && stacks) stacks->Give(Dyn::Str(s.at("id"))); } } void Admission::SetOthers(SeatPool::Others value) { const std::lock_guard guard(lock); others = std::move(value); } SeatPool::Others Admission::OthersNow() const { const std::lock_guard guard(lock); return others; } std::int64_t Admission::Inflight() const { const std::lock_guard guard(lock); return inflight; } double Admission::Held() const { const std::lock_guard guard(lock); return held; } // ---- Runner ----------------------------------------------------------------------- Runner::Runner(RunnerOptions given) : options(std::move(given)) { if (options.bin.empty()) { const std::optional configured = Platform::GetEnv("REVIEW_GATES_CLAUDE_BIN"); options.bin = configured && !configured->empty() ? *configured : DefaultBin(); } if (options.opencodeBin.empty()) options.opencodeBin = OpenCode::DefaultBin(); if (!options.log) options.log = [](const std::string& line) { std::println("{}", line); }; for (const std::string_view d : {"prompts", "out", "err"}) fs::create_directories(options.scratch / d); env = Platform::Environment(); env["CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC"] = "1"; env["DISABLE_AUTOUPDATER"] = "1"; for (const std::string_view key : {std::string_view("MAX_THINKING_TOKENS"), std::string_view("CLAUDE_CODE_OAUTH_TOKEN"), std::string_view("ANTHROPIC_API_KEY"), std::string_view("ANTHROPIC_AUTH_TOKEN"), std::string_view("ANTHROPIC_BASE_URL"), PgWire::UrlEnv, PgWire::BrokerEnv}) { env.erase(std::string(key)); } } std::int32_t Runner::KillChildren() { const std::lock_guard guard(lock); std::int32_t n = 0; for (const auto& [pid, live] : running) { Process::Signal(pid, SIGTERM); ++n; } return n; } std::string Runner::EnvOf(const Json& job) const { const auto found = options.environments.find(Dyn::Str(Dyn::At(job, "lensFile"))); return found == options.environments.end() ? "none" : found->second; } std::map Runner::ProcessEnv(const Json& job, const std::map& base) const { if (!EnvironmentOf(EnvOf(job)).db) return base; const Json& s = Dyn::Get(job, "stack"); std::optional url = options.dbUrl; if (Dyn::Truthy(s)) { const Json& own = Dyn::Get(s, "dbUrl"); url = Dyn::Truthy(own) ? std::optional(Dyn::Str(own)) : std::nullopt; } std::map out = base; out[std::string(PgWire::BrokerEnv)] = options.broker && url && !url->empty() ? options.broker->Address(*url) : std::string(); return out; } std::optional Runner::LoginOf(const Json& job) const { const Json& s = Dyn::Get(job, "stack"); if (!Dyn::Truthy(s)) return options.loginFixture; const Json& login = Dyn::Get(s, "login"); return login.is_null() ? std::nullopt : std::optional(Dyn::Str(login)); } std::optional Runner::McpConfig(const Json& job) const { if (EnvironmentOf(EnvOf(job)).mcp != "playwright") return std::nullopt; return Mcp::ConfigJson(Mcp::PlaywrightConfig(options.functional, Mcp::OutputDir(options.scratch, Dyn::Str(Dyn::At(job, "label"))).string())); } std::vector Runner::Cmd(const Json& job, const fs::path& systemPath, std::string_view schema, std::span trees, std::int64_t maxTurns, bool persist, const std::string& sessionId, bool resume) const { std::vector c = {options.bin, "--print", "--model", Dyn::Str(Dyn::At(job, "model")), "--effort", options.effort}; c.insert(c.end(), {resume ? std::string("--resume") : std::string("--session-id"), sessionId}); if (!persist && !resume) c.emplace_back("--no-session-persistence"); const std::optional mcp = McpConfig(job); // --safe-mode drops every MCP server (measured 2026-10-15, CLI // 2.1.300); a pass with the browser runs without it - // --strict-mcp-config and --setting-sources '' still keep the // consumer's own servers, hooks and settings out. if (!mcp) c.emplace_back("--safe-mode"); c.insert(c.end(), {"--strict-mcp-config", "--setting-sources", "", "--disable-slash-commands", "--append-system-prompt-file", systemPath.string(), "--exclude-dynamic-system-prompt-sections"}); if (mcp) c.insert(c.end(), {"--mcp-config", *mcp}); for (const std::string& d : ExtraDirs(trees)) c.insert(c.end(), {"--add-dir", d}); std::vector extra; for (const Json& x : Dyn::Or(Dyn::Get(options.functional, "extraTools"), Dyn::EmptyList())) extra.push_back(Dyn::Str(x)); const std::vector allow = AllowedTools(Dyn::Str(job.at("lensFile")), Dyn::Str(Dyn::At(job, "kind")), trees, extra, LoginOf(job), EnvOf(job)); c.insert(c.end(), {"--tools", "Read,Glob,Grep,Bash", "--allowedTools", Dyn::Join(allow, ","), "--disallowedTools", Dyn::Join(Tools::Deny(), ","), "--permission-prompts", "none", "--max-turns", std::to_string(maxTurns), "--json-schema", std::string(schema), "--output-format", "stream-json", "--verbose"}); return c; } void Runner::SetSeats(const std::vector& seats) { const std::lock_guard guard(lock); seatsById.clear(); for (const Json& s : seats) seatsById[Dyn::Str(s.at("id"))] = {{"id", s.at("id")}, {"kind", s.at("kind")}, {"secret", Dyn::Get(s, "secret")}}; } const std::string& Runner::Bin() const { return options.bin; } std::map& Runner::Env() { return env; } const fs::path& Runner::Scratch() const { return options.scratch; } std::optional Runner::Wait(std::int64_t pid, std::int64_t timeoutMin, bool& timedOut) { std::optional code = Process::Wait(pid, std::chrono::milliseconds(timeoutMin * 60 * 1000)); if (!code) { timedOut = true; // The supervisor takes the tree down on SIGTERM and reports the // kill; it gets ten seconds for that. Process::Signal(pid, SIGTERM); code = Process::Wait(pid, std::chrono::milliseconds(10000)); if (!code) { Process::Signal(pid, SIGKILL); code = Process::Wait(pid); } } { const std::lock_guard guard(lock); running.erase(pid); } return code ? std::optional(Process::SupervisedReturnCode(*code)) : std::nullopt; } std::pair Runner::Run(const Json& job, const std::string& prompt, const fs::path& systemPath, std::string_view schema, std::span trees, std::int64_t maxTurns, std::int64_t timeoutMin, bool persist) { const std::string label = FileLabel(job); const fs::path pp = options.scratch / "prompts" / std::format("{}.txt", label); WriteText(pp, prompt); const fs::path op = options.scratch / "out" / std::format("{}.jsonl", label); const fs::path ep = options.scratch / "err" / std::format("{}.txt", label); std::string sid = Uuid4(); Json rec = {{"label", job.at("label")}, {"sessionId", sid}, {"startedAt", Platform::LocalTime("%Y-%m-%dT%H:%M:%S")}, {"maxTurns", maxTurns}, {"timeoutMin", timeoutMin}, {"outFile", Dyn::RelPath(op.string(), options.repo.string())}, {"promptBytes", Dyn::Length(prompt)}}; if (options.dryRun) { std::vector fail; for (const std::string& x : Dyn::Split(Platform::GetEnv("REVIEW_GATES_SIMULATE_FAIL").value_or(""), ",")) { if (!x.empty()) fail.push_back(x); } const Json& unit = Dyn::Get(job, "unit"); if ((unit.is_string() && std::ranges::contains(fail, unit.get())) || (job.at("kind") == "whole" && std::ranges::contains(fail, std::string("whole")))) { rec["status"] = "error"; rec["reason"] = "simulated failure (REVIEW_GATES_SIMULATE_FAIL)"; return {rec, nullptr}; } rec["status"] = "dry-run"; rec["reason"] = ""; return {rec, nullptr}; } Json seat; if (Dyn::Truthy(Dyn::Get(job, "seat"))) { const std::lock_guard guard(lock); const auto found = seatsById.find(Dyn::Str(job.at("seat"))); if (found != seatsById.end()) seat = found->second; } if (options.harness == "opencode") return RunOpenCode(job, prompt, systemPath, trees, timeoutMin, seat, std::move(rec), op, ep, sid); std::map processEnv = ProcessEnv(job, Dyn::Truthy(seat) ? SeatPool::ChildEnv(seat, env, options.scratch) : env); bool resume = false; if (Dyn::Truthy(seat)) { const std::string seatId = Dyn::Str(seat.at("id")); const bool local = seat.at("kind") == "local"; rec["seat"] = seatId; persist = persist || !local; const Json& resumeSession = Dyn::Get(job, "resumeSession"); if (Dyn::Truthy(resumeSession) && !local) { const std::string fromSeat = Dyn::StrOr(Dyn::Get(job, "resumeSeat"), ""); const std::string session = Dyn::Str(resumeSession); if (SeatPool::CarryTranscript(options.scratch, fromSeat, seatId, session, options.log)) { sid = session; resume = true; WriteText(pp, SeatPool::ResumePrompt); rec["sessionId"] = sid; rec["resumedFrom"] = {{"seat", Dyn::Get(job, "resumeSeat")}, {"attempt", Integer(Dyn::Has(job, "attempt") ? job.at("attempt") : Json(1)) - 1}}; rec["promptBytes"] = Dyn::Length(SeatPool::ResumePrompt); } else { options.log(std::format("{}: no transcript of {} on seat {} - the pass restarts on seat {}", Dyn::Str(job.at("label")), Dyn::Slice(session, 0, 8), Dyn::Str(Dyn::Get(job, "resumeSeat")), seatId)); } } } const std::vector cmd = Cmd(job, systemPath, schema, trees, maxTurns, persist, sid, resume); bool timedOut = false; std::optional code; const std::string key = std::format("{}\n{}", Dyn::Str(job.at("label")), sid); { Process::Spec spec {.argv = Process::Supervised(cmd), .environment = processEnv, .cwd = options.repo}; spec.input = {.kind = Process::Stream::Kind::File, .file = pp}; spec.output = {.kind = Process::Stream::Kind::File, .file = op}; spec.error = {.kind = Process::Stream::Kind::File, .file = ep}; spec.deathSignal = SIGTERM; const Process::Child child = Process::Start(spec); { const std::lock_guard guard(lock); running[child.pid] = true; } // The supervisor's pid: the driver's measure leaves the supervisor // itself out, so a peak is claude's tree alone, as the calibration // records it. if (options.sampler) options.sampler->Watch(key, child.pid); code = Wait(child.pid, timeoutMin, timedOut); const std::optional peak = options.sampler ? options.sampler->Done(key) : std::nullopt; rec["memPeakMb"] = peak ? Json(*peak) : Json(); } const std::string out = ReadLenient(op); const std::string err = ReadLenient(ep); const Json parsed = Stream::ParseStream(out); auto [status, reason] = Stream::Classify(parsed, code, err, timedOut); const std::string environment = EnvOf(job); if (status == "ok") std::tie(status, reason) = Exercised(parsed, environment, Dyn::Str(job.at("kind")), Plan::IsLens(Dyn::Str(job.at("lensFile"))) && Plan::LensOf(Dyn::Str(job.at("lensFile"))).empirical); const Json& rate = parsed.at("rateLimit"); rec["endedAt"] = Platform::LocalTime("%Y-%m-%dT%H:%M:%S"); rec["exit"] = code ? Json(*code) : Json(); rec["status"] = status; rec["reason"] = reason; rec["usage"] = parsed.at("usage"); rec["firstTurn"] = parsed.at("firstTurn"); rec["turns"] = parsed.at("turns"); rec["toolCalls"] = parsed.at("toolCalls"); rec["toolErrors"] = parsed.at("toolErrors"); rec["costUsd"] = parsed.at("costUsd"); rec["denials"] = parsed.at("denials"); rec["rateLimit"] = Stream::Windows(rate); rec["rateStatus"] = Dyn::Get(Dyn::Or(rate, Dyn::EmptyDict()), "status"); rec["rateType"] = Dyn::Get(Dyn::Or(rate, Dyn::EmptyDict()), "rateLimitType"); rec["mcp"] = McpConfig(job).has_value(); rec["environment"] = environment; rec["browser"] = Dyn::Get(parsed, "browser"); rec["db"] = Dyn::Get(parsed, "db"); if (status != "ok") return {rec, nullptr}; const std::optional structured = Stream::StructuredOutput(parsed.at("result")); return {rec, structured ? *structured : Json()}; } std::pair Runner::RunOpenCode(const Json& job, const std::string& prompt, const fs::path& systemPath, std::span trees, std::int64_t timeoutMin, const Json& seat, Json rec, const fs::path& out, const fs::path& err, const std::string& sessionId) { // The same prompts, the allowlist as permission rules, the result // through the StructuredOutput tool. No MCP server: a lens whose // environment needs one never runs on this harness. if (!Dyn::Truthy(seat) || seat.at("kind") != "openrouter") { rec["status"] = "error"; rec["reason"] = "the opencode harness runs on an OpenRouter seat only (--seats openrouter)"; return {rec, nullptr}; } const fs::path& home = opencodeHome; const std::string outText = out.string(); const fs::path so = std::format("{}.so.json", outText.substr(0, outText.size() - std::string_view(".jsonl").size())); const std::map processEnv = OpenCode::ChildEnv(OpenCode::BaseEnv(home, Dyn::Str(seat.at("secret")), env), systemPath, so, AllowedTools(Dyn::Str(job.at("lensFile")), Dyn::Str(job.at("kind")), trees), home / "data" / sessionId); const std::string variant = options.variant.value_or(""); const std::vector cmd = OpenCode::Cmd(options.opencodeBin, Dyn::Str(job.at("model")), home / "view", Dyn::Str(job.at("label")), variant, prompt); rec["seat"] = seat.at("id"); rec["harness"] = "opencode"; rec["variant"] = options.variant ? Json(*options.variant) : Json(); rec["environment"] = EnvOf(job); // The line salvage knows an OpenCode output by, with DumpSpaced's // separators. WriteText(out, std::format("{}\n", DumpSpaced({{"type", "review_gates_meta"}, {"model", job.at("model")}, {"harness", "opencode"}, {"variant", rec.at("variant")}}))); bool timedOut = false; std::optional code; { Process::Spec spec {.argv = Process::Supervised(cmd), .environment = processEnv, .cwd = home / "view"}; spec.input = {.kind = Process::Stream::Kind::Null}; spec.output = {.kind = Process::Stream::Kind::Append, .file = out}; spec.error = {.kind = Process::Stream::Kind::File, .file = err}; spec.deathSignal = SIGTERM; const Process::Child child = Process::Start(spec); { const std::lock_guard guard(lock); running[child.pid] = true; } code = Wait(child.pid, timeoutMin, timedOut); } const Json parsed = OpenCode::ParseEvents(ReadLenient(out), so); const auto [status, reason] = OpenCode::Classify(parsed, code.value_or(-1), ReadLenient(err), timedOut); rec["endedAt"] = Platform::LocalTime("%Y-%m-%dT%H:%M:%S"); rec["exit"] = code ? Json(*code) : Json(); rec["status"] = status; rec["reason"] = reason; rec["usage"] = parsed.at("usage"); rec["firstTurn"] = parsed.at("firstTurn"); rec["turns"] = parsed.at("turns"); rec["toolCalls"] = parsed.at("toolCalls"); rec["toolErrors"] = parsed.at("toolErrors"); rec["costUsd"] = parsed.at("costUsd"); rec["denials"] = Json::array(); rec["rateLimit"] = Json::object(); rec["rateStatus"] = nullptr; rec["rateType"] = nullptr; rec["opencodeSession"] = parsed.at("sessionId"); rec["stop"] = parsed.at("stop"); return {rec, status == "ok" ? parsed.at("structured") : Json()}; } // ---- Scheduler -------------------------------------------------------------------- Scheduler::Scheduler(SchedulerOptions options) : o(std::move(options)) { if (!o.log) o.log = [](const std::string& line) { std::println("{}", line); }; if (!o.now) o.now = [] { return Platform::Now(); }; if (o.seats && !o.seats->empty()) { if (!Dyn::Truthy(o.seatConfig)) o.seatConfig = SeatPool::PoolConfig(Platform::Environment()); if (!Dyn::Truthy(o.apiConfig)) o.apiConfig = SeatPool::ApiConfig(Platform::Environment()); if (o.runner) o.runner->SetSeats(*o.seats); } abort = o.abort; } void Scheduler::Log(const std::string& line) const { o.log(line); } void Scheduler::Abort(const std::string& reason) { { const std::lock_guard guard(lock); abort = reason; } wake.notify_all(); } std::optional Scheduler::Aborted() const { const std::lock_guard guard(lock); return abort; } const std::vector& Scheduler::DoneList() const { return done; } Json Scheduler::PauseUntil() const { return pauseUntil; } void Scheduler::RaisePause(double until) { if (until > Number(pauseUntil)) pauseUntil = until; } void Scheduler::Sleep(double seconds) { if (seconds <= 0) return; std::unique_lock guard(lock); wake.wait_for(guard, std::chrono::duration(seconds), [&] { return abort.has_value(); }); } bool Scheduler::Capped(const Json& job) { return o.admission && !o.admission->Can(job); } bool Scheduler::Launchable(Json& job, const std::vector& live, double now, const std::optional& seatId) { // The API serves a prompt prefix from its cache only once the first // response that carries it has begun, so a wave of passes launched // together all WRITE the same prefix. One pass per cold prefix goes // first; the others follow its first response, its end, or // warmTimeout. With seats the prefix is warmed once per seat. if (!o.cacheKey) return true; const std::optional k = o.cacheKey(job); if (!k) return true; const std::pair key = {*k, seatId.value_or("")}; const std::string where = seatId ? std::format(" on seat {}", *seatId) : std::string(); const auto known = warm.find(key); if (now - (known == warm.end() ? 0.0 : known->second) < o.warmTtl) return true; const bool answered = o.warmProbe && std::ranges::any_of(live, [&](Json* r) { if (r == &job) return false; const std::optional rk = o.cacheKey(*r); return rk && *rk == key.first && Dyn::StrOr(Dyn::Get(*r, "seat"), "") == key.second && o.warmProbe(*r); }); if (answered) { warm[key] = now; warming.erase(key); Log(std::format("pool: prompt prefix cached{} (first response in) - the passes that share it follow", where)); return true; } const auto w = warming.find(key); if (w == warming.end()) { warming[key] = {Dyn::Str(job.at("label")), now}; Log(std::format("pool: {} goes first{} - its prompt prefix is not cached yet there", Dyn::Str(job.at("label")), where)); return true; } if (now - w->second.second > o.warmTimeout) { warm[key] = now; const std::string first = w->second.first; warming.erase(w); Log(std::format("pool: no first response from {}{} after {} s - the passes that share its prefix follow anyway", first, where, Plain(o.warmTimeout))); return true; } return false; } void Scheduler::Progress(const std::vector& running, const std::vector& queued) { if (!o.progressPath) return; try { const std::optional reason = Aborted(); Json runningJson = Json::array(); for (const std::string& label : running) runningJson.push_back(label); Json queuedJson = Json::array(); for (const std::string& label : queued) queuedJson.push_back(label); const Json progress = {{"time", Platform::LocalTime("%H:%M:%S")}, {"running", runningJson}, {"queued", queuedJson}, {"done", done.size()}, {"pauseUntil", pauseUntil}, {"abort", reason ? Json(*reason) : Json()}}; Platform::WriteAtomic(*o.progressPath, DumpSpaced(progress), 0666); } catch (const std::exception&) { // A record, never a dependency. } } const std::vector& Scheduler::Run(std::vector jobs, const ExecJob& exec, const OnDone& onDone) { struct Slot { // The pool's copy; the worker runs on its own. Json job; std::thread thread; bool finished = false; std::uint64_t order = 0; std::pair result; std::optional error; }; std::deque queue(std::make_move_iterator(jobs.begin()), std::make_move_iterator(jobs.end())); std::list running; std::uint64_t finishes = 0; const bool withSeats = o.seats && !o.seats->empty(); auto labelsOf = [](const auto& items, auto get) { std::vector labels; for (const auto& item : items) labels.push_back(Dyn::Str(get(item).at("label"))); return labels; }; auto runningLabels = [&] { return labelsOf(running, [](const Slot& s) -> const Json& { return s.job; }); }; auto queuedLabels = [&] { return labelsOf(queue, [](const Json& j) -> const Json& { return j; }); }; auto everySeat = [&] { std::vector parts; for (const Json& s : *o.seats) parts.push_back(std::format("{} {} ({})", Dyn::Str(s.at("id")), Dyn::Str(s.at("state")), Dyn::Slice(Dyn::StrOr(Dyn::Get(s, "error"), ""), 0, 60))); return std::format("every seat is out: {}", Dyn::Join(parts, "; ")); }; for (;;) { if (queue.empty() && running.empty()) break; if (Aborted()) break; const double now = Platform::Now(); bool paused = now < Number(pauseUntil) && !queue.empty(); if (paused && running.empty()) { Log(std::format("pool: paused until {} (rate window), {} queued", Clock(Number(pauseUntil)), queue.size())); Sleep(std::min(60.0, Number(pauseUntil) - now)); continue; } while (!queue.empty() && (!o.concurrency || static_cast(running.size()) < *o.concurrency) && !paused) { std::vector live; for (Slot& slot : running) live.push_back(&slot.job); // Without a board every queued pass waiting for a stack or // memory waits for a completion; with one the check needs the // other drivers' passes as they are now, which only the fold // in Place has. if (!o.board && std::ranges::all_of(queue, [&](const Json& q) { return Capped(q); })) break; Json pickedJob; Json* pickedSeat = nullptr; bool picked = false; // The first queued pass the admission lets in that may launch // on the best seat with room, popped and admitted. auto choose = [&](const std::vector& cands, const SeatPool::Others* others) -> std::optional> { if (others && o.admission) o.admission->SetOthers(*others); for (Json* seat : cands) { const std::optional sid = seat ? std::optional(Dyn::Str(seat->at("id"))) : std::nullopt; for (auto it = queue.begin(); it != queue.end(); ++it) { if (Capped(*it) || !Launchable(*it, live, Platform::Now(), sid)) continue; pickedJob = std::move(*it); queue.erase(it); if (!pickedJob.contains("attempt")) pickedJob["attempt"] = 1; if (o.admission) o.admission->Take(pickedJob); pickedSeat = seat; picked = true; return std::pair(seat, &pickedJob); } } return std::nullopt; }; std::vector cands = {nullptr}; std::int64_t full = 0; if (withSeats) { if (o.board) { // The board folded, the seats ranked, the admission // checked against every driver's passes and the pick // registered in one locked section; a pool width, when // given, is the machine's cap. const SeatPool::Board::Placed placed = o.board->Place(*o.seats, o.now(), o.seatConfig, o.apiConfig, [&](const std::vector& c, const SeatPool::Others& others) { return choose(c, &others); }, o.concurrency); cands = placed.ranked; full = placed.full; } else { cands = SeatPool::Ranked(*o.seats, o.now(), o.seatConfig, o.apiConfig); if (!cands.empty()) choose(cands, nullptr); } if (full) { if (roomWait.is_null()) roomWait = pauseUntil; RaisePause(o.now() + o.noRoomWait); Log(std::format("pool: {} passes in flight on this machine, the cap is {} - re-checked in {:g} s, {} queued", full, o.concurrency.value_or(0), o.noRoomWait, queue.size())); paused = true; break; } if (cands.empty()) { if (SeatPool::AllOut(*o.seats)) { Abort(everySeat()); Log(std::format("pool: {}", *Aborted())); break; } // Wait for the earliest reset, never less than // noRoomWait; with other drivers' passes in flight on a // usable seat their completions free room. std::int64_t held = 0; for (const Json& s : *o.seats) { if (s.at("state") != "out" && s.at("state") != "invalid") held += Integer(SeatPool::Foreign(s).at("count")); } const std::optional resume = SeatPool::EarliestResume(*o.seats, o.now(), o.board ? std::optional(o.noRoomWait) : std::nullopt); if (roomWait.is_null()) roomWait = pauseUntil; RaisePause(std::max(resume.value_or(0.0), o.now() + o.noRoomWait)); Log(std::format("pool: no seat has room for a new launch - waiting until {}, {} queued{}", Clock(Number(pauseUntil)), queue.size(), held ? std::format(" ({} passes of other drivers hold the room - re-checked in {:g} s)", held, o.noRoomWait) : std::string())); paused = true; break; } } else { choose(cands, nullptr); } // Every queued pass waits: for the one warming its prefix, for // a stack, for memory. if (!picked) break; if (pickedSeat) { (*pickedSeat)["running"] = Integer(pickedSeat->at("running")) + 1; pickedJob["seat"] = pickedSeat->at("id"); } Slot& slot = running.emplace_back(); slot.job = std::move(pickedJob); const std::int64_t attempt = Integer(slot.job.at("attempt")); slot.thread = std::thread([this, &slot, &exec, &finishes, copy = slot.job, attempt] { std::pair result; std::optional error; try { result = exec(copy, attempt); } catch (const std::exception& e) { error = e.what(); } { const std::lock_guard guard(lock); slot.result = std::move(result); slot.error = std::move(error); slot.finished = true; slot.order = ++finishes; } wake.notify_all(); }); heldOff = false; } if (Aborted()) break; Progress(runningLabels(), queuedLabels()); if (running.empty()) { if (!queue.empty() && !paused) { // Nothing of this driver runs and nothing could launch: // what holds the room is other drivers' passes - their // completions free it; re-checked, not spun on. if (!heldOff) { const SeatPool::Others others = o.admission ? o.admission->OthersNow() : SeatPool::Others(); std::string what = "nothing in flight to wait for"; if (others.count) { what = std::format("{} pass{} of other drivers on this machine hold{} {:.0f} MB", others.count, others.count != 1 ? "es" : "", others.count == 1 ? "s" : "", others.mb); if (!others.stacks.empty()) what += std::format(" and stack(s) {}", Dyn::Join(std::vector(others.stacks.begin(), others.stacks.end()), ", ")); } Log(std::format("pool: nothing can launch, {} queued - {}; re-checked every {:g} s", queue.size(), what, o.noRoomWait)); heldOff = true; } Sleep(o.noRoomWait); } continue; } std::vector::iterator> finished; { std::unique_lock guard(lock); auto anyFinished = [&] { return std::ranges::any_of(running, [](const Slot& s) { return s.finished; }); }; wake.wait_for(guard, std::chrono::duration(warming.empty() ? 30.0 : o.warmPoll), [&] { return anyFinished() || abort.has_value(); }); for (auto it = running.begin(); it != running.end(); ++it) { if (it->finished) finished.push_back(it); } } std::ranges::sort(finished, {}, [](const auto& it) { return it->order; }); if (!finished.empty() && !roomWait.is_null()) { pauseUntil = roomWait; roomWait = nullptr; Log("pool: a pass finished - seat room is re-checked"); } if (paused && finished.empty()) Log(std::format("pool: paused until {} (rate window), {} running, {} queued", Clock(Number(pauseUntil)), running.size(), queue.size())); for (const auto& it : finished) { it->thread.join(); Json j = std::move(it->job); Json rec; Json raw; if (it->error) { rec = {{"label", j.at("label")}, {"status", "error"}, {"reason", std::format("driver exception: {}", *it->error)}}; } else { rec = std::move(it->result.first); raw = std::move(it->result.second); } running.erase(it); if (o.admission) o.admission->Give(j); if (o.cacheKey) { if (const std::optional k = o.cacheKey(j)) { const std::pair key = {*k, Dyn::StrOr(Dyn::Get(j, "seat"), "")}; if (o.warmProbe && o.warmProbe(j)) warm[key] = Platform::Now(); if (const auto w = warming.find(key); w != warming.end() && w->second.first == Dyn::Str(j.at("label"))) warming.erase(w); } } const std::int64_t attempt = Integer(j.at("attempt")); rec["attempt"] = attempt; const std::string st = Dyn::StrOr(Dyn::Get(rec, "status"), ""); const std::string label = Dyn::Str(j.at("label")); Json fh = Json::object(); if (withSeats && Dyn::Truthy(Dyn::Get(j, "seat"))) { Json* seat = nullptr; for (Json& s : *o.seats) { if (s.at("id") == j.at("seat")) { seat = &s; break; } } if (seat) { (*seat)["running"] = std::max(0, Integer(seat->at("running")) - 1); if (!rec.contains("seat")) rec["seat"] = seat->at("id"); const bool requeue = SeatPool::Settle(*seat, rec, o.now(), o.seatConfig, o.apiConfig, o.ledger, o.log); // Off the board whether or not it is placed again. if (o.board) o.board->Finish(*seat, label, o.now()); if (requeue) { if (attempt <= SeatPool::MaxAttempts && !SeatPool::AllOut(*o.seats)) { Json next = j; next["attempt"] = attempt + 1; next["seat"] = nullptr; if (Dyn::Truthy(Dyn::Get(rec, "sessionId")) && seat->at("kind") != "local" && !Dyn::Truthy(Dyn::Get(rec, "salvaged"))) { next["resumeSession"] = rec.at("sessionId"); next["resumeSeat"] = seat->at("id"); } queue.push_front(std::move(next)); continue; } if (SeatPool::AllOut(*o.seats)) { Abort(everySeat()); Log(std::format("pool: {}", *Aborted())); } done.push_back({std::move(j), std::move(rec), std::move(raw)}); continue; } } if (st == "limit" || st == "credits" || st == "auth") { done.push_back({std::move(j), std::move(rec), std::move(raw)}); continue; } } else if (st == "credits") { Abort(std::format("credit balance exhausted ({})", Dyn::Str(Dyn::Get(rec, "reason")))); Log(std::format("pool: {}", *Aborted())); done.push_back({std::move(j), std::move(rec), std::move(raw)}); continue; } else { const Json& win = Dyn::Or(Dyn::Get(rec, "rateLimit"), Dyn::EmptyDict()); fh = Dyn::Or(Dyn::Get(win, "five_hour"), Dyn::EmptyDict()); const Json& utilization = Dyn::Get(fh, "utilization"); if (!utilization.is_null() && Number(utilization) >= o.pauseAt && Dyn::Truthy(Dyn::Get(fh, "resetsAt"))) { RaisePause(Number(fh.at("resetsAt")) + 30); Log(std::format("pool: five_hour window at {}% - pausing new launches until reset", Dyn::Str(utilization))); } if (st == "limit" && Dyn::StrOr(Dyn::Get(rec, "rateType"), "").starts_with("seven_day")) { Abort(std::format("seven-day window rejected ({})", Dyn::Str(Dyn::Get(rec, "reason")))); Log(std::format("pool: {}", *Aborted())); done.push_back({std::move(j), std::move(rec), std::move(raw)}); continue; } if (st == "auth") { Abort(std::format("authentication failure ({})", Dyn::Str(Dyn::Get(rec, "reason")))); Log(std::format("pool: {}", *Aborted())); done.push_back({std::move(j), std::move(rec), std::move(raw)}); continue; } } if (st == "limit" && attempt <= 4) { const Json& reset = Dyn::Get(fh, "resetsAt"); constexpr std::array Backoff = {300, 900, 1800, 1800}; const double delay = Dyn::Truthy(reset) ? std::max(60.0, Number(reset) - Platform::Now() + 30) : Backoff[static_cast(std::min(attempt - 1, 3))]; if (waitedMin + delay / 60 > o.maxWaitMin) { Log(std::format("pool: {} limit - max wait exceeded, giving up", label)); done.push_back({std::move(j), std::move(rec), std::move(raw)}); continue; } waitedMin += delay / 60; RaisePause(Platform::Now() + delay); Log(std::format("pool: {} hit a limit - requeued after {} min (attempt {})", label, static_cast(delay / 60), attempt + 1)); Json next = j; next["attempt"] = attempt + 1; queue.push_front(std::move(next)); continue; } const std::string reason = Dyn::Has(rec, "reason") ? Dyn::Str(rec.at("reason")) : std::string(); if (st == "error" && reason.contains("max_turns")) { Log(std::format("pool: {} ran out of turns ({}) - not retried, unit unaddressed", label, Dyn::Str(Dyn::Get(j, "maxTurns")))); } else if ((st == "error" || st == "timeout") && attempt <= 1 && !Aborted()) { Json next = j; next["attempt"] = attempt + 1; if (st == "timeout") next["timeoutMin"] = static_cast(Number(Dyn::Has(j, "timeoutMin") ? j.at("timeoutMin") : Json(20)) * 1.5); Log(std::format("pool: {} {} ({}) - one retry", label, st, Dyn::Slice(reason, 0, 80))); queue.push_back(std::move(next)); continue; } const Done& entry = done.emplace_back(std::move(j), std::move(rec), std::move(raw)); std::vector follow; try { follow = onDone(entry.job, entry.rec, entry.raw); } catch (const std::exception& e) { Log(std::format("pool: on_done failed for {}: {}", label, e.what())); follow.clear(); } std::ranges::move(follow, std::back_inserter(queue)); } Progress(runningLabels(), queuedLabels()); } // As ThreadPoolExecutor's exit: the passes in flight end first. for (Slot& slot : running) { if (slot.thread.joinable()) slot.thread.join(); } return done; } }