137 lines
13 KiB
PHP
137 lines
13 KiB
PHP
<?php
|
|
declare(strict_types=1);
|
|
|
|
namespace app\adminapi\service {
|
|
function hrtime(bool $number=false){return \app\common\service\hrtime($number);}
|
|
function usleep(int $us):void{$GLOBALS['h26Ns']+=$us*1000;}
|
|
}
|
|
namespace app\common\service {
|
|
require __DIR__.'/fixtures/CancellableCurlMultiStub.php';
|
|
function hrtime(bool $number=false){$n=$GLOBALS['h26Ns'];return $number?$n:[intdiv($n,1000000000),$n%1000000000];}
|
|
function config(string $name):array{return ['enable'=>true,'base_url'=>'https://fixture.invalid/v1/chat-messages','timeout'=>90,'models'=>['qwen'=>['name'=>'fixture','api_key'=>'fixture-only']]];}
|
|
function curl_init():\stdClass{return (object)['errno'=>0];}
|
|
function curl_setopt_array(\stdClass $h,array $options):bool{$h->options=$options;return true;}
|
|
function curl_exec(\stdClass $h):bool{
|
|
$GLOBALS['h26InCurl']=true;$GLOBALS['h26Closed']=false;++$GLOBALS['h26Posts'];$n=$GLOBALS['h26Posts'];
|
|
try{
|
|
($h->options[CURLOPT_HEADERFUNCTION])($h,"HTTP/1.1 200 OK\r\n");$write=$h->options[CURLOPT_WRITEFUNCTION];
|
|
$frames=[['event'=>'workflow_started','task_id'=>'own-task-'.$n,'workflow_run_id'=>'own-run-'.$n],
|
|
['event'=>'message','task_id'=>'own-task-'.$n,'answer'=>'原始合成字节。']];
|
|
if($GLOBALS['h26Mode']!=='no_producer')$frames[]=['event'=>'workflow_finished','task_id'=>'own-task-'.$n,
|
|
'workflow_run_id'=>$GLOBALS['h26Mode']==='foreign_producer'?'foreign-run':'own-run-'.$n,'data'=>['status'=>'stopped']];
|
|
if(in_array($GLOBALS['h26Mode'],['unidentified','terminal_only'],true)){
|
|
$frames=[['event'=>'message','answer'=>'原始合成字节。']];
|
|
if($GLOBALS['h26Mode']==='terminal_only')$frames[]=['event'=>'workflow_finished','workflow_run_id'=>'unbound-run','data'=>['status'=>'stopped']];
|
|
}
|
|
$frames[]=['event'=>'message_end'];
|
|
foreach($frames as$frame){$GLOBALS['h26Ns']+=1000000;$write($h,'data: '.json_encode($frame,JSON_UNESCAPED_UNICODE)."\n\n");}
|
|
return true;
|
|
}finally{$GLOBALS['h26InCurl']=false;}
|
|
}
|
|
function curl_errno(\stdClass $h):int{return $h->errno;}
|
|
function curl_getinfo(\stdClass $h,int $option):int|float{return $option===CURLINFO_HTTP_CODE?200:0.001;}
|
|
function curl_close(\stdClass $h):void{$GLOBALS['h26Closed']=true;}
|
|
class DifyWorkflowRunStatus{
|
|
public static function lookup(array $id,int $ms):array{
|
|
\h26Expect(!$GLOBALS['h26InCurl']&&$GLOBALS['h26Closed'],'No GET inside cURL callbacks or before actual handle closure.');
|
|
\h26Expect(count($GLOBALS['h26Timing'])===$GLOBALS['h26Posts'],'Timing callback is published before GET.');
|
|
$bound=(new \ReflectionProperty($GLOBALS['h26Audit'],'generations'))->getValue($GLOBALS['h26Audit']);
|
|
\h26Expect(!empty($bound[$GLOBALS['h26Posts']]['closed']),'Own generation closure is recorded before GET.');
|
|
\h26Expect($ms>0&&$ms<=1000,'Exact original bounded adapter cap.');
|
|
++$GLOBALS['h26Gets'];$GLOBALS['h26Ns']+=20000000;
|
|
$mode=$GLOBALS['h26Mode'];
|
|
return ['ok'=>$mode!=='false_terminal','workflow_run_id'=>$mode==='foreign_get'?'foreign-run':$id['workflow_run_id'],
|
|
'status'=>($mode==='running'||($mode==='two_generations'&&$GLOBALS['h26Posts']===2))?'running':'stopped'];
|
|
}
|
|
}
|
|
}
|
|
namespace {
|
|
require dirname(__DIR__).'/vendor/autoload.php';
|
|
require dirname(__DIR__,2).'/app/artifacts/ai_concurrency/private_tcm_r26_wire_audit.php';
|
|
use app\adminapi\service\AssistantRequestControl as Control;
|
|
use app\common\service\DifyChatService;
|
|
$checks=0;$number=0;
|
|
function h26Expect(bool $ok,string $message):void{++$GLOBALS['checks'];if(!$ok)throw new \RuntimeException($message);}
|
|
$directory=sys_get_temp_dir().'/r26-private-harness-'.bin2hex(random_bytes(8));
|
|
$source=file_get_contents(dirname(__DIR__,2).'/app/artifacts/ai_concurrency/private_tcm_r26_acceptance_probe.php');
|
|
$literals=[];
|
|
foreach(['on_submission','on_task','on_terminal','on_transfer_end','on_upstream_timing','can_start_attempt']as$key){
|
|
h26Expect(preg_match_all('/\''.preg_quote($key,'/').'\' => (static (?:function.*?\n }|fn \(\): bool => \$control->canStartAttempt\(\))),/s',$source,$m)===1,'One literal production-probe callback: '.$key);
|
|
$literals[$key]=$m[1][0];
|
|
h26Expect(str_contains(file_get_contents(dirname(__DIR__,2).'/app/artifacts/ai_concurrency/private_tcm_r26_deadline_probe.php'),$literals[$key]),'Deadline and acceptance probes share exact canonical callback: '.$key);
|
|
}
|
|
$make=static function(string $mode)use($directory,&$number,$literals):array{
|
|
$GLOBALS['h26Ns']=100000000000;foreach(['h26Posts','h26Gets']as$key)$GLOBALS[$key]=0;
|
|
$GLOBALS['h26InCurl']=false;$GLOBALS['h26Closed']=false;$GLOBALS['h26Timing']=[];$GLOBALS['h26Mode']=$mode;
|
|
$uuid=sprintf('f2345678-1234-4234-8234-%012d',++$number);
|
|
$control=new Control(7,11,$uuid,$directory.'/controls');h26Expect($control->begin(),'Isolated request begins.');
|
|
$path=$directory.'/controls/'.hash('sha256','7:11:'.$uuid).'.json';
|
|
$privateWireAudit=new \PrivateTcmR26WireAudit($directory.'/audits',$uuid,'fixture_'.$number,'qwen',$path);
|
|
$GLOBALS['h26Audit']=$privateWireAudit;
|
|
$GLOBALS['h26AuditPath']=$directory.'/audits/'.hash('sha256',$uuid).'.private.jsonl';
|
|
$profile='qwen';$upstreamUser='fixture-user';$marker=$directory.'/marker-'.$number;$lastTask='';
|
|
$stages=[];$upstreamAttemptObservations=[];$started=$GLOBALS['h26Ns'];$options=[];
|
|
foreach($literals as$key=>$literal)$options[$key]=eval('return '.$literal.';');
|
|
$timing=$options['on_upstream_timing'];
|
|
$options['on_upstream_timing']=static function($value)use($timing):void{$timing($value);$GLOBALS['h26Timing'][]=$value->snapshot();};
|
|
$options['on_private_wire_request']=[$privateWireAudit,'observe'];$options['timeout_ms']=90000;$options['request_deadline_ns']=190000000000;
|
|
return [$control,$privateWireAudit,$options,$path];
|
|
};
|
|
$records=static fn():array=>array_map(static fn($line)=>json_decode($line,true,64,JSON_THROW_ON_ERROR),file($GLOBALS['h26AuditPath'],FILE_IGNORE_NEW_LINES));
|
|
try{
|
|
foreach(['stopped','running','false_terminal','foreign_get','foreign_producer','no_producer','two_generations']as$mode){
|
|
[$control,$audit,$options,$path]=$make($mode);$text='';
|
|
$out=DifyChatService::streamChat('qwen',[],'fixture query','fixture-user',static function(string $s)use(&$text):bool{$text.=$s;return true;},static fn():bool=>false,[],$options);
|
|
h26Expect(!empty($out['ok'])&&$text==='原始合成字节。','Lifecycle observer preserves original delivered bytes.');
|
|
if($mode==='two_generations'){
|
|
h26Expect($control->canStartAttempt(),'Exact first terminal permits only eligible next generation.');
|
|
DifyChatService::streamChat('qwen',[],"fixture\n<REPORT_CONTINUATION_DATA>\n{}\n</REPORT_CONTINUATION_DATA>",'fixture-user',static fn():bool=>true,static fn():bool=>false,[],$options);
|
|
}
|
|
$control->finish('done');$before=$GLOBALS['h26Gets'];$control->reconcileClosedTasks([\app\common\service\DifyWorkflowRunStatus::class,'lookup']);
|
|
h26Expect($GLOBALS['h26Gets']===$before,'Final recovery cannot grant EOF a new verification budget.');
|
|
$audit->captureControl();$state=$control->publicState();$summary=$audit->finish($state);$rows=$records();
|
|
h26Expect($summary['persisted']&&$summary['audit_sha256']===hash_file('sha256',$GLOBALS['h26AuditPath']),'Exact durable digest includes all lifecycle/footer records.');
|
|
$pending=in_array($mode,['running','false_terminal','foreign_get','two_generations'],true);
|
|
h26Expect($summary['independent_cleanup_confirmed']===!$pending&&$summary['pending_generation_count']===($pending?1:0),'No stopped producer receipt or false/foreign response fabricates independent cleanup: '.$mode);
|
|
h26Expect($summary['lifecycle_consistent']&&$summary['closed_generation_count']===$GLOBALS['h26Posts'],'All original transfer and aggregate counts agree.');
|
|
$generation=end($summary['generations']);$snapshot=$generation['lifecycle'];
|
|
h26Expect($snapshot['producer_terminal_status']===(in_array($mode,['foreign_producer','no_producer'],true)?'':'stopped'),'Only bound producer receipts are audited.');
|
|
h26Expect($snapshot['persisted_run_status']===(in_array($mode,['foreign_get','false_terminal'],true)?'':($pending?'running':'stopped')),'Only exact adapter observations enter lifecycle evidence.');
|
|
h26Expect(!str_contains(json_encode($summary),'own-task')&&!str_contains(json_encode($summary),'own-run')&&!str_contains(json_encode($summary),'fixture-user'),'Public audit summary never reveals original identities.');
|
|
$raw=file_get_contents($GLOBALS['h26AuditPath']);foreach(['foreign-run','fixture-user','原始合成字节','fixture-only','fixture query']as$secret)h26Expect(!str_contains($raw,$secret),'Private audit excludes foreign identities/text/credentials: '.$secret);
|
|
if($mode==='two_generations')h26Expect($summary['independently_terminal_generation_count']===1&&array_column($summary['generations'],'generation')===[1,2],'Separate first-terminal/second-running original generations remain distinct.');
|
|
if($mode==='running'){
|
|
h26Expect($snapshot['lifecycle_conflict']&&count($snapshot['observations'])===2,'Contrary running observations and producer receipt are retained.');
|
|
h26Expect(!$control->canStartAttempt(),'Finished request cannot restart or hide pending generation.');
|
|
}
|
|
h26Expect($audit->finish(['upstream_pending_task_count'=>999])===$summary,'Footer is immutable across repeated finish calls.');
|
|
}
|
|
|
|
foreach(['unidentified','terminal_only']as$mode){
|
|
[$c,$audit,$options,$path]=$make($mode);$text='';
|
|
$reply=DifyChatService::streamChat('qwen',[],'fixture','fixture-user',static function(string $s)use(&$text):bool{$text.=$s;return true;},static fn():bool=>false,[],$options);
|
|
h26Expect(!empty($reply['ok'])&&$text==='原始合成字节。','Unidentified actual delivered bytes preserved.');
|
|
h26Expect($GLOBALS['h26Gets']===0&&!$c->canStartAttempt(),'Unknown/terminal-only metadata never guesses pair or permits successor.');
|
|
$c->finish('done');$audit->captureControl();$state=$c->publicState();$summary=$audit->finish($state);
|
|
h26Expect($summary['persisted']&&$summary['lifecycle_consistent']&&!$summary['independent_cleanup_confirmed'],'Unknown audit is durable and pending, never independent terminal.');
|
|
h26Expect($summary['submitted_pending_attempt_count']===1&&$summary['submitted_without_identity_count']===1&&$summary['pending_generation_count']===1,'Unknown audit/control pending counts agree.');
|
|
h26Expect($summary['generations'][0]['submitted']&&$summary['generations'][0]['transfer_closed']&&!$summary['generations'][0]['no_execution_confirmed'],'Empty-task EOF retains submitted owned admission.');
|
|
foreach(['own-task','own-run','unbound-run','submission_id']as$v)h26Expect(!str_contains(json_encode($summary),$v),'Private token/original identity is not in safe summary.');
|
|
}
|
|
|
|
[$c,$audit,$options,$path]=$make('running');
|
|
DifyChatService::streamChat('qwen',[],'fixture','fixture-user',static fn():bool=>true,static fn():bool=>false,[],$options);
|
|
$c->finish('error');$audit->captureControl();$state=$c->publicState();$state['upstream_pending_task_count']=0;$state['upstream_terminal_confirmed']=true;
|
|
$bad=$audit->finish($state);h26Expect(!$bad['lifecycle_consistent']&&!$bad['independent_cleanup_confirmed']&&$bad['pending_generation_count']===1,'False terminal footer cannot erase known pending generation.');
|
|
|
|
[$c,$audit,$options,$path]=$make('stopped');file_put_contents($path,'{"handoff_cleanup":true}');$audit->captureControl();$bad=$audit->finish(['upstream_pending_task_count'=>0]);
|
|
h26Expect(!$bad['persisted']&&!$bad['independent_cleanup_confirmed'],'Optional audit failure is bounded and cannot certify cleanup.');
|
|
echo "Private r26 actual probe callbacks / lifecycle audit: {$checks} assertions passed; no network/bootstrap/model.\n";
|
|
}finally{
|
|
$real=realpath($directory);$temp=realpath(sys_get_temp_dir());
|
|
if($real!==false){if($temp===false||!str_starts_with(str_replace('\\','/',$real),rtrim(str_replace('\\','/',$temp),'/').'/r26-private-harness-'))throw new \RuntimeException('Cleanup boundary');
|
|
$iterator=new \RecursiveIteratorIterator(new \RecursiveDirectoryIterator($real,\FilesystemIterator::SKIP_DOTS),\RecursiveIteratorIterator::CHILD_FIRST);
|
|
foreach($iterator as$f){if($f->isLink())throw new \RuntimeException('Unexpected temp symlink');$f->isDir()?rmdir($f->getPathname()):unlink($f->getPathname());}rmdir($real);}
|
|
}
|
|
}
|