// 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}, {"kind", a.kind}, {"subsystem", opt(a.subsystem)}, {"artifact", a.artifact}, {"hash", opt(a.hash)}, {"reviewers", a.reviewers}, {"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 ? Json(*a.concurrency) : Json()}, {"effort", a.effort}, {"reads", a.reads}, {"no_reuse", a.noReuse}, {"no_salvage", a.noSalvage}, {"persona_blocks", a.personaBlocks}, {"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 Machine& machine) { 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", artifact ? Freeze::ReviewedBy(*artifact, repo).repo : 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 = machine.available ? machine.available() : Memory::Available(); const std::int64_t reserve = machine.reserve ? machine.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()), reviewed(repo), t0(Platform::Now()) { log = [this](const std::string& message) { Message(message); }; appReady = [](std::string_view url, std::string_view path) { return AppReady(url, path); }; probe = [](std::string_view url) { return PgWire::Probe(url); }; query = [](std::string_view url, std::string_view sql) { return PgWire::Query(url, sql); }; bypassesRls = [](std::string_view url) { return PgWire::BypassesRls(url); }; migrationsAtHead = [this] { return MigrationsAtHead(); }; migrationsDrift = [this](const std::string& url) { return MigrationsDrift(url); }; treeDrift = [this] { return TreeDrift(); }; promptFor = [this](const Json& job) { return PromptFor(job); }; runPass = [this](const Json& job, const std::string& prompt, const fs::path& systemPath, std::string_view schema, std::int64_t maxTurns, std::int64_t timeoutMin, bool persist) { return runner->Run(job, prompt, systemPath, schema, trees, maxTurns, timeoutMin, persist); }; resumable = [this](const std::string& name, const std::string& lensFile, std::string_view title) { return Resumable(name, lensFile, title); }; instrumentPreflight = [this](std::span lensFiles) { InstrumentPreflight(lensFiles); }; } 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, a.mode); environments = Config::Environments(cfg); } bool Run::Kernel() const { return Plan::IsKernelKind(a.kind); } 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(a.kind, 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); if (a.kind == "commits") { // The lenses read the live checkout named first in --trees: it // must be the range HEAD the artifact was cut from, and clean - // anything else is evidence the review never saw (re-checked // before every launch). 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)); } // A reply's gate-state sits beside the draft; a directory artifact's // inside it. gateDir = Dyn::JoinPath(a.kind == "reply" ? Dyn::DirName(a.artifact) : a.artifact, "gate-state"); fs::create_directories(gateDir); Inventory::Built built; try { built = Inventory::Build(a.kind, a.artifact); } catch (const Inventory::InvariantError& e) { Die(e.what()); } catch (const Dyn::Error& e) { // A patch that is not UTF-8 is no artifact to review: a preflight // failure. if (e.Kind() != "UnicodeDecodeError") throw; Die(std::format("the artifact's patches are not UTF-8 ({}) - re-freeze", 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. // The lock, the run numbers and the dump names are the kind's: the // gate-state is shared by every mode of a kind, so one full run per // hash holds across them and their rounds count on together; the dump // says which mode ran (2026-11-01, the first self-review). Json holder; if (!a.dryRun && !testMode) { try { runLock = std::make_unique(gateDir, a.kind, 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, a.kind, h8); if (found) holder = *found; } std::int64_t runN = State::NextRunNumber(gateDir, a.kind, 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; }; // Nor the number of a dump on record: past a killed run's gap, a later // run that bought nothing (no outputs in its dir) would otherwise have // its number taken again and its dump written over. auto dumped = [&](std::int64_t n) { return fs::exists(gateDir / std::format("{}-{}-run-{}.json", a.kind, h8, n)) || fs::exists(gateDir / std::format("{}-{}-run-{}-test.json", a.kind, h8, n)); }; while (!a.dryRun) { const fs::path candidate = scratchRoot / std::format("{}-run-{}", h8, runN); if (!dumped(runN) && !hasOutputs(candidate) && SeatPool::ClaimScratch(candidate)) break; ++runN; } scratch = scratchRoot / std::format("{}-run-{}{}", h8, runN, a.dryRun ? "-dry" : ""); fs::create_directories(scratch); if (a.dryRun) { // Every dry run on a hash shares this dir: the report counts // this run's prompts, not the last one's too. std::error_code error; fs::remove_all(scratch / "prompts", error); } 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("{}-{}-run-{}{}.json", a.kind, 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{}: {}{} {} @ {}", Version(), PromptVersion, a.mode, a.mode == a.kind ? std::string() : std::format(" (a {})", a.kind), Dyn::RelPath(a.artifact, repo.string()), hash)); std::error_code canonical; if (a.kind == "commits" && fs::weakly_canonical(reviewed, canonical) != fs::weakly_canonical(repo, canonical)) log(std::format("repository under review: {} (read at {})", reviewed.string(), trees.front())); 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))); // The mode's lenses: the others are another mode's and say nothing // about this run. const std::vector modeLenses = Plan::KindLenses(a.kind); std::vector lensParts; for (const std::string& lf : modeLenses) { const std::string name = Plan::LensName(lf); lensParts.push_back(std::ranges::contains(off, lf) ? std::format("{} OFF (not in modes.{}.lenses)", name, a.mode) : std::format("{} ({})", name, Dyn::Str(Dyn::At(environments, lf)))); } log(std::format("lenses: {}", Dyn::Join(lensParts, ", "))); std::vector running; for (const std::string& lf : modeLenses) { if (!std::ranges::contains(off, lf)) running.push_back(lf); } const std::vector exclude = Strings(cfg.at("functional").at("exclude")); if (Kernel()) { // The kernel modes excuse nothing: a unit no lens owns is a hole. notReviewed.clear(); log(std::format("inventory: {} units ({})", units.size(), Dyn::RelPath(invPath.string(), repo.string()))); } else { notReviewed = Plan::NotReviewed(units, lines, running, exclude); log(std::format("inventory: {} units ({}); {} units no lens that runs reads", 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, a.kind); 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, ", "))); if (Kernel()) { // The references the artifact cites, prefetched into the // gate-state (a dry run reads an existing index and fetches // nothing), and each reviewer's calibration map. const std::string text = a.kind == "reply" ? Dyn::ReadText(fs::path(a.artifact)) : Dyn::Join(lines, "\n"); refUrls = Refs::ParseRefs(text); const fs::path refsDir = gateDir / "refs"; const fs::path indexPath = gateDir / std::format("refs-{}.json", h8); if (a.dryRun && fs::exists(indexPath)) { refIndex = Dyn::LoadJsonFile(indexPath); log(std::format("refs: {} entries loaded from the existing index (dry run, no fetch)", refIndex.size())); } else { refIndex = Refs::Prefetch(repo, refUrls, refsDir, indexPath, log); } refDigest = Dyn::Truthy(refIndex) ? Refs::Digest(refIndex) : std::string("REFERENCES prefetched: (none - the artifact cites no lore URLs)"); const std::string since = Refs::SinceOf(a.date); corpora.clear(); for (const std::string& name : reviewers) { auto [path, corpus] = Refs::PersonaCorpus(repo, name, refsDir, since, log); if (corpus.empty()) { // A private contributor (no lore record): their mail is // indexed in the ledger's live-threads table. const std::string s = a.subsystem.value_or(""); corpus = std::format("lore.py search for mail from {} since {}: NO RESULTS - this person has no mail on lore in the last year. " "Their mail is PRIVATE: the Live threads table in journal/{}/ledger.md indexes every message with its capture path " "under journal/{}/captures/, and their submissions are frozen under journal/{}/patch/. Calibrate on those, verbatim.", name, since, s, s, s); } corpora[name] = corpus; } anchorText.clear(); for (const Json& u : units) { if (u.at("id") == "anchor") { anchorText = Plan::UnitText(lines, u); break; } } } 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 {}code lenses needs {} clean reads, one per round; this is round {} on this hash{}", Kernel() ? "" : "static ", x, round, newRound || !sameHash ? std::string() : std::string(" (continuing an unfinished run)"))); } // The repository's own lens sections (lenses..section): each // file is appended to its lens's shipped charter (Prompt::Charter), so // the charter sha in every reuse key covers it. The config refused a // path outside the repository and a file the run could not read. const std::map sections = Config::LensSectionsOf(cfg, repo); Prompt::SetLensSections(sections); if (!sections.empty()) { std::vector own; for (const std::string& lf : modeLenses) { if (sections.contains(lf)) own.push_back(Plan::LensName(lf)); } log(std::format("lenses: the repository's sections appended to {}", own.empty() ? std::string("no charter of this mode") : Dyn::Join(own, ", "))); } 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(a.kind); systemPaths.clear(); systemByLens.clear(); runLayer = Kernel() ? Prompt::RunLayerKernel(a.kind, Dyn::RelPath(a.artifact, repo.string()), a.subsystem.value_or(""), a.base, reviewers, a.date, trees, hash, lines, units, refDigest, contextPaths) : Prompt::RunLayer(Dyn::RelPath(a.artifact, repo.string()), a.base, a.date, trees, hash, lines, units, contextPaths, Basename(reviewed), a.subsystem.value_or("")); // 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", a.kind)}, {"whole", Schema::SchemaJson("whole", a.kind)}}; // Memory: every pass's tree measured, its peak recorded per repository // and environment; the budget per gate. calibration = std::make_unique(scratchRoot / "memory.json", reviewed, 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. Only a lens this mode runs counts: `off` is the // mode's own kind's, as every reader of it reads it. 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(modeLenses, it.key()) && !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(), }); // The seats file (--env-file, else the lookup from the working // directory), read once: its placement settings, under the process // environment, go to the placement readers alone (SeatPool::SettingsOf: // ApiConfig here, PoolConfig here and in the Scheduler) and its other // REVIEW_GATES_* lines are named, its tokens over the process // environment's into the seats below (SeatPool::WithFileTokens). const SeatPool::SeatsFile seatsFile = SeatPool::ResolveSeatsFile(a.envFile, fs::current_path()); const SeatPool::Settings settings = SeatPool::SettingsOf(seatsFile); settingsEnv = settings.env; if (seatsFile.path) { log(std::format("settings: {} from {} (under the process environment, for the placement settings)", settings.fromFile.empty() ? std::string("none") : Dyn::Join(settings.fromFile, ", "), seatsFile.path->string())); if (!settings.ignored.empty()) { log(std::format("seats file: {} in {} - no placement setting, read from the process environment alone", Dyn::Join(settings.ignored, ", "), seatsFile.path->string())); } } if (seatsFile.passedOver) log(std::format("seats file: {} is no seats file (no seat, no placement setting) - passed over", seatsFile.passedOver->string())); const std::map processEnv = Platform::Environment(); const std::optional envFile = seatsFile.path ? std::optional(seatsFile.path->string()) : std::nullopt; apiCfg = SeatPool::ApiConfig(settingsEnv); if (a.apiBudget) apiCfg["budgetUsd"] = *a.apiBudget; ledger = std::make_unique(SeatPool::Ledger::Load(SeatPool::LedgerPath(scratchRoot), apiCfg)); try { // The tokens from the file's one read, over the process // environment's; a read that failed raises its error here. seats = SeatPool::LoadSeats(SeatPool::WithFileTokens(processEnv, seatsFile), a.seats.value_or(""), ledger.get(), apiCfg); } catch (const std::invalid_argument& e) { Die(e.what()); } catch (const Dyn::Error& e) { // The errors that mean the file could not be read: the OS errors, a // file that is not UTF-8 (UnicodeDecodeError), and ValueError. constexpr std::array Caught = {"FileNotFoundError", "PermissionError", "IsADirectoryError", "OSError", "UnicodeDecodeError", "ValueError"}; if (!std::ranges::contains(Caught, e.Kind())) 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(settingsEnv, a.effort); board = std::make_unique(SeatPool::BoardPath(), std::string(), std::nullopt, log); std::vector runParts = {a.mode}; if (a.subsystem && !a.subsystem->empty()) runParts.push_back(*a.subsystem); runParts.push_back(Basename(Dyn::RStrip(a.artifact, "/"))); runParts.emplace_back("@"); runParts.push_back(h8); board->Join({{"run", Dyn::Join(runParts, " ")}, {"dump", Basename(dumpPath)}, {"scratch", scratch.string()}, {"effort", a.effort}, {"settings", {{"maxUtil", pc.at("maxUtil")}, {"apiConcurrency", apiCfg.at("concurrency")}, {"lensCostPct", pc.at("lensCostPct")}, // Shown as the driver's width by every driver's `seats`. {"concurrency", a.concurrency ? Json(*a.concurrency) : Json(std::format("{} per account", Dyn::Str(pc.at("oauthConcurrency"))))}}}}, 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, {}, ledger.get()); 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(settingsEnv, a.effort), 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"; })) { std::vector days; for (const Json& s : seats) { if (s.at("kind") == "api") days.push_back(std::format("{} {}", Dyn::Str(s.at("id")), Dyn::Str(Dyn::Get(s, "grantDay")))); } line += std::format("; {}, budget ${:g} per api seat per cycle from its grant day ({}; ledger {})", Dyn::Str(apiCfg.at("order")), Number(apiCfg.at("budgetUsd")), Dyn::Join(days, ", "), 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(); } MigrationsHead Run::MigrationsAtHead() { constexpr std::string_view M = "supabase/migrations/"; 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)})); return {atHead, inRange, headFp}; } 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 (a tree's HEAD does not move during a run - TreeDrift // aborts it first). constexpr std::string_view M = "supabase/migrations/"; const std::pair key = {trees.front(), a.base}; if (!migCache.contains(key)) migCache[key] = migrationsAtHead(); 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 = 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 = 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] = 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(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] = 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 persona = Dyn::StrOr(Dyn::Get(job, "persona"), ""); const auto corpus = corpora.find(persona); const std::string ll = Prompt::LensLayer(lensFile, charters.at(lensFile), Dispositions::Block(inForce, dispRows, lensFile, hash, a.subsystem.value_or("")), env.app ? appBlock : std::string(), summary, db, persona, corpus == corpora.end() ? std::string() : corpus->second); 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, Kernel())}; } 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), a.kind == "series"); if (a.kind == "series" && !a.base.empty()) { // A kernel export carries a base-commit line only when it was // cut with --base, and then in the cover letter: the // submission base the run was given is part of every code // key as well. commitCode->base = commitCode->base.empty() ? a.base : std::format("{},{}", commitCode->base, a.base); } } 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")); if (a.kind != "commits") { // Only the push gates reuse a whole pass; a kernel whole pass // runs on every hash. job["reuseKey"] = nullptr; job["otherKeys"] = Json::array(); return std::nullopt; } 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)); if (Kernel()) { // A series' code unit is keyed on the code (the push gates' // rule, 2026-10-06.6); every text unit on its text and what it // cites: the [n] URLs, the anchor when it is dated, the R-units // it names. if (a.kind == "series" && Dyn::Get(job, "unitKind") == "code") return Plan::CodeReuseKey(job, lensFile, text, RangeCode(), ids, charterSha.at(lensFile), PromptVersion, env); return Plan::ReuseKey(job, lensFile, text, Plan::TextDeps(text, units, lines, refUrls, anchorText), ids, charterSha.at(lensFile), PromptVersion); } 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. if (Kernel()) return {}; 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["persona"] = Dyn::Get(job, "persona"); 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(a.kind, 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 = a.kind == "commits" ? treeDrift() : std::nullopt) { 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] = runPass(job, prompt, systemPath, schemas.at(whole ? "whole" : "block"), 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) { std::vector> specs; for (const std::string& lf : lensFiles) specs.emplace_back(lf, std::string()); return RunGate(title, std::move(specs)); } std::vector Run::RunGate(std::string_view title, std::vector> specs) { // specs -> the merged lens results, one per lens instance. std::vector names; for (const auto& [lf, persona] : specs) names.push_back(Plan::LensName(lf, persona)); log(std::format("{}: {}", title, Dyn::Join(names, ", "))); std::vector resumed; std::vector> live; for (const auto& [lf, persona] : specs) { if (std::optional r = resumable(Plan::LensName(lf, persona), lf, title)) { resumed.push_back(std::move(*r)); } else { live.emplace_back(lf, persona); } } specs = live; // Nothing runs: no stack, database or application is needed for a // result already bought. if (specs.empty()) return resumed; std::vector lensFiles; for (const auto& [lf, persona] : specs) { if (!std::ranges::contains(lensFiles, lf)) lensFiles.push_back(lf); } 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 auto& [lf, persona] : specs) { const std::string name = Plan::LensName(lf, persona); parts[name]; missing[name]; planned[name]; Plan::LensPlan plan = Plan::PlanLens(lf, units, lines, unitFilter, exclude, persona); 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); // A persona's digest is its summary, else the review it // would send. std::string digest = Dyn::StrOr(Dyn::Get(part, "summary"), ""); if (digest.empty()) digest = Dyn::StrOr(Dyn::Get(part, "predictedReview"), ""); summaries[name] = Dyn::Slice(digest, 0, 2500); } std::vector blocks = pendingBlocks[name]; if (Dyn::Truthy(Dyn::Get(job, "persona")) && a.personaBlocks != "all") { // The units the persona's whole pass marked not-mine get no // block pass from it: each becomes a pruned part with that // verdict. std::vector notMine; for (const Json& id : Dyn::Or(Dyn::Get(part, "notMine"), Dyn::EmptyList())) notMine.push_back(Dyn::Str(id)); std::ranges::sort(notMine); notMine.erase(std::ranges::unique(notMine).begin(), notMine.end()); if (!notMine.empty()) { const double ratio = static_cast(notMine.size()) / static_cast(std::max(1, blocks.size())); log(std::format("{}: {} units marked not-mine by the whole pass ({:.0f}%){}; those block passes are pruned", name, notMine.size(), ratio * 100, ratio > 0.5 ? " - WARNING above 50%" : "")); for (const std::string& uid : notMine) { parts[name].push_back({{"partLabel", std::format("{}:not-mine:{}", name, uid)}, {"partKind", "pruned"}, {"partUnit", uid}, {"verdict", "SEND-READY"}, {"findings", Json::array()}, {"notes", Json::array()}, {"coverage", Json::array({{{"item", uid}, {"verdict", "not-mine"}, {"basis", "whole pass: not this reviewer's tree, file or decision"}}})}, {"coverageMissing", Json::array()}, {"reused", nullptr}}); } std::vector kept; for (const Json& b : blocks) { if (!std::ranges::contains(notMine, Dyn::Str(b.at("unit")))) kept.push_back(b); } blocks = kept; } } return PrepareBlocks(name, blocks, parts); }; const bool warm = !a.dryRun && (!testMode || a.warm); std::unique_ptr admission = Admission(lensFiles); 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, .seatConfig = SeatPool::PoolConfig(settingsEnv, a.effort), .apiConfig = apiCfg, .ledger = ledger.get(), .admission = admission.get(), .board = board.get()}); { // A stop before this gate, or while its pool was set up, stays in // force: the pool starts aborted, so the next gate never runs. const std::lock_guard guard(shared); if (stopped) scheduler.Abort(*stopped); 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 auto& [lf, persona] : specs) { const std::string name = Plan::LensName(lf, persona); 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() { Freeze::Reviewed frozen; try { frozen = Freeze::ReviewedBy(a.artifact, repo); } catch (const std::runtime_error& e) { Die(e.what()); } reviewed = frozen.repo; trees = CommaList(a.trees.value_or(frozen.tree)); contextPaths = CommaList(a.contextPaths); for (const std::string& c : contextPaths) { if (!fs::exists(c)) Die(std::format("--context-paths: {} does not exist", c)); } reviewers = CommaList(a.reviewers); if (!reviewers.empty() && std::ranges::contains(off, std::string(Plan::PersonaLens))) { Die(std::format("--reviewers names {} but modes.{}.lenses does not list the persona lens ({}): list it, or drop --reviewers", Dyn::Join(reviewers, ", "), a.mode, Plan::LensName(Plan::PersonaLens))); } Gate0(); const std::vector modeLenses = Plan::KindLenses(a.kind); 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 foreign; for (const std::string& lf : only) { if (!std::ranges::contains(modeLenses, lf)) foreign.push_back(lf); } if (!foreign.empty()) Die(std::format("--only names lens file(s) the {} mode never runs (nothing would run): {}", a.mode, Dyn::Join(foreign, ", "))); 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 modes.{}.lenses does not list: the config is the record - edit it for a test run", Dyn::Join(isOff, ", "), a.mode)); } } 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, ", "))); std::vector foreign; for (const std::string& lf : without) { if (!std::ranges::contains(modeLenses, lf)) foreign.push_back(lf); } if (!foreign.empty()) Die(std::format("--without names lens file(s) the {} mode never runs: {}", a.mode, Dyn::Join(foreign, ", "))); 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(a.kind, only, without, off); if (gates.empty()) { std::vector why; std::vector offHere; for (const std::string& lf : off) { if (std::ranges::contains(modeLenses, lf)) offHere.push_back(lf); } if (offHere.size() == modeLenses.size()) { why.emplace_back(std::format("modes.{}.lenses lists no lens of this kind (every lens is off until a mode lists it)", a.mode)); } else if (!offHere.empty()) { std::vector names; for (const std::string& lf : offHere) 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; } } personaRan = false; const std::string personaFile(Plan::PersonaLens); if (!Dyn::Truthy(stoppedAt) && !reviewers.empty() && (only.empty() || std::ranges::contains(only, personaFile)) && !std::ranges::contains(without, personaFile)) { // The persona gate: one instance of the persona lens per // reviewer, after every other gate passed. std::vector> specs; for (const std::string& r : reviewers) specs.emplace_back(personaFile, r); const std::vector res = RunGate(Plan::PersonaGate, specs); std::ranges::copy(res, std::back_inserter(results)); Checkpoint(Plan::PersonaGate); std::vector notReturned; for (const Json& r : res) { if (Dyn::Truthy(r.at("coverageMissing"))) notReturned.push_back(Dyn::Str(r.at("lens"))); } if (abort) { stoppedAt = std::format("{} - ABORTED ({})", Plan::PersonaGate, *abort); incomplete.push_back({{"gate", Plan::PersonaGate}, {"reason", *abort}}); } else if (!notReturned.empty()) { stoppedAt = std::format("{} - INCOMPLETE ({} not returned)", Plan::PersonaGate, Dyn::Join(notReturned, ", ")); incomplete.push_back({{"gate", Plan::PersonaGate}, {"notReturned", Array(notReturned)}}); } else if (std::ranges::any_of(res, [](const Json& r) { return r.at("verdict") == "FIX FIRST"; })) { stoppedAt = Plan::PersonaGate; } else { personaRan = true; } } 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 if (personaRan || a.kind == "commits") { // commits: no persona gate - the gates clean = push-ready (the // owner is the last gate). overall = "SEND-READY"; } else { overall = "GATES CLEAN - NOT SEND-READY (no reviewers passed, Gate 4 did not run)"; } 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(a.kind, 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")}}); } const std::vector modeLenses = Plan::KindLenses(a.kind); Json lensesOff = Json::array(); for (const std::string& lf : off) { if (std::ranges::contains(modeLenses, lf)) lensesOff.push_back(Plan::LensName(lf)); } Json environmentNames = Json::object(); for (auto it = environments.begin(); it != environments.end(); ++it) { if (std::ranges::contains(modeLenses, it.key())) environmentNames[Plan::LensName(it.key())] = it.value(); } Json refs; if (Kernel()) { std::int64_t fetched = 0; Json failed = Json::array(); for (auto it = refIndex.begin(); it != refIndex.end(); ++it) { const Json& status = Dyn::Get(it.value(), "status"); if (status == "cached" || status == "fetched") ++fetched; if (status == "failed") failed.push_back(it.key()); } refs = {{"index", std::format("gate-state/refs-{}.json", h8)}, {"fetched", fetched}, {"failed", failed}}; } Json dump = { {"driver", DriverLine()}, {"mode", a.mode}, {"kind", a.kind}, {"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()}}}, {"refs", refs}, {"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"); // The line number prefix (`0001| `, five digits from line 10000 on - the self-review of 2026-11-01 found the fixed width). for (std::size_t i = 1; i < artLines.size(); ++i) { const std::size_t bar = artLines[i].find("| "); art.push_back(bar == std::string::npos ? artLines[i] : artLines[i].substr(bar + 2)); } } 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)}}; if (Dyn::Truthy(Dyn::Get(rec, "persona"))) sets[key]["persona"] = rec.at("persona"); ++n; } // One clean read per round, not for a reply: 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. The persona pass is recorded once the persona gate // passed. const Json read = a.kind == "reply" ? Json() : Json({{"hash", h8}, {"round", round}, {"run", Basename(dumpPath)}}); const Json persona = personaRan ? Json({{"hash", hash}, {"reviewers", Array(reviewers)}, {"date", a.date}}) : Json(); const std::vector drops(dirty.begin(), dirty.end()); Json merged = State::SaveState(statePath, sets, drops, read, persona); // 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; // However this ends - a return, or an exception on its way to Main's // handler - the signal handler loses its run, and whatever is left // below this subreaper (a daemon a git call started) goes with it; // never a process it did not start. struct Cleanup { ~Cleanup() { Current = nullptr; Process::KillDescendants(); } } const cleanup; // `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(); return code; } }