<?php
declare(strict_types=1);
require __DIR__.'/gate.php';
require __DIR__.'/catalog-amendment.php';
umask(0077);
$lane=getenv('NF_LANE');$phase=getenv('NF_PHASE');$root=getenv('NF_ROOT');
nf_check(in_array($lane,['B','C'],true)&&in_array($phase,['preflight','execute'],true),'Explicit lane/phase');
$plan=nf_plan(__DIR__.'/probe-plan.json');$spec=$plan['lanes'][$lane]['five_float_group'];nf_shape($spec,$lane);
$amend=$lane==='C'?nf_amend_load(__DIR__.'/catalog-amendment-C.json',$plan):null;
$journalPath=__DIR__.'/private-'.$phase.'-'.$lane.'.jsonl';
$handle=fopen($journalPath,'xb');nf_check($handle!==false,'No journal overwrite/replay');
chmod($journalPath,0600);$seq=0;$probeReads=0;$naturalReads=0;$dml=0;$ddl=0;$rollbacks=0;$created=false;$createdMeta=null;$completedCases=0;$unknown=false;$db=null;$cfg=null;
if($lane==='C'){$probeReads=2;$naturalReads=10;}
$emit=static function(string $event,array $data=[])use($handle,&$seq,$lane,$phase):void{
 $line=json_encode(['seq'=>++$seq,'utc'=>gmdate('c'),'lane'=>$lane,'phase'=>$phase,'event'=>$event,'data'=>$data],JSON_UNESCAPED_UNICODE|JSON_PRESERVE_ZERO_FRACTION|JSON_THROW_ON_ERROR)."\n";
 nf_check(fwrite($handle,$line)===strlen($line)&&fflush($handle),'Durable journal append');
};
$typed=static function($v):array{
 $out=['php_type'=>gettype($v),'value'=>$v];
 if(is_float($v)){$out['float17g']=sprintf('%.17g',$v);$out['native_double_hex']=bin2hex(pack('d',$v));$out['byte_order']=bin2hex(pack('L',1))==='01000000'?'little':'big';}
 return $out;
};
$parameters=static function(array $binds)use($typed):array {
 $out=[];foreach($binds as $b){
  $original=array_key_exists('original_value',$b)?$b['original_value']:$b['value'];
  $kind=$b['php_type']??(is_null($original)?'NULL':(is_int($original)?'integer':'string'));
  $value=$kind==='float'?(float)$original:$original;
  $code=$b['pdo_type_code']??($b['pdo_type']==='PDO::PARAM_INT'?PDO::PARAM_INT:($b['pdo_type']==='PDO::PARAM_NULL'?PDO::PARAM_NULL:PDO::PARAM_STR));
  $out[]=['original'=>$b,'runtime'=>$typed($value),'value'=>$value,'pdo_type_code'=>$code];
 }return $out;
};
$error=static function(Throwable $e):array {return ['class'=>get_class($e),'code'=>$e->getCode(),'message'=>$e->getMessage(),'errorInfo'=>$e instanceof PDOException?($e->errorInfo??null):null];};
$connect=static function()use(&$cfg):PDO{
 $opts=[PDO::ATTR_ERRMODE=>PDO::ERRMODE_EXCEPTION,PDO::ATTR_CASE=>PDO::CASE_LOWER,PDO::ATTR_ORACLE_NULLS=>PDO::NULL_NATURAL,PDO::ATTR_STRINGIFY_FETCHES=>false,PDO::ATTR_EMULATE_PREPARES=>false];
 foreach(($cfg['params']??[])as $k=>$v)$opts[$k]=$v;
 return new PDO($cfg['_nf_dsn'],$cfg['username'],$cfg['password'],$opts);
};
$query=static function(PDO $pdo,string $sql,array $binds=[],string $bucket='probe',int $cap=256)use(&$probeReads,&$naturalReads,$parameters,$emit,$typed,$error,$lane,$amend):array {
 $originalSql=$sql;$sql=nf_catalog_sql($sql,$lane,$amend);
 nf_check(preg_match('/^SELECT\b/i',$sql)===1||preg_match('/^SHOW (WARNINGS|CREATE TABLE)\b/',$sql)===1,'Read/diagnostic allowlist');
 if($bucket==='natural'){nf_check(++$naturalReads<=40,'Natural query budget40');}else nf_check(++$probeReads<=299,'Probe query budget299');
 $actual=$parameters($binds);$emit('read-start',['sql'=>$sql,'original_frozen_sql'=>$originalSql,'catalog_amendment_sha256'=>$sql!==$originalSql?NF_AMEND_SHA:null,'parameters'=>$actual,'bucket'=>$bucket]);
 try{
  $s=$pdo->prepare($sql);foreach($actual as $i=>$v)$s->bindValue($i+1,$v['value'],$v['pdo_type_code']);
  $s->execute();$rows=[];$types=[];$bytes=0;
  while(($row=$s->fetch(PDO::FETCH_ASSOC))!==false){
   $row=array_change_key_case($row,CASE_LOWER);
   foreach($row as &$v)if(is_resource($v))$v=stream_get_contents($v);unset($v);
   $rows[]=$row;$types[]=array_map($typed,$row);
   nf_check(count($rows)<=$cap,'Bounded row cap');
   $bytes+=strlen(json_encode($row,JSON_UNESCAPED_UNICODE|JSON_PRESERVE_ZERO_FRACTION));nf_check($bytes<=524288,'Bounded512KB result');
  }
  $s->closeCursor();$emit('read-success',['sql'=>$sql,'rows'=>$rows,'typed_rows'=>$types,'row_count'=>count($rows),'errorInfo'=>$s->errorInfo()]);
  return $rows;
 }catch(Throwable $e){$emit('read-failure',['sql'=>$sql,'error'=>$error($e)]);throw $e;}
};
$natural=static function(PDO $pdo)use($plan,$lane,$query):array{
 $out=[];foreach($plan['lanes'][$lane]['natural_protection']['queries']as $q)$out[]=$query($pdo,$q['sql'],$q['bindings'],'natural',$q['max_rows']);return $out;
};
$meta=static function(PDO $pdo)use($spec,$query):array{
 $out=[];foreach($spec['schema_fingerprint_queries']as $q)$out[]=$query($pdo,$q['sql'],$q['bindings'],'probe',$q['max_rows']);return $out;
};
$zero=static function(PDO $pdo)use($spec,$query):void { $r=$query($pdo,$spec['statements']['row_count'],[],'probe',1);nf_check(count($r)===1,'Row count shape');nf_num($r[0]['n'],0,'Own ordinary row count must be zero');};
$execDDL=static function(string $sql)use(&$ddl,$spec,&$db,$emit,$error,&$unknown):void {
 nf_check(in_array($sql,[$spec['statements']['create_ordinary'],$spec['statements']['drop_ordinary']],true),'Exact ordinary DDL only');
 nf_check(++$ddl<=2,'DDL cap2');$emit('DDL-start',['sql'=>$sql,'sha256'=>hash('sha256',$sql)]);
 try{$s=$db->prepare($sql);$s->execute();$emit('DDL-success',['sql'=>$sql,'errorInfo'=>$s->errorInfo()]);$s->closeCursor();}
 catch(Throwable $e){$unknown=true;$emit('DDL-failure-unknown-state',['sql'=>$sql,'error'=>$error($e)]);throw $e;}
};
$rollback=static function()use(&$db,&$rollbacks,$emit):void{
 nf_check(++$rollbacks<=48,'Rollback cap48');$emit('rollback-start');
 nf_check($db->rollBack(),'Rollback succeeded');$emit('rollback-success');
};
$cleanup=static function()use(&$db,&$created,&$createdMeta,&$unknown,$connect,$meta,$zero,$execDDL,$emit,$query,$spec,$lane):void{
 if(!$created)return;
 nf_check(!$unknown,'Unknown state forbids cleanup DROP');
 // Caller has rolled back every owned transaction. Do not manipulate another connection.
 $zero($db);$independent=$connect();$zero($independent);
 $now=$meta($db);$independentMeta=$meta($independent);
 nf_fingerprint_gate($now,$lane);nf_fingerprint_gate($independentMeta,$lane);
 nf_check($createdMeta!==null&&$now===$createdMeta&&$independentMeta===$createdMeta,'Creation ledger/full actual definition fingerprint exact before DROP');
 $emit('cleanup-ownership-gate',['created'=>true,'metadata_sha256'=>hash('sha256',json_encode($createdMeta,JSON_PRESERVE_ZERO_FRACTION)),'independent_zero'=>true]);
 $execDDL($spec['statements']['drop_ordinary']);$created=false;
 $args=[['value'=>$lane==='B'?strtolower($spec['table_name']):$spec['table_name'],'pdo_type'=>'PDO::PARAM_STR']];
 $after=$query($independent,$spec['statements']['drop_readback'],$args,'probe',1);nf_num($after[0]['n'],0,'Fresh independent catalog object0');
 $emit('cleanup-complete',['fresh_object_count'=>0]);
};
try{
 if($lane==='C')$emit('approved-catalog-amendment',['sha256'=>NF_AMEND_SHA,'prior_probe_queries'=>2,'prior_natural_queries'=>10,'prior_DDL'=>0,'prior_DML'=>0]);
 nf_check($root===($lane==='B'?'/tmp/zhct-http-store-mysql.dQTTBP':'/tmp/zhct-dm-poc-20260928.fcpiiN/store'),'Root path before private require');
 $cfg=(static function(string $file):array{return require $file;})($root.'/application/database.php');
 if($lane==='B'){
  nf_check($root==='/tmp/zhct-http-store-mysql.dQTTBP','Exact existing B root');
  $local=(static function(string $file):array{return require $file;})('/workspace/zhct/zhctproject/store/application/config/local/database.php');
  foreach(['hostname','database','username','password']as $k)nf_check((string)($cfg[$k]??'')===(string)($local[$k]??''),'B local config scope key '.$k);
  nf_check((string)(($cfg['hostport']??'')?:3306)===(string)(($local['hostport']??'')?:3306),'B local port scope');
  $emit('approved-B-config-recovery',['four_keys_match'=>true,'effective_default3306_port_match'=>true,'old_worker_charset_equals_local'=>(string)($cfg['charset']??'')===(string)($local['charset']??''),'old_worker_raw_port_equals_local'=>(string)($cfg['hostport']??'')===(string)($local['hostport']??''),'effective_config'=>'current code local; original charset inherited without flag/setter']);
  $cfg=$local;
  $cfg['_nf_dsn']='mysql:host='.$cfg['hostname'].';port='.(($cfg['hostport']??'')?:3306).';dbname='.$cfg['database'];
  if(!empty($cfg['charset']))$cfg['_nf_dsn'].=';charset='.$cfg['charset'];
 }else{
  nf_check($root==='/tmp/zhct-dm-poc-20260928.fcpiiN/store','Exact existing C root');
  nf_check($cfg['hostname']==='127.0.0.1'&&(string)$cfg['hostport']==='5237'&&$cfg['database']==='ZHCT_APP','Exact approved C target');
  $cfg['_nf_dsn']='dm:host=127.0.0.1;port=5237;schema=ZHCT_APP';
 }
 nf_check(!isset($cfg['params'][1002]),'No PDO init SQL command allowed');
 $db=$connect();$emit('connection-scope-confirmed',['scope_matches'=>true,'charset_from_existing_config'=>true,'new_charset_override'=>false,'php'=>PHP_VERSION,'int_size'=>PHP_INT_SIZE,'precision'=>ini_get('precision'),'serialize_precision'=>ini_get('serialize_precision'),'runner_sha256'=>hash_file('sha256',__FILE__),'plan_sha256'=>NF_PLAN_SHA]);
 $attrs=[];foreach([PDO::ATTR_DRIVER_NAME,PDO::ATTR_CLIENT_VERSION,PDO::ATTR_SERVER_VERSION,PDO::ATTR_EMULATE_PREPARES,PDO::ATTR_STRINGIFY_FETCHES,PDO::ATTR_CASE,PDO::ATTR_ERRMODE]as $a){
  try{$attrs[$a]=['value'=>$db->getAttribute($a),'getter_supported'=>true];}catch(Throwable $e){$attrs[$a]=['getter_supported'=>false,'error'=>$error($e)];}
 }$emit('PDO-attributes',['attributes'=>$attrs]);
 foreach($plan['identity_queries'][$lane]as $q){$r=$query($db,$q['sql'],$q['bindings'],'probe',$q['max_rows']);if($lane==='B')nf_check(strpos($r[0]['version'],'5.6.')===0,'B exact MySQL5.6');elseif(isset($r[0]['schema_name']))nf_check($r[0]['schema_name']==='ZHCT_APP','C current schema');}
 $before=$natural($db);nf_natural_gate($before,$lane);$emit('natural-before',['sha256'=>hash('sha256',json_encode($before,JSON_PRESERVE_ZERO_FRACTION)),'fresh_types_passed'=>true]);
 foreach($plan['preflight_and_cleanup']['collision_before_create'][$lane]as $q){$args=array_map(static function($v){return ['value'=>$v,'pdo_type'=>'PDO::PARAM_STR'];},$q['values']);nf_check($query($db,$q['sql'],$args,'probe',32)===[],'Collision0 exact object names');}
 // Catalog SQL preparation/execute can be proven against absent own table; actual SHOW CREATE/GET_DDL waits for CREATE.
 foreach(array_slice($spec['schema_fingerprint_queries'],0,-1)as $q)nf_check($query($db,$q['sql'],$q['bindings'],'probe',$q['max_rows'])===[],'Absent own catalog rows empty');
 $emit('preflight-complete',['probe_queries'=>$probeReads,'natural_queries'=>$naturalReads,'DDL'=>0,'DML'=>0]);
 if($phase==='execute'){
  $execDDL($spec['statements']['create_ordinary']);$created=true;
  $createdMeta=$meta($db);nf_fingerprint_gate($createdMeta,$lane);$emit('created-schema-gate',['sha256'=>hash('sha256',json_encode($createdMeta,JSON_PRESERVE_ZERO_FRACTION)),'passed'=>true]);
  foreach($spec['steps']as $case){
   nf_check($db->beginTransaction(),'Begin own transaction');$emit('case-start',['case_id'=>$case['case_id']]);
   foreach($case['DML']as $op){
    nf_check(++$dml<=72,'DML cap72');$actual=$parameters($op['bindings']);$emit('DML-start',['case_id'=>$case['case_id'],'role'=>$op['role'],'sql'=>$op['sql'],'parameters'=>$actual,'column_original_values'=>$op['column_original_values']??null]);
    $success=false;$domain=false;
    try{$s=$db->prepare($op['sql']);foreach($actual as $i=>$v)nf_check($s->bindValue($i+1,$v['value'],$v['pdo_type_code']),'Bind success');$success=$s->execute();nf_check($success,'DML returned false');$emit('DML-success',['case_id'=>$case['case_id'],'role'=>$op['role'],'rowCount'=>$s->rowCount(),'errorInfo'=>$s->errorInfo()]);$s->closeCursor();}
    catch(PDOException $e){$domain=nf_domain($e,$lane);$emit('DML-failure',['case_id'=>$case['case_id'],'role'=>$op['role'],'domain_error'=>$domain,'error'=>$error($e)]);if(!$domain){$unknown=true;throw $e;}}
    if($lane==='B'){$warnings=$query($db,'SHOW WARNINGS',[],'probe',11);$emit('warnings',['case_id'=>$case['case_id'],'role'=>$op['role'],'rows'=>$warnings,'overflow'=>count($warnings)>10]);nf_check(count($warnings)<=10,'Warnings overflow stop');}
    nf_check($op['role']!=='zero_seed_insert'||$success,'Zero seed must succeed; no hidden dependent skip');
   }
   foreach($case['readbacks']as $name)$query($db,$spec['statements'][$name],$case['readback_bindings'],'probe',1);
   $rollback();$zero($db);$completedCases++;$emit('case-complete',['case_id'=>$case['case_id'],'completed_cases'=>$completedCases]);
  }
  nf_check($dml===72&&$completedCases===48&&$rollbacks===48,'Exact approved sequence complete');
  $cleanup();
 }
 $after=$natural($db);$stable=$before===$after;$emit('natural-after',['sha256'=>hash('sha256',json_encode($after,JSON_PRESERVE_ZERO_FRACTION)),'equals_before'=>$stable,'drift_classification'=>$stable?'none':'external_or_unattributed; no restoration authorized']);
 $emit('complete',['completed_cases'=>$completedCases,'DDL'=>$ddl,'DML_attempts'=>$dml,'rollbacks'=>$rollbacks,'probe_queries'=>$probeReads,'natural_queries'=>$naturalReads,'natural_equal'=>$stable,'objects_removed'=>!$created]);
 echo json_encode(['lane'=>$lane,'phase'=>$phase,'status'=>'COMPLETE','cases'=>$completedCases,'DDL'=>$ddl,'DML_attempts'=>$dml,'probe_queries'=>$probeReads,'natural_queries'=>$naturalReads,'raw_journal_sha256'=>hash_file('sha256',$journalPath)]),"\n";
}catch(Throwable $e){
 $emit('stopped',['error'=>$error($e),'unknown'=>$unknown,'created'=>$created,'completed_cases'=>$completedCases,'DDL'=>$ddl,'DML_attempts'=>$dml,'probe_queries'=>$probeReads,'natural_queries'=>$naturalReads]);
 // An ownership/schema/transport gate failure never causes an automatic DROP or replay.
 if($db!==null){try{if($db->inTransaction()){$emit('emergency-owned-rollback-start');$db->rollBack();$emit('emergency-owned-rollback-success');}}catch(Throwable $rollbackError){$emit('emergency-rollback-failure',['error'=>$error($rollbackError)]);}}
 echo json_encode(['lane'=>$lane,'phase'=>$phase,'status'=>'STOPPED','created'=>$created,'unknown'=>$unknown,'cases'=>$completedCases,'DDL'=>$ddl,'DML_attempts'=>$dml,'raw_journal_sha256'=>hash_file('sha256',$journalPath)]),"\n";
 exit(2);
}