<?php
require_once __DIR__ . '/../web/server/worker.php';
$originalCwd=getcwd();$tmp=sys_get_temp_dir().'/exoplasim-worker-'.bin2hex(random_bytes(5));mkdir($tmp,0700,true);
$storage=['root'=>$tmp,'database'=>'data/lab.sqlite3','jobs'=>'jobs','runtime'=>'runtime','catalog'=>'catalog/schema.json'];
foreach(['data','jobs','runtime','catalog'] as $dir) mkdir($tmp.'/'.$dir,0750,true);
$schema=['fields'=>[],'model'=>[],'run'=>[],'model_defaults'=>[],'run_defaults'=>[]];file_put_contents($tmp.'/catalog/schema.json',json_encode($schema));
$science=$tmp.'/science';mkdir($science,0750,true);
$script=<<<'PY'
import argparse,json,os,pathlib
p=argparse.ArgumentParser();p.add_argument('--config');p.add_argument('--job');a=p.parse_args();j=json.load(open(a.job));print('ENV_MARKER='+os.environ.get('EXOPLASIM_TEST_INHERIT','missing'),flush=True);d=('e' if j['kind']=='planet' else 'f')*32;root=pathlib.Path(j['directory'])/'artifacts'/'datasets'/d;root.mkdir(parents=True,exist_ok=True);json.dump({'id':d,'fields':[]},open(root/'index.json','w'));event={'cycle':1,'cycles':1,'dataset':d};event.update({'planet':True,'spec':j['spec']} if j['kind']=='planet' else {});print(json.dumps(event),flush=True)
PY;
file_put_contents($science.'/model.py',$script);file_put_contents($science.'/planet.py',$script);
$config=['workspace'=>'local','storage'=>$storage,'python'=>'/usr/bin/python3','science_scripts'=>$science,'project_root'=>dirname(__DIR__),'poll_seconds'=>1,'job_timeout_seconds'=>60,'minimum_free_bytes'=>0,'queue_limit'=>10];
file_put_contents($tmp.'/config.json',json_encode($config));putenv('EXOPLASIM_CONFIG='.$tmp.'/config.json');
function expect($condition,$message){if(!$condition)throw new RuntimeException($message);}
try {
    $worker=new LabWorker();
    expect($worker->acquire(),'First worker must acquire flock.');
    $second=new LabWorker();expect(!$second->acquire(),'Second worker must not acquire flock.');
    $db=LabDb::connection();$now=gmdate('c');
    $db->exec("INSERT INTO jobs(id,owner,name,spec,status,total_cycles,created_at,updated_at) VALUES('".str_repeat('a',32)."','local','bad','{\"invalid\":true}','queued',1,'$now','$now')");
    $installation=fopen(lab_config()['runtime_lock'],'c');flock($installation,LOCK_EX);
    expect($worker->claim()===null,'Installation lock must prevent a worker claim.');
    expect($db->query("SELECT status FROM jobs WHERE id='".str_repeat('a',32)."'")->fetchColumn()==='queued','Installation must leave the queued job intact.');
    fclose($installation);
    $claimed=$worker->claim();expect($claimed['id']===str_repeat('a',32),'Worker must claim queued job after installation lock is released.');
    $worker->recover();$status=$db->query("SELECT status FROM jobs WHERE id='".str_repeat('a',32)."'")->fetchColumn();expect($status==='interrupted','Recovery must interrupt running job.');
    $db->exec("UPDATE jobs SET status='running' WHERE id='".str_repeat('a',32)."'");
    $reflection=new ReflectionClass(LabWorker::class);$method=$reflection->getMethod('run');$method->setAccessible(true);
    $method->invoke($worker,['id'=>str_repeat('a',32),'owner'=>'local','name'=>'bad','kind'=>'model','spec'=>['invalid'=>true],'total_cycles'=>1]);
    $row=$db->query("SELECT status,error FROM jobs WHERE id='".str_repeat('a',32)."'")->fetch();expect($row['status']==='failed'&&str_contains($row['error'],'validation'),'Invalid stored spec must fail before task launch.');
    putenv('EXOPLASIM_TEST_INHERIT=inherited-value');chdir($tmp);putenv('EXOPLASIM_CONFIG=config.json');$id=str_repeat('b',32);$spec=json_encode(['model'=>[],'run'=>[],'parameters'=>[]]);$s=$db->prepare("INSERT INTO jobs(id,owner,name,kind,spec,status,total_cycles,created_at,updated_at) VALUES(?,?,?,'model',?,'running',1,?,?)");$s->execute([$id,'local','test',$spec,$now,$now]);
    $method->invoke($worker,['id'=>$id,'owner'=>'local','name'=>'test','kind'=>'model','spec'=>json_decode($spec,true),'total_cycles'=>1]);
    $complete=$db->prepare('SELECT status,progress,completed_cycles FROM jobs WHERE id=?');$complete->execute([$id]);$result=$complete->fetch();$complete->closeCursor();expect($result['status']==='completed'&&(float)$result['progress']===100.0&&(int)$result['completed_cycles']===1,'Worker must drain final event and complete job.');
    $published=json_decode(file_get_contents($tmp.'/jobs/'.$id.'/artifacts/index.json'),true);expect(($published['latest']??null)===str_repeat('f',32),'PHP worker must publish the job results index.');
    expect(str_contains(file_get_contents($tmp.'/jobs/'.$id.'/worker.log'),'ENV_MARKER=inherited-value'),'Worker must preserve inherited process environment.');
    $planetId=str_repeat('e',32);$planetSpec=json_encode(['model'=>[],'run'=>[],'parameters'=>[],'surface'=>[]]);$p=$db->prepare("INSERT INTO jobs(id,owner,name,kind,spec,status,total_cycles,created_at,updated_at) VALUES(?,?,?,'planet',?,'running',1,?,?)");$p->execute([$planetId,'local','generated',$planetSpec,$now,$now]);
    $method->invoke($worker,['id'=>$planetId,'owner'=>'local','name'=>'generated','kind'=>'planet','spec'=>json_decode($planetSpec,true),'total_cycles'=>1]);
    $savedQuery=$db->query("SELECT experiment_id,status FROM jobs WHERE id='$planetId'");$saved=$savedQuery->fetch();$savedQuery->closeCursor();expect($saved['status']==='completed'&&preg_match('/^[a-f0-9]{32}$/',$saved['experiment_id']??''),'Planet job must save an experiment and link it to the job.');
    $planetIndex=json_decode(file_get_contents($tmp.'/jobs/'.$planetId.'/artifacts/index.json'),true);expect(($planetIndex['latest']??null)===str_repeat('e',32),'PHP worker must publish planet result index.');expect(count($planetIndex['datasets'])===1,'Planet event must publish its result index once.');
    $noOutputId=str_repeat('9',32);$s->execute([$noOutputId,'local','silent',$spec,$now,$now]);
    file_put_contents($science.'/model.py',implode(PHP_EOL,["import argparse","p=argparse.ArgumentParser();p.add_argument('--config');p.add_argument('--job');p.parse_args()"]));
    $method->invoke($worker,['id'=>$noOutputId,'owner'=>'local','name'=>'silent','kind'=>'model','spec'=>json_decode($spec,true),'total_cycles'=>1]);$silent=$db->query("SELECT status,error FROM jobs WHERE id='$noOutputId'")->fetch();expect($silent['status']==='failed'&&str_contains($silent['error'],'without publishing'),'Zero exit without results must fail.');
    $badKindId=str_repeat('8',32);$s->execute([$badKindId,'local','kind',$spec,$now,$now]);$method->invoke($worker,['id'=>$badKindId,'owner'=>'local','name'=>'kind','kind'=>'other','spec'=>json_decode($spec,true),'total_cycles'=>1]);expect($db->query("SELECT status FROM jobs WHERE id='$badKindId'")->fetchColumn()==='failed','Unsupported job kind must fail.');
    $badJsonId=str_repeat('7',32);$db->exec("INSERT INTO jobs(id,owner,name,spec,status,total_cycles,created_at,updated_at) VALUES('$badJsonId','local','json','{','queued',1,'$now','$now')");$badJson=$worker->claim();expect(($badJson['spec_error']??'')!=='','Malformed stored JSON must not terminate claim.');$method->invoke($worker,$badJson);expect($db->query("SELECT status FROM jobs WHERE id='$badJsonId'")->fetchColumn()==='failed','Malformed stored JSON must fail its job.');
    file_put_contents($tmp.'/large.log',str_repeat('x',1048576));$tailMethod=(new ReflectionClass(LabWorker::class))->getMethod('tail');$tailMethod->setAccessible(true);$tail=$tailMethod->invoke($worker,$tmp.'/large.log');expect(strlen($tail)===8192&&$tail===str_repeat('x',8192),'Log tail must read only its bounded suffix.');
    $sleepScript=<<<'PY'
import argparse,time
p=argparse.ArgumentParser();p.add_argument('--config');p.add_argument('--job');p.parse_args();time.sleep(20)
PY;
file_put_contents($science.'/model.py',$sleepScript);
$timeoutId=str_repeat('c',32);$s->execute([$timeoutId,'local','timeout',$spec,$now,$now]);
$property=(new ReflectionClass(LabWorker::class))->getProperty('config');$workerConfig=$property->getValue($worker);$workerConfig['job_timeout_seconds']=1;$property->setValue($worker,$workerConfig);
$begin=microtime(true);$method->invoke($worker,['id'=>$timeoutId,'owner'=>'local','name'=>'timeout','kind'=>'model','spec'=>json_decode($spec,true),'total_cycles'=>1]);$timedQuery=$db->query("SELECT status,error FROM jobs WHERE id='$timeoutId'");$timed=$timedQuery->fetch();$timedQuery->closeCursor();expect($timed['status']==='failed'&&str_contains($timed['error'],'time limit')&&microtime(true)-$begin<8,'Worker must enforce timeout and stop the process group.');
    file_put_contents($science.'/model.py',$sleepScript);$cancelId=str_repeat('d',32);$s->execute([$cancelId,'local','cancel',$spec,$now,$now]);
    $pid=pcntl_fork();if($pid===0){usleep(500000);$cancelDb=new PDO('sqlite:'.$tmp.'/data/lab.sqlite3');$cancelDb->exec("UPDATE jobs SET cancel_requested=1 WHERE id='$cancelId'");exit(0);}
    $begin=microtime(true);$method->invoke($worker,['id'=>$cancelId,'owner'=>'local','name'=>'cancel','kind'=>'model','spec'=>json_decode($spec,true),'total_cycles'=>1]);pcntl_waitpid($pid,$childStatus);$cancelled=$db->query("SELECT status FROM jobs WHERE id='$cancelId'")->fetchColumn();expect($cancelled==='cancelled'&&microtime(true)-$begin<8,'Worker must honor cancellation and stop the process group.');
    echo "PHP worker tests passed\n";
} finally {
    putenv('EXOPLASIM_CONFIG');chdir($originalCwd);
    $it=new RecursiveIteratorIterator(new RecursiveDirectoryIterator($tmp,FilesystemIterator::SKIP_DOTS),RecursiveIteratorIterator::CHILD_FIRST);
    foreach($it as $file){$file->isDir()?rmdir($file->getPathname()):unlink($file->getPathname());}
    rmdir($tmp);
}