<?php
use Illuminate\Support\Str;
if($phase==='ledger-cycle-worker'){
    $input=realpath($argv[3]??'');if(!$input||!str_starts_with($input,realpath($s['root']).DIRECTORY_SEPARATOR))throw new LogicException('Private worker input required');$w=json_decode(file_get_contents($input),true,64,JSON_THROW_ON_ERROR);
    if(!preg_match('/^cycle_(correct|close|stale)_[ab]_[a-f0-9]{12}$/D',$w['label'])||!in_array($w['action'],['correct-post','period-close'],true))throw new LogicException('Worker request refused');
    DB::select('SELECT set_config(?,?,false)',['application_name',$w['label']]);DB::statement("SET lock_timeout='30s'");
    try{$r=$command('prepare',$w['action'],$w['body'],$w['key']);$result=['status'=>200,'result'=>$r];}catch(DomainException $e){$result=['status'=>409,'message'=>$e->getMessage()];}
    file_put_contents($input.'.result',json_encode($result,JSON_THROW_ON_ERROR));exit;
}
$output=$s['root'].'/ledger-cycle-race.json';if(file_exists($output))throw new LogicException('Completed race preserved');$prepared=json_decode(file_get_contents($s['root'].'/ledger-cycle-race-prepared.json'),true,64,JSON_THROW_ON_ERROR);
$race=function(string $name,string $action,array $body,int $number=2,?Closure $change=null)use($s,$file){
    $workers=[];$inputs=[];$observed=[];DB::beginTransaction();
    try{
        $parent=(int)DB::selectOne('SELECT pg_backend_pid() pid')->pid;DB::table('companies')->where('id',1)->lockForUpdate()->firstOrFail();if($change)$change();
        foreach(array_slice(['a','b'],0,$number) as $n){$input=$s['root'].'/cycle-'.$name.'-'.$n.'-'.bin2hex(random_bytes(4)).'.json';file_put_contents($input,json_encode(['label'=>'cycle_'.$name.'_'.$n.'_'.$s['tag'],'action'=>$action,'body'=>$body,'key'=>(string)Str::uuid()]));$log=$input.'.log';$p=proc_open([PHP_BINARY,__DIR__.'/payable-history-sgf-proof.php',$file,'ledger-cycle-worker',$input],[0=>['file','NUL','r'],1=>['file',$log,'a'],2=>['file',$log,'a']],$pipes,dirname(__DIR__),null,['bypass_shell'=>true]);if(!is_resource($p))throw new LogicException('Worker launch failed');$workers[$n]=$p;$inputs[$n]=$input;}
        config(['database.connections.cycle_monitor'=>DB::connection()->getConfig()]);$monitor=DB::connection('cycle_monitor');$blocked=[];$until=microtime(true)+15;
        do{
            $monitor->select('SELECT pg_stat_clear_snapshot()');foreach($inputs as $n=>$input){$r=$monitor->selectOne('SELECT pid,wait_event_type,wait_event,query,pg_blocking_pids(pid)::text blockers FROM pg_stat_activity WHERE datname=current_database() AND application_name=?',['cycle_'.$name.'_'.$n.'_'.$s['tag']]);if($r)$observed[$n]=['pid'=>(int)$r->pid,'wait'=>$r->wait_event_type,'event'=>$r->wait_event,'company'=>str_contains($r->query,'"companies"')&&str_contains(strtolower($r->query),'for update'),'blockers'=>json_decode(str_replace(['{','}'],['[',']'],$r->blockers))];}
            foreach($observed as $n=>$r){$reach=in_array($parent,$r['blockers'],true);foreach($observed as $other)if(in_array($other['pid'],$r['blockers'],true)&&in_array($parent,$other['blockers'],true))$reach=true;if($r['wait']==='Lock'&&$r['company']&&$reach)$blocked[$n]=['chain_reaches_parent'=>true,'event'=>$r['event']];}
            if(count($blocked)===$number)break;usleep(20000);
        }while(microtime(true)<$until);
        if(count($blocked)!==$number)throw new LogicException('Actual company lock waits required');
        $change?DB::commit():DB::rollBack();$results=[];foreach($inputs as $n=>$input){$until=microtime(true)+20;while(!file_exists($input.'.result')&&microtime(true)<$until)usleep(20000);if(!file_exists($input.'.result'))throw new LogicException('Worker did not complete after release');$results[$n]=json_decode(file_get_contents($input.'.result'),true,64,JSON_THROW_ON_ERROR);}
        return ['blocked'=>$blocked,'results'=>$results];
    }finally{while(DB::transactionLevel()>0)DB::rollBack();foreach($workers as $p){if(proc_get_status($p)['running'])proc_terminate($p);proc_close($p);}DB::purge('cycle_monitor');}
};
$correct=$race('correct','correct-post',$prepared['correction_body']);
if($correct['results']['a']['status']!==200||$correct['results']['b']['status']!==200||$correct['results']['a']['result']['result_id']!==$correct['results']['b']['result']['result_id']||DB::table('gl_correction_results')->count()!==1||DB::table('gl_journals')->count()!==5)throw new LogicException('Concurrent corrective pair duplicated');
$p=$command('prepare','period-prepare',['period_key'=>'QA-2026-10','reason'=>'Capture before concurrent evidence change.']);$command('validate','period-validate',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent pre-change close review.']);
$stale=$race('stale','period-close',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'reason'=>'Must revalidate after writer commits changed rule.'],1,function(){ $r=(array)DB::table('gl_rules')->where('rule_key','QA-ALLOCATION')->first();unset($r['id']);$r['version']++;$r['reason']='Explicit synthetic concurrent rule version.';DB::table('gl_rules')->insert($r); });
if($stale['results']['a']['status']!==409||DB::table('gl_period_closures')->exists())throw new LogicException('Changed evidence must refuse closing after lock release');
$p=$command('prepare','period-prepare',['period_key'=>'QA-2026-10','reason'=>'Fresh conference after actual concurrent change.']);$command('validate','period-validate',['check_id'=>$p['check_id'],'revision'=>$p['revision'],'decision'=>'approved','reason'=>'Independent review of updated period evidence.']);
$closeBody=['check_id'=>$p['check_id'],'revision'=>$p['revision'],'reason'=>'Apply updated artificial period close once.'];$closed=$race('close','period-close',$closeBody);
if($closed['results']['a']['status']!==200||$closed['results']['b']['status']!==200||$closed['results']['a']['result']['closure_id']!==$closed['results']['b']['result']['closure_id']||DB::table('gl_period_closures')->count()!==1||DB::table('gl_periods')->count()!==2)throw new LogicException('Concurrent close duplicated version');
DB::select('SELECT gl_assert_balanced(id) FROM gl_journals');
file_put_contents($output,json_encode(['passed'=>true,'corrective_pair'=>$correct,'stale_after_committed_change'=>$stale,'period_close'=>$closed,'journals'=>5,'lines'=>10,'correction_results'=>1,'closures'=>1,'period_versions'=>2,'old_check_id'=>$stale['results']['a']['status'],'final_check_id'=>$p['check_id'],'final_revision'=>$p['revision']],JSON_PRETTY_PRINT|JSON_THROW_ON_ERROR));echo "Five real workers: corrective pair once, stale close409 after committed change, fresh concurrent close once.\n";