105 lines
8.2 KiB
PHP
105 lines
8.2 KiB
PHP
<?php
|
|
declare(strict_types=1);
|
|
|
|
// Actual transport and exact canonical controller closures, with in-process I/O
|
|
// stubs only. No socket, bootstrap, runtime configuration or model is used.
|
|
namespace app\adminapi\service {
|
|
function hrtime(bool $number = false) { return \app\common\service\hrtime($number); }
|
|
function usleep(int $us): void { $GLOBALS['wireR26Ns'] += $us * 1000; }
|
|
}
|
|
namespace app\common\service {
|
|
require __DIR__ . '/fixtures/CancellableCurlMultiStub.php';
|
|
function hrtime(bool $number = false) { $n=$GLOBALS['wireR26Ns']; 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['wireR26Posts']; $GLOBALS['wireR26InCurl']=true;
|
|
$GLOBALS['wireR26Timeouts'][]=$h->options[CURLOPT_TIMEOUT_MS];
|
|
try {
|
|
($h->options[CURLOPT_HEADERFUNCTION])($h,"HTTP/1.1 200 OK\r\n");
|
|
$write=$h->options[CURLOPT_WRITEFUNCTION];
|
|
foreach ([['event'=>'workflow_started','task_id'=>'fixture-task','workflow_run_id'=>'fixture-run'],
|
|
['event'=>'message','task_id'=>'fixture-task','answer'=>'原始临床字节。'],
|
|
['event'=>'workflow_finished','task_id'=>'fixture-task','workflow_run_id'=>'fixture-run','data'=>['status'=>$GLOBALS['wireR26Producer']]],
|
|
['event'=>'message_end']] as $frame) {
|
|
$GLOBALS['wireR26Ns']+=1000000;
|
|
$write($h,'data: '.json_encode($frame,JSON_UNESCAPED_UNICODE)."\n\n");
|
|
}
|
|
return true;
|
|
} finally {$GLOBALS['wireR26InCurl']=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['wireR26Closed']=true;}
|
|
class DifyWorkflowRunStatus {
|
|
public static function lookup(array $id,int $ms):array {
|
|
\wireR26Expect(!$GLOBALS['wireR26InCurl'] && $GLOBALS['wireR26Closed'],'No verification I/O before curl_close or within callbacks.');
|
|
\wireR26Expect($GLOBALS['wireR26TimingPublished'],'Transport timing observer precedes independent verification.');
|
|
\wireR26Expect($id===['profile'=>'qwen','user'=>'fixture-user','task_id'=>'fixture-task','workflow_run_id'=>'fixture-run','generation'=>1], 'Canonical callback retains exact pair, profile, user and generation.');
|
|
\wireR26Expect($ms>0 && $ms<=1000,'Adapter receives original bounded I/O cap.');
|
|
++$GLOBALS['wireR26Gets']; $GLOBALS['wireR26Ns']+=50000000;
|
|
return ['ok'=>true,'workflow_run_id'=>$id['workflow_run_id'],'status'=>$GLOBALS['wireR26Persisted']];
|
|
}
|
|
}
|
|
}
|
|
namespace {
|
|
require dirname(__DIR__).'/vendor/autoload.php';
|
|
use app\adminapi\service\AssistantRequestControl;
|
|
use app\common\service\DifyChatService;
|
|
$checks=0;
|
|
function wireR26Expect(bool $ok,string $why):void {++$GLOBALS['checks'];if(!$ok)throw new \RuntimeException($why);}
|
|
$source=file_get_contents(dirname(__DIR__).'/app/adminapi/controller/tcm/DiagnosisController.php');
|
|
// Execute the literal production closures rather than a hand-copied adapter.
|
|
// Failure to find exactly one current wiring is an integration-test failure.
|
|
$extract=static function(string $key)use($source):string {
|
|
$pattern=$key==='on_task'
|
|
? '/\$options\[\'on_task\'\] = (static function.*?\n });/s'
|
|
: '/\''.preg_quote($key,'/').'\' => (static function.*?\n }),/s';
|
|
wireR26Expect(preg_match_all($pattern,$source,$matches)===1,'One canonical '.$key.' closure.');
|
|
return $matches[1][0];
|
|
};
|
|
$closures=[];foreach(['on_task','on_terminal','on_transfer_end','can_start_attempt'] as $key)$closures[$key]=$extract($key);
|
|
$directory=sys_get_temp_dir().'/r26-wiring-'.bin2hex(random_bytes(8));
|
|
try {
|
|
foreach(['running','succeeded'] as $case=>$persisted) {
|
|
$GLOBALS['wireR26Ns']=100000000000;
|
|
foreach(['wireR26Posts','wireR26Gets'] as $key)$GLOBALS[$key]=0;
|
|
foreach(['wireR26Closed','wireR26InCurl','wireR26TimingPublished']as$key)$GLOBALS[$key]=false;
|
|
$GLOBALS['wireR26Producer']=$persisted==='running'?'stopped':'succeeded';
|
|
$GLOBALS['wireR26Persisted']=$persisted; $GLOBALS['wireR26Timeouts']=[];
|
|
$uuid=sprintf('d2345678-1234-4234-8234-%012d',$case+1);
|
|
$control=new AssistantRequestControl(7,11,$uuid,$directory);
|
|
wireR26Expect($control->begin(),'Synthetic request begins.');
|
|
$profile='qwen';$upstreamUser='fixture-user';$options=[];
|
|
foreach($closures as $key=>$literal)$options[$key]=eval('namespace app\\adminapi\\controller\\tcm; use app\\common\\service\\DifyWorkflowRunStatus; return '.$literal.';');
|
|
$snapshot=[];$observerNs=0;
|
|
$options['on_upstream_timing']=static function($timing)use(&$snapshot,&$observerNs):void {
|
|
wireR26Expect($GLOBALS['wireR26Closed'] && $GLOBALS['wireR26Gets']===0,'Transport clock captured after close and before any lookup.');
|
|
$snapshot=$timing->snapshot();$observerNs=$GLOBALS['wireR26Ns'];$GLOBALS['wireR26TimingPublished']=true;
|
|
};
|
|
$options['request_deadline_ns']=190000000000;$options['timeout_ms']=90000;
|
|
$received='';
|
|
$out=DifyChatService::streamChat('qwen',[],'fixture query','fixture-user',static function(string $text)use(&$received):bool{$received.=$text;return true;},static fn():bool=>false,[],$options);
|
|
wireR26Expect($received==='原始临床字节。' && !empty($out['ok']),'Original delivered and returned clinical bytes survive lifecycle verification.');
|
|
wireR26Expect($GLOBALS['wireR26Posts']===1 && $GLOBALS['wireR26Gets']===($persisted==='running'?2:1),'One model transfer and bounded own-run reads.');
|
|
wireR26Expect($observerNs<=$GLOBALS['wireR26Ns']-50000000 && $snapshot['duration_ms']===4,'Lookup delay cannot shift transport origin/duration.');
|
|
wireR26Expect($GLOBALS['wireR26Timeouts']===[92500] && $options['timeout_ms']===90000 && $options['request_deadline_ns']===190000000000,'Original 90-second generation budget, existing 2.5-second drain cap and absolute deadline remain fixed.');
|
|
$p=$control->publicState();
|
|
wireR26Expect($p['generation_transfer_closed'] && $p['producer_terminal_received'] && $p['persisted_run_status']===$persisted,'Canonical callbacks persist distinct producer/run/EOF receipts.');
|
|
wireR26Expect($options['can_start_attempt']()===($persisted==='succeeded'),'Canonical attempt gate follows independent proof.');
|
|
if($persisted==='running') {
|
|
$again=DifyChatService::streamChat('qwen',[],'another query','fixture-user',static fn():bool=>true,static fn():bool=>false,[],$options);
|
|
wireR26Expect(($again['error_code']??'')==='UPSTREAM_CLEANUP_PENDING' && $GLOBALS['wireR26Posts']===1 && $GLOBALS['wireR26Gets']===2,'Blocked successor makes zero extra transport or lookup calls.');
|
|
wireR26Expect($p['lifecycle_conflict'] && $p['upstream_pending_task_count']===1,'Original contradiction is visible as safe status and retains private pair.');
|
|
}
|
|
}
|
|
echo "R26 canonical controller / transport wiring: {$checks} assertions passed (in-process stubs only).\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-wiring-'))throw new \RuntimeException('Cleanup boundary');
|
|
foreach(new \DirectoryIterator($real)as$f){if($f->isDot())continue;if($f->isLink()||!$f->isFile())throw new \RuntimeException('Unexpected temp entry');unlink($f->getPathname());}rmdir($real);}
|
|
}
|
|
}
|