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;