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