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®
// The pool at work: re-placement after a seat's failure, the Runner on a
// fake claude (the seat, the MCP check, resume, the database broker), a
// no-room wait other drivers hold, the machine-wide cap, the stack keys, the
// admission against other drivers' memory and stacks, and the race for the
// last stack.
//
// The test binary is the supervisor every pass runs under (`__supervise`);
// another driver is a live `sleep 300`.
# include <signal.h>
# include <stdlib.h>
# include <sys/stat.h>
# include <unistd.h>
import std ;
import Crafter . Review ;
namespace fs = std : : filesystem ;
using namespace Crafter : : Review ;
using namespace std : : chrono_literals ;
namespace {
std : : int32_t Failures = 0 ;
const std : : string Fn ( Plan : : FunctionalLens ) ;
void Check ( std : : string_view name , bool cond , std : : string_view detail = { } ) {
std : : println ( " {}{}{} " , cond ? " ok " : " FAIL " , name , ! cond & & ! detail . empty ( ) ? std : : format ( " ({}) " , detail ) : std : : string ( ) ) ;
if ( ! cond ) + + Failures ;
}
void Write ( const fs : : path & path , std : : string_view text ) {
fs : : create_directories ( path . parent_path ( ) ) ;
std : : ofstream ( path , std : : ios : : binary ) < < text ;
}
std : : string Read ( const fs : : path & path ) {
return Platform : : ReadFile ( path ) . value_or ( " " ) ;
}
Json With ( Json value , const Json & changes ) {
for ( auto it = changes . begin ( ) ; it ! = changes . end ( ) ; + + it ) value [ it . key ( ) ] = it . value ( ) ;
return value ;
}
bool AnyContains ( const std : : vector < std : : string > & lines , std : : string_view part ) {
return std : : ranges : : any_of ( lines , [ & ] ( const std : : string & line ) { return line . contains ( part ) ; } ) ;
}
// The log a pool writes into from its threads.
struct Logs {
std : : mutex lock ;
std : : vector < std : : string > lines ;
Pool : : Log Sink ( ) {
return [ this ] ( const std : : string & line ) {
const std : : lock_guard guard ( lock ) ;
lines . push_back ( line ) ;
} ;
}
std : : vector < std : : string > Copy ( ) {
const std : : lock_guard guard ( lock ) ;
return lines ;
}
void Clear ( ) {
const std : : lock_guard guard ( lock ) ;
lines . clear ( ) ;
}
} ;
struct Sleeper {
std : : int64_t pid = Process : : Start ( { . argv = { " sleep " , " 300 " } } ) . pid ;
std : : string key = std : : format ( " {}:{} " , pid , Platform : : ProcStart ( pid ) . value_or ( 0 ) ) ;
void Kill ( ) const {
Process : : Signal ( pid , SIGKILL ) ;
Process : : Wait ( pid ) ;
}
} ;
double Seconds ( std : : chrono : : steady_clock : : time_point since ) {
return std : : chrono : : duration < double > ( std : : chrono : : steady_clock : : now ( ) - since ) . count ( ) ;
}
const Pool : : Scheduler : : OnDone NoFollowUp = [ ] ( const Json & , const Json & , const Json & ) { return std : : vector < Json > { } ; } ;
void TestPool ( ) {
std : : println ( " pool: admission (stacks, memory), a no-room wait ends on the next completion, one pass per cold prefix goes first " ) ;
std : : mutex lock ;
std : : map < std : : string , std : : int32_t > running = { { " F " , 0 } , { " S " , 0 } } ;
std : : map < std : : string , std : : int32_t > peak = { { " F " , 0 } , { " S " , 0 } } ;
auto execJob = [ & ] ( const Json & job , std : : int64_t ) {
const std : : string lens = job . at ( " lens " ) . get < std : : string > ( ) ;
{
const std : : lock_guard guard ( lock ) ;
peak [ lens ] = std : : max ( peak [ lens ] , + + running [ lens ] ) ;
}
std : : this_thread : : sleep_for ( 150 ms ) ;
{
const std : : lock_guard guard ( lock ) ;
- - running [ lens ] ;
}
return std : : pair < Json , Json > ( { { " label " , job . at ( " label " ) } , { " status " , " ok " } } , nullptr ) ;
} ;
std : : vector < Json > slots ;
for ( const std : : int32_t n : { 1 , 2 } ) slots . push_back ( { { " id " , std : : format ( " slot {} " , n ) } , { " appUrl " , std : : format ( " http://localhost:{} " , n ) } , { " apiUrls " , Json : : array ( ) } , { " dbUrl " , nullptr } , { " login " , nullptr } , { " slot " , true } } ) ;
Pool : : Stacks stacks ( slots ) ;
auto plenty = [ ] ( ) - > std : : optional < double > { return 1e6 ; } ;
Pool : : Admission adm ( [ ] ( const Json & ) { return 100.0 ; } , [ ] ( const Json & j ) { return j . at ( " lens " ) = = " F " ; } , & stacks , 10'000 .0 , 0 , plenty ) ;
std : : vector < Json > jobs ;
for ( std : : int32_t i = 0 ; i < 6 ; + + i ) jobs . push_back ( { { " label " , std : : format ( " f{} " , i ) } , { " lens " , " F " } } ) ;
for ( std : : int32_t i = 0 ; i < 3 ; + + i ) jobs . push_back ( { { " label " , std : : format ( " s{} " , i ) } , { " lens " , " S " } } ) ;
Pool : : Scheduler s ( { . concurrency = 8 , . log = [ ] ( const std : : string & ) { } , . admission = & adm } ) ;
s . Run ( jobs , execJob , NoFollowUp ) ;
std : : vector < std : : string > free = stacks . Free ( ) ;
std : : ranges : : sort ( free ) ;
Check ( " admission: the passes that need a stack never outnumber the stacks (2), the rest fill the pool, everything finishes, every stack is back " ,
s . DoneList ( ) . size ( ) = = 9 & & peak [ " F " ] = = 2 & & peak [ " S " ] = = 3 & & free = = std : : vector < std : : string > { " slot 1 " , " slot 2 " } & & adm . Inflight ( ) = = 0 & & adm . Held ( ) = = 0 ,
std : : format ( " F {} S {} " , peak [ " F " ] , peak [ " S " ] ) ) ;
running = { { " F " , 0 } , { " S " , 0 } } ;
peak = { { " F " , 0 } , { " S " , 0 } } ;
Pool : : Admission memory ( [ ] ( const Json & ) { return 300.0 ; } , [ ] ( const Json & ) { return false ; } , nullptr , 1000.0 , 0 , plenty ) ;
jobs . clear ( ) ;
for ( std : : int32_t i = 0 ; i < 7 ; + + i ) jobs . push_back ( { { " label " , std : : format ( " s{} " , i ) } , { " lens " , " S " } } ) ;
Pool : : Scheduler m ( { . concurrency = 8 , . log = [ ] ( const std : : string & ) { } , . admission = & memory } ) ;
m . Run ( jobs , execJob , NoFollowUp ) ;
Check ( " admission: memory - passes at 300 MB in a 1000 MB budget run three at once, and all of them finish " , m . DoneList ( ) . size ( ) = = 7 & & peak [ " S " ] = = 3 , std : : format ( " S {} " , peak [ " S " ] ) ) ;
Logs logs ;
const double now = Platform : : Now ( ) ;
std : : vector < Json > seat = { { { " id " , " 3 " } , { " kind " , " oauth " } , { " secret " , " x " } , { " state " , " ok " } , { " util " , 80.0 } , { " binding " , " five_hour " } , { " resetsAt " , nullptr } ,
{ " windows " , { { " five_hour " , { { " utilization " , 80.0 } , { " resetsAt " , now + 3600 } } } , { " seven_day " , { { " utilization " , 30.0 } , { " resetsAt " , now + 86400 } } } } } ,
{ " pauseUntil " , 0.0 } , { " running " , 0 } , { " procs " , 0 } , { " spentUsd " , 0.0 } , { " error " , nullptr } , { " probedWith " , nullptr } } } ;
Json cfg = SeatPool : : PoolConfig ( { } ) ;
cfg [ " lensCostPct " ] = 2 ;
cfg [ " maxUtil " ] = 85 ;
Pool : : Scheduler r ( { . concurrency = 8 , . log = logs . Sink ( ) , . seats = & seat , . seatConfig = cfg , . apiConfig = SeatPool : : ApiConfig ( { } ) } ) ;
const auto t0 = std : : chrono : : steady_clock : : now ( ) ;
jobs . clear ( ) ;
for ( std : : int32_t i = 0 ; i < 3 ; + + i ) jobs . push_back ( { { " label " , std : : format ( " r{} " , i ) } } ) ;
r . Run ( jobs , [ ] ( const Json & j , std : : int64_t ) {
std : : this_thread : : sleep_for ( 200 ms ) ;
return std : : pair < Json , Json > ( { { " label " , j . at ( " label " ) } , { " status " , " ok " } } , nullptr ) ;
} , NoFollowUp ) ;
const std : : vector < std : : string > waited = logs . Copy ( ) ;
Check ( " a no-room wait ends on the next completion, not at the window reset " ,
r . DoneList ( ) . size ( ) = = 3 & & Seconds ( t0 ) < 5 & & AnyContains ( waited , " no seat has room " ) & & AnyContains ( waited , " seat room is re-checked " ) & & r . PauseUntil ( ) = = 0 , Dyn : : Join ( waited , " | " ) ) ;
2026-10-31 17:33:11 +01:00
// 2026-10-31: no machine-wide width by default - each account takes at
// most 128 passes at once, every driver's counted.
{
std : : vector < Json > two ;
for ( const std : : string_view id : { std : : string_view ( " 1 " ) , std : : string_view ( " 2 " ) } ) {
Json s = SeatPool : : NewSeat ( std : : string ( id ) , " oauth " , std : : format ( " t{} " , id ) ) ;
s [ " state " ] = " ok " ;
s [ " windows " ] = { { " five_hour " , { { " utilization " , 10 } , { " resetsAt " , now + 3600 } } } , { " seven_day " , { { " utilization " , 10 } , { " resetsAt " , now + 86400 } } } } ;
two . push_back ( std : : move ( s ) ) ;
}
std : : mutex lock ;
std : : map < std : : string , std : : int64_t > inSeat ;
std : : map < std : : string , std : : int64_t > peakSeat ;
std : : int64_t inFlight = 0 ;
std : : int64_t peak = 0 ;
Pool : : Scheduler wide ( { . log = [ ] ( const std : : string & ) { } , . seats = & two , . seatConfig = SeatPool : : PoolConfig ( { } ) , . apiConfig = SeatPool : : ApiConfig ( { } ) } ) ;
std : : vector < Json > many ;
for ( std : : int32_t i = 0 ; i < 300 ; + + i ) many . push_back ( { { " label " , std : : format ( " w{} " , i ) } } ) ;
wide . Run ( many , [ & ] ( const Json & j , std : : int64_t ) {
const std : : string seat = Dyn : : Str ( j . at ( " seat " ) ) ;
{
const std : : lock_guard guard ( lock ) ;
peakSeat [ seat ] = std : : max ( peakSeat [ seat ] , + + inSeat [ seat ] ) ;
peak = std : : max ( peak , + + inFlight ) ;
}
std : : this_thread : : sleep_for ( 300 ms ) ;
{
const std : : lock_guard guard ( lock ) ;
- - inSeat [ seat ] ;
- - inFlight ;
}
return std : : pair < Json , Json > ( { { " label " , j . at ( " label " ) } , { " status " , " ok " } } , nullptr ) ;
} , NoFollowUp ) ;
Check ( " no machine-wide width by default: two accounts take 128 passes each, 256 at once, and all 300 finish " ,
wide . DoneList ( ) . size ( ) = = 300 & & peak = = 256 & & peakSeat [ " 1 " ] = = 128 & & peakSeat [ " 2 " ] = = 128 , std : : format ( " peak {}, per seat {} and {} " , peak , peakSeat [ " 1 " ] , peakSeat [ " 2 " ] ) ) ;
}
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 : : map < std : : string , double > started ;
std : : map < std : : string , double > answered ;
logs . Clear ( ) ;
Pool : : Scheduler w ( { . concurrency = 8 ,
. log = logs . Sink ( ) ,
. cacheKey = [ ] ( Json & j ) - > std : : optional < std : : string > {
const Json & k = Dyn : : Get ( j , " key " ) ;
return k . is_null ( ) ? std : : nullopt : std : : optional < std : : string > ( k . get < std : : string > ( ) ) ;
} ,
. warmProbe = [ & ] ( const Json & j ) {
const std : : lock_guard guard ( lock ) ;
return answered . contains ( j . at ( " label " ) . get < std : : string > ( ) ) ;
} ,
. warmPoll = 0.02 } ) ;
jobs . clear ( ) ;
for ( std : : int32_t i = 0 ; i < 5 ; + + i ) jobs . push_back ( { { " label " , std : : format ( " a{} " , i ) } , { " key " , " A " } , { " respondAfter " , i = = 0 ? 0.3 : 0.05 } } ) ;
jobs . push_back ( { { " label " , " n0 " } , { " key " , nullptr } } ) ;
w . Run ( jobs , [ & ] ( const Json & j , std : : int64_t ) {
const std : : string label = j . at ( " label " ) . get < std : : string > ( ) ;
{
const std : : lock_guard guard ( lock ) ;
started [ label ] = Platform : : Now ( ) ;
}
std : : this_thread : : sleep_for ( std : : chrono : : duration < double > ( Dyn : : Has ( j , " respondAfter " ) ? j . at ( " respondAfter " ) . get < double > ( ) : 0.05 ) ) ;
{
const std : : lock_guard guard ( lock ) ;
answered [ label ] = Platform : : Now ( ) ;
}
std : : this_thread : : sleep_for ( 300 ms ) ;
return std : : pair < Json , Json > ( { { " label " , j . at ( " label " ) } , { " status " , " ok " } } , nullptr ) ;
} , NoFollowUp ) ;
const double first = answered . at ( " a0 " ) ;
bool followed = true ;
for ( std : : int32_t i = 1 ; i < 5 ; + + i ) {
const double at = started . at ( std : : format ( " a{} " , i ) ) ;
followed = followed & & first < = at & & at < first + 0.3 ;
}
Check ( " warm-up: the first pass of a cold prefix starts alone, the others follow its first response, a pass without a key is never held " ,
w . DoneList ( ) . size ( ) = = 6 & & followed & & started . at ( " n0 " ) < first & & AnyContains ( logs . Copy ( ) , " a0 goes first " ) ) ;
}
void TestSeatPool ( const fs : : path & td ) {
std : : println ( " seat pool: re-placement on another seat, Runner.run on a fake claude " ) ;
const Json acfg = SeatPool : : ApiConfig ( { { " REVIEW_GATES_API_BUDGET_USD " , " 10 " } , { " REVIEW_GATES_API_LENS_COST_USD " , " 1 " } , { " REVIEW_GATES_API_CONCURRENCY " , " 2 " } } ) ;
const Json cfg = Seats : : ConfigFromEnv ( { } ) ;
Logs logs ;
std : : mutex lock ;
std : : vector < std : : pair < std : : string , std : : string > > order ;
std : : vector < Json > pool = { SeatPool : : NewSeat ( " api1 " , " api " , " k " ) , SeatPool : : NewSeat ( " 1 " , " oauth " , " t1 " ) } ;
for ( Json & s : pool ) s [ " state " ] = " ok " ;
SeatPool : : Ledger ledger ( td / " l2.json " , acfg ) ;
Pool : : Scheduler sch ( { . concurrency = 4 , . log = logs . Sink ( ) , . seats = & pool , . seatConfig = cfg , . apiConfig = With ( acfg , { { " budgetUsd " , 100 } } ) , . ledger = & ledger } ) ;
sch . Run ( { { { " label " , " j1 " } } } , [ & ] ( const Json & job , std : : int64_t ) {
{
const std : : lock_guard guard ( lock ) ;
order . emplace_back ( job . at ( " label " ) . get < std : : string > ( ) , job . at ( " seat " ) . get < std : : string > ( ) ) ;
}
if ( job . at ( " seat " ) = = " api1 " ) return std : : pair < Json , Json > ( { { " label " , job . at ( " label " ) } , { " status " , " credits " } , { " reason " , " Your credit balance is too low " } , { " costUsd " , 0.0 } , { " rateLimit " , Json : : object ( ) } } , nullptr ) ;
return std : : pair < Json , Json > ( { { " label " , job . at ( " label " ) } , { " status " , " ok " } , { " costUsd " , 0.5 } , { " rateLimit " , { { " five_hour " , { { " utilization " , 30 } , { " resetsAt " , 9'999'999'999LL } } } } } } , nullptr ) ;
} , NoFollowUp ) ;
std : : map < std : : string , std : : string > statuses ;
for ( const Pool : : Done & d : sch . DoneList ( ) ) statuses [ d . rec . at ( " label " ) . get < std : : string > ( ) ] = d . rec . at ( " status " ) . get < std : : string > ( ) ;
Check ( " a process that fails on its seat is placed again on another; the first seat is out " ,
statuses = = std : : map < std : : string , std : : string > { { " j1 " , " ok " } } & & pool [ 0 ] . at ( " state " ) = = " out " & & ! order . empty ( ) & & order . front ( ) . second = = " api1 " & & order . back ( ) . second = = " 1 " ) ;
const fs : : path fake = td / " claude " ;
const fs : : path argsf = td / " args.txt " ;
Write ( fake , " #!/bin/sh \n printf \" %s \\ n \" \" $@ \" > \" $FAKE_ARGS \" \n printf \" %s \" \" $REVIEW_GATES_DB \" > \" $FAKE_ARGS.db \" \n cat > /dev/null \n "
" if [ -n \" $FAKE_MCP \" ]; then echo '{ \" type \" : \" system \" , \" subtype \" : \" init \" , \" mcp_servers \" :[{ \" name \" : \" playwright \" , \" status \" : \" connected \" }], \" tools \" :[ \" Read \" , \" mcp__playwright__browser_snapshot \" ]}'; "
" echo '{ \" type \" : \" assistant \" , \" message \" :{ \" role \" : \" assistant \" , \" content \" :[{ \" type \" : \" tool_use \" , \" id \" : \" t1 \" , \" name \" : \" mcp__playwright__browser_snapshot \" , \" input \" :{}}]}}'; "
" echo '{ \" type \" : \" user \" , \" message \" :{ \" role \" : \" user \" , \" content \" :[{ \" type \" : \" tool_result \" , \" tool_use_id \" : \" t1 \" , \" content \" : \" - page \" }]}}'; "
" elif [ -n \" $FAKE_SQL \" ]; then echo '{ \" type \" : \" system \" , \" subtype \" : \" init \" , \" mcp_servers \" :[], \" tools \" :[ \" Read \" , \" Bash \" ]}'; printf \" %s \\ n \" \" $FAKE_SQL \" ; "
" else echo '{ \" type \" : \" system \" , \" subtype \" : \" init \" , \" mcp_servers \" :[], \" tools \" :[ \" Read \" ]}'; fi \n "
" echo '{ \" type \" : \" result \" , \" subtype \" : \" success \" , \" is_error \" :false, \" result \" : \" ok \" , \" structured_output \" :{ \" verdict \" : \" SEND-READY \" , \" findings \" :[], \" notes \" :[], \" coverage \" :[]}, "
" \" usage \" :{ \" input_tokens \" :1, \" output_tokens \" :1}, \" total_cost_usd \" :0.01}' \n " ) ;
fs : : permissions ( fake , fs : : perms : : owner_all | fs : : perms : : group_read | fs : : perms : : group_exec | fs : : perms : : others_read | fs : : perms : : others_exec ) ;
setenv ( " FAKE_ARGS " , argsf . c_str ( ) , 1 ) ;
const fs : : path rscratch = td / " rscratch " ;
const std : : string cwd = fs : : current_path ( ) . string ( ) ;
Pool : : Runner runner ( { . repo = cwd , . scratch = rscratch , . bin = fake . string ( ) , . effort = " xhigh " , . log = logs . Sink ( ) , . dryRun = false ,
. functional = { { " mcp " , { { " command " , " npx " } , { " args " , nullptr } } } , { " extraTools " , Json : : array ( ) } } , . environments = { { Fn , " browser " } } } ) ;
runner . SetSeats ( { SeatPool : : NewSeat ( " api1 " , " api " , " k1 " ) , SeatPool : : NewSeat ( " api2 " , " api " , " k2 " ) } ) ;
const fs : : path sysf = td / " sys.md " ;
Write ( sysf , " x " ) ;
const std : : vector < std : : string > trees = { " . " } ;
auto args = [ & ] { return Dyn : : Split ( Read ( argsf ) , " \n " ) ; } ;
auto has = [ ] ( const std : : vector < std : : string > & list , std : : string_view item ) { return std : : ranges : : contains ( list , item ) ; } ;
auto [ rec , raw ] = runner . Run ( { { " label " , " f1 " } , { " model " , " claude-opus-5-5 " } , { " lensFile " , " repo-bug-prover.md " } , { " kind " , " block " } , { " seat " , " api1 " } , { " attempt " , 1 } } , " the prompt " , sysf , " {} " , trees , 20 , 1 ) ;
std : : vector < std : : string > a = args ( ) ;
Check ( " Runner.run on a fake claude: ok, the seat on the record, persisted, no --mcp-config for a static pass, environment none " ,
rec . at ( " status " ) = = " ok " & & rec . at ( " seat " ) = = " api1 " & & ! has ( a , " --no-session-persistence " ) & & has ( a , " --session-id " ) & & ! has ( a , " --mcp-config " ) & & rec . at ( " mcp " ) = = false & & rec . at ( " environment " ) = = " none " ,
DumpSpaced ( rec ) ) ;
std : : tie ( rec , raw ) = runner . Run ( { { " label " , " f2 " } , { " model " , " claude-opus-5-5 " } , { " lensFile " , Fn } , { " kind " , " block " } , { " seat " , " api2 " } , { " attempt " , 1 } } , " the prompt " , sysf , " {} " , trees , 20 , 1 ) ;
a = args ( ) ;
Check ( " ...a functional pass whose init line shows no connected server is an error (no browser = no functional evidence), with --mcp-config passed, its playwright output dir made and environment browser " ,
rec . at ( " mcp " ) = = true & & rec . at ( " environment " ) = = " browser " & & rec . at ( " status " ) = = " error " & & rec . at ( " reason " ) . get < std : : string > ( ) . contains ( " did not connect " ) & & has ( a , " --mcp-config " )
& & fs : : is_directory ( rscratch / " playwright " / " f2 " ) ) ;
// The runner captured its environment at construction; the fake reads
// this one.
runner . Env ( ) [ " FAKE_MCP " ] = " 1 " ;
std : : tie ( rec , raw ) = runner . Run ( { { " label " , " f2 " } , { " model " , " claude-opus-5-5 " } , { " lensFile " , Fn } , { " kind " , " block " } , { " seat " , " api2 " } , { " attempt " , 1 } } , " the prompt " , sysf , " {} " , trees , 20 , 1 ) ;
runner . Env ( ) . erase ( " FAKE_MCP " ) ;
Check ( " ...and ok when the server connected " , rec . at ( " status " ) = = " ok " & & ! raw . is_null ( ) ) ;
const std : : string sid1 = rec . at ( " sessionId " ) . get < std : : string > ( ) ;
Write ( rscratch / " seat-api2 " / " projects " / " -home-repo " / ( std : : format ( " {}.jsonl " , sid1 ) ) , " {} \n " ) ;
std : : tie ( rec , raw ) = runner . Run ( { { " label " , " f2 " } , { " model " , " claude-opus-5-5 " } , { " lensFile " , Fn } , { " kind " , " block " } , { " seat " , " api1 " } , { " attempt " , 2 } , { " resumeSession " , sid1 } , { " resumeSeat " , " api2 " } } ,
" the prompt " , sysf , " {} " , trees , 20 , 1 ) ;
a = args ( ) ;
const auto resumeAt = std : : ranges : : find ( a , " --resume " ) ;
Check ( " resumed pass: --resume <id>, the transcript carried to the new seat, the resume prompt " ,
rec . at ( " sessionId " ) = = sid1 & & resumeAt ! = a . end ( ) & & resumeAt + 1 ! = a . end ( ) & & * ( resumeAt + 1 ) = = sid1 & & fs : : exists ( rscratch / " seat-api1 " / " projects " / " -home-repo " / ( std : : format ( " {}.jsonl " , sid1 ) ) )
& & Read ( rscratch / " prompts " / " f2.txt " ) = = SeatPool : : ResumePrompt ) ;
// A pass with the database under test: the broker's address and a token
// in its process alone, never the URL; and no evidence when no SQL call
// reached the database.
const std : : string durl = " postgresql://postgres:hunter2@127.0.0.1:54322/postgres " ;
PgWire : : Broker dbr ;
dbr . Start ( ) ;
Pool : : Runner rdb ( { . repo = cwd , . scratch = td / " dscratch " , . bin = fake . string ( ) , . effort = " xhigh " , . log = logs . Sink ( ) , . dryRun = false ,
. environments = { { " repo-bug-prover.md " , " supabase " } } , . dbUrl = durl , . broker = & dbr } ) ;
rdb . SetSeats ( { SeatPool : : NewSeat ( " api1 " , " api " , " k1 " ) } ) ;
auto sqlLines = [ ] ( const std : : string & content ) {
return std : : format ( " {} \n {} " , DumpSpaced ( { { " type " , " assistant " } ,
{ " message " , { { " role " , " assistant " } , { " content " , Json : : array ( { { { " type " , " tool_use " } , { " id " , " q1 " } , { " name " , " Bash " } , { " input " , { { " command " , " /app/bin/crafter-review sql <<'SQL' \n select 1; \n SQL " } } } } } ) } } } } ) ,
DumpSpaced ( { { " type " , " user " } , { " message " , { { " role " , " user " } , { " content " , Json : : array ( { { { " type " , " tool_result " } , { " tool_use_id " , " q1 " } , { " content " , content } } } ) } } } } ) ) ;
} ;
rdb . Env ( ) [ " FAKE_SQL " ] = sqlLines ( std : : format ( " {} - cannot connect to 127.0.0.1:54322 " , PgWire : : NoDatabase ) ) ;
std : : tie ( rec , raw ) = rdb . Run ( { { " label " , " d1 " } , { " model " , " claude-opus-5-5 " } , { " lensFile " , " repo-bug-prover.md " } , { " kind " , " block " } , { " seat " , " api1 " } , { " attempt " , 1 } } , " the prompt " , sysf , " {} " , trees , 20 , 1 ) ;
const std : : string argsText = Read ( argsf ) ;
const std : : string gotEnv = Read ( std : : format ( " {}.db " , argsf . string ( ) ) ) ;
Check ( " Runner.run, supabase: the db pass's process gets the broker and its token (no URL, no password, on no command line); a pass whose every SQL call missed the database is an error with no result " ,
gotEnv = = dbr . Address ( durl ) & & dbr . UrlOf ( gotEnv . substr ( gotEnv . rfind ( ' # ' ) + 1 ) ) = = durl & & ! ( gotEnv + argsText ) . contains ( " hunter2 " ) & & has ( Dyn : : Split ( argsText , " \n " ) , " --safe-mode " ) & & rec . at ( " status " ) = = " error "
& & rec . at ( " reason " ) . get < std : : string > ( ) . contains ( " reached the database " ) & & raw . is_null ( ) & & Dyn : : Equal ( rec . at ( " db " ) , { { " calls " , 1 } , { " answered " , 0 } } ) & & rec . at ( " environment " ) = = " supabase " ,
DumpSpaced ( rec ) ) ;
rdb . Env ( ) [ " FAKE_SQL " ] = sqlLines ( std : : format ( " {} to postgres on 127.0.0.1:54322 as postgres - ONE transaction \n SELECT 1 " , PgWire : : Answered ) ) ;
std : : tie ( rec , raw ) = rdb . Run ( { { " label " , " d1 " } , { " model " , " claude-opus-5-5 " } , { " lensFile " , " repo-bug-prover.md " } , { " kind " , " block " } , { " seat " , " api1 " } , { " attempt " , 1 } } , " the prompt " , sysf , " {} " , trees , 20 , 1 ) ;
rdb . Env ( ) . erase ( " FAKE_SQL " ) ;
Check ( " ...and ok once the database answered " , rec . at ( " status " ) = = " ok " & & ! raw . is_null ( ) & & Dyn : : Equal ( rec . at ( " db " ) , { { " calls " , 1 } , { " answered " , 1 } } ) ) ;
std : : tie ( rec , raw ) = rdb . Run ( { { " label " , " d2 " } , { " model " , " claude-opus-5-5 " } , { " lensFile " , " repo-better-solution.md " } , { " kind " , " block " } , { " seat " , " api1 " } , { " attempt " , 1 } } , " the prompt " , sysf , " {} " , trees , 20 , 1 ) ;
Check ( " ...while a lens without the database gets no broker at all " , Read ( std : : format ( " {}.db " , argsf . string ( ) ) ) . empty ( ) & & rec . at ( " status " ) = = " ok " & & rec . at ( " environment " ) = = " none " ) ;
dbr . Close ( ) ;
unsetenv ( " FAKE_ARGS " ) ;
}
void TestDrivers ( const fs : : path & td ) {
std : : println ( " several drivers at once: the pool against the board - a wait other drivers hold, the machine's cap, stacks and memory across drivers " ) ;
const fs : : path path = td / " seat-board.json " ;
Logs logs ;
const double now = Platform : : Now ( ) ;
const Json cfg = SeatPool : : PoolConfig ( { } ) ;
const Json acfg = SeatPool : : ApiConfig ( { { " REVIEW_GATES_API_CONCURRENCY " , " 8 " } , { " REVIEW_GATES_API_LENS_COST_USD " , " 1 " } } ) ;
auto mk = [ & ] ( const std : : string & id , const std : : string & secret , std : : int64_t fh , std : : int64_t sd ) {
Json s = SeatPool : : NewSeat ( id , " oauth " , secret ) ;
s [ " state " ] = " ok " ;
s [ " windows " ] = { { " five_hour " , { { " utilization " , fh } , { " resetsAt " , now + 3600 } } } , { " seven_day " , { { " utilization " , sd } , { " resetsAt " , now + 86400 } } } } ;
const auto [ util , binding ] = Seats : : SeatUtil ( s . at ( " windows " ) ) ;
s [ " util " ] = util ;
s [ " binding " ] = binding ;
return s ;
} ;
std : : deque < Json > jobs ;
std : : mutex jobsLock ;
auto chooserFor = [ & ] ( const std : : string & id , const std : : string & label , const Json & extra = Json : : object ( ) ) {
return SeatPool : : Board : : Chooser ( [ & jobs , & jobsLock , id , label , extra ] ( const std : : vector < Json * > & ranked , const SeatPool : : Others & ) - > std : : optional < std : : pair < Json * , Json * > > {
for ( Json * s : ranked ) {
if ( s - > at ( " id " ) = = id ) {
const std : : lock_guard guard ( jobsLock ) ;
jobs . push_back ( With ( { { " label " , label } , { " timeoutMin " , 20 } } , extra ) ) ;
return std : : pair ( s , & jobs . back ( ) ) ;
}
}
return std : : nullopt ;
} ) ;
} ;
auto place = [ & ] ( SeatPool : : Board & board , Json & seat , const Json & c , const SeatPool : : Board : : Chooser & chooser ) {
std : : vector < Json > one = { seat } ;
board . Place ( one , now , c , acfg , chooser ) ;
seat = one . front ( ) ;
} ;
SeatPool : : Board A ( path , { } , std : : nullopt , logs . Sink ( ) ) ;
A . Join ( { { " run " , " A " } } , now ) ;
const Sleeper child2 ;
SeatPool : : Board D ( path , child2 . key , std : : nullopt , logs . Sink ( ) ) ;
D . Join ( { { " run " , " D " } } , now ) ;
const Json cfg3 = With ( cfg , { { " lensCostPct " , 2 } , { " maxUtil " , 85 } } ) ;
std : : vector < Json > seat3 = { mk ( " 3 " , " t3 " , 80 , 30 ) } ;
Json sd3 = mk ( " 3 " , " t3 " , 80 , 30 ) ;
place ( D , sd3 , cfg3 , chooserFor ( " 3 " , " h0 " ) ) ;
// 80 + 4 held: 80 + 4 + 2 > 85, no room for A until D's passes finish.
place ( D , sd3 , cfg3 , chooserFor ( " 3 " , " h1 " ) ) ;
auto execJob = [ & ] ( const Json & job , std : : int64_t ) {
std : : this_thread : : sleep_for ( 100 ms ) ;
return std : : pair < Json , Json > ( { { " label " , job . at ( " label " ) } , { " status " , " ok " } , { " rateLimit " , { { " five_hour " , { { " utilization " , 81 } , { " resetsAt " , now + 3600 } } } } } } , nullptr ) ;
} ;
{
std : : jthread freer ( [ & ] {
std : : this_thread : : sleep_for ( 500 ms ) ;
D . Finish ( sd3 , " h0 " , Platform : : Now ( ) ) ;
D . Finish ( sd3 , " h1 " , Platform : : Now ( ) ) ;
} ) ;
logs . Clear ( ) ;
Pool : : Scheduler sch ( { . concurrency = 8 , . log = logs . Sink ( ) , . seats = & seat3 , . seatConfig = cfg3 , . apiConfig = acfg , . board = & A , . noRoomWait = 0.2 } ) ;
const auto t0 = std : : chrono : : steady_clock : : now ( ) ;
sch . Run ( { { { " label " , " a0 " } , { " timeoutMin " , 20 } } } , execJob , NoFollowUp ) ;
const double dt = Seconds ( t0 ) ;
const Json bd = Dyn : : LoadJsonFile ( path ) ;
const Json & inflight = bd . at ( " seats " ) . at ( seat3 [ 0 ] . at ( " boardKey " ) . get < std : : string > ( ) ) . at ( " inflight " ) ;
bool mine = false ;
for ( auto it = inflight . begin ( ) ; it ! = inflight . end ( ) ; + + it ) mine = mine | | it . key ( ) . starts_with ( A . Me ( ) ) ;
const std : : vector < std : : string > lines = logs . Copy ( ) ;
Check ( " a no-room wait held by another driver's passes ends when they finish, not at the window reset " ,
sch . DoneList ( ) . size ( ) = = 1 & & dt < 5 & & AnyContains ( lines , " hold the room " ) & & ! mine , std : : format ( " {:.1f}s {} " , dt , Dyn : : Join ( lines , " | " ) ) ) ;
Check ( " ...and the finished pass's reading is on the board " , bd . at ( " seats " ) . at ( seat3 [ 0 ] . at ( " boardKey " ) . get < std : : string > ( ) ) . at ( " windows " ) . at ( " five_hour " ) . at ( " utilization " ) = = 81 ) ;
}
place ( D , sd3 , cfg3 , chooserFor ( " 3 " , " w0 " ) ) ;
// Two of D's on the machine: a width of 2 holds A's pass until one
// finishes.
place ( D , sd3 , cfg3 , chooserFor ( " 3 " , " w1 " ) ) ;
{
std : : jthread freer ( [ & ] {
std : : this_thread : : sleep_for ( 500 ms ) ;
D . Finish ( sd3 , " w0 " , Platform : : Now ( ) ) ;
} ) ;
logs . Clear ( ) ;
std : : vector < Json > seat3b = { mk ( " 3 " , " t3 " , 10 , 30 ) } ;
Pool : : Scheduler sch ( { . concurrency = 2 , . log = logs . Sink ( ) , . seats = & seat3b , . seatConfig = cfg3 , . apiConfig = acfg , . board = & A , . noRoomWait = 0.2 } ) ;
const auto t0 = std : : chrono : : steady_clock : : now ( ) ;
sch . Run ( { { { " label " , " a1 " } , { " timeoutMin " , 20 } } } , execJob , NoFollowUp ) ;
const double dt = Seconds ( t0 ) ;
const std : : vector < std : : string > lines = logs . Copy ( ) ;
Check ( " the machine-wide cap holds a launch while other drivers fill it, and releases it when one of theirs finishes " ,
sch . DoneList ( ) . size ( ) = = 1 & & 0.4 < dt & & dt < 5 & & AnyContains ( lines , " the cap is 2 " ) , std : : format ( " {:.1f}s {} " , dt , Dyn : : Join ( lines , " | " ) ) ) ;
}
D . Finish ( sd3 , " w1 " , Platform : : Now ( ) ) ;
// Memory and stacks across drivers (2026-10-31).
Check ( " stack keys: one application whatever the spelling (localhost/127.0.0.1, case, trailing slash, the default port); another port another stack; no URL, its id " ,
Pool : : StackKey ( { { " id " , " a " } , { " appUrl " , " http://localhost:5173/ " } } ) = = Pool : : StackKey ( { { " id " , " b " } , { " appUrl " , " HTTP://127.0.0.1:5173 " } } )
& & Pool : : StackKey ( { { " id " , " b " } , { " appUrl " , " HTTP://127.0.0.1:5173 " } } ) = = " http://127.0.0.1:5173 " & & Pool : : StackKey ( { { " id " , " c " } , { " appUrl " , " http://localhost " } } ) = = " http://127.0.0.1:80 "
& & Pool : : StackKey ( { { " id " , " d " } , { " appUrl " , " http://localhost:5273 " } } ) ! = Pool : : StackKey ( { { " id " , " a " } , { " appUrl " , " http://localhost:5173 " } } ) & & Pool : : StackKey ( { { " id " , " s1 " } } ) = = " s1 " ) ;
// What the board's fold left in A.others when another driver had two
// passes in flight (test_drivers checks the fold itself).
SeatPool : : Others held ;
held . count = 2 ;
held . mb = 1050.0 ;
held . stacks = { " http://127.0.0.1:5173 " } ;
auto plenty = [ ] ( ) - > std : : optional < double > { return 1e6 ; } ;
Pool : : Stacks stx ( { { { " id " , " slot 1 " } , { " appUrl " , " http://localhost:5173/ " } } , { { " id " , " slot 2 " } , { " appUrl " , " http://localhost:5273 " } } } ) ;
Pool : : Admission adm ( [ ] ( const Json & j ) { return j . at ( " mb " ) . get < double > ( ) ; } , [ ] ( const Json & j ) { return Dyn : : Truthy ( Dyn : : Get ( j , " app " ) ) ; } , & stx , 2000.0 , 0 , plenty , logs . Sink ( ) ) ;
adm . SetOthers ( held ) ;
Json j1 = { { " label " , " x " } , { " mb " , 840 } , { " app " , true } } ;
const bool ok = adm . Can ( j1 ) ;
adm . Take ( j1 ) ;
Check ( " Admission: a stack another driver holds is not free here - the pass takes the other; their memory counts against the budget beside this driver's (1050 + 840 + 200 > 2000, + 100 fits) " ,
ok & & j1 . at ( " stack " ) . at ( " id " ) = = " slot 2 " & & j1 . at ( " stackKey " ) = = " http://127.0.0.1:5273 " & & ! adm . Can ( { { " label " , " y " } , { " mb " , 840 } , { " app " , true } } ) & & ! adm . Can ( { { " label " , " z " } , { " mb " , 200 } } )
& & adm . Can ( { { " label " , " z " } , { " mb " , 100 } } ) ,
DumpSpaced ( Dyn : : Get ( j1 , " stack " ) ) ) ;
adm . Give ( j1 ) ;
logs . Clear ( ) ;
const Json big = { { " label " , " b " } , { " mb " , 5000 } } ;
const bool waits = ! adm . Can ( big ) ;
adm . SetOthers ( SeatPool : : Others ( ) ) ;
Check ( " Admission: nothing of this driver's in flight but other drivers' passes on the machine - a pass over the budget waits for them instead of running alone; with the machine idle it runs alone " ,
waits & & adm . Can ( big ) & & ! j1 . contains ( " stackKey " ) & & logs . Copy ( ) . empty ( ) ) ;
// The pool: a stack another driver holds - the pass waits (re-checked,
// not spun on) and runs when theirs finishes.
Json sE = mk ( " 1 " , " t1 " , 10 , 50 ) ;
place ( D , sE , cfg , chooserFor ( " 1 " , " hold " , { { " _admitted " , 840.0 } , { " stackKey " , " http://127.0.0.1:5173 " } } ) ) ;
Pool : : Stacks stw ( { { { " id " , " slot 1 " } , { " appUrl " , " http://localhost:5173 " } } } ) ;
Pool : : Admission admw ( [ ] ( const Json & ) { return 100.0 ; } , [ ] ( const Json & ) { return true ; } , & stw , 10'000 .0 , 0 , plenty , logs . Sink ( ) ) ;
double seenAt = 0 ;
Json seenEntry ;
const std : : string keyE = sE . at ( " boardKey " ) . get < std : : string > ( ) ;
auto execBoard = [ & ] ( const Json & job , std : : int64_t ) {
seenAt = Platform : : Now ( ) ;
seenEntry = Dyn : : Get ( Dyn : : LoadJsonFile ( path ) . at ( " seats " ) . at ( keyE ) . at ( " inflight " ) , std : : format ( " {}/{} " , A . Me ( ) , job . at ( " label " ) . get < std : : string > ( ) ) ) ;
std : : this_thread : : sleep_for ( 100 ms ) ;
return std : : pair < Json , Json > ( { { " label " , job . at ( " label " ) } , { " status " , " ok " } } , nullptr ) ;
} ;
{
std : : jthread freer ( [ & ] {
std : : this_thread : : sleep_for ( 600 ms ) ;
D . Finish ( sE , " hold " , Platform : : Now ( ) ) ;
} ) ;
logs . Clear ( ) ;
std : : vector < Json > seats = { mk ( " 1 " , " t1 " , 10 , 50 ) } ;
Pool : : Scheduler sch ( { . concurrency = 8 , . log = logs . Sink ( ) , . seats = & seats , . seatConfig = cfg , . apiConfig = acfg , . admission = & admw , . board = & A , . noRoomWait = 0.2 } ) ;
const double t0 = Platform : : Now ( ) ;
sch . Run ( { { { " label " , " w0 " } , { " timeoutMin " , 20 } } } , execBoard , NoFollowUp ) ;
const Json bd = Dyn : : LoadJsonFile ( path ) ;
bool mine = false ;
for ( const Json & s : bd . at ( " seats " ) ) {
for ( auto it = Dyn : : Or ( Dyn : : Get ( s , " inflight " ) , Dyn : : EmptyDict ( ) ) . begin ( ) ; it ! = Dyn : : Or ( Dyn : : Get ( s , " inflight " ) , Dyn : : EmptyDict ( ) ) . end ( ) ; + + it ) mine = mine | | it . key ( ) . starts_with ( A . Me ( ) ) ;
}
const std : : vector < std : : string > lines = logs . Copy ( ) ;
const auto once = std : : ranges : : count_if ( lines , [ ] ( const std : : string & l ) { return l . contains ( " nothing can launch " ) ; } ) ;
Check ( " the pool: a pass whose only stack another driver holds waits, re-checked every no_room_wait and logged once, and launches when theirs finishes - on the board with its memory and its stack " ,
sch . DoneList ( ) . size ( ) = = 1 & & 0.5 < seenAt - t0 & & seenAt - t0 < 5 & & once = = 1 & & AnyContains ( lines , " stack(s) http://127.0.0.1:5173 " ) & & Dyn : : Equal ( Dyn : : Get ( seenEntry , " mb " ) , 100.0 )
& & Dyn : : Get ( seenEntry , " stack " ) = = " http://127.0.0.1:5173 " & & stw . Can ( ) & & ! mine ,
std : : format ( " {:.2f}s {} {} " , seenAt - t0 , DumpSpaced ( seenEntry ) , Dyn : : Join ( lines , " | " ) ) ) ;
}
// Two drivers racing for the last stack: one gets it.
const Sleeper child3 ;
SeatPool : : Board F ( path , child3 . key , std : : nullopt , logs . Sink ( ) ) ;
F . Join ( { { " run " , " F " } } , now ) ;
std : : mutex picksLock ;
std : : vector < bool > picks ;
auto racer = [ & ] ( SeatPool : : Board & board , const std : : string & label ) {
Pool : : Stacks st ( { { { " id " , " slot 1 " } , { " appUrl " , " http://localhost:5999 " } } } ) ;
Pool : : Admission ad ( [ ] ( const Json & ) { return 100.0 ; } , [ ] ( const Json & ) { return true ; } , & st , 10'000 .0 , 0 , plenty ) ;
Json job = { { " label " , label } , { " timeoutMin " , 20 } } ;
std : : vector < Json > one = { mk ( " 1 " , " t1 " , 10 , 50 ) } ;
const SeatPool : : Board : : Placed got = board . Place ( one , now , cfg , acfg , [ & ] ( const std : : vector < Json * > & cands , const SeatPool : : Others & others ) - > std : : optional < std : : pair < Json * , Json * > > {
ad . SetOthers ( others ) ;
std : : this_thread : : sleep_for ( 50 ms ) ;
if ( ! ad . Can ( job ) ) return std : : nullopt ;
ad . Take ( job ) ;
return std : : pair ( cands . front ( ) , & job ) ;
} ) ;
const std : : lock_guard guard ( picksLock ) ;
picks . push_back ( got . picked . has_value ( ) ) ;
} ;
{
const std : : jthread ta ( racer , std : : ref ( A ) , std : : string ( " ra " ) ) ;
const std : : jthread tf ( racer , std : : ref ( F ) , std : : string ( " rf " ) ) ;
}
Check ( " two drivers racing for the last stack on the machine: exactly one gets it " , std : : ranges : : count ( picks , true ) = = 1 ) ;
F . Leave ( ) ;
child3 . Kill ( ) ;
A . Leave ( ) ;
child2 . Kill ( ) ;
}
}
int main ( int argc , char * * argv ) {
const std : : vector < std : : string_view > args ( argv , argv + argc ) ;
if ( args . size ( ) > 1 & & args [ 1 ] = = Process : : SuperviseVerb ) return Process : : SuperviseMain ( std : : span ( args ) . subspan ( 2 ) ) ;
Platform : : SetMode ( Platform : : Mode : : Native ) ;
const fs : : path td = fs : : temp_directory_path ( ) / std : : format ( " crafter-review-scheduler-{} " , Platform : : PublishedPid ( ) ) ;
fs : : remove_all ( td ) ;
fs : : create_directories ( td ) ;
// No check reaches the machine's own board: the drivers running on it are
// real.
setenv ( " REVIEW_GATES_BOARD " , ( td / " seat-board.json " ) . c_str ( ) , 1 ) ;
try {
TestPool ( ) ;
TestSeatPool ( td ) ;
TestDrivers ( td ) ;
} catch ( const std : : exception & error ) {
std : : println ( std : : cerr , " scheduler: {} " , error . what ( ) ) ;
return 1 ;
}
fs : : remove_all ( td ) ;
return Failures = = 0 ? 0 : 1 ;
}