From 0f803d90bd355cf05ad53969dbc1dba5ca01536f Mon Sep 17 00:00:00 2001 From: Jorijn van der Graaf Date: Sat, 31 Oct 2026 09:16:41 +0100 Subject: [PATCH] feat(run): `crafter-review run` and `slots`, the driver of the gates Run: gate 0 (the artifact, the trees, the inventory, the run lock, the scratch dir, dispositions, reuse state, resume, reads per round, seats, the board), the stack and migration preflights, the prompts, reuse of whole and block verdicts, salvage, a gate on the pool, checkpoints, the dump, the dry-run report and the unit-state store; slots_plan and app_ready beside it. The CLI parses run's 28 options, types and choices included. A full run takes its lifetime lock beside the board before it joins it. A signal - or the stop file `crafter-review stop` writes, looked at every 2 s - stops the run once gate 0 is through, and the stop stays in force for the gates after; before that it ends the process with its descendants. Co-Authored-By: Claude Opus 5.5 --- implementations/Crafter.Review-Driver.cpp | 212 +- implementations/Crafter.Review-Run.cpp | 2171 +++++++++++++++++++ implementations/Crafter.Review-SeatPool.cpp | 4 + interfaces/Crafter.Review-Run.cppm | 249 +++ interfaces/Crafter.Review-SeatPool.cppm | 1 + interfaces/Crafter.Review.cppm | 1 + project.cpp | 6 +- tests/Drivers/main.cpp | 4 +- tests/OpenCode/main.cpp | 4 +- 9 files changed, 2639 insertions(+), 13 deletions(-) create mode 100644 implementations/Crafter.Review-Run.cpp create mode 100644 interfaces/Crafter.Review-Run.cppm diff --git a/implementations/Crafter.Review-Driver.cpp b/implementations/Crafter.Review-Driver.cpp index f680970..22b13ef 100644 --- a/implementations/Crafter.Review-Driver.cpp +++ b/implementations/Crafter.Review-Driver.cpp @@ -12,6 +12,9 @@ import :Config; import :Freeze; import :PgWire; import :Totp; +import :Plan; +import :OpenCode; +import :Run; import std; namespace fs = std::filesystem; @@ -51,12 +54,25 @@ inventory unit, against a frozen commit range.)"; } // ---- the option parser these commands use --------------------------------- + enum class Kind { + String, + Int, + Float, + }; + struct OptionSpec { std::string_view name; // A value follows (`--x V` or `--x=V`); else a store_true flag. bool takesValue; std::string_view metavar; std::string_view help; + // An integer or a float option: the value must parse as + // Dyn::ParseInt / Dyn::ParseFloat takes it. + Kind kind = Kind::String; + std::span choices = {}; + bool required = false; + // Left out of --help. + bool hidden = false; }; struct Parsed { @@ -78,7 +94,9 @@ inventory unit, against a frozen commit range.)"; std::string UsageLine(std::string_view prog, std::span specs) { std::string line = std::format("usage: {} [-h]", prog); for (const OptionSpec& spec : specs) { - line += spec.takesValue ? std::format(" [{} {}]", spec.name, spec.metavar) : std::format(" [{}]", spec.name); + if (spec.hidden) continue; + const std::string text = spec.takesValue ? std::format("{} {}", spec.name, spec.metavar) : std::string(spec.name); + line += spec.required ? std::format(" {}", text) : std::format(" [{}]", text); } return line; } @@ -86,6 +104,7 @@ inventory unit, against a frozen commit range.)"; std::string Help(std::string_view prog, std::string_view description, std::span specs) { std::string text = std::format("{}\n\n{}\n\noptions:\n -h, --help show this help message and exit\n", UsageLine(prog, specs), description); for (const OptionSpec& spec : specs) { + if (spec.hidden) continue; const std::string left = spec.takesValue ? std::format("{} {}", spec.name, spec.metavar) : std::string(spec.name); text += left.size() < 20 ? std::format(" {:<20} {}\n", left, spec.help) : std::format(" {}\n {}\n", left, spec.help); } @@ -142,24 +161,51 @@ inventory unit, against a frozen commit range.)"; parsed.flags.insert(key); continue; } + std::string value; if (equals != std::string_view::npos) { - parsed.values[key] = std::string(arg.substr(equals + 1)); + value = std::string(arg.substr(equals + 1)); } else if (i + 1 < args.size() && !(args[i + 1].starts_with('-') && args[i + 1].size() > 1)) { - parsed.values[key] = std::string(args[++i]); + value = std::string(args[++i]); } else { parsed.error = std::format("argument {}: expected one argument", spec.name); return parsed; } + if (spec.kind == Kind::Int && !Dyn::ParseInt(value)) { + parsed.error = std::format("argument {}: invalid int value: {}", spec.name, Dyn::Repr(value)); + return parsed; + } + if (spec.kind == Kind::Float && !Dyn::ParseFloat(value)) { + parsed.error = std::format("argument {}: invalid float value: {}", spec.name, Dyn::Repr(value)); + return parsed; + } + if (!spec.choices.empty() && !std::ranges::contains(spec.choices, value)) { + std::vector quoted; + for (const std::string_view c : spec.choices) quoted.push_back(Dyn::Repr(c)); + parsed.error = std::format("argument {}: invalid choice: {} (choose from {})", spec.name, Dyn::Repr(value), Dyn::Join(quoted, ", ")); + return parsed; + } + parsed.values[key] = std::move(value); } - if (!unrecognized.empty()) parsed.error = std::format("unrecognized arguments: {}", Dyn::Join(unrecognized, " ")); + if (!unrecognized.empty()) { + parsed.error = std::format("unrecognized arguments: {}", Dyn::Join(unrecognized, " ")); + return parsed; + } + std::vector missing; + for (const OptionSpec& spec : specs) { + if (spec.required && !parsed.values.contains(spec.name.substr(2))) missing.emplace_back(spec.name); + } + if (!missing.empty()) parsed.error = std::format("the following arguments are required: {}", Dyn::Join(missing, ", ")); return parsed; } // The consumer config, loaded before every command but `config`: - // warnings on stderr, a problem stops it. - std::optional LoadConfig(const std::optional& configPath) { + // warnings on stderr (`run` logs them with its other lines), a problem + // stops it. + std::optional LoadConfig(const std::optional& configPath, bool warn = true) { Config::Loaded loaded = Config::Load(fs::current_path(), configPath ? fs::path(*configPath) : fs::path()); - for (const Json& warning : Dyn::Or(Dyn::Get(loaded.config, "_warnings"), Dyn::EmptyList())) std::println(std::cerr, "config: warning: {}", warning.get()); + if (warn) { + for (const Json& warning : Dyn::Or(Dyn::Get(loaded.config, "_warnings"), Dyn::EmptyList())) std::println(std::cerr, "config: warning: {}", warning.get()); + } if (!loaded.problems.empty()) { Die(std::format("config: {}", Dyn::Join(loaded.problems, "; "))); return std::nullopt; @@ -179,6 +225,152 @@ inventory unit, against a frozen commit range.)"; {"--verify", false, "", "push-time check: HEAD, a clean tree and a fresh export still match freeze.json"}, }}; + constexpr std::array Modes = {"commits"}; + constexpr std::array Harnesses = {"claude", "opencode"}; + + const std::vector& RunOptions() { + static const std::string effort = std::format("reasoning effort of every lens process (default {})", Plan::Effort); + static const std::string reads = + std::format("clean reads a code unit of the static code lenses needs before its verdict is reused, one per round (default {})", Plan::Reads); + static const std::string models = [] { + std::vector keys; + for (const Plan::ModelTier& m : Plan::Models()) keys.push_back(m.tier); + return std::format("override the lens model table, e.g. fable=claude-fable-5-1 (keys: {})", Dyn::Join(keys, ",")); + }(); + static const std::string variant = std::format("opencode harness: the model's reasoning variant (default {})", OpenCode::DefaultVariant); + static const std::vector specs = { + {"--mode", true, "MODE", "", Kind::String, Modes, false, true}, + {"--artifact", true, "ARTIFACT", "the frozen artifact dir (.review-gates/push//)", Kind::String, {}, true}, + {"--hash", true, "HASH", ""}, + {"--dispositions", true, "DISPOSITIONS", ""}, + {"--base", true, "BASE", ""}, + {"--date", true, "DATE", ""}, + {"--trees", true, "TREES", "the checkout the lenses read, at the range HEAD: the repository itself (`.`, default) or the frozen worktree"}, + {"--context-paths", true, "CONTEXT_PATHS", "comma-separated paths of material the artifact discusses (captures, evidence) - mechanical pointers, no narrative"}, + {"--only", true, "ONLY", "comma-separated lens files (test only)"}, + {"--without", true, "WITHOUT", "comma-separated lens files to leave out of this run by owner ruling (logged; the run stays a full run)"}, + {"--units", true, "UNITS", "comma-separated unit ids (test only)"}, + {"--warm", false, "", "test runs only: let one pass per prompt prefix go first, as a full run does"}, + {"--concurrency", true, "CONCURRENCY", + "lens processes in flight at once across all seats (default 128, 2026-10-15; the pool still places each one by seat headroom, launches a pass only when it fits in free " + "memory at its environment's measured cost, and gives a pass with the browser a stack of its own)", + Kind::Int}, + {"--effort", true, "EFFORT", effort}, + {"--reads", true, "READS", reads, Kind::Int}, + {"--no-reuse", false, "", ""}, + {"--no-salvage", false, "", "do not reuse finished process outputs of an earlier run on this hash"}, + {"--dry-run", false, "", ""}, + {"--claude-bin", true, "CLAUDE_BIN", ""}, + {"--models", true, "MODELS", models}, + {"--scratch", true, "SCRATCH", ""}, + {"--seats", true, "SEATS", "the seats the lenses run on: auto (every Anthropic seat in the env file, else local), local, api, oauth, openrouter (never in auto), or a list like api1,1,2,local"}, + {"--env-file", true, "ENV_FILE", "dotenv file with CLAUDE_CODE_OAUTH_TOKEN1..6 / ANTHROPIC_API_KEY1..6 (default $SEATS_ENV_FILE, else the repository .env)"}, + {"--no-probe", false, "", "skip the one-request probe of every seat at the start"}, + {"--api-budget", true, "API_BUDGET", "USD an API-credit seat may spend per cycle (default REVIEW_GATES_API_BUDGET_USD or 200)", Kind::Float}, + {"--harness", true, "{claude,opencode}", + "the agent harness of every lens process: claude (default) or opencode (OpenRouter seats only; no lens whose environment needs an MCP server)", Kind::String, Harnesses}, + {"--variant", true, "VARIANT", variant}, + {"--opencode-bin", true, "OPENCODE_BIN", ""}, + }; + return specs; + } + + std::int32_t RunCommand(std::span args, const std::optional& configPath) { + constexpr std::string_view Prog = "crafter-review run"; + const std::vector& specs = RunOptions(); + const Parsed parsed = ParseOptions(args, specs); + if (parsed.help) { + std::print("{}", Help(Prog, "one claude process per inventory unit per lens, the gates, the dump (process/prepush.md)", specs)); + return 0; + } + if (!parsed.error.empty()) return UsageError(UsageLine(Prog, specs), Prog, parsed.error); + const std::optional config = LoadConfig(configPath, false); + if (!config) return 3; + Run::Arguments a; + a.config = configPath; + a.mode = parsed.Value("mode").value_or("commits"); + a.artifact = parsed.Value("artifact").value_or(""); + a.hash = parsed.Value("hash"); + a.dispositions = parsed.Value("dispositions"); + a.base = parsed.Value("base").value_or(""); + a.date = parsed.Value("date").value_or(Platform::TodayIso()); + a.trees = parsed.Value("trees"); + a.contextPaths = parsed.Value("context-paths").value_or(""); + a.only = parsed.Value("only"); + a.without = parsed.Value("without"); + if (const std::optional units = parsed.Value("units"); units && !units->empty()) { + // A trailing comma names no unit. + a.units = std::set(); + for (const std::string& part : Dyn::Split(*units, ",")) { + if (std::string id = Dyn::Strip(part); !id.empty()) a.units->insert(std::move(id)); + } + } + a.warm = parsed.Flag("warm"); + if (const std::optional n = parsed.Value("concurrency")) a.concurrency = *Dyn::ParseInt(*n); + a.effort = parsed.Value("effort").value_or(""); + if (a.effort.empty()) a.effort = std::string(Plan::DefaultEffort()); + a.noReuse = parsed.Flag("no-reuse"); + a.noSalvage = parsed.Flag("no-salvage"); + a.dryRun = parsed.Flag("dry-run"); + a.claudeBin = parsed.Value("claude-bin"); + a.models = parsed.Value("models").value_or(""); + a.scratch = parsed.Value("scratch"); + a.seats = parsed.Value("seats"); + if (!a.seats) a.seats = Platform::GetEnv("REVIEW_GATES_SEATS"); + a.envFile = parsed.Value("env-file"); + a.noProbe = parsed.Flag("no-probe"); + if (const std::optional budget = parsed.Value("api-budget")) a.apiBudget = *Dyn::ParseFloat(*budget); + a.harness = parsed.Value("harness").value_or("claude"); + a.variant = parsed.Value("variant"); + if (!a.variant || a.variant->empty()) a.variant = a.harness == "opencode" ? std::optional(std::string(OpenCode::DefaultVariant)) : std::nullopt; + a.opencodeBin = parsed.Value("opencode-bin"); + if (const std::optional reads = parsed.Value("reads")) { + const std::int64_t n = *Dyn::ParseInt(*reads); + if (n < 1) return Die("--reads: a number from 1 up"); + a.reads = n; + } + for (const std::string& part : Dyn::Split(a.models, ",")) { + const std::string kv = Dyn::Strip(part); + if (kv.empty()) continue; + const std::size_t equals = kv.find('='); + if (equals == std::string::npos) throw Dyn::Error("ValueError", "not enough values to unpack (expected 2, got 1)"); + const std::string key = kv.substr(0, equals); + std::vector& tiers = Plan::Models(); + const auto tier = std::ranges::find_if(tiers, [&](const Plan::ModelTier& m) { return m.tier == key; }); + if (tier == tiers.end()) { + std::vector known; + for (const Plan::ModelTier& m : tiers) known.push_back(m.tier); + std::println(std::cerr, "--models: unknown key {} (known: {})", Dyn::Repr(key), Dyn::Join(known, ", ")); + return 1; + } + tier->id = kv.substr(equals + 1); + } + return Run::Main(std::move(a), *config); + } + + constexpr std::array SlotsOptions = {{ + {"--artifact", true, "ARTIFACT", "a frozen artifact dir: no more slots than its browser passes can use at once"}, + }}; + + std::int32_t SlotsCommand(std::span args, const std::optional& configPath) { + constexpr std::string_view Prog = "crafter-review slots"; + const Parsed parsed = ParseOptions(args, SlotsOptions); + if (parsed.help) { + std::print("{}", Help(Prog, "how many stacks (slots.*) fit in free memory, and the commands that bring up the missing ones", SlotsOptions)); + return 0; + } + if (!parsed.error.empty()) return UsageError(UsageLine(Prog, SlotsOptions), Prog, parsed.error); + const std::optional config = LoadConfig(configPath); + if (!config) return 3; + const std::optional artifact = parsed.Value("artifact"); + try { + Run::SlotsPlan(*config, fs::current_path(), artifact ? std::optional(*artifact) : std::nullopt); + } catch (const Inventory::InvariantError& e) { + return Die(e.what()); + } + return 0; + } + std::int32_t FreezeCommand(std::span args, const std::optional& configPath) { constexpr std::string_view Prog = "crafter-review freeze"; const Parsed parsed = ParseOptions(args, FreezeOptions); @@ -215,6 +407,10 @@ inventory unit, against a frozen commit range.)"; // command anyone types. if (!rest.empty() && rest.front() == Process::SuperviseVerb) return Process::SuperviseMain(rest.subspan(1)); Platform::Init(); + // Before any other thread: SIGINT, SIGTERM and SIGHUP go to a thread of + // their own, which stops a run once it is under way and otherwise ends + // the process. + Process::InstallSignalThread(Run::OnSignal); // The options before the command: --config FILE or --config=FILE, and // any unique prefix of a long option. std::optional configPath; @@ -259,6 +455,8 @@ inventory unit, against a frozen commit range.)"; return Config::Main(fs::current_path(), configPath ? fs::path(*configPath) : fs::path()); } if (command == "freeze") return FreezeCommand(after, configPath); + if (command == "run") return RunCommand(after, configPath); + if (command == "slots") return SlotsCommand(after, configPath); } catch (const Dyn::Error& error) { // An error no command catches: its message, exit 1. std::println(std::cerr, "crafter-review: {}: {}", error.Kind(), error.what()); diff --git a/implementations/Crafter.Review-Run.cpp b/implementations/Crafter.Review-Run.cpp new file mode 100644 index 0000000..2ee647f --- /dev/null +++ b/implementations/Crafter.Review-Run.cpp @@ -0,0 +1,2171 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +module; +#include +#include + +module Crafter.Review:Run_impl; +import :Run; +import :Json; +import :Text; +import :Core; +import :Platform; +import :Process; +import :Http; +import :Schema; +import :Plan; +import :Artifact; +import :Inventory; +import :Dispositions; +import :State; +import :Normalize; +import :Stream; +import :Prompt; +import :Config; +import :Memory; +import :PgWire; +import :Freeze; +import :Seats; +import :SeatPool; +import :OpenCode; +import :Pool; +import std; + +namespace fs = std::filesystem; + +namespace Crafter::Review::Run { + namespace { + [[noreturn]] void Die(const std::string& message) { + throw Preflight(message); + } + + void WriteText(const fs::path& path, std::string_view text) { + std::ofstream file(path, std::ios::binary | std::ios::trunc); + file << text; + } + + // A temp file renamed into place: what a reader opens is whole. + void WriteReplace(const fs::path& path, std::string_view text) { + const fs::path temp = std::format("{}.tmp", path.string()); + WriteText(temp, text); + fs::rename(temp, path); + } + + std::string Basename(const fs::path& path) { + return Dyn::BaseName(path.string()); + } + + 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; + } + + // A number or a numeric string, as an integer. + std::int64_t Int(const Json& value) { + if (value.is_number_integer()) return value.get(); + if (value.is_number()) return static_cast(value.get()); + if (value.is_boolean()) return value.get() ? 1 : 0; + if (value.is_string()) { + if (const std::optional parsed = Dyn::ParseInt(value.get_ref())) return *parsed; + throw Dyn::Error("ValueError", Dyn::IntError(value.get_ref())); + } + throw Dyn::Error("TypeError", std::format("int() argument must be a string, a bytes-like object or a real number, not '{}'", Dyn::TypeName(value))); + } + + // Floor division of floats through fmod, exact where a plain quotient + // would round, and with the sign of zero kept. + double FloorDiv(double vx, double wx) { + double mod = std::fmod(vx, wx); + double div = (vx - mod) / wx; + if (mod != 0 && ((wx < 0) != (mod < 0))) { + mod += wx; + div -= 1.0; + } + if (div != 0) { + double floordiv = std::floor(div); + if (div - floordiv > 0.5) floordiv += 1.0; + return floordiv; + } + return std::copysign(0.0, vx / wx); + } + + std::vector Strings(const Json& array) { + std::vector out; + for (const Json& item : Dyn::Or(array, Dyn::EmptyList())) out.push_back(Dyn::Str(item)); + return out; + } + + Json Array(std::span values) { + Json out = Json::array(); + for (const std::string& v : values) out.push_back(v); + return out; + } + + // The text a JSON object key is written as. + std::string KeyText(const Json& key) { + return key.is_string() ? key.get() : DumpCompact(key); + } + + // [x.strip() for x in text.split(',') if x.strip()] + std::vector CommaList(std::string_view text) { + std::vector out; + for (const std::string& part : Dyn::Split(text, ",")) { + std::string item = Dyn::Strip(part); + if (!item.empty()) out.push_back(std::move(item)); + } + return out; + } + + // label.replace('/', '_').replace(' ', '_') + std::string FileLabel(const Json& job) { + return Dyn::Replace(Dyn::Replace(Dyn::Str(Dyn::At(job, "label")), "/", "_"), " ", "_"); + } + + // The run a signal or `crafter-review stop` reaches. + std::atomic Current = nullptr; + + // The default: the process ends; what it started goes with it. + [[noreturn]] void EndNow(std::int32_t signal) { + Process::KillDescendants(); + std::_Exit(128 + signal); + } + + void Trigger(const std::string& why, std::int32_t signal) { + Run* run = Current.load(); + if (!run || !run->Ready() || signal == SIGHUP) EndNow(signal ? signal : SIGTERM); + run->Stop(why, signal); + } + + std::string ScratchRoot(const std::optional& given) { + if (given && !given->empty()) return *given; + if (const std::optional configured = Platform::GetEnv("REVIEW_GATES_SCRATCH"); configured && !configured->empty()) return *configured; + return (fs::path(Platform::Home()) / ".cache" / "review-gates").string(); + } + + std::map DefaultCosts() { + std::map out; + for (const Plan::Environment& e : Plan::Environments) out[std::string(e.name)] = e.memMb; + return out; + } + + std::vector ModelIds() { + std::vector out; + for (const Plan::ModelTier& m : Plan::Models()) { + if (!std::ranges::contains(out, m.id)) out.push_back(m.id); + } + return out; + } + } + + void OnSignal(std::int32_t signal) { + Trigger(std::format("signal {}", signal), signal); + } + + Json ArgsJson(const Arguments& a) { + auto opt = [](const std::optional& v) { return v ? Json(*v) : Json(); }; + Json units; + if (a.units) { + units = Json::array(); + for (const std::string& u : *a.units) units.push_back(u); + } + return { + {"config", opt(a.config)}, + {"mode", a.mode}, + {"artifact", a.artifact}, + {"hash", opt(a.hash)}, + {"dispositions", opt(a.dispositions)}, + {"base", a.base}, + {"date", a.date}, + {"trees", opt(a.trees)}, + {"context_paths", a.contextPaths}, + {"only", opt(a.only)}, + {"without", opt(a.without)}, + {"units", units}, + {"warm", a.warm}, + {"concurrency", a.concurrency}, + {"effort", a.effort}, + {"reads", a.reads}, + {"no_reuse", a.noReuse}, + {"no_salvage", a.noSalvage}, + {"dry_run", a.dryRun}, + {"claude_bin", opt(a.claudeBin)}, + {"models", a.models}, + {"scratch", opt(a.scratch)}, + {"seats", opt(a.seats)}, + {"env_file", opt(a.envFile)}, + {"no_probe", a.noProbe}, + {"api_budget", a.apiBudget ? Json(*a.apiBudget) : Json()}, + {"harness", a.harness}, + {"variant", opt(a.variant)}, + {"opencode_bin", opt(a.opencodeBin)}, + }; + } + + std::pair AppReady(std::string_view url, std::string_view path, double timeout) { + const std::string full = std::format("{}/{}", Dyn::RStrip(url, "/"), Dyn::LStrip(path.empty() ? std::string_view("/") : path, "/")); + try { + const Http::Response r = Http::UrlOpen(full, {{"User-Agent", "review-gates"}}, timeout); + return {r.status < 500, std::format("HTTP {} from {}", r.status, full)}; + } catch (const Http::HttpError& e) { + return {e.Code() < 500, std::format("HTTP {} from {}", e.Code(), full)}; + } catch (const std::exception& e) { + return {false, std::format("{}: {}", full, e.what())}; + } + } + + std::pair> SlotsPlan(const Json& cfg, const fs::path& repo, const std::optional& artifact, const std::optional& scratchRoot, const Log& given) { + const Log log = given ? given : Log([](const std::string& line) { std::println("{}", line); }); + const Json& sl = cfg.at("slots"); + if (!Dyn::Truthy(Dyn::Get(sl, "dir"))) { + log("slots: the config names no slots.dir - the one stack of functional.* and supabase.*, one browser pass at a time on it"); + return {1, {}}; + } + const Config::Slots slotted = Config::SlotStacks(cfg, repo); + const Json& up = slotted.stacks; + for (const std::string& x : slotted.problems) log(std::format("slots: {}", x)); + const std::vector off = Config::LensesOff(cfg); + const Json envs = Config::Environments(cfg); + std::set appEnvs; + for (auto it = envs.begin(); it != envs.end(); ++it) { + const Plan::Environment* e = Plan::FindEnvironment(Dyn::Str(it.value())); + if (!std::ranges::contains(off, it.key()) && e && e->app) appEnvs.insert(Dyn::Str(it.value())); + } + if (appEnvs.empty()) { + log("slots: no lens that runs drives the application - no slot needed"); + return {0, {}}; + } + const fs::path root = scratchRoot && !scratchRoot->empty() ? *scratchRoot : fs::path(ScratchRoot(std::nullopt)); + const Memory::Calibration calib(root / "memory.json", repo, DefaultCosts()); + std::optional best; + for (const std::string& e : appEnvs) { + Memory::Calibration::Cost cost = calib.CostOf(e); + if (!best || Number(cost.megabytes) > Number(best->megabytes)) best = std::move(cost); + } + const double passMb = Number(best->megabytes); + const std::optional avail = Memory::Available(); + const std::int64_t reserve = Memory::Reserve(); + const auto upCount = static_cast(up.size()); + if (!avail) { + log("slots: no /proc/meminfo - cannot size the slots by memory"); + return {std::max(1, upCount), {}}; + } + const double room = static_cast(*avail - reserve) - static_cast(upCount) * passMb; + const std::int64_t more = std::max(0, static_cast(FloorDiv(room, Number(sl.at("memMb")) + passMb))); + std::int64_t want = upCount + more; + std::optional planned; + if (artifact) { + const Inventory::Built built = Inventory::Build("commits", artifact->string()); + const std::vector exclude = Strings(cfg.at("functional").at("exclude")); + std::int64_t most = 0; + for (auto it = envs.begin(); it != envs.end(); ++it) { + const Plan::Environment* e = Plan::FindEnvironment(Dyn::Str(it.value())); + if (std::ranges::contains(off, it.key()) || !e || !e->app) continue; + const Plan::LensPlan plan = Plan::PlanLens(it.key(), built.out.at("units"), built.lines, {}, exclude); + most = std::max(most, std::ranges::count_if(plan.blocks, [](const Json& b) { return b.at("kind") == "block"; })); + } + planned = most; + want = std::min(want, std::max(1, *planned)); + } + log(std::format("memory: {} MB free, reserve {} MB; a slot {} MB (slots.memMb), a pass with the browser ~{:.0f} MB ({})", *avail, reserve, Dyn::Str(sl.at("memMb")), passMb, best->source)); + std::vector ids; + for (const Json& s : up) ids.push_back(Dyn::Str(s.at("id"))); + log(std::format("slots up: {}{}; room for {} more by memory{}", upCount, ids.empty() ? std::string() : std::format(" ({})", Dyn::Join(ids, ", ")), more, + planned ? std::format("; the range has {} browser block pass(es) to run at once at most", *planned) : std::string())); + std::set have; + for (const std::string& id : ids) { + const std::vector words = Dyn::Split(id); + if (words.empty()) continue; + if (const std::optional n = Dyn::ParseInt(words.back()); n && std::ranges::all_of(Dyn::Decode(words.back()), Dyn::IsDecimal)) have.insert(*n); + } + std::vector fresh; + for (std::int64_t n = 1; static_cast(have.size() + fresh.size()) < want; ++n) { + if (!have.contains(n)) fresh.push_back(n); + } + const std::vector cmds = Config::SlotCommands(cfg, "start", fresh); + if (!fresh.empty() && cmds.empty()) { + std::vector numbers; + for (const std::int64_t n : fresh) numbers.push_back(std::to_string(n)); + log(std::format("slots: the config names no slots.start - bring up slots {} by hand", Dyn::Join(numbers, ", "))); + } + const Json& startMb = Dyn::Get(sl, "startMb"); + std::int64_t batch = 1; + if (Dyn::Truthy(startMb)) { + const double free = std::max(0.0, static_cast(*avail - reserve) - static_cast(upCount) * passMb); + batch = std::max(1, std::min(fresh.empty() ? 1 : static_cast(fresh.size()), static_cast(FloorDiv(free, Number(startMb))))); + } + if (!fresh.empty()) { + log(std::format("slots: bring up {} more -> {}, at most {} at once ({}", fresh.size(), want, batch, + Dyn::Truthy(startMb) ? std::format("slots.startMb {} MB each while it comes up)", Dyn::Str(startMb)) : std::string("no slots.startMb: one at a time)"))); + } else { + log(std::format("slots: {} up, nothing more to bring up", upCount)); + } + for (const std::string& c : cmds) log(std::format(" {}", c)); + return {want, cmds}; + } + + // ---- the Run --------------------------------------------------------------------------- + Run::Run() : repo(fs::current_path()), t0(Platform::Now()) { + log = [this](const std::string& message) { Message(message); }; + } + + Run::Run(Arguments given, Json config) : Run() { + a = std::move(given); + cfg = std::move(config); + // The lens files the config turns off, and the environment of each. + off = Config::LensesOff(cfg); + environments = Config::Environments(cfg); + } + + Run::~Run() { + // atexit's board.leave: everything of this driver off the board. + if (board) board->Leave(); + if (broker) broker->Close(); + } + + void Run::Message(const std::string& message) { + const std::string line = std::format("{} {}", Platform::LocalTime("%H:%M:%S"), message); + const std::lock_guard guard(logLock); + logs.push_back(line); + std::println("{}", line); + std::fflush(stdout); + if (logFile) { + std::ofstream file(*logFile, std::ios::binary | std::ios::app); + file << line << '\n'; + } + } + + bool Run::Ready() const { + return ready.load(); + } + + std::optional Run::StopFile() const { + const std::lock_guard guard(shared); + return stopFile; + } + + void Run::Stop(const std::string& why, std::int32_t signal) { + const std::int32_t n = runner ? runner->KillChildren() : 0; + const std::string reason = signal ? std::format("stopped by signal {} ({} running processes killed)", signal, n) : std::format("stopped by {} ({} running processes killed)", why, n); + { + const std::lock_guard guard(shared); + stopped = reason; + if (sched) sched->Abort(reason); + } + log(signal ? std::format("signal {}: {} running processes killed - aborting", signal, n) : std::format("{}: {} running processes killed - aborting", why, n)); + } + + // ---- Gate 0 ------------------------------------------------------------------------- + void Run::Gate0() { + if (!fs::exists(a.artifact)) Die(std::format("artifact {} does not exist", a.artifact)); + hash = Artifact::ArtifactHash("commits", a.artifact); + if (a.hash && !a.hash->empty() && *a.hash != hash) Die(std::format("--hash {} does not match the artifact ({})", Dyn::Slice(*a.hash, 0, 12), hash.substr(0, 12))); + h8 = hash.substr(0, 8); + const std::optional patchHead = Artifact::PatchHead(a.artifact); + if (!patchHead || patchHead->empty()) Die("the last patch has no `From ` line - not a git format-patch export (use `freeze`)"); + head = *patchHead; + for (const std::string& t : trees) { + if (!Artifact::IsToplevel(t)) Die(std::format("--trees: {} is not the toplevel of a git checkout", t)); + } + if (const std::optional drift = TreeDrift()) Die(std::format("{} - re-freeze, or check out the frozen worktree (freeze --worktree)", *drift)); + gateDir = Dyn::JoinPath(a.artifact, "gate-state"); + fs::create_directories(gateDir); + Inventory::Built built; + try { + built = Inventory::Build("commits", a.artifact); + } catch (const Inventory::InvariantError& e) { + Die(e.what()); + } + units = built.out.at("units"); + lines = built.lines; + const fs::path invPath = gateDir / std::format("inventory-{}.json", h8); + if (fs::exists(invPath)) { + const Json prev = Dyn::LoadJsonFile(invPath); + auto shape = [](const Json& list) { + Json out = Json::array(); + for (const Json& u : list) out.push_back(Json::array({u.at("id"), u.at("lines")})); + return out; + }; + if (!Dyn::Equal(shape(prev.at("units")), shape(units))) Die(std::format("stored inventory {} differs from the partition of this artifact", invPath.string())); + } else { + WriteText(invPath, DumpFile(built.out)); + } + testMode = (a.only && !a.only->empty()) || (a.units && !a.units->empty()); + // One full run per hash (2026-10-18): a second would number itself the + // same, clobber the resume file and buy every pass twice. The lock is + // the kernel's: a killed driver releases it, no lens child inherits it. + // A test or dry run takes none but numbers itself past a live holder. + Json holder; + if (!a.dryRun && !testMode) { + try { + runLock = std::make_unique(gateDir, "commits", h8); + } catch (const State::RunLocked& e) { + const std::string how = Platform::InSandbox() ? std::format("wait for it or stop it (`{} stop `)", Invocation()) : std::string("wait for it or kill it"); + Die(std::format("a full run on this hash is already live in {}: {} - {}; a dead holder releases the lock", Dyn::RelPath(gateDir.string(), repo.string()), e.what(), how)); + } + } else { + const std::optional found = State::LockHolder(gateDir, "commits", h8); + if (found) holder = *found; + } + std::int64_t runN = State::NextRunNumber(gateDir, "commits", h8); + if (Dyn::Truthy(holder) && Dyn::Truthy(Dyn::Get(holder, "run_n"))) runN = std::max(runN, Int(holder.at("run_n")) + 1); + const fs::path scratchRoot = ScratchRoot(a.scratch); + // A run killed before its first checkpoint left no dump but did leave + // outputs: never reuse its scratch dir (salvage reads it); and never + // share one with a live driver (the .driver marker). + auto hasOutputs = [](const fs::path& dir) { + std::error_code error; + for (fs::directory_iterator it(dir / "out", error), end; !error && it != end; it.increment(error)) { + const std::string name = it->path().filename().string(); + if (!name.starts_with('.') && name.ends_with(".jsonl")) return true; + } + return false; + }; + while (!a.dryRun) { + const fs::path candidate = scratchRoot / std::format("{}-run-{}", h8, runN); + if (!hasOutputs(candidate) && SeatPool::ClaimScratch(candidate)) break; + ++runN; + } + scratch = scratchRoot / std::format("{}-run-{}{}", h8, runN, a.dryRun ? "-dry" : ""); + fs::create_directories(scratch); + if (!a.dryRun) { + // A stop meant for a run this dir held before is not this run's. + std::error_code error; + fs::remove(scratch / "stop", error); + // The marker a later run reads before salvaging this dir's outputs. + Json marker = Json::object(); + for (auto it = environments.begin(); it != environments.end(); ++it) marker[Plan::LensName(it.key())] = it.value(); + WriteText(scratch / "environments", DumpSpaced(marker)); + } + { + const std::lock_guard guard(shared); + stopFile = scratch / "stop"; + } + dumpPath = a.dryRun ? scratch / "dry-run.json" : gateDir / std::format("commits-{}-run-{}{}.json", h8, runN, testMode ? "-test" : ""); + if (runLock) { + runLock->WriteInfo({{"pid", Platform::PublishedPid()}, + {"since", Platform::LocalTime("%Y-%m-%dT%H:%M:%S")}, + {"artifact", Dyn::RelPath(a.artifact, repo.string())}, + {"scratch", scratch.string()}, + {"run_n", runN}}); + } + logFile = scratch / "log.txt"; + log(std::format("crafter-review {} prompt v{}: commits {} @ {}", Version(), PromptVersion, Dyn::RelPath(a.artifact, repo.string()), hash)); + log(std::format("config: {}", Dyn::Truthy(Dyn::Get(cfg, "_path")) ? Dyn::Str(cfg.at("_path")) : std::string("defaults (no .review-gates/config.json)"))); + for (const Json& w : Dyn::Or(Dyn::Get(cfg, "_warnings"), Dyn::EmptyList())) log(std::format("config: warning: {}", Dyn::Str(w))); + std::vector lensParts; + for (const Plan::Lens& lens : Plan::Lenses) { + const std::string lf(lens.file); + const std::string name = Plan::LensName(lf); + lensParts.push_back(std::ranges::contains(off, lf) ? std::format("{} OFF (lenses.{}.run = false)", name, name) : std::format("{} ({})", name, Dyn::Str(Dyn::At(environments, lf)))); + } + log(std::format("lenses: {}", Dyn::Join(lensParts, ", "))); + std::vector running; + for (const Plan::Lens& lens : Plan::Lenses) { + if (!std::ranges::contains(off, std::string(lens.file))) running.emplace_back(lens.file); + } + const std::vector exclude = Strings(cfg.at("functional").at("exclude")); + notReviewed = Plan::NotReviewed(units, lines, running, exclude); + log(std::format("inventory: {} units ({}); {} units no lens that runs reads (text, documentation, the units of a lens the config turns off)", units.size(), Dyn::RelPath(invPath.string(), repo.string()), + notReviewed.size())); + if (a.units) { + std::set known; + for (const Json& u : units) known.insert(Dyn::Str(u.at("id"))); + std::vector unknown; + for (const std::string& u : *a.units) { + if (!known.contains(u)) unknown.push_back(u); + } + if (!unknown.empty()) Die(std::format("--units names ids not in the inventory: {}", Dyn::Join(unknown, ", "))); + } + dispEntries = Dispositions::Load(a.dispositions.value_or("")); + dispRows = Dispositions::Classify(dispEntries, hash, a.artifact); + inForce = Dispositions::InForce(dispEntries, dispRows); + std::vector rowParts; + for (const Json& r : dispRows) rowParts.push_back(std::format("{}:{}", Dyn::Str(r.at("id")), Dyn::Str(r.at("status")))); + log(std::format("dispositions: {} in force of {} ({})", inForce.size(), dispEntries.size(), rowParts.empty() ? std::string("none") : Dyn::Join(rowParts, ", "))); + statePath = gateDir / "unit-state.json"; + state = a.noReuse ? State::EmptyState() : State::LoadState(statePath); + salvageDirs.clear(); + if (Dyn::Truthy(Dyn::Get(state, "entries"))) log(std::format("unit-state: {} stored unit results available for reuse", state.at("entries").size())); + resumePath = gateDir / std::format("resume-{}.json", h8); + resume = nullptr; + if (!testMode && !a.dryRun && !a.noReuse && fs::exists(resumePath)) { + try { + const Json r = Dyn::LoadJsonFile(resumePath); + if (Dyn::Get(r, "hash") == hash && Dyn::Get(r, "results").is_array() && Dyn::Truthy(r.at("results"))) { + resume = r; + std::vector names; + for (const Json& x : r.at("results")) names.push_back(Dyn::Str(Dyn::At(x, "lens"))); + log(std::format("resume: {} lens result(s) from {} on this hash ({}) - fully returned lenses are re-normalised, not re-run", r.at("results").size(), Dyn::Str(Dyn::Get(r, "run")), + Dyn::Join(names, ", "))); + } + } catch (const std::exception& e) { + log(std::format("resume: {} unreadable ({}) - ignored", resumePath.string(), e.what())); + } + } + x = a.reads; + if (x > 1) { + Json prev; + try { + if (fs::exists(resumePath)) prev = Dyn::LoadJsonFile(resumePath); + } catch (const std::exception&) { + prev = nullptr; + } + const bool sameHash = Dyn::Truthy(prev) && Dyn::Get(prev, "hash") == hash; + if (sameHash) { + newRound = Dyn::Truthy(Dyn::Get(prev, "complete")); + const Json& r = Dyn::Get(prev, "round"); + round = (Dyn::Truthy(r) ? Int(r) : 1) + (newRound ? 1 : 0); + } + if (!a.dryRun) WriteText(scratch / "round", std::to_string(round)); + log(std::format("reads: every code unit of the static code lenses needs {} clean reads, one per round; this is round {} on this hash{}", x, round, + newRound || !sameHash ? std::string() : std::string(" (continuing an unfinished run)"))); + } + for (const Plan::Lens& lens : Plan::Lenses) { + const std::string lf(lens.file); + charters[lf] = Prompt::Charter(lf); + charterSha[lf] = Plan::Sha(charters[lf]); + } + systemText = Prompt::SystemText(); + systemPaths.clear(); + systemByLens.clear(); + runLayer = Prompt::RunLayer(Dyn::RelPath(a.artifact, repo.string()), a.base, a.date, trees, hash, lines, units, contextPaths, Basename(repo)); + // The application under test: the login fixture, the TOTP helper. + const Json& fcfg = cfg.at("functional"); + loginFixture = Config::LoginFixture(cfg, repo); + stacks = LoadStacks(); + const Json first = *stacks->First(); + appBlock = Prompt::AppBlock(fcfg, loginFixture, Pool::TotpCommand(loginFixture), slotMode); + schemas = {{"block", Schema::SchemaJson("block")}, {"whole", Schema::SchemaJson("whole")}}; + // Memory: every pass's tree measured, its peak recorded per repository + // and environment; the budget per gate. + calibration = std::make_unique(scratchRoot / "memory.json", repo, DefaultCosts()); + if (!a.dryRun) { + // A pass runs under the supervisor: its tree is claude's, the + // supervisor itself left out, so a peak measures claude's tree + // alone. + sampler = std::make_unique(3.0, [](std::int64_t pid, const std::map>& kids) { + std::vector tree = Platform::Tree(pid, kids); + std::erase(tree, pid); + return Platform::PssMb(tree); + }); + } + // The database URLs stay here: a db pass gets the broker's address and + // its stack's token. + bool anyDb = false; + for (auto it = environments.begin(); it != environments.end(); ++it) { + const Plan::Environment* e = Plan::FindEnvironment(Dyn::Str(it.value())); + anyDb = anyDb || (e && e->db && !std::ranges::contains(off, it.key())); + } + if (!a.dryRun && anyDb) { + broker = std::make_unique(); + broker->Start(); + } + std::map environmentMap; + for (auto it = environments.begin(); it != environments.end(); ++it) environmentMap[it.key()] = Dyn::Str(it.value()); + const Json& firstLogin = Dyn::Get(first, "login"); + const Json& firstDb = Dyn::Get(first, "dbUrl"); + runner = std::make_unique(Pool::RunnerOptions { + .repo = repo, + .scratch = scratch, + .bin = a.claudeBin.value_or(""), + .effort = a.effort, + .log = log, + .dryRun = a.dryRun, + .harness = a.harness, + .variant = a.variant, + .opencodeBin = a.opencodeBin.value_or(""), + .functional = fcfg, + .loginFixture = firstLogin.is_null() ? std::nullopt : std::optional(Dyn::Str(firstLogin)), + .environments = environmentMap, + .dbUrl = Dyn::Truthy(firstDb) ? std::optional(Dyn::Str(firstDb)) : std::nullopt, + .sampler = sampler.get(), + .broker = broker.get(), + }); + const std::map processEnv = Platform::Environment(); + apiCfg = SeatPool::ApiConfig(processEnv); + if (a.apiBudget) apiCfg["budgetUsd"] = *a.apiBudget; + ledger = std::make_unique(SeatPool::Ledger::Load(SeatPool::LedgerPath(scratchRoot), apiCfg)); + std::optional envFile = a.envFile; + if (!envFile) { + if (const std::optional configured = Platform::GetEnv("SEATS_ENV_FILE"); configured && !configured->empty()) { + envFile = configured; + } else if (fs::exists(repo / ".env")) { + envFile = (repo / ".env").string(); + } + } + try { + seats = SeatPool::LoadSeats(processEnv, envFile && !envFile->empty() ? std::optional(*envFile) : std::nullopt, a.seats.value_or(""), ledger.get(), apiCfg); + } catch (const std::invalid_argument& e) { + Die(e.what()); + } catch (const Dyn::Error& e) { + if (e.Kind() != "FileNotFoundError" && e.Kind() != "PermissionError" && e.Kind() != "IsADirectoryError" && e.Kind() != "OSError") throw; + Die(e.what()); + } + // The board (2026-10-31): the other drivers on this machine, what they + // hold on these seats, the memory and the stacks their passes hold - + // the pool places against all of it; a dry run launches nothing and + // stays off it. + if (!a.dryRun) { + const std::optional self = Platform::Self(); + if (!self) Die("no identity another driver could check (no bwrapinfo.json in this sandbox, as under `flatpak build`): a full run needs one - run it with `flatpak run`"); + try { + lifetime = std::make_unique(SeatPool::BoardPath().parent_path() / "drivers", *self); + } catch (const std::runtime_error& e) { + Die(e.what()); + } + const Json pc = SeatPool::PoolConfig(processEnv); + board = std::make_unique(SeatPool::BoardPath(), std::string(), std::nullopt, log); + board->Join({{"run", std::format("commits {} @ {}", Basename(Dyn::RStrip(a.artifact, "/")), h8)}, + {"dump", Basename(dumpPath)}, + {"scratch", scratch.string()}, + {"effort", a.effort}, + {"settings", {{"maxUtil", pc.at("maxUtil")}, {"apiConcurrency", apiCfg.at("concurrency")}, {"lensCostPct", pc.at("lensCostPct")}, {"concurrency", a.concurrency}}}}, + Platform::Now()); + } + const bool openrouter = std::ranges::any_of(seats, [](const Json& s) { return s.at("kind") == "openrouter"; }); + if (a.harness == "opencode" && !openrouter) Die("--harness opencode runs on OpenRouter seats only: add --seats openrouter (OPENROUTER_KEY in the env file)"); + const std::vector modelIds = ModelIds(); + if (openrouter && !std::ranges::all_of(Plan::Models(), [](const Plan::ModelTier& m) { return m.id.contains('/'); })) { + std::vector sorted = modelIds; + std::ranges::sort(sorted); + Die(std::format("OpenRouter seats need OpenRouter model ids (vendor/model): --models opus=...,fable=... (now {})", Dyn::Join(sorted, ", "))); + } + if (!a.dryRun && !a.noProbe) SeatPool::Probe(seats, runner->Bin(), 30, modelIds, log); + if (board) { + // Our readings onto the board (a probed-ok seat clears a verdict + // another driver left), theirs into our seats. + const double now = Platform::Now(); + board->Publish(seats, now, !a.noProbe); + board->Fold(seats, now); + for (const std::string& line : board->OthersLines(seats, now, SeatPool::PoolConfig(processEnv), apiCfg, a.concurrency)) log(line); + } + if (a.harness == "opencode") { + runner->opencodeHome = OpenCode::PrepareHome(scratch, repo, modelIds); + log(std::format("harness: opencode ({}), variant {}, routing {}; home {} (view dir = the project: no CLAUDE.md reaches the model)", a.opencodeBin.value_or(OpenCode::DefaultBin()), + a.variant.value_or("None"), DumpSpaced(OpenCode::Routing()), Dyn::RelPath(runner->opencodeHome.string(), repo.string()))); + if (!a.dryRun) { + const Json& keySeat = *std::ranges::find_if(seats, [](const Json& s) { return s.at("kind") == "openrouter"; }); + const auto [ok, detail] = OpenCode::Warm(a.opencodeBin.value_or(OpenCode::DefaultBin()), runner->opencodeHome, OpenCode::BaseEnv(runner->opencodeHome, Dyn::Str(keySeat.at("secret")), runner->Env()), + modelIds.front(), log); + if (!ok) Die(std::format("opencode warm-up failed: {}", Dyn::Slice(detail, -400))); + } + } else if (openrouter) { + log("harness: claude on OpenRouter's Anthropic-format endpoint - provider routing (zero data retention) cannot be set per request there; the account's privacy setting decides"); + } + std::vector names; + for (const Json& s : seats) names.push_back(std::format("{} ({})", Dyn::Str(s.at("id")), Dyn::Str(s.at("kind")))); + std::string line = std::format("seats: {} from {}", Dyn::Join(names, ", "), envFile && !envFile->empty() ? *envFile : std::string("this machine")); + if (std::ranges::any_of(seats, [](const Json& s) { return s.at("kind") == "api"; })) { + line += std::format("; {}, budget ${:g} per api seat per cycle (ledger {})", Dyn::Str(apiCfg.at("order")), Number(apiCfg.at("budgetUsd")), ledger->Path().string()); + } + if (openrouter) { + line += std::format("; budget ${:g} per OpenRouter seat per cycle, {} concurrent (ledger {})", Number(apiCfg.at("orBudgetUsd")), Dyn::Str(apiCfg.at("orConcurrency")), ledger->Path().string()); + } + log(line); + for (const std::string& row : Dyn::Split(SeatPool::Table(seats), "\n")) log(std::format(" {}", row)); + // From here a signal or `crafter-review stop` stops the run: every + // running process killed, the pool aborted. + ready = true; + std::vector dirs; + std::error_code error; + for (fs::directory_iterator it(scratchRoot, error), end; !error && it != end; it.increment(error)) { + const std::string name = it->path().filename().string(); + if (name.starts_with(std::format("{}-run-", h8)) && it->path() != scratch && !name.ends_with("-dry")) dirs.push_back(scratchRoot / name); + } + // Newest first, in run order: run-10 before run-9. + std::ranges::sort(dirs, [&](const fs::path& l, const fs::path& r) { + auto n = [&](const fs::path& p) { return Dyn::ParseInt(Dyn::Slice(p.filename().string(), static_cast(h8.size() + 5))).value_or(0); }; + return std::pair(n(l), l.string()) > std::pair(n(r), r.string()); + }); + salvageDirs = dirs; + if (!salvageDirs.empty() && !a.noSalvage && !testMode && !a.dryRun) { + std::int64_t outputs = 0; + for (const fs::path& d : salvageDirs) { + for (fs::directory_iterator it(d / "out", error), end; !error && it != end; it.increment(error)) { + const std::string name = it->path().filename().string(); + if (!name.starts_with('.') && name.ends_with(".jsonl")) ++outputs; + } + error.clear(); + } + log(std::format("salvage: {} earlier scratch dir(s) on this hash with {} process outputs - finished ok outputs are reused, not re-bought", salvageDirs.size(), outputs)); + } + log(std::format("run layer: {} bytes (~{} tokens); scratch {}", Dyn::Length(runLayer), static_cast(static_cast(Dyn::Length(runLayer)) / 3.7), scratch.string())); + } + + std::map Run::AppLenses(std::span lensFiles) const { + std::map out; + for (const std::string& lf : lensFiles) { + const std::string e = Dyn::Str(Dyn::At(environments, lf)); + if (Plan::FindEnvironment(e)->app) out[lf] = e; + } + return out; + } + + std::map Run::DbLenses(std::span lensFiles) const { + std::map out; + for (const std::string& lf : lensFiles) { + const std::string e = Dyn::Str(Dyn::At(environments, lf)); + if (Plan::FindEnvironment(e)->db) out[lf] = e; + } + return out; + } + + void Run::InstrumentPreflight(std::span lensFiles) { + // The harness can give the instrument (the opencode harness runs no + // MCP server, and the database check reads the claude stream), the + // application answers, the database answers. + if (a.harness == "opencode") { + std::vector need; + for (const std::string& lf : lensFiles) { + const Plan::Environment* e = Plan::FindEnvironment(Dyn::Str(Dyn::At(environments, lf))); + if (!e->mcp.empty() || e->db) need.push_back(lf); + } + if (!need.empty()) { + Die(std::format("the {} environment of {} needs the claude harness (an MCP server, or the database check that reads the claude stream), which the opencode harness is not: run with --without {} on that harness", + Dyn::Str(environments.at(need.front())), Plan::LensName(need.front()), Dyn::Join(need, ","))); + } + } + if (!AppLenses(lensFiles).empty() || !DbLenses(lensFiles).empty()) StackPreflight(lensFiles); + } + + std::optional Run::MigrationVersion(std::string_view file) { + // re.match(r'^\d+_.*\.sql$', basename) + const std::string base = Dyn::BaseName(file); + const std::size_t digits = Dyn::DecimalRun(base, 0); + if (digits == 0 || digits >= base.size() || base[digits] != '_' || !base.ends_with(".sql") || base.size() < digits + 5 || base.contains('\n')) return std::nullopt; + return Dyn::Split(base, "_", 1).front(); + } + + std::pair, std::optional> Run::MigrationsDrift(const std::string& url) { + // The HEAD side is the same for every stack: worked out once per tree + // and base. + constexpr std::string_view M = "supabase/migrations/"; + const std::pair key = {trees.front(), a.base}; + if (!migCache.contains(key)) { + auto git = [&](std::vector args) { + std::vector argv = {"git", "-C", trees.front()}; + std::ranges::move(args, std::back_inserter(argv)); + return Process::Run(argv).out; + }; + std::map atHead; + for (const std::string& f : Dyn::Split(git({"ls-tree", "--name-only", "HEAD", std::string(M)}))) { + if (const std::optional v = MigrationVersion(f)) atHead[*v] = f; + } + std::vector inRange; + if (!a.base.empty() && !atHead.empty()) { + for (const std::string& f : Dyn::Split(git({"diff", "--name-only", "--diff-filter=AM", std::format("{}..HEAD", a.base), "--", std::string(M)}))) { + if (MigrationVersion(f)) inRange.push_back(f); + } + } + std::map headFp; + for (const std::string& f : inRange) headFp[f] = PgWire::Fingerprint(git({"show", std::format("HEAD:{}", f)})); + migCache[key] = {atHead, inRange, headFp}; + } + const auto& [atHead, inRange, headFp] = migCache.at(key); + if (atHead.empty()) return {std::nullopt, std::format("no {} at the range HEAD - applied migrations not compared", M)}; + std::vector>> rows; + try { + try { + rows = PgWire::Query(url, "select version, array_to_string(statements, ' ') from supabase_migrations.schema_migrations").second; + } catch (const PgWire::SqlError& e) { + // No statements column (an old CLI): versions alone. + if (!e.fields.contains('C') || e.fields.at('C') != "42703") throw; + rows = PgWire::Query(url, "select version, null from supabase_migrations.schema_migrations").second; + } + } catch (const PgWire::SqlError& e) { + const std::string code = e.fields.contains('C') ? e.fields.at('C') : std::string(); + if (code == "42P01" || code == "3F000") return {std::nullopt, std::string("no supabase_migrations.schema_migrations in the database - applied migrations not compared")}; + return {std::format("its applied migrations could not be read ({})", e.what()), std::nullopt}; + } catch (const PgWire::DbError& e) { + return {std::format("its applied migrations could not be read ({})", e.what()), std::nullopt}; + } + std::map> applied; + for (const auto& row : rows) applied[row.at(0).value_or("None")] = row.size() > 1 ? row.at(1) : std::nullopt; + std::vector missing; + std::vector extra; + for (const auto& [v, f] : atHead) { + if (!applied.contains(v)) missing.push_back(v); + } + for (const auto& [v, statements] : applied) { + if (!atHead.contains(v)) extra.push_back(v); + } + std::vector changed; + std::int64_t compared = 0; + for (const std::string& f : inRange) { + const std::string v = *MigrationVersion(f); + const auto found = applied.find(v); + if (found == applied.end() || !found->second) continue; + ++compared; + if (PgWire::Fingerprint(*found->second) != headFp.at(f)) changed.push_back(v); + } + std::ranges::sort(changed); + if (!missing.empty() || !extra.empty() || !changed.empty()) { + auto cut = [](const std::vector& xs) { + const std::vector head(xs.begin(), xs.begin() + static_cast(std::min(8, xs.size()))); + return Dyn::Join(head, ", ") + (xs.size() > 8 ? std::format(" and {} more", xs.size() - 8) : std::string()); + }; + std::string problem = "its migrations are not the range HEAD's"; + if (!missing.empty()) problem += std::format(" - at HEAD, not applied: {}", cut(missing)); + if (!extra.empty()) problem += std::format(" - applied, not at HEAD: {}", cut(extra)); + if (!changed.empty()) problem += std::format(" - applied with other content than HEAD has: {}", cut(changed)); + problem += " - reset the stack from the frozen tree (supabase.start / functional.start)"; + return {problem, std::nullopt}; + } + return {std::nullopt, std::format("its {} applied migrations are the range HEAD's{}", atHead.size(), inRange.empty() ? std::string() : std::format(" ({} of the range compared by content)", compared))}; + } + + std::unique_ptr Run::LoadStacks() { + // The slots that are up when the config names slots.dir, else the one + // stack of functional.* and supabase.dbUrl. A dry run with slots + // configured and none up plans on the one stack. + const Json& f = cfg.at("functional"); + Json apiUrls = Json::array(); + for (const Json& u : Dyn::Or(Dyn::Get(f, "apiUrls"), Dyn::EmptyList())) apiUrls.push_back(u); + const Json one = {{"id", "the stack"}, {"appUrl", f.at("appUrl")}, {"apiUrls", apiUrls}, {"dbUrl", cfg.at("supabase").at("dbUrl")}, + {"login", loginFixture ? Json(*loginFixture) : Json()}, {"slot", false}}; + if (!Dyn::Truthy(Dyn::Get(cfg.at("slots"), "dir"))) return std::make_unique(std::vector {one}); + slotMode = true; + const Config::Slots found = Config::SlotStacks(cfg, repo); + slotProblems = found.problems; + std::vector list; + for (const Json& s : found.stacks) list.push_back(s); + if (list.empty()) { + Json dry = one; + dry["id"] = "slot 1 (dry run: none is up)"; + dry["slot"] = true; + list.push_back(dry); + } + return std::make_unique(list); + } + + std::string Run::StacksLine() const { + if (!slotMode) { + return std::format("one stack ({}), one pass at a time on it; login fixture {}", Dyn::Str(stacks->First()->at("appUrl")), + loginFixture ? *loginFixture : std::string("NONE found - behaviours behind a login will come back unverified")); + } + std::vector ids; + for (const Json& s : stacks->All()) ids.push_back(Dyn::Str(s.at("id"))); + return std::format("{} slot(s) under {} ({}), one pass at a time on each", stacks->Size(), Dyn::Str(cfg.at("slots").at("dir")), Dyn::Join(ids, ", ")); + } + + std::vector Run::DbStacks(std::span lensFiles) const { + // Every stack for a lens that drives the application and has the + // database (its pass may hold any of them), the first for a lens with + // the database alone. + std::vector out; + for (const auto& [lf, e] : DbLenses(lensFiles)) { + const std::vector& all = stacks->All(); + const std::size_t count = Plan::FindEnvironment(e)->app ? all.size() : std::min(1, all.size()); + for (std::size_t i = 0; i < count; ++i) { + if (std::ranges::none_of(out, [&](const Json& o) { return Dyn::Equal(o, all[i]); })) out.push_back(all[i]); + } + } + return out; + } + + void Run::StackPreflight(std::span lensFiles) { + // Before a gate whose lenses use a stack: the slots are up (slot + // mode), and every stack the gate's passes use answers - its + // application, its database with the range HEAD's migrations, its + // login in its own database's auth.users. + const std::map app = AppLenses(lensFiles); + const std::map db = DbLenses(lensFiles); + std::vector whoParts; + std::vector seen; + for (const std::string& lf : lensFiles) { + const auto inApp = app.find(lf); + const auto inDb = db.find(lf); + if (inApp == app.end() && inDb == db.end()) continue; + whoParts.push_back(std::format("{} ({})", Plan::LensName(lf), inApp != app.end() ? inApp->second : inDb->second)); + } + const std::string who = Dyn::Join(whoParts, ", "); + const Json& sl = cfg.at("slots"); + if (!slotProblems.empty()) Die(std::format("slots: {}", Dyn::Join(slotProblems, "; "))); + if (slotMode && !a.dryRun && std::ranges::none_of(stacks->All(), [](const Json& s) { return Dyn::Truthy(Dyn::Get(s, "slot")) && !Dyn::Str(s.at("id")).ends_with("none is up)"); })) { + Die(std::format("stacks ({}): no slot is up under {} (no /{}) - bring them up from the frozen tree first: `{} slots` says how many fit and prints the commands", who, Dyn::Str(sl.at("dir")), + Config::SlotFile, Invocation())); + } + const std::vector dbs = DbStacks(lensFiles); + std::vector nodb; + for (const Json& s : dbs) { + if (!Dyn::Truthy(Dyn::Get(s, "dbUrl"))) nodb.push_back(Dyn::Str(s.at("id"))); + } + if (!nodb.empty()) { + Die(std::format("stacks ({}): no database for {} (no dbUrl in {}) - a lens with the database runs its SQL on its own stack's, never on another's: bring the slot up again, or give the lens an environment without the database", + who, Dyn::Join(nodb, ", "), Config::SlotFile)); + } + std::set roles; + for (const Json& s : dbs) roles.insert(PgWire::ParseUrl(Dyn::Str(s.at("dbUrl"))).user); + dbRole = roles.size() == 1 ? std::optional(*roles.begin()) : std::nullopt; + if (a.dryRun) { + dbChecked = std::nullopt; + dbBypass = std::nullopt; + std::vector redacted; + for (const Json& s : dbs) redacted.push_back(PgWire::Redact(Dyn::Str(s.at("dbUrl")))); + log(std::format("stacks ({}): dry run - not probed; {}{}", who, StacksLine(), dbs.empty() ? std::string() : std::format("; database {}", Dyn::Join(redacted, ", ")))); + return; + } + const Json& fcfg = cfg.at("functional"); + std::vector apps; + if (!app.empty()) apps = stacks->All(); + for (const Json& s : apps) { + const Json& readyPath = Dyn::Get(fcfg, "readyPath"); + const auto [ok, detail] = AppReady(Dyn::Str(s.at("appUrl")), Dyn::Truthy(readyPath) ? Dyn::Str(readyPath) : std::string("/")); + if (!ok) { + const Json& how = Dyn::Truthy(Dyn::Get(s, "slot")) ? Dyn::Get(sl, "start") : Dyn::Get(fcfg, "start"); + Die(std::format("stacks ({}): {}: the application does not answer ({}) - start it from the frozen tree first{}", who, Dyn::Str(s.at("id")), detail, + Dyn::Truthy(how) ? std::format(" ({})", Dyn::Str(how)) : std::string())); + } + } + std::set> bypass; + std::optional firstDetail; + std::optional note; + for (const Json& s : dbs) { + const std::string url = Dyn::Str(s.at("dbUrl")); + const auto [ok, detail] = PgWire::Probe(url); + if (!ok) { + const Json& supabaseStart = Dyn::Get(cfg.at("supabase"), "start"); + const Json& how = Dyn::Truthy(Dyn::Get(s, "slot")) ? Dyn::Get(sl, "start") : (Dyn::Truthy(supabaseStart) ? supabaseStart : Dyn::Get(fcfg, "start")); + Die(std::format("stacks ({}): {}: the database does not answer ({}) - start the stack from the frozen tree first{}", who, Dyn::Str(s.at("id")), detail, + Dyn::Truthy(how) ? std::format(" ({})", Dyn::Str(how)) : std::string())); + } + const auto [drift, n] = MigrationsDrift(url); + if (drift) Die(std::format("stacks ({}): {}: the database: {}", who, Dyn::Str(s.at("id")), *drift)); + if (!firstDetail || firstDetail->empty()) firstDetail = detail; + if (!note || note->empty()) note = n; + bypass.insert(PgWire::BypassesRls(url)); + } + if (!dbs.empty()) { + dbChecked = dbs.size() == 1 ? std::format("it answers ({}); {}", firstDetail.value_or("None"), note.value_or("None")) + : std::format("the database of each of the {} stacks answers ({} for {}); {}", dbs.size(), firstDetail.value_or("None"), Dyn::Str(dbs.front().at("id")), note.value_or("None")); + dbBypass = bypass.size() == 1 ? *bypass.begin() : std::nullopt; + } + std::vector notes; + for (const Json& s : apps) { + if (!Dyn::Truthy(Dyn::Get(s, "login"))) { + notes.push_back(std::format("{} has no login fixture - behaviours behind a login will come back unverified", Dyn::Str(s.at("id")))); + } else if (Dyn::Truthy(Dyn::Get(s, "dbUrl")) && (Dyn::Truthy(Dyn::Get(s, "slot")) || std::ranges::any_of(dbs, [&](const Json& d) { return Dyn::Equal(d, s); }))) { + if (const std::optional gone = LoginMissing(s)) { + Die(std::format("stacks ({}): {}: its login {} ({}) is not in that stack's auth.users - the fixture is left from another bring-up: bring the stack up again", who, Dyn::Str(s.at("id")), + Dyn::Str(s.at("login")), *gone)); + } + } + } + // What was checked, no more. + const std::string dbline = dbs.empty() ? std::string() + : std::format("; {} answer{} ({}); {}", dbs.size() == 1 ? std::string("its database") : std::format("the databases of {} stacks", dbs.size()), dbs.size() == 1 ? "s" : "", + firstDetail.value_or("None"), note.value_or("None")); + if (!app.empty()) { + const Json& exclude = fcfg.at("exclude"); + log(std::format("stacks ({}): {}; every application answers{}{}; exclude {}", who, StacksLine(), dbline, notes.empty() ? std::string() : std::format("; {}", Dyn::Join(notes, "; ")), + Dyn::Truthy(exclude) ? Dyn::ReprValue(exclude) : std::string("[]"))); + } else { + log(std::format("stacks ({}): a pass with the database alone uses {}'s, every call rolled back{}", who, Dyn::Str(dbs.front().at("id")), dbline)); + } + } + + std::optional Run::LoginMissing(const Json& stack) const { + // The login's email when the stack's database answers and has no such + // user, else nullopt (not checked). + std::string email; + try { + const Json fixture = Dyn::LoadJsonFile(repo / Dyn::Str(stack.at("login"))); + email = Dyn::Lower(Dyn::Strip(Dyn::StrOr(Dyn::Get(fixture, "email"), ""))); + } catch (const std::exception&) { + return std::nullopt; + } + if (email.empty() || PgWire::CheckUrl(Dyn::Get(stack, "dbUrl"))) return std::nullopt; + try { + const auto [columns, rows] = PgWire::Query(Dyn::Str(stack.at("dbUrl")), std::format("select 1 from auth.users where lower(email) = '{}'", Dyn::Replace(email, "'", "''"))); + return rows.empty() ? std::optional(email) : std::nullopt; + } catch (const PgWire::SqlError&) { + return std::nullopt; + } catch (const PgWire::DbError&) { + return std::nullopt; + } + } + + std::optional Run::TreeDrift() { + // A moved HEAD or a dirty tree under the lenses is evidence drift, + // like an artifact re-hash mismatch; the paths the config allows are + // not drift. + try { + const std::string th = Artifact::TreeHead(trees.front()); + if (th != head) return std::format("tree {} is at {}, the artifact was cut from {}", trees.front(), Dyn::Slice(th, 0, 12), Dyn::Slice(head, 0, 12)); + const auto [counted, ignored] = Freeze::DirtyPaths(trees.front(), Strings(cfg.at("allowDirty"))); + if (!counted.empty()) { + std::vector shown; + for (std::size_t i = 0; i < std::min(3, counted.size()); ++i) shown.push_back(Dyn::Slice(counted[i], 3)); + return std::format("tree {} has uncommitted tracked changes ({})", trees.front(), Dyn::Join(shown, ", ")); + } + } catch (const std::exception& e) { + return std::format("tree check failed: {}", e.what()); + } + return std::nullopt; + } + + // ---- prompts ------------------------------------------------------------------------ + std::pair Run::PromptFor(const Json& job) { + // The system file holds the common text, the run layer and the lens + // layer - identical for every pass of the lens, written once, served + // from the prompt cache; the user message is the scope of this pass. + const std::lock_guard guard(shared); + const std::string lens = Dyn::Str(Dyn::At(job, "lens")); + const std::string lensFile = Dyn::Str(Dyn::At(job, "lensFile")); + std::string summary; + if (job.at("kind") != "whole") { + if (const auto found = summaries.find(lens); found != summaries.end()) summary = found->second; + } + const Plan::Environment& env = *Plan::FindEnvironment(Dyn::Str(Dyn::At(environments, lensFile))); + const std::string db = env.db ? Prompt::DbBlock(cfg.at("supabase"), Pool::SqlCommand(), dbRole, env.app, dbChecked, dbBypass) : std::string(); + const std::string ll = Prompt::LensLayer(lensFile, charters.at(lensFile), Dispositions::Block(inForce, dispRows, lensFile, hash), env.app ? appBlock : std::string(), summary, db); + lensLayersBytes[lens] = static_cast(Dyn::Length(ll)); + const std::string sysText = Prompt::SystemPrompt(systemText, runLayer, ll); + const std::string sha = Plan::Sha(sysText); + auto found = std::ranges::find_if(systemPaths, [&](const auto& entry) { return entry.first == sha; }); + if (found == systemPaths.end()) { + const fs::path path = scratch / std::format("system-{}.md", sha.substr(0, 8)); + WriteText(path, sysText); + systemPaths.emplace_back(sha, path); + found = systemPaths.end() - 1; + } + auto byLens = std::ranges::find_if(systemByLens, [&](const auto& entry) { return entry.first == lens; }); + if (byLens == systemByLens.end()) { + systemByLens.emplace_back(lens, std::set {}); + byLens = systemByLens.end() - 1; + } + byLens->second.insert(sha); + std::string stackText; + const Json& stack = Dyn::Get(job, "stack"); + if (slotMode && Dyn::Truthy(stack)) { + const Json& login = Dyn::Get(stack, "login"); + stackText = Prompt::StackText(stack, Pool::TotpCommand(login.is_null() ? std::nullopt : std::optional(Dyn::Str(login)))); + } + return {found->second, Prompt::UserPrompt(Prompt::Tail(job, units, lines, summary), job, stackText)}; + } + + std::optional Run::CacheKey(Json& job) { + if (Dyn::Get(job, "cacheKey").is_null()) job["cacheKey"] = Basename(PromptFor(job).first); + return Dyn::Str(job.at("cacheKey")); + } + + bool Run::WarmProbe(const Json& job) const { + const std::optional text = Platform::ReadFile(scratch / "out" / std::format("{}.jsonl", FileLabel(job))); + return text && (text->contains("\"type\":\"assistant\"") || text->contains("\"type\":\"step_start\"")); + } + + // ---- reuse -------------------------------------------------------------------------- + const Plan::CommitCode& Run::RangeCode() { + if (!commitCode) commitCode = Plan::CommitCodeOf(Artifact::PatchTexts(a.artifact)); + return *commitCode; + } + + std::optional Run::ReuseWhole(const std::string& name, Json& job) { + // A lens with wholeReuse: the stored clean whole pass for this code as + // a part, re-normalised against today's dispositions, with its summary + // for the block passes. nullopt: the whole pass runs. + const std::string lensFile = Dyn::Str(job.at("lensFile")); + std::vector ids; + for (const Json& d : inForce) ids.push_back(Dyn::Str(d.at("id"))); + std::map> keys; + for (const Plan::Environment& e : Plan::Environments) keys[std::string(e.name)] = Plan::WholeCodeKey(job, lensFile, RangeCode(), ids, charterSha.at(lensFile), PromptVersion, e.name); + const Json& configured = Dyn::Get(environments, lensFile); + const std::optional key = keys.at(configured.is_null() ? std::string(Plan::LensOf(lensFile).env) : Dyn::Str(configured)); + job["reuseKey"] = key ? Json(*key) : Json(); + std::set others; + for (const auto& [e, k] : keys) { + if (k && !k->empty() && k != key) others.insert(*k); + } + job["otherKeys"] = Array(std::vector(others.begin(), others.end())); + const Json& hit = key ? Dyn::Get(state.at("entries"), *key) : Dyn::EmptyDict(); + if (!Dyn::Truthy(hit) || a.dryRun) return std::nullopt; + Json v = {{"verdict", "SEND-READY"}, {"findings", Dyn::At(hit, "findings")}, {"notes", Dyn::At(hit, "notes")}, {"coverage", Dyn::At(hit, "coverage")}, {"summary", Dyn::Get(hit, "summary")}}; + Json expected = Json::array(); + std::vector allIds; + const std::vector wanted = Strings(job.at("expectedIds")); + for (const Json& u : units) { + allIds.push_back(Dyn::Str(u.at("id"))); + if (std::ranges::contains(wanted, allIds.back())) expected.push_back(u); + } + Json n = Normalize::NormalizeOutput(v, inForce, lensFile, hash, expected, allIds); + if (n.at("verdict") != "SEND-READY" || Dyn::Truthy(n.at("coverageMissing"))) { + log(std::format("{}: stored result re-blocked against today's dispositions or slots - re-running", Dyn::Str(job.at("label")))); + return std::nullopt; + } + const std::string judgedAt = Dyn::Slice(Dyn::StrOr(Dyn::Has(hit, "hash") ? hit.at("hash") : Json(""), ""), 0, 8); + n["partLabel"] = job.at("label"); + n["partKind"] = "whole"; + n["partUnit"] = nullptr; + n["status"] = "reused"; + n["reused"] = std::format("unit-state {} (run {} at {})", *key, Dyn::Str(Dyn::Get(hit, "run")), judgedAt); + n["usage"] = nullptr; + n["turns"] = 0; + { + const std::lock_guard guard(shared); + summaries[name] = Dyn::Slice(Dyn::StrOr(Dyn::Get(n, "summary"), ""), 0, 2500); + } + records.push_back({{"label", job.at("label")}, {"lens", name}, {"lensFile", lensFile}, {"unit", nullptr}, {"kind", "whole"}, {"status", "reused"}, {"reuseKey", *key}, {"gate", Dyn::Get(job, "gate")}}); + log(std::format("{}: whole pass REUSED - the code of the range is the code {} judged at {}", name, Dyn::Str(Dyn::Get(hit, "run")), judgedAt)); + return n; + } + + std::optional Run::ReuseKey(const Json& job, const std::optional& environment) { + // The block's reuse key in this run's environment for its lens (or in + // the one named). + const std::string lensFile = Dyn::Str(job.at("lensFile")); + if (job.at("kind") != "block" || !Plan::LensOf(lensFile).reuse) return std::nullopt; + const auto u = std::ranges::find_if(units, [&](const Json& x) { return x.at("id") == job.at("unit"); }); + if (u == units.end()) throw Dyn::Error("StopIteration", ""); + const std::string text = Plan::UnitText(lines, *u); + std::vector ids; + for (const Json& d : inForce) ids.push_back(Dyn::Str(d.at("id"))); + const Json& configured = Dyn::Get(environments, lensFile); + const std::string env = environment && !environment->empty() ? *environment : (configured.is_null() ? std::string() : Dyn::Str(configured)); + return Plan::BlockReuseKey(job, lensFile, text, RangeCode(), ids, charterSha.at(lensFile), PromptVersion, env); + } + + std::vector Run::OtherKeys(const Json& job, const std::optional& key) { + // The block's keys in every other environment: a defect its pass + // returns here drops the clean verdicts those hold for the same code - + // a finding is not undone by switching the lens's environment. + std::set out; + for (const Plan::Environment& e : Plan::Environments) { + const std::optional k = ReuseKey(job, std::string(e.name)); + if (k && !k->empty() && k != key) out.insert(*k); + } + return {out.begin(), out.end()}; + } + + Json Run::DirEnvironments(const fs::path& dir) { + // {lens name: environment} a scratch dir's passes ran with; a dir + // without the marker ran every lens in its default environment. + const std::lock_guard guard(shared); + if (!envMarkers.contains(dir)) { + try { + const Json m = Dyn::LoadJsonFile(dir / "environments"); + envMarkers[dir] = m.is_object() ? m : Json::object(); + } catch (const std::exception&) { + envMarkers[dir] = Json::object(); + } + } + return envMarkers.at(dir); + } + + std::int64_t Run::DirRound(const fs::path& dir) { + const std::optional text = Platform::ReadFile(dir / "round"); + if (!text) return 1; + return Dyn::ParseInt(Dyn::Strip(*text)).value_or(1); + } + + std::int64_t Run::ReadsNeeded(const Json& job) const { + // Clean reads this pass's unit needs before its verdict is reused: x + // for a code unit of a lens marked `reads`. + return x > 1 && job.at("kind") == "block" && Dyn::Get(job, "unitKind") == "code" && Plan::LensOf(Dyn::Str(job.at("lensFile"))).reads ? x : 1; + } + + std::optional> Run::Salvage(const Json& job) { + // A finished, ok output for this label from an earlier scratch dir on + // the same hash: reused instead of launched. + if (testMode || a.dryRun || a.noSalvage) return std::nullopt; + const std::string lensFile = Dyn::Str(job.at("lensFile")); + const Json& configured = Dyn::Get(environments, lensFile); + const std::string env = configured.is_null() ? std::string("none") : Dyn::Str(configured); + const Plan::Lens& lens = Plan::LensOf(lensFile); + for (const fs::path& d : salvageDirs) { + if (x > 1 && DirRound(d) != round && ReadsNeeded(job) > 1) continue; + const std::string name = Plan::LensName(lensFile); + const Json markers = DirEnvironments(d); + const std::string was = Dyn::Has(markers, name) ? Dyn::Str(markers.at(name)) : std::string(lens.env); + if (was != env) { + // Evidence bought in one environment is never reused in another. + bool first = false; + { + const std::lock_guard guard(shared); + first = envSkips.insert({d.string(), name}).second; + } + if (first) log(std::format("{}: nothing salvaged from {} - its passes ran in the {} environment, this run gives the lens {}", name, Basename(d), was, env)); + continue; + } + const fs::path f = d / "out" / std::format("{}.jsonl", FileLabel(job)); + if (!fs::exists(f)) continue; + Json ps; + Json raw; + try { + const std::string text = Dyn::UniversalNewlines(Dyn::DecodeReplace(Platform::ReadFile(f).value_or(""))); + if (text.starts_with("{\"type\": \"review_gates_meta\"")) { + const Json meta = Json::parse(text.substr(0, text.find('\n'))); + const std::string fs = f.string(); + ps = OpenCode::ParseEvents(text, std::format("{}.so.json", fs.substr(0, fs.size() - std::string_view(".jsonl").size()))); + ps["init"] = {{"model", Dyn::Get(meta, "model")}}; + ps["denials"] = Json::array(); + ps["result"] = nullptr; + const auto [st, reason] = OpenCode::Classify(ps, 0, ""); + if (st == "ok") raw = ps.at("structured"); + } else { + ps = Stream::ParseStream(text); + const auto [st, reason] = Stream::Classify(ps, 0, ""); + if (st == "ok") { + const std::optional structured = Stream::StructuredOutput(ps.at("result")); + if (structured) raw = *structured; + } + } + } catch (const std::exception&) { + continue; + } + if (raw.is_null()) continue; + const std::string got = Dyn::StrOr(Dyn::Get(Dyn::Or(Dyn::Get(ps, "init"), Dyn::EmptyDict()), "model"), ""); + if (got != Dyn::Str(job.at("model"))) { + log(std::format("{}: not salvaged from {} - model {} != {}", Dyn::Str(job.at("label")), Dyn::RelPath(f.string(), repo.string()), got.empty() ? std::string("?") : got, Dyn::Str(job.at("model")))); + continue; + } + const auto [st, reason] = Pool::Exercised(ps, env, Dyn::Str(job.at("kind")), lens.empirical); + if (st != "ok") { + // An output whose instrument was missing is no evidence, + // salvaged or not. + log(std::format("{}: not salvaged from {} - {}", Dyn::Str(job.at("label")), Dyn::RelPath(f.string(), repo.string()), Dyn::Slice(reason, 0, 120))); + continue; + } + Json rec = {{"label", job.at("label")}, + {"sessionId", ps.at("sessionId")}, + {"status", "ok"}, + {"reason", ""}, + {"salvaged", Dyn::RelPath(f.string(), repo.string())}, + {"maxTurns", Dyn::Get(job, "maxTurns")}, + {"environment", env}, + {"usage", ps.at("usage")}, + {"firstTurn", ps.at("firstTurn")}, + {"turns", ps.at("turns")}, + {"toolCalls", ps.at("toolCalls")}, + {"toolErrors", ps.at("toolErrors")}, + {"costUsd", 0.0}, + {"costUsdOriginal", ps.at("costUsd")}, + {"denials", ps.at("denials")}, + {"rateLimit", Json::object()}, + {"promptBytes", nullptr}}; + return std::pair(rec, raw); + } + return std::nullopt; + } + + // ---- one process ---------------------------------------------------------------------- + std::pair Run::ExecJob(const Json& job, std::int64_t) { + auto stamp = [&](Json& rec) { + rec["lens"] = job.at("lens"); + rec["lensFile"] = job.at("lensFile"); + rec["kind"] = job.at("kind"); + rec["unit"] = Dyn::Get(job, "unit"); + rec["unitSha"] = Dyn::Get(job, "unitSha"); + rec["model"] = job.at("model"); + rec["effort"] = a.effort; + rec["gate"] = Dyn::Get(job, "gate"); + rec["reuseKey"] = Dyn::Get(job, "reuseKey"); + rec["otherKeys"] = Dyn::Get(job, "otherKeys"); + rec["readNo"] = Dyn::Get(job, "readNo"); + }; + if (std::optional> s = Salvage(job)) { + auto [rec, raw] = std::move(*s); + stamp(rec); + { + const std::lock_guard guard(shared); + jobsRun.push_back(job); + } + log(std::format("{}: salvaged from {}", Dyn::Str(job.at("label")), Dyn::Str(rec.at("salvaged")))); + return {rec, raw}; + } + auto abortPool = [&](const std::string& reason) { + const std::lock_guard guard(shared); + if (sched) sched->Abort(reason); + }; + const std::string now = Artifact::ArtifactHash("commits", a.artifact); + if (now != hash) { + abortPool(std::format("artifact changed during the run ({} != {})", now.substr(0, 8), h8)); + return {{{"label", job.at("label")}, {"status", "error"}, {"reason", "artifact changed during the run - aborted"}}, nullptr}; + } + if (const std::optional drift = TreeDrift()) { + abortPool(std::format("evidence drift during the run ({})", *drift)); + return {{{"label", job.at("label")}, {"status", "error"}, {"reason", std::format("{} - aborted", *drift)}}, nullptr}; + } + const auto [systemPath, prompt] = PromptFor(job); + const bool whole = job.at("kind") == "whole"; + auto [rec, raw] = runner->Run(job, prompt, systemPath, schemas.at(whole ? "whole" : "block"), trees, Dyn::Has(job, "maxTurns") ? Int(job.at("maxTurns")) : 20, + Dyn::Has(job, "timeoutMin") ? Int(job.at("timeoutMin")) : 20, whole); + if (Dyn::Truthy(Dyn::Get(job, "stack"))) rec["stack"] = job.at("stack").at("id"); + if (Dyn::Truthy(Dyn::Get(rec, "memPeakMb"))) { + const Json& configured = Dyn::Get(environments, Dyn::Str(job.at("lensFile"))); + calibration->Record(configured.is_null() ? std::string("none") : Dyn::Str(configured), Number(rec.at("memPeakMb"))); + } + stamp(rec); + { + const std::lock_guard guard(shared); + const auto found = lensLayersBytes.find(Dyn::Str(job.at("lens"))); + rec["lensLayerBytes"] = found == lensLayersBytes.end() ? Json() : Json(found->second); + jobsRun.push_back(job); + } + rec["systemFile"] = Dyn::RelPath(systemPath.string(), repo.string()); + return {rec, raw}; + } + + Json Run::PartFrom(const Json& job, const Json& rec, Json raw) { + Json expected = Json::array(); + std::vector allIds; + const std::vector wanted = Strings(job.at("expectedIds")); + for (const Json& u : units) { + allIds.push_back(Dyn::Str(u.at("id"))); + if (std::ranges::contains(wanted, allIds.back())) expected.push_back(u); + } + Json n = Normalize::NormalizeOutput(raw, inForce, Dyn::Str(job.at("lensFile")), hash, expected, allIds); + for (const Json& l : n.at("normalizeLogs")) log(Dyn::Str(l)); + if (Dyn::Truthy(Dyn::Get(job, "readNo"))) { + for (Json& f : n["findings"]) f["read"] = job.at("readNo"); + } + n["partLabel"] = job.at("label"); + n["partKind"] = job.at("kind"); + n["partUnit"] = Dyn::Get(job, "unit"); + n["status"] = Dyn::Has(rec, "status") ? rec.at("status") : Json("ok"); + n["usage"] = Dyn::Get(rec, "usage"); + n["turns"] = Dyn::Get(rec, "turns"); + n["reused"] = nullptr; + return n; + } + + Json Run::DryPart(const Json& job) { + Json coverage = Json::array(); + for (const Json& i : job.at("expectedIds")) coverage.push_back({{"item", i}, {"verdict", "clear"}, {"basis", "dry run"}}); + return {{"partLabel", job.at("label")}, {"partKind", job.at("kind")}, {"partUnit", Dyn::Get(job, "unit")}, {"verdict", "SEND-READY"}, {"findings", Json::array()}, {"notes", Json::array()}, + {"coverage", coverage}, {"coverageMissing", Json::array()}, {"status", "dry-run"}, {"summary", "dry run"}, {"reused", nullptr}}; + } + + // ---- a gate ----------------------------------------------------------------------------- + std::vector Run::RunGate(std::string_view title, std::vector lensFiles) { + // lensFiles -> the merged lens results, one per lens. + std::vector names; + for (const std::string& lf : lensFiles) names.push_back(Plan::LensName(lf)); + log(std::format("{}: {}", title, Dyn::Join(names, ", "))); + std::vector resumed; + std::vector live; + for (const std::string& lf : lensFiles) { + if (std::optional r = Resumable(Plan::LensName(lf), lf, title)) { + resumed.push_back(std::move(*r)); + } else { + live.push_back(lf); + } + } + lensFiles = live; + // Nothing runs: no stack, database or application is needed for a + // result already bought. + if (lensFiles.empty()) return resumed; + InstrumentPreflight(lensFiles); + std::map> parts; + std::map> missing; + std::map> planned; + std::map> pendingBlocks; + std::vector initial; + const std::vector exclude = Strings(cfg.at("functional").at("exclude")); + std::vector unitFilter; + if (a.units) unitFilter.assign(a.units->begin(), a.units->end()); + for (const std::string& lf : lensFiles) { + const std::string name = Plan::LensName(lf); + parts[name]; + missing[name]; + planned[name]; + Plan::LensPlan plan = Plan::PlanLens(lf, units, lines, unitFilter, exclude); + if (!plan.whole.is_null()) { + plan.whole["gate"] = title; + for (const Json& id : plan.whole.at("expectedIds")) planned[name].insert(Dyn::Str(id)); + } + for (Json& j : plan.blocks) { + j["gate"] = title; + for (const Json& id : j.at("expectedIds")) planned[name].insert(Dyn::Str(id)); + } + if (!plan.whole.is_null()) { + pendingBlocks[name] = plan.blocks; + if (std::optional part = ReuseWhole(name, plan.whole)) { + parts[name].push_back(std::move(*part)); + std::ranges::move(PrepareBlocks(name, plan.blocks, parts), std::back_inserter(initial)); + } else { + initial.push_back(plan.whole); + } + } else { + std::ranges::move(PrepareBlocks(name, plan.blocks, parts), std::back_inserter(initial)); + } + } + auto onDone = [&](const Json& job, const Json& rec, const Json& raw) -> std::vector { + const std::string name = Dyn::Str(job.at("lens")); + records.push_back(rec); + const std::string status = Dyn::StrOr(Dyn::Get(rec, "status"), ""); + if (status != "ok" && status != "dry-run") { + for (const Json& id : job.at("expectedIds")) missing[name].insert(Dyn::Str(id)); + if (job.at("kind") == "whole") { + for (const Json& b : pendingBlocks[name]) missing[name].insert(Dyn::Str(b.at("unit"))); + } + log(std::format("{}: {} - {}", Dyn::Str(job.at("label")), Dyn::Str(Dyn::Get(rec, "status")), Dyn::Slice(Dyn::Has(rec, "reason") ? Dyn::Str(rec.at("reason")) : std::string(), 0, 120))); + return {}; + } + Json part = raw.is_null() ? DryPart(job) : PartFrom(job, rec, raw); + for (const Json& id : Dyn::Or(Dyn::Get(part, "coverageMissing"), Dyn::EmptyList())) missing[name].insert(Dyn::Str(id)); + parts[name].push_back(part); + if (job.at("kind") != "whole") return {}; + wholeParts[name] = part; + if (part.at("verdict") != "SEND-READY") { + log(std::format("{}: whole pass FIX FIRST ({} undischarged) - block passes skipped until the whole pass is accepted", name, Dyn::Str(Dyn::Get(part, "undischarged")))); + parts[name].push_back({{"partLabel", std::format("{}:blocks", name)}, {"partKind", "skipped"}, {"verdict", nullptr}, {"findings", Json::array()}, {"notes", Json::array()}, + {"coverage", Json::array()}, {"coverageMissing", Json::array()}}); + return {}; + } + { + const std::lock_guard guard(shared); + summaries[name] = Dyn::Slice(Dyn::StrOr(Dyn::Get(part, "summary"), ""), 0, 2500); + } + return PrepareBlocks(name, pendingBlocks[name], parts); + }; + const bool warm = !a.dryRun && (!testMode || a.warm); + std::unique_ptr admission = Admission(lensFiles); + std::optional before; + { + const std::lock_guard guard(shared); + before = stopped; + } + Pool::Scheduler scheduler({.runner = runner.get(), + .concurrency = a.concurrency, + .log = log, + .progressPath = scratch / "progress.json", + .cacheKey = warm ? std::function(Json&)>([this](Json& job) { return CacheKey(job); }) : nullptr, + .warmProbe = warm ? std::function([this](const Json& job) { return WarmProbe(job); }) : nullptr, + .seats = &seats, + .apiConfig = apiCfg, + .ledger = ledger.get(), + .admission = admission.get(), + .board = board.get(), + .abort = before}); + { + const std::lock_guard guard(shared); + sched = &scheduler; + } + scheduler.Run(initial, [this](const Json& job, std::int64_t attempt) { return ExecJob(job, attempt); }, onDone); + { + const std::lock_guard guard(shared); + sched = nullptr; + } + if (const std::optional aborted = scheduler.Aborted()) abort = aborted; + std::vector out; + for (const std::string& lf : lensFiles) { + const std::string name = Plan::LensName(lf); + const bool stoppedEarly = std::ranges::any_of(parts[name], [](const Json& p) { return Dyn::Get(p, "partKind") == "skipped"; }); + std::vector real; + for (const Json& p : parts[name]) { + if (Dyn::Get(p, "partKind") != "skipped") real.push_back(p); + } + std::set covered; + for (const Json& p : real) { + for (const Json& c : Dyn::Or(Dyn::Get(p, "coverage"), Dyn::EmptyList())) covered.insert(Dyn::Str(c.at("item"))); + } + if (!stoppedEarly) { + for (const std::string& id : planned[name]) { + if (!covered.contains(id)) missing[name].insert(id); + } + } + const std::vector missingList(missing[name].begin(), missing[name].end()); + Json m = Normalize::MergeParts(name, lf, real, missingList, stoppedEarly ? Json("whole pass FIX FIRST - block passes skipped") : Json()); + if (!Plan::LensOf(lf).whole.empty() && std::ranges::none_of(real, [](const Json& p) { return Dyn::Get(p, "partKind") == "whole"; })) { + Json coverageMissing = Json::array(); + for (const Json& id : m.at("coverageMissing")) { + if (std::ranges::none_of(coverageMissing, [&](const Json& c) { return c == id; })) coverageMissing.push_back(id); + } + if (std::ranges::none_of(coverageMissing, [](const Json& c) { return c == "(whole pass)"; })) coverageMissing.push_back("(whole pass)"); + m["coverageMissing"] = coverageMissing; + m["contractViolations"].push_back({{"lens", lf}, {"kind", "whole pass did not return - lens NOT RETURNED"}}); + } + std::size_t logged = 0; + for (const Json& p : real) logged += Dyn::Or(Dyn::Get(p, "normalizeLogs"), Dyn::EmptyList()).size(); + const Json& merged = m.at("normalizeLogs"); + for (std::size_t i = logged; i < merged.size(); ++i) log(Dyn::Str(merged.at(i))); + log(std::format("{}: {} - findings {} (undischarged {}), passes {}, missing {}", name, Dyn::Str(m.at("verdict")), m.at("findings").size(), Dyn::Str(m.at("undischarged")), real.size(), m.at("coverageMissing").size())); + out.push_back(std::move(m)); + } + std::ranges::move(out, std::back_inserter(resumed)); + return resumed; + } + + std::unique_ptr Run::Admission(std::span lensFiles) { + // Free memory less the reserve is the budget, each pass held at its + // environment's measured cost; a pass that drives an application + // holds a stack of its own. Free memory is the reading of now - or, + // while other drivers' passes are in flight, the board's reading from + // before they were. + std::set envs; + for (const std::string& lf : lensFiles) envs.insert(Dyn::Str(Dyn::At(environments, lf))); + const std::optional free = Memory::Available(); + const std::int64_t reserve = Memory::Reserve(); + std::optional avail = free ? std::optional(static_cast(*free)) : std::nullopt; + std::optional how; + bool recorded = false; + if (free && board) { + const auto [base, why] = board->MemoryBase(*free, Platform::Now()); + avail = base; + how = why; + recorded = base != static_cast(*free); + } + const std::optional budget = avail ? std::optional(std::max(0.0, *avail - static_cast(reserve))) : std::nullopt; + std::map cost; + for (const std::string& e : envs) cost.emplace(e, calibration->CostOf(e)); + std::vector partsText; + for (const std::string& e : envs) { + const double mb = Number(cost.at(e).megabytes); + std::string line = std::format("{} ~{:.0f} MB a pass - {}", e, mb, cost.at(e).source); + if (budget) line += std::format(" -> {} at once by memory", static_cast(FloorDiv(*budget, mb))); + if (Plan::FindEnvironment(e)->app) { + const auto size = static_cast(stacks->Size()); + line += std::format(", {} stack(s) -> {} at once", size, budget ? std::min(size, std::max(1, static_cast(FloorDiv(*budget, mb)))) : size); + } + partsText.push_back(line); + } + (void)recorded; + log(std::format("memory: {}{}", avail ? std::format("{:.0f} MB free{}, reserve {} MB -> budget {:.0f} MB; ", *avail, how ? std::format(" ({})", *how) : std::string(), reserve, *budget) + : std::string("no /proc/meminfo - memory not limited; "), + Dyn::Join(partsText, "; "))); + std::map costOf; + for (const auto& [e, c] : cost) costOf[e] = Number(c.megabytes); + const Json envCopy = environments; + return std::make_unique([costOf, envCopy](const Json& j) { return costOf.at(Dyn::Str(Dyn::At(envCopy, Dyn::Str(j.at("lensFile"))))); }, + [envCopy](const Json& j) { return Plan::FindEnvironment(Dyn::Str(Dyn::At(envCopy, Dyn::Str(j.at("lensFile")))))->app; }, stacks.get(), budget, + static_cast(reserve), Pool::Admission::Available(), log); + } + + std::optional Run::Resumable(const std::string& name, const std::string& lensFile, std::string_view title) { + // A lens result from the resume file (same hash) that returned fully: + // discharge re-derived against the dispositions in force NOW, reused. + if (!Dyn::Truthy(resume)) return std::nullopt; + if (x > 1 && newRound && Plan::LensOf(lensFile).reads) return std::nullopt; + const Json* src = nullptr; + for (const Json& r : resume.at("results")) { + if (Dyn::Get(r, "lens") == name) { + src = &r; + break; + } + } + if (!src || Dyn::Truthy(Dyn::Get(*src, "coverageMissing")) || !(Dyn::Get(*src, "verdict") == "SEND-READY" || Dyn::Get(*src, "verdict") == "FIX FIRST")) return std::nullopt; + // A checkpoint written before 2026-10-16.1 carries no environments: + // its lenses ran in their defaults. + const Json& markers = Dyn::Or(Dyn::Get(resume, "environments"), Dyn::EmptyDict()); + const std::string fallback(Plan::LensOf(lensFile).env); + const std::string was = Dyn::Has(markers, name) ? Dyn::Str(markers.at(name)) : fallback; + const Json& configured = Dyn::Get(environments, lensFile); + const std::string now = configured.is_null() ? fallback : Dyn::Str(configured); + if (was != now) { + log(std::format("{}: not resumed - its result on this hash was bought in the {} environment, this run gives the lens {}", name, was, now)); + return std::nullopt; + } + Json r = *src; + std::set ids; + for (const Json& d : inForce) ids.insert(Dyn::Str(d.at("id"))); + std::int64_t undischarged = 0; + for (Json& f : r["findings"]) { + const Json& by = Dyn::Get(f, "dischargedBy"); + f["discharged"] = Dyn::Truthy(by) && ids.contains(Dyn::Str(by)); + if (!f.at("discharged").get()) ++undischarged; + } + r["undischarged"] = undischarged; + r["verdict"] = undischarged ? "FIX FIRST" : "SEND-READY"; + r["resumed"] = Dyn::Str(Dyn::Get(resume, "run")); + records.push_back({{"label", std::format("{}:resumed", name)}, {"lens", name}, {"lensFile", lensFile}, {"kind", "resumed"}, {"status", "resumed"}, {"gate", title}, {"from", Dyn::Get(resume, "run")}}); + log(std::format("{}: RESUMED from {} - {} (findings {}, undischarged {}, re-normalised against {} disposition(s) in force)", name, Dyn::Str(Dyn::Get(resume, "run")), Dyn::Str(r.at("verdict")), + r.at("findings").size(), undischarged, ids.size())); + return r; + } + + void Run::Checkpoint(std::string_view title) { + if (testMode || a.dryRun) return; + Json markers = Json::object(); + for (auto it = environments.begin(); it != environments.end(); ++it) markers[Plan::LensName(it.key())] = it.value(); + Json checkpoint = {{"hash", hash}, {"date", a.date}, {"run", Basename(dumpPath)}, {"after", title}, {"results", results}, {"environments", markers}}; + if (x > 1) { + checkpoint["round"] = round; + checkpoint["complete"] = false; + } + WriteReplace(resumePath, DumpFile(checkpoint)); + StoreState(); + WriteText(dumpPath, DumpFile(BuildDump(std::format("IN PROGRESS - after {} (partial dump, the run continues)", title), nullptr, Json::array()))); + log(std::format("checkpoint after {}: {} + partial dump written", title, Dyn::RelPath(resumePath.string(), repo.string()))); + } + + std::vector Run::PrepareBlocks(const std::string& name, std::vector blocks, std::map>& parts) { + // Auto-n/a and reuse hits become parts at once; the rest are jobs for + // the pool. + std::vector jobs; + std::vector allIds; + for (const Json& u : units) allIds.push_back(Dyn::Str(u.at("id"))); + for (Json& j : blocks) { + if (j.at("kind") == "auto-na") { + parts[name].push_back({{"partLabel", j.at("label")}, {"partKind", "auto-na"}, {"partUnit", j.at("unit")}, {"verdict", "SEND-READY"}, {"findings", Json::array()}, {"notes", Json::array()}, + {"coverage", Json::array({{{"item", j.at("unit")}, {"verdict", "n/a"}, {"basis", Dyn::Has(j, "basis") ? j.at("basis") : Json("mechanical")}}})}, + {"coverageMissing", Json::array()}, {"reused", nullptr}}); + continue; + } + const std::optional key = ReuseKey(j); + Json hit = key ? Dyn::Get(state.at("entries"), *key) : Json(); + const Json was = Dyn::Truthy(hit) ? Dyn::Get(hit, "unit") : Json(); + if (Dyn::Truthy(hit) && !Dyn::Equal(was, j.at("unit"))) hit = Plan::Rebind(hit, Dyn::Str(j.at("unit"))); + const std::int64_t need = ReadsNeeded(j); + const std::int64_t have = Plan::ReadsOf(hit); + if (need > 1 && (!Dyn::Truthy(hit) || (have < need && !Plan::ReadIn(hit, h8, round)))) { + j["readNo"] = have + 1; + hit = nullptr; + } + if (Dyn::Truthy(hit) && !a.dryRun) { + Json v = {{"verdict", "SEND-READY"}, {"findings", Dyn::At(hit, "findings")}, {"notes", Dyn::At(hit, "notes")}, {"coverage", Json::array({Dyn::At(hit, "coverage")})}}; + Json expected = Json::array(); + for (const Json& u : units) { + if (u.at("id") == j.at("unit")) expected.push_back(u); + } + Json n = Normalize::NormalizeOutput(v, inForce, Dyn::Str(j.at("lensFile")), hash, expected, allIds); + if (n.at("verdict") == "SEND-READY" && !Dyn::Truthy(n.at("coverageMissing"))) { + std::string reused = std::format("unit-state {} (run {} at {}", *key, Dyn::Str(Dyn::Get(hit, "run")), Dyn::Slice(Dyn::StrOr(Dyn::Has(hit, "hash") ? hit.at("hash") : Json(""), ""), 0, 8)); + if (!Dyn::Equal(was, j.at("unit"))) reused += std::format(", stored as {}", Dyn::Str(was)); + if (need > 1) reused += std::format(", {} of {} clean reads", have, need); + reused += ")"; + n["partLabel"] = j.at("label"); + n["partKind"] = "block"; + n["partUnit"] = j.at("unit"); + n["status"] = "reused"; + n["reused"] = reused; + n["usage"] = nullptr; + n["turns"] = 0; + parts[name].push_back(n); + records.push_back({{"label", j.at("label")}, {"lens", name}, {"lensFile", j.at("lensFile")}, {"unit", j.at("unit")}, {"kind", "block"}, {"status", "reused"}, {"reuseKey", *key}, + {"gate", Dyn::Get(j, "gate")}}); + continue; + } + log(std::format("{}: stored result re-blocked against today's dispositions - re-running", Dyn::Str(j.at("label")))); + } + j["reuseKey"] = key ? Json(*key) : Json(); + j["otherKeys"] = key ? Array(OtherKeys(j, key)) : Json::array(); + jobs.push_back(j); + } + return jobs; + } + + // ---- the run ------------------------------------------------------------------------------ + std::int32_t Run::Execute() { + trees = CommaList(a.trees.value_or(".")); + contextPaths = CommaList(a.contextPaths); + for (const std::string& c : contextPaths) { + if (!fs::exists(c)) Die(std::format("--context-paths: {} does not exist", c)); + } + Gate0(); + std::vector only; + if (a.only && !a.only->empty()) { + only = CommaList(*a.only); + std::vector bad; + for (const std::string& lf : only) { + if (!Plan::IsLens(lf)) bad.push_back(lf); + } + if (!bad.empty()) Die(std::format("--only names unknown lens file(s): {}", Dyn::Join(bad, ", "))); + std::vector isOff; + for (const std::string& lf : only) { + if (std::ranges::contains(off, lf)) isOff.push_back(lf); + } + if (!isOff.empty()) { + Die(std::format("--only names {}, which the config turns off (lenses.{}.run = false): the config is the record - edit it for a test run", Dyn::Join(isOff, ", "), Plan::LensName(isOff.front()))); + } + } + std::vector without; + if (a.without && !a.without->empty()) { + for (const std::string& part : Dyn::Split(*a.without, ",")) without.push_back(Dyn::Strip(part)); + std::vector bad; + for (const std::string& lf : without) { + if (!Plan::IsLens(lf)) bad.push_back(lf); + } + if (!bad.empty()) Die(std::format("--without names unknown lens file(s): {}", Dyn::Join(bad, ", "))); + log(std::format("without: {} - left out of this run by owner ruling (recorded in the dump's args); the run stays a full run", Dyn::Join(without, ", "))); + } + const std::vector gates = Plan::GatesFor(only, without, off); + if (gates.empty()) { + std::vector why; + if (!off.empty()) { + std::vector names; + for (const std::string& lf : off) names.push_back(Plan::LensName(lf)); + why.push_back(std::format("the config turns off {}", Dyn::Join(names, ", "))); + } + if (!without.empty()) why.push_back(std::format("--without drops {}", Dyn::Join(without, ", "))); + if (!only.empty()) why.push_back(std::format("--only keeps {}", Dyn::Join(only, ", "))); + Die(std::format("no lens runs: {}", Dyn::Join(why, "; "))); + } + Json stoppedAt; + Json incomplete = Json::array(); + for (const Plan::Gate& gate : gates) { + const std::vector lensFiles(gate.lenses.begin(), gate.lenses.end()); + const std::vector res = RunGate(gate.title, lensFiles); + std::ranges::copy(res, std::back_inserter(results)); + Checkpoint(gate.title); + if (abort) { + stoppedAt = std::format("{} - ABORTED ({})", gate.title, *abort); + incomplete.push_back({{"gate", gate.title}, {"reason", *abort}}); + break; + } + std::vector notReturned; + Json missingByLens = Json::object(); + for (const Json& r : res) { + if (Dyn::Truthy(r.at("coverageMissing"))) { + notReturned.push_back(Dyn::Str(r.at("lens"))); + missingByLens[Dyn::Str(r.at("lens"))] = r.at("coverageMissing"); + } + } + if (!notReturned.empty()) { + stoppedAt = std::format("{} - INCOMPLETE ({} not returned)", gate.title, Dyn::Join(notReturned, ", ")); + incomplete.push_back({{"gate", gate.title}, {"notReturned", Array(notReturned)}, {"missing", missingByLens}}); + log(std::format("{}: a gate that cannot fire is indistinguishable from a gate that passed - NOT a pass", Dyn::Str(stoppedAt))); + break; + } + if (std::ranges::any_of(res, [](const Json& r) { return r.at("verdict") == "FIX FIRST"; })) { + stoppedAt = gate.title; + break; + } + } + const bool anyFailed = std::ranges::any_of(results, [](const Json& r) { return r.at("verdict") == "FIX FIRST"; }); + if (x > 1 && !a.dryRun && !testMode) { + // The file after this run's store: a --no-reuse run counts its own + // reads there too. + const Json merged = StoreState(); + const std::vector exclude = Strings(cfg.at("functional").at("exclude")); + for (const Json& r : results) { + const std::string lf = Dyn::StrOr(Dyn::Get(r, "lensFile"), ""); + if (!Plan::IsLens(lf) || !Plan::LensOf(lf).reads) continue; + for (const Json& j : Plan::PlanLens(lf, units, lines, {}, exclude).blocks) { + std::int64_t n = x; + if (ReadsNeeded(j) > 1) { + const std::optional k = ReuseKey(j); + n = Plan::ReadsOf(k ? Dyn::Get(merged.at("entries"), *k) : Json()); + } + if (n < x) { + const std::string lens = Dyn::Str(r.at("lens")); + if (!owed.contains(lens)) owed[lens] = {{"units", 0}, {"reads", 0}, {"pending", Json::array()}}; + Json& o = owed[lens]; + o["units"] = o.at("units").get() + 1; + o["reads"] = o.at("reads").get() + (x - n); + o["pending"].push_back(j.at("unit")); + } + } + } + } + std::string overall; + if (a.dryRun) { + overall = "DRY RUN - NOT SEND-READY (no process launched)"; + } else if (testMode) { + overall = "PARTIAL - NOT SEND-READY (test run with --only/--units)"; + } else if (!incomplete.empty()) { + overall = anyFailed ? "FIX FIRST" : "INCOMPLETE - NOT SEND-READY (a gate did not fully run)"; + } else if (anyFailed || Dyn::Truthy(stoppedAt)) { + overall = "FIX FIRST"; + } else if (!owed.empty()) { + std::int64_t owedUnits = 0; + std::int64_t owedReads = 0; + for (const Json& o : owed) { + owedUnits += o.at("units").get(); + owedReads += o.at("reads").get(); + } + overall = std::format("CONFIRMING - NOT SEND-READY (gates clean; {} code unit(s) owe {} clean read(s) to reach {}: run the same line again)", owedUnits, owedReads, x); + } else { + // Gates 1-3 clean = push-ready (Gate 4 is the owner). + overall = "SEND-READY"; + } + return Finish(overall, stoppedAt, incomplete); + } + + std::int32_t Run::Finish(std::string overall, const Json& stoppedAt, Json incomplete) { + if (broker) broker->Close(); + if (Artifact::ArtifactHash("commits", a.artifact) != hash) { + overall = "INCOMPLETE - NOT SEND-READY (artifact changed during the run)"; + incomplete.push_back({{"gate", "end"}, {"reason", "artifact hash changed during the run"}}); + } + Json dump = BuildDump(overall, stoppedAt, incomplete); + if (a.dryRun) dump["dryRun"] = DryReport(); + WriteText(dumpPath, DumpFile(dump)); + if (!testMode && !a.dryRun) { + StoreState(); + if (x > 1 && incomplete.empty() && !abort) { + Json r; + try { + r = Dyn::LoadJsonFile(resumePath); + } catch (const std::exception&) { + r = {{"hash", hash}, {"date", a.date}, {"run", Basename(dumpPath)}, {"after", nullptr}, {"results", results}}; + } + if (Dyn::Get(r, "hash") == hash) { + r["round"] = round; + r["complete"] = true; + WriteReplace(resumePath, DumpFile(r)); + } + } + } + std::println("{}", State::Summary(dump)); + if (a.dryRun) std::println("{}", DumpFile(dump.at("dryRun"))); + std::fflush(stdout); + if (overall == "SEND-READY") return 0; + return overall == "FIX FIRST" ? 1 : 2; + } + + Json Run::BuildDump(std::string_view overall, const Json& stoppedAt, const Json& incomplete) { + const Json cov = Normalize::Matrix(units, results, notReviewed); + std::vector recs; + std::vector runJobs; + { + const std::lock_guard guard(shared); + recs = records; + runJobs = jobsRun; + } + // {lens: records}, in the order first met. + std::vector>> perLens; + for (const Json& r : recs) { + const std::string key = KeyText(Dyn::Get(r, "lens")); + auto found = std::ranges::find_if(perLens, [&](const auto& entry) { return entry.first == key; }); + if (found == perLens.end()) { + perLens.emplace_back(key, std::vector {}); + found = perLens.end() - 1; + } + found->second.push_back(r); + } + std::int64_t firsts = 0; + std::int64_t hits = 0; + const double prefix = 0.8 * (static_cast(Dyn::Length(runLayer)) / 3.7); + for (const Json& r : recs) { + const Json& first = Dyn::Get(r, "firstTurn"); + if (!Dyn::Truthy(first)) continue; + ++firsts; + const Json& read = Dyn::Get(first, "cache_read_input_tokens"); + if ((read.is_null() ? 0.0 : Number(read)) >= prefix) ++hits; + } + Json reusedUnits = Json::object(); + for (const Json& r : results) { + Json reused = Json::array(); + for (const Json& p : r.at("parts")) { + if (Dyn::Truthy(Dyn::Get(p, "reused"))) reused.push_back(Dyn::Truthy(Dyn::Get(p, "unit")) ? p.at("unit") : Json("(whole pass)")); + } + if (!reused.empty()) reusedUnits[Dyn::Str(r.at("lens"))] = reused; + } + Json dispositionsInForce = Json::array(); + for (const Json& d : inForce) { + const auto row = std::ranges::find_if(dispRows, [&](const Json& x) { return x.at("id") == d.at("id"); }); + if (row == dispRows.end()) throw Dyn::Error("StopIteration", ""); + dispositionsInForce.push_back({{"id", d.at("id")}, {"kind", Dyn::Get(d, "kind")}, {"status", row->at("status")}}); + } + Json lapsed = Json::array(); + for (const Json& r : dispRows) { + if (r.at("status") == "lapsed") lapsed.push_back(r); + } + Json contests = Json::array(); + Json violations = Json::array(); + Json crossLens = Json::array(); + for (const Json& r : results) { + for (const Json& c : r.at("contests")) { + Json entry = c; + entry["lens"] = r.at("lens"); + contests.push_back(entry); + } + for (const Json& v : r.at("contractViolations")) violations.push_back(v); + for (const Json& c : r.at("crossLensDischarges")) crossLens.push_back(c); + } + Json perLensUsage = Json::object(); + for (const auto& [lens, list] : perLens) perLensUsage[lens] = State::UsageTotals(list); + Json resumedLenses = Json::array(); + for (const Json& r : results) { + if (Dyn::Truthy(Dyn::Get(r, "resumed"))) resumedLenses.push_back(r.at("lens")); + } + Json stackList = Json::array(); + for (const Json& s : stacks->All()) { + const Json& db = Dyn::Get(s, "dbUrl"); + stackList.push_back({{"id", s.at("id")}, {"appUrl", s.at("appUrl")}, {"dbUrl", Dyn::Truthy(db) ? Json(PgWire::Redact(Dyn::Str(db))) : Json()}, {"login", Dyn::Get(s, "login")}}); + } + Json lensesOff = Json::array(); + for (const std::string& lf : off) lensesOff.push_back(Plan::LensName(lf)); + Json environmentNames = Json::object(); + for (auto it = environments.begin(); it != environments.end(); ++it) environmentNames[Plan::LensName(it.key())] = it.value(); + Json dump = { + {"driver", DriverLine()}, + {"mode", "commits"}, + {"artifact", Dyn::RelPath(a.artifact, repo.string())}, + {"hash", hash}, + {"date", a.date}, + {"args", ArgsJson(a)}, + {"config", Config::Redacted(cfg)}, + {"configPath", Dyn::Get(cfg, "_path")}, + {"overall", overall}, + {"stoppedAt", stoppedAt}, + {"incomplete", incomplete}, + {"reusedUnits", reusedUnits}, + {"dispositionsInForce", dispositionsInForce}, + {"lapsed", lapsed}, + {"contests", contests}, + {"contractViolations", violations}, + {"crossLensDischarges", crossLens}, + {"coverage", cov}, + {"results", results}, + {"processes", recs}, + {"usage", {{"run", State::UsageTotals(recs)}, {"perLens", perLensUsage}, {"prefixHitRate", firsts ? Json(Dyn::Round(static_cast(hits) / static_cast(firsts), 2)) : Json()}}}, + {"estimate", runJobs.empty() ? Json() : Plan::Estimate(runJobs, static_cast(Dyn::Length(runLayer)), lensLayersBytes)}, + {"wallMin", (Platform::Now() - t0) / 60}, + {"scratch", scratch.string()}, + {"logs", Array(logs)}, + {"dumpPath", Dyn::RelPath(dumpPath.string(), repo.string())}, + {"resumedLenses", resumedLenses}, + {"seats", SeatPool::Summary(seats)}, + {"apiBudget", apiCfg}, + {"harness", a.harness}, + {"variant", a.variant ? Json(*a.variant) : Json()}, + {"functional", {{"appUrl", cfg.at("functional").at("appUrl")}, {"loginFixture", loginFixture ? Json(*loginFixture) : Json()}}}, + {"stacks", stackList}, + {"supabase", {{"dbUrl", PgWire::Redact(Dyn::Str(cfg.at("supabase").at("dbUrl")))}}}, + {"lensesOff", lensesOff}, + {"environments", environmentNames}, + }; + if (x > 1) { + Json readThisRun = Json::object(); + for (const Json& r : recs) { + if (!Dyn::Truthy(Dyn::Get(r, "readNo"))) continue; + const std::string lens = KeyText(Dyn::Get(r, "lens")); + if (readThisRun.contains(lens)) continue; + std::int64_t n = 0; + for (const Json& q : recs) { + if (Dyn::Equal(Dyn::Get(q, "lens"), Dyn::Get(r, "lens")) && Dyn::Truthy(Dyn::Get(q, "readNo")) && Dyn::Get(q, "status") == "ok") ++n; + } + readThisRun[lens] = n; + } + dump["reads"] = {{"x", x}, {"round", round}, {"continued", Dyn::Truthy(resume) && !newRound}, {"owed", owed}, {"readThisRun", readThisRun}}; + } + return dump; + } + + Json Run::DryReport() { + // Every appended system prompt starts with the common text + the one + // run layer, one system prompt per lens (two for a gated lens: whole, + // and blocks with the summary), no user message carries the run layer, + // the inlined artifact re-slices to the file. + const fs::path pdir = scratch / "prompts"; + std::vector files; + std::error_code error; + for (fs::directory_iterator it(pdir, error), end; !error && it != end; it.increment(error)) { + const std::string name = it->path().filename().string(); + if (!name.starts_with('.') && name.ends_with(".txt")) files.push_back(it->path()); + } + std::ranges::sort(files); + const std::string runPrefix = std::format("{}{}{}{}", systemText, Prompt::SepRun, runLayer, Prompt::SepLens); + Json badPrefix = Json::array(); + for (const auto& [sha, path] : systemPaths) { + if (!Platform::ReadFile(path).value_or("").starts_with(runPrefix)) badPrefix.push_back(Basename(path)); + } + const std::string runHead = Dyn::Slice(runLayer, 0, 2000); + for (const fs::path& f : files) { + if (Platform::ReadFile(f).value_or("").contains(runHead)) badPrefix.push_back(Basename(f)); + } + std::set> tuples; + std::set models; + for (const Json& j : jobsRun) { + models.insert(Dyn::Str(j.at("model"))); + const auto found = std::ranges::find_if(systemByLens, [&](const auto& entry) { return entry.first == Dyn::Str(j.at("lens")); }); + if (found == systemByLens.end()) continue; + for (const std::string& sha : found->second) tuples.insert({Dyn::Str(j.at("model")), sha, j.at("kind") == "whole" ? "whole" : "block"}); + } + const std::string marker = "\nARTIFACT (frozen, hash "; + const std::size_t at = runLayer.find(marker); + std::vector art; + if (at != std::string::npos) { + const std::string after = runLayer.substr(at + marker.size()); + const std::string body = after.substr(0, after.find("\n\nINVENTORY - ")); + const std::vector artLines = Dyn::Split(body, "\n"); + for (std::size_t i = 1; i < artLines.size(); ++i) art.push_back(Dyn::Slice(artLines[i], 6)); + } + for (const Json& u : units) { + const Json& range = u.at("lines"); + if (!Dyn::Truthy(range.at(0))) continue; + const auto [begin, end] = Dyn::SliceBounds(art.size(), Int(range.at(0)) - 1, Int(range.at(1))); + const std::vector slice(art.begin() + static_cast(begin), art.begin() + static_cast(end)); + if (Dyn::Join(slice, "\n") != Plan::UnitText(lines, u)) Die(std::format("re-slice mismatch on {}", Dyn::Str(u.at("id")))); + } + Json perLens = Json::object(); + for (const auto& [lens, shas] : systemByLens) perLens[lens] = shas.size(); + Json byKind = Json::object(); + for (const std::string_view kind : {"whole", "block", "slot"}) byKind[std::string(kind)] = std::ranges::count_if(jobsRun, [&](const Json& j) { return j.at("kind") == kind; }); + Json byLens = Json::object(); + for (const Json& j : jobsRun) { + const std::string lens = Dyn::Str(j.at("lens")); + byLens[lens] = (byLens.contains(lens) ? byLens.at(lens).get() : 0) + 1; + } + auto passes = [&](auto pred) { + return std::ranges::count_if(jobsRun, [&](const Json& j) { return pred(*Plan::FindEnvironment(Dyn::Str(Dyn::At(environments, Dyn::Str(j.at("lensFile")))))); }); + }; + Json rep = { + {"prompts", files.size()}, + {"systemPrompts", systemPaths.size()}, + {"runLayerPrefixViolations", badPrefix}, + {"lensLayersPerLens", perLens}, + {"modelSystemSchemaTuples", tuples.size()}, + {"models", models.size()}, + {"resliceOk", true}, + {"jobsByKind", byKind}, + {"jobsByLens", byLens}, + {"mcpPasses", passes([](const Plan::Environment& e) { return !e.mcp.empty(); })}, + {"dbPasses", passes([](const Plan::Environment& e) { return e.db; })}, + }; + bool multi = false; + bool atMostTwo = true; + for (const Json& v : perLens) { + if (v.get() > 1) { + multi = true; + atMostTwo = atMostTwo && v.get() <= 2; + } + } + rep["lensLayerNote"] = multi ? "gated lenses show 2 (whole layer + block layer with the summary)" : "one lens layer per lens"; + rep["ok"] = badPrefix.empty() && atMostTwo && tuples.size() <= 2 * systemPaths.size(); + return rep; + } + + bool Run::Storable(const Json& rec) const { + // A block record whose result may be reused: its own pass came back + // without an undischarged finding, and its unit is clean in the merged + // result. + const auto r = std::ranges::find_if(results, [&](const Json& x) { return x.at("lens") == rec.at("lens"); }); + if (r == results.end()) return true; + const Json* own = nullptr; + for (const Json& p : r->at("parts")) { + if (Dyn::Get(p, "label") == rec.at("label")) { + own = &p; + break; + } + } + const Json* cov = nullptr; + for (const Json& c : r->at("coverage")) { + if (c.at("item") == rec.at("unit")) { + cov = &c; + break; + } + } + std::vector citing; + for (const Json& f : r->at("findings")) { + const Json& items = Dyn::Or(Dyn::Get(f, "items"), Dyn::EmptyList()); + if (std::ranges::any_of(items, [&](const Json& i) { return i == rec.at("unit"); })) citing.push_back(&f); + } + if (!own || Dyn::Get(*own, "verdict") != "SEND-READY" || !cov || cov->at("verdict") == "MISSING" || std::ranges::any_of(citing, [](const Json* f) { return !Dyn::Truthy(Dyn::Get(*f, "discharged")); })) return false; + if (std::ranges::contains(Schema::CoverageBlocking, Dyn::Str(cov->at("verdict")))) { + return cov->at("verdict") == "defect" && !citing.empty() && std::ranges::all_of(citing, [](const Json* f) { return Dyn::Truthy(Dyn::Get(*f, "dischargedBy")); }); + } + return true; + } + + Json Run::StoreState() { + // Reusable clean block results -> unit-state.json (full runs only). A + // pass is stored only when the pass ITSELF returned no undischarged + // finding, and a key only when EVERY pass under it in this run is + // storable. The file is merged, not overwritten (2026-10-31). + std::int64_t n = 0; + std::set dirty; + for (const Json& rec : records) { + if (Dyn::Truthy(Dyn::Get(rec, "reuseKey")) && Dyn::Get(rec, "status") == "ok" && Dyn::Get(rec, "kind") != "whole" && !Storable(rec)) dirty.insert(Dyn::Str(rec.at("reuseKey"))); + } + // A defect on a unit's code stands whatever environment found it: the + // clean verdicts the lens holds for the same code in its other + // environments go too, and so do a lens's other whole verdicts when + // its whole pass ran and was not clean. + for (const Json& rec : records) { + if (Dyn::Get(rec, "status") != "ok" || !Dyn::Truthy(Dyn::Get(rec, "otherKeys"))) continue; + const Json* part = nullptr; + if (Dyn::Get(rec, "kind") == "whole") { + const auto found = wholeParts.find(Dyn::Str(rec.at("lens"))); + if (found != wholeParts.end()) part = &found->second; + } + const Json& key = Dyn::Get(rec, "reuseKey"); + if ((key.is_string() && dirty.contains(key.get())) || (part && (part->at("verdict") != "SEND-READY" || Dyn::Truthy(part->at("coverageMissing"))))) { + for (const Json& k : rec.at("otherKeys")) dirty.insert(Dyn::Str(k)); + } + } + Json sets = Json::object(); + for (const Json& rec : records) { + const Json& keyValue = Dyn::Get(rec, "reuseKey"); + if (!Dyn::Truthy(keyValue) || Dyn::Get(rec, "status") != "ok") continue; + const std::string key = Dyn::Str(keyValue); + if (dirty.contains(key)) continue; + const auto r = std::ranges::find_if(results, [&](const Json& x) { return x.at("lens") == rec.at("lens"); }); + if (r == results.end()) continue; + if (Dyn::Get(rec, "kind") == "whole") { + const auto found = wholeParts.find(Dyn::Str(rec.at("lens"))); + if (found != wholeParts.end()) { + const Json& part = found->second; + if (Dyn::Get(part, "status") == "ok" && part.at("verdict") == "SEND-READY" && !Dyn::Truthy(part.at("coverageMissing"))) { + sets[key] = {{"lens", rec.at("lensFile")}, {"unit", nullptr}, {"kind", "whole"}, {"hash", hash}, {"run", Basename(dumpPath)}, {"date", a.date}, {"coverage", part.at("coverage")}, + {"findings", Normalize::Dedupe(part.at("findings").get>())}, {"notes", Normalize::Dedupe(part.at("notes").get>())}, + {"summary", Dyn::Get(part, "summary")}}; + ++n; + } + } + continue; + } + const Json* cov = nullptr; + for (const Json& c : r->at("coverage")) { + if (c.at("item") == rec.at("unit")) { + cov = &c; + break; + } + } + Json citing = Json::array(); + for (const Json& f : r->at("findings")) { + const Json& items = Dyn::Or(Dyn::Get(f, "items"), Dyn::EmptyList()); + if (std::ranges::any_of(items, [&](const Json& i) { return i == rec.at("unit"); })) citing.push_back(f); + } + if (!Storable(rec)) continue; + std::vector notes; + for (const Json& note : r->at("notes")) { + const Json& items = Dyn::Or(Dyn::Get(note, "items"), Dyn::EmptyList()); + if (std::ranges::any_of(items, [&](const Json& i) { return i == rec.at("unit"); })) notes.push_back(note); + } + sets[key] = {{"lens", rec.at("lensFile")}, {"unit", rec.at("unit")}, {"unitSha", Dyn::Get(rec, "unitSha")}, {"hash", hash}, {"run", Basename(dumpPath)}, {"date", a.date}, + {"coverage", cov ? *cov : Json()}, {"findings", citing}, {"notes", Normalize::Dedupe(notes)}}; + ++n; + } + // One clean read per round: the count comes from the FILE at write + // time, so a read another driver stored is kept; a second store of the + // same run, or the relaunch of an unfinished run, adds nothing. + const Json read = {{"hash", h8}, {"round", round}, {"run", Basename(dumpPath)}}; + const std::vector drops(dirty.begin(), dirty.end()); + Json merged = State::SaveState(statePath, sets, drops, read); + // A --no-reuse run keeps its empty view for the reuse decisions; its + // reads are counted from the file. + if (!a.noReuse) state = merged; + log(std::format("unit-state: {} reusable unit results stored ({})", n, Dyn::RelPath(statePath.string(), repo.string()))); + return merged; + } + + std::int32_t Main(Arguments a, const Json& cfg) { + Process::BecomeSubreaper(); + Run run(std::move(a), cfg); + Current = &run; + // `crafter-review stop`: the file is looked at every 2 s, and a stop + // is handled exactly like SIGTERM. + std::jthread watcher([&run](const std::stop_token& token) { + std::mutex sleep; + std::condition_variable_any wake; + while (!token.stop_requested()) { + { + std::unique_lock guard(sleep); + wake.wait_for(guard, token, std::chrono::seconds(2), [] { return false; }); + } + if (token.stop_requested()) return; + const std::optional file = run.StopFile(); + std::error_code error; + if (file && fs::exists(*file, error)) { + Trigger(std::string("crafter-review stop"), 0); + return; + } + } + }); + std::int32_t code = 0; + try { + code = run.Execute(); + } catch (const Preflight& e) { + std::println(std::cerr, "preflight: {}", e.what()); + code = 3; + } + watcher.request_stop(); + watcher.join(); + Current = nullptr; + return code; + } +} diff --git a/implementations/Crafter.Review-SeatPool.cpp b/implementations/Crafter.Review-SeatPool.cpp index 8ec8979..19d0215 100644 --- a/implementations/Crafter.Review-SeatPool.cpp +++ b/implementations/Crafter.Review-SeatPool.cpp @@ -319,6 +319,10 @@ namespace Crafter::Review::SeatPool { return data; } + const fs::path& Ledger::Path() const { + return path; + } + // ---- the seats --------------------------------------------------------------------- Json NewSeat(const std::string& id, const std::string& kind, const std::string& secret) { return { diff --git a/interfaces/Crafter.Review-Run.cppm b/interfaces/Crafter.Review-Run.cppm new file mode 100644 index 0000000..cdaa28f --- /dev/null +++ b/interfaces/Crafter.Review-Run.cppm @@ -0,0 +1,249 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts® + +export module Crafter.Review:Run; +import std; +import :Json; +import :Plan; +import :Pool; +import :SeatPool; +import :Memory; +import :PgWire; +import :State; +import :Platform; + +// Run: the driver for a commit range about to be pushed +// (process/prepush.md). One claude process per inventory unit per lens, +// whole passes where a lens has one (each gating its block passes), +// fail-closed coverage per unit, the matrix, per-unit reuse across hashes, a +// seat pool. +// +// Gates: 1 repo-better-solution (approach) -> 2 repo-functional (the range +// exercised in the running application) -> 3 repo-bug-prover and +// repo-security -> 4 the owner. SEND-READY = push-ready. Exit codes: 0 +// SEND-READY, 1 FIX FIRST, 2 INCOMPLETE / PARTIAL / CONFIRMING (gates clean, +// code units still owe clean reads - run the same line again), 3 preflight. +// +// A Run is built empty and filled field by field, as the tests fill one, so +// its attributes are public; Main builds and runs the real one. +export namespace Crafter::Review::Run { + using Log = std::function; + + // `crafter-review run`'s options: the dump records them under `args`, in + // this order. + struct Arguments { + std::optional config; + std::string mode = "commits"; + std::string artifact; + std::optional hash; + std::optional dispositions; + std::string base; + std::string date; + std::optional trees; + std::string contextPaths; + std::optional only; + std::optional without; + // --units, split on commas. + std::optional> units; + bool warm = false; + std::int64_t concurrency = 128; + std::string effort; + std::int64_t reads = Plan::Reads; + bool noReuse = false; + bool noSalvage = false; + bool dryRun = false; + std::optional claudeBin; + std::string models; + std::optional scratch; + std::optional seats; + std::optional envFile; + bool noProbe = false; + std::optional apiBudget; + std::string harness = "claude"; + std::optional variant; + std::optional opencodeBin; + }; + // vars(a) without `cmd`, the units sorted. + Json ArgsJson(const Arguments& a); + + // die(): "preflight: " on stderr, exit 3. + class Preflight : public std::runtime_error { + public: + using std::runtime_error::runtime_error; + }; + + // (ok, detail): does the application under test answer at ? + // Below 500 is ok. + std::pair AppReady(std::string_view url, std::string_view path = "/", double timeout = 10); + + // `slots`: how many stacks to bring up - as many as free memory holds, + // and no more than the range's browser passes can use at once - and the + // commands for the ones not up yet. (want, commands). + std::pair> SlotsPlan(const Json& cfg, const std::filesystem::path& repo, const std::optional& artifact = std::nullopt, + const std::optional& scratchRoot = std::nullopt, const Log& log = {}); + + class Run { + public: + Run(); + Run(Arguments a, Json cfg); + ~Run(); + Run(const Run&) = delete; + Run& operator=(const Run&) = delete; + + // The whole run: the gates, the dump, the summary; the exit code. + std::int32_t Execute(); + // A signal or `crafter-review stop`: every running pass killed, the + // pool aborted, and the stop stays in force for the gates after. + void Stop(const std::string& why, std::int32_t signal = 0); + // The file `crafter-review stop` writes, once the scratch dir is + // known. + std::optional StopFile() const; + // The handler a signal reaches: false before gate 0 set it up (the + // signal then ends the process). + bool Ready() const; + + // ---- gate 0 and its preflights --------------------------------------- + void Gate0(); + std::map AppLenses(std::span lensFiles) const; + std::map DbLenses(std::span lensFiles) const; + void InstrumentPreflight(std::span lensFiles); + static std::optional MigrationVersion(std::string_view file); + // (problem, note): the migrations at the range HEAD against the ones + // the database applied. + std::pair, std::optional> MigrationsDrift(const std::string& url); + std::unique_ptr LoadStacks(); + std::string StacksLine() const; + std::vector DbStacks(std::span lensFiles) const; + void StackPreflight(std::span lensFiles); + std::optional LoginMissing(const Json& stack) const; + std::optional TreeDrift(); + + // ---- prompts --------------------------------------------------------------- + // (appended system prompt file, stdin user message) of a pass. + std::pair PromptFor(const Json& job); + std::optional CacheKey(Json& job); + bool WarmProbe(const Json& job) const; + + // ---- reuse ----------------------------------------------------------------- + const Plan::CommitCode& RangeCode(); + std::optional ReuseWhole(const std::string& name, Json& job); + std::optional ReuseKey(const Json& job, const std::optional& environment = std::nullopt); + std::vector OtherKeys(const Json& job, const std::optional& key); + std::optional> Salvage(const Json& job); + Json DirEnvironments(const std::filesystem::path& dir); + static std::int64_t DirRound(const std::filesystem::path& dir); + std::int64_t ReadsNeeded(const Json& job) const; + + // ---- one process, a gate ------------------------------------------------- + std::pair ExecJob(const Json& job, std::int64_t attempt); + Json PartFrom(const Json& job, const Json& rec, Json raw); + static Json DryPart(const Json& job); + std::vector RunGate(std::string_view title, std::vector lensFiles); + std::unique_ptr Admission(std::span lensFiles); + std::optional Resumable(const std::string& name, const std::string& lensFile, std::string_view title); + void Checkpoint(std::string_view title); + std::vector PrepareBlocks(const std::string& name, std::vector blocks, std::map>& parts); + + // ---- the end --------------------------------------------------------------- + std::int32_t Finish(std::string overall, const Json& stoppedAt, Json incomplete); + Json BuildDump(std::string_view overall, const Json& stoppedAt, const Json& incomplete); + Json DryReport(); + bool Storable(const Json& rec) const; + Json StoreState(); + + void Message(const std::string& message); + + // ---- the run's state ----------------------------------------------------- + Arguments a; + Json cfg = Json::object(); + std::vector off; + // {lens file: environment}, in the config's lens order: the dump, the + // resume file and the scratch dir's marker write it in that order. + Json environments = Json::object(); + std::filesystem::path repo; + std::vector logs; + double t0 = 0; + std::optional logFile; + std::vector records; + std::vector results; + std::map lensLayersBytes; + std::vector jobsRun; + std::map summaries; + std::optional abort; + Pool::Scheduler* sched = nullptr; + std::optional commitCode; + std::int64_t x = 1; + std::int64_t round = 1; + bool newRound = false; + Json owed = Json::object(); + std::map wholeParts; + std::optional dbChecked; + std::optional dbBypass; + std::optional dbRole; + std::unique_ptr broker; + std::unique_ptr stacks; + bool slotMode = false; + std::vector slotProblems; + std::unique_ptr board; + std::unique_ptr runLock; + std::unique_ptr lifetime; + std::vector trees; + std::vector contextPaths; + std::string hash; + std::string h8; + std::string head; + std::filesystem::path gateDir; + Json units = Json::array(); + std::vector lines; + bool testMode = false; + std::filesystem::path scratch; + std::filesystem::path dumpPath; + std::vector notReviewed; + std::vector dispEntries; + std::vector dispRows; + std::vector inForce; + std::filesystem::path statePath; + Json state = State::EmptyState(); + std::vector salvageDirs; + std::filesystem::path resumePath; + Json resume; + std::map charters; + std::map charterSha; + std::string systemText; + // {sha: path}, in the order first written. + std::vector> systemPaths; + // {lens: the shas of its system files}, lenses in the order first met. + std::vector>> systemByLens; + std::string runLayer; + std::optional loginFixture; + std::string appBlock; + std::map schemas; + std::unique_ptr calibration; + std::unique_ptr sampler; + std::unique_ptr runner; + Json apiCfg = Json::object(); + std::unique_ptr ledger; + std::vector seats; + std::map envMarkers; + std::set> envSkips; + std::map, std::tuple, std::vector, std::map>> migCache; + // The log every line goes to (the tests put their own here). + Log log; + + private: + // What exec_job's threads share with the pool's: the system files, + // the lens layers, the jobs run. + mutable std::recursive_mutex shared; + mutable std::mutex logLock; + std::optional stopped; + std::atomic ready = false; + std::optional stopFile; + }; + + // The `run` command: argv parsed by the CLI; the exit code. + std::int32_t Main(Arguments a, const Json& cfg); + // What SIGINT, SIGTERM and SIGHUP do (Process::InstallSignalThread): + // stop the run once gate 0 is through; before that, and for every other + // command, the process ends, its descendants with it. + void OnSignal(std::int32_t signal); +} diff --git a/interfaces/Crafter.Review-SeatPool.cppm b/interfaces/Crafter.Review-SeatPool.cppm index daace60..90d1c39 100644 --- a/interfaces/Crafter.Review-SeatPool.cppm +++ b/interfaces/Crafter.Review-SeatPool.cppm @@ -66,6 +66,7 @@ export namespace Crafter::Review::SeatPool { Json MarkOut(const std::string& seatId, std::string_view reason, std::optional now = std::nullopt); void Save(); const Json& Data() const; + const std::filesystem::path& Path() const; private: std::filesystem::path path; diff --git a/interfaces/Crafter.Review.cppm b/interfaces/Crafter.Review.cppm index 5ffb78f..9d24c64 100644 --- a/interfaces/Crafter.Review.cppm +++ b/interfaces/Crafter.Review.cppm @@ -31,4 +31,5 @@ export import :SeatPool; export import :OpenCode; export import :Pool; export import :Freeze; +export import :Run; export import :Driver; diff --git a/project.cpp b/project.cpp index 17a6db5..3b3f927 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", @@ -43,9 +43,10 @@ namespace { "interfaces/Crafter.Review-OpenCode", "interfaces/Crafter.Review-Pool", "interfaces/Crafter.Review-Freeze", + "interfaces/Crafter.Review-Run", "interfaces/Crafter.Review-Driver", }; - constexpr std::array Implementations = { + constexpr std::array Implementations = { "implementations/Crafter.Review-Text", "implementations/Crafter.Review-Crypto", "implementations/Crafter.Review-Platform", @@ -72,6 +73,7 @@ namespace { "implementations/Crafter.Review-OpenCode", "implementations/Crafter.Review-Pool", "implementations/Crafter.Review-Freeze", + "implementations/Crafter.Review-Run", "implementations/Crafter.Review-Driver", }; diff --git a/tests/Drivers/main.cpp b/tests/Drivers/main.cpp index a49ebdc..939a3f6 100644 --- a/tests/Drivers/main.cpp +++ b/tests/Drivers/main.cpp @@ -75,7 +75,7 @@ namespace { } }; - std::int32_t Run() { + std::int32_t Checks() { 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()); @@ -488,7 +488,7 @@ int main(int argc, char** argv) { return 0; } try { - return Run(); + return Checks(); } catch (const std::exception& error) { std::println(std::cerr, "drivers: {}", error.what()); return 1; diff --git a/tests/OpenCode/main.cpp b/tests/OpenCode/main.cpp index 2fb038f..ddd25ed 100644 --- a/tests/OpenCode/main.cpp +++ b/tests/OpenCode/main.cpp @@ -49,7 +49,7 @@ namespace { return Dyn::Replace(text, td.string(), ""); } - std::int32_t Run() { + std::int32_t Checks() { std::println("opencode harness + OpenRouter seats: loader, environments, permissions, event parsing, the tool schema"); Platform::SetMode(Platform::Mode::Native); // The directory modes below are the generator's, made under 022. @@ -221,7 +221,7 @@ int main(int argc, char** argv) { const std::vector args(argv, argv + argc); if (args.size() > 1 && args[1] == Process::SuperviseVerb) return Process::SuperviseMain(std::span(args).subspan(2)); try { - return Run(); + return Checks(); } catch (const std::exception& error) { std::println(std::cerr, "opencode: {}", error.what()); return 1;