Catcrafts/Crafter.Review
Watch
2
0
Fork
You've already forked Crafter.Review
 0
Code Issues 7 Pull requests Projects Releases 3 Packages Wiki Activity Actions

feat(pool): the pool - stacks, admission, the runner and the scheduler

Browse source 
The allowlists and the instrument checks (mcp_connected, db_answered,
exercised), the stack keys, Stacks and Admission, the Runner that runs one
claude or opencode process per pass, and the Scheduler with its retries,
rate-window pauses, the prompt-prefix warm-up, the seats and the board.
One thread per pass; each worker runs on its own copy of the job.

Every pass runs under the supervisor, so a timeout or an abort takes its
whole tree down; a timed-out pass reports exit -9, a SIGKILLed child's
code. The helpers the lenses run are this binary's `sql` and `totp`.
:Process gains an Append stream for the OpenCode meta line.

The tests cover the pool and its share of the commits-mode, seat-pool and
several-drivers checks, a fake claude under the supervisor included.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Jorijn van der Graaf 2026-10-31 08:55:00 +01:00
commit 554a5ddd48
Signed by: jorijnvdgraaf
GPG key ID: 2937E59CDCC1BCFB
8 changed files with 2132 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

269
interfaces/Crafter.Review-Pool.cppm Normal file
View file

@ -0,0 +1,269 @@
// SPDX-License-Identifier: AGPL-3.0-only
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
export module Crafter.Review:Pool;
import std;
import :Json;
import :SeatPool;
import :Memory;
import :PgWire;
// One `claude --print` process per pass, a pool with gating, retries,
// rate-window pauses and progress; with seats every process runs on the seat
// picked for it and a seat's limit pauses that seat, not the pool. What else
// holds a launch back is the pass's environment, never its lens (Admission):
// a pass that drives an application waits for a stack of its own (Stacks),
// and every pass waits until it fits in free memory at its environment's
// measured cost. A pass with the database alone shares the first stack:
// every SQL call is its own transaction, rolled back.
//
// Every pass runs under the supervisor (`crafter-review __supervise`), so a
// timeout, an abort or the driver's death takes its whole tree down - the
// MCP server and Chromium with it.
export namespace Crafter::Review::Pool {
using Log = std::function<void(const std::string&)>;
// (status, reason)
using Verdict = std::pair<std::string, std::string>;
// seats.find_claude_bin
std::string DefaultBin();
// The exact Bash prefix a lens with the database under test runs SQL with:
// this binary's `sql`, spelled absolute (allowlisted as a prefix).
std::string SqlCommand();
// The exact Bash line a lens with an application under test runs for a
// one-time code: this binary's `totp <fixture>`; nullopt without one.
std::optional<std::string> TotpCommand(const std::optional<std::string>& fixture);
// Trees outside the working directory: the CLI must be told (--add-dir)
// and Read allowed there, because --setting-sources '' drops the
// project's additionalDirectories and a read there would prompt.
std::vector<std::string> ExtraDirs(std::span<const std::string> trees);
// A pass with the browser that returned ok, fail closed: the Playwright
// server connected with its tools (2026-10-15: --safe-mode had dropped it
// and four units came back n/a), and the browser answered - an empirical
// lens's whole pass must have called it, and no pass may have had every
// browser call fail.
Verdict McpConnected(const Json& parsed, std::string_view kind = "block", bool empirical = true);
// A pass that ran the SQL helper and never got the database's answer had
// no database: its verdict is not evidence from one.
Verdict DbAnswered(const Json& parsed, std::string_view kind = "block", bool empirical = false);
// Every instrument the environment gives must show it was there.
Verdict Exercised(const Json& parsed, std::string_view environment, std::string_view kind = "block", bool empirical = false);
// A stack's name on the board, the same in every driver: its
// application's address, localhost and 127.0.0.1 one host, a default port
// spelled out; a stack without a URL is its id.
std::string StackKey(const Json& stack);
// The read-only tools, git -C on every tree in both spellings, and what
// the environment adds: the browser tools and curl, the SQL helper, and
// with an application the one-time-code command and the config's extras.
std::vector<std::string> AllowedTools(std::string_view lensFile, std::string_view kind, std::span<const std::string> trees, std::span<const std::string> extra = {},
const std::optional<std::string>& loginFixture = std::nullopt, std::string_view environment = "none");
// The stacks a run's passes use (2026-10-22): the one the config names,
// or the slots brought up side by side. A pass whose environment drives an
// application holds a stack of its own from launch to end, least recently
// returned first; another driver's pass holds one too (`held`, the
// StackKey of every stack the other drivers on this machine hold).
class Stacks {
public:
explicit Stacks(std::vector<Json> stacks);
std::size_t Size() const;
// nullptr when there is none.
const Json* First() const;
const std::vector<Json>& All() const;
const std::string& KeyOf(const std::string& id) const;
// Is a stack free (prefer: the one a resumed session ran on - only
// that one will do)?
bool Can(const std::optional<std::string>& prefer = std::nullopt, const std::set<std::string>& held = {}) const;
Json Take(const std::optional<std::string>& prefer = std::nullopt, const std::set<std::string>& held = {});
void Give(const std::string& id);
std::vector<std::string> Free() const;
private:
std::vector<Json> all;
std::map<std::string, std::string> keys;
std::vector<std::string> free;
mutable std::mutex lock;
};
// May the pool launch this pass now? No per-lens number (2026-10-22): an
// environment that drives an application needs a stack of its own, and the
// pass must fit the memory budget - free memory at the gate's start less
// the reserve, less what the passes in flight were admitted at - with the
// machine still holding its cost free beyond the reserve right now. With
// nothing in flight on the machine a pass is admitted whatever the memory
// says, and the log says so once. `others` is what the other drivers'
// passes hold, as the board's last fold left it.
class Admission {
public:
using CostOf = std::function<double(const Json& job)>;
using NeedsStack = std::function<bool(const Json& job)>;
using Available = std::function<std::optional<double>()>;
Admission(CostOf costOf, NeedsStack needsStack, Stacks* stacks = nullptr, std::optional<double> budget = std::nullopt, double reserve = 0, Available available = {}, Log log = {});
bool Can(const Json& job);
// The pass's memory and its stack taken: _admitted, stack, prevStack
// and stackKey on the job.
void Take(Json& job);
void Give(Json& job);
void SetOthers(SeatPool::Others others);
SeatPool::Others OthersNow() const;
std::int64_t Inflight() const;
double Held() const;
private:
CostOf costOf;
NeedsStack needsStack;
Stacks* stacks;
std::optional<double> budget;
double reserve;
Available available;
Log log;
std::int64_t inflight = 0;
double held = 0;
bool warned = false;
SeatPool::Others others;
mutable std::mutex lock;
std::optional<std::string> Prefer(const Json& job) const;
};
struct RunnerOptions {
std::filesystem::path repo;
std::filesystem::path scratch;
// "": REVIEW_GATES_CLAUDE_BIN, else DefaultBin().
std::string bin;
std::string effort = "xhigh";
Log log;
bool dryRun = false;
std::string harness = "claude";
std::optional<std::string> variant;
std::string opencodeBin;
// The config's functional section: the MCP server, the extra tools.
Json functional = Json::object();
std::optional<std::string> loginFixture;
// {lens file: environment}
std::map<std::string, std::string> environments;
// The first stack's: a pass with the database alone runs there.
std::optional<std::string> dbUrl;
Memory::Sampler* sampler = nullptr;
// The database URLs stay in the driver; a db pass gets the broker's
// address and a token.
PgWire::Broker* broker = nullptr;
};
class Runner {
public:
explicit Runner(RunnerOptions options);
// Every running pass stopped (its supervisor takes the tree down); how
// many.
std::int32_t KillChildren();
std::string EnvOf(const Json& job) const;
// A pass's process environment: the seat's (base), plus - when its
// environment reaches the database - the broker's address and the
// token of the database of the stack it holds (none for a stack
// without one), else the run's first stack's.
std::map<std::string, std::string> ProcessEnv(const Json& job, const std::map<std::string, std::string>& base) const;
// The login fixture of the stack it holds, else the run's.
std::optional<std::string> LoginOf(const Json& job) const;
// The --mcp-config of a pass, nullopt for an environment without the
// browser.
std::optional<std::string> McpConfig(const Json& job) const;
std::vector<std::string> Cmd(const Json& job, const std::filesystem::path& systemPath, std::string_view schema, std::span<const std::string> trees, std::int64_t maxTurns, bool persist,
const std::string& sessionId, bool resume = false) const;
// One pass: (record, the structured output or null).
std::pair<Json, Json> Run(const Json& job, const std::string& prompt, const std::filesystem::path& systemPath, std::string_view schema, std::span<const std::string> trees,
std::int64_t maxTurns, std::int64_t timeoutMin, bool persist = false);
// The seats a job's `seat` names, as they were when the pool began:
// only what never changes (id, kind, secret) is read from them.
void SetSeats(const std::vector<Json>& seats);
const std::string& Bin() const;
// The environment every pass starts from.
std::map<std::string, std::string>& Env();
const std::filesystem::path& Scratch() const;
std::filesystem::path opencodeHome;
private:
RunnerOptions options;
std::map<std::string, std::string> env;
std::map<std::string, Json> seatsById;
std::map<std::int64_t, bool> running;
mutable std::mutex lock;
std::pair<Json, Json> RunOpenCode(const Json& job, const std::string& prompt, const std::filesystem::path& systemPath, std::span<const std::string> trees, std::int64_t timeoutMin,
const Json& seat, Json rec, const std::filesystem::path& out, const std::filesystem::path& err, const std::string& sessionId);
std::optional<std::int32_t> Wait(std::int64_t pid, std::int64_t timeoutMin, bool& timedOut);
};
struct SchedulerOptions {
Runner* runner = nullptr;
std::int64_t concurrency = 8;
Log log;
std::optional<std::filesystem::path> progressPath;
double pauseAt = 85;
double maxWaitMin = 360;
// The key of a pass's prompt prefix (nullopt: none), which the driver
// keeps on the job as `cacheKey`; and whether a running pass's first
// response is in.
std::function<std::optional<std::string>(Json& job)> cacheKey;
std::function<bool(const Json& job)> warmProbe;
double warmTtl = 240;
double warmTimeout = 180;
double warmPoll = 5;
std::vector<Json>* seats = nullptr;
// null: from the environment.
Json seatConfig;
Json apiConfig;
SeatPool::Ledger* ledger = nullptr;
std::function<double()> now;
Admission* admission = nullptr;
SeatPool::Board* board = nullptr;
double noRoomWait = 30;
// A stop that came before this pool (a signal, `crafter-review stop`):
// the pool starts aborted.
std::optional<std::string> abort;
};
struct Done {
Json job;
Json rec;
Json raw;
};
// Runs jobs on a pool; onDone(job, rec, raw) -> the follow-up jobs (the
// blocks after a passing whole pass). Retries: error/timeout once (x1.5
// budget for a timeout); limit -> pause until resetsAt (or a backoff) and
// requeue (<= 4); auth or a seven_day rejection -> abort. With seats every
// launch is placed (with the board: placed, admitted and registered under
// its lock, the pool width the machine's cap), every finished process
// settled onto its seat and taken off the board. One thread per pass.
class Scheduler {
public:
using ExecJob = std::function<std::pair<Json, Json>(const Json& job, std::int64_t attempt)>;
using OnDone = std::function<std::vector<Json>(const Json& job, const Json& rec, const Json& raw)>;
explicit Scheduler(SchedulerOptions options);
const std::vector<Done>& Run(std::vector<Json> jobs, const ExecJob& exec, const OnDone& onDone);
// From any thread: the pool stops launching and returns once what runs
// has ended.
void Abort(const std::string& reason);
std::optional<std::string> Aborted() const;
const std::vector<Done>& DoneList() const;
// pause_until: 0 until a pause raised it.
Json PauseUntil() const;
private:
SchedulerOptions o;
std::vector<Done> done;
std::optional<std::string> abort;
Json pauseUntil = 0;
Json roomWait;
double waitedMin = 0;
bool heldOff = false;
std::map<std::pair<std::string, std::string>, double> warm;
std::map<std::pair<std::string, std::string>, std::pair<std::string, double>> warming;
mutable std::mutex lock;
std::condition_variable wake;
bool Capped(const Json& job);
bool Launchable(Json& job, const std::vector<Json*>& running, double now, const std::optional<std::string>& seatId);
void RaisePause(double until);
void Sleep(double seconds);
void Progress(const std::vector<std::string>& running, const std::vector<std::string>& queued);
void Log(const std::string& line) const;
};
}

3
interfaces/Crafter.Review-Process.cppm
View file

@ -32,6 +32,9 @@ export namespace Crafter::Review::Process {
// Opened for reading (stdin) or created and truncated (stdout,
// stderr).
File,
// stdout or stderr after what the file already holds: a line the
// driver wrote before the child starts stays first.
Append,
// A pipe whose other end Start() returns.
Pipe,
};

1
interfaces/Crafter.Review.cppm
View file

@ -29,5 +29,6 @@ export import :Mcp;
export import :Seats;
export import :SeatPool;
export import :OpenCode;
export import :Pool;
export import :Freeze;
export import :Driver;