202 lines
13 KiB
PHP
202 lines
13 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace app\common\service {
|
|
function config(string $name): array { return $GLOBALS['drainConfig']; }
|
|
}
|
|
|
|
namespace {
|
|
require dirname(__DIR__) . '/vendor/autoload.php';
|
|
|
|
use app\adminapi\service\AssistantRequestControl;
|
|
use app\common\service\DifyChatService;
|
|
|
|
function drainExpect(bool $ok, string $message): void {
|
|
if (!$ok) throw new RuntimeException($message);
|
|
}
|
|
|
|
// A real loopback HTTP peer. It accepts the stop on another socket while the
|
|
// generation stays open, then either sends terminal metadata or stays silent.
|
|
if (($argv[1] ?? '') === 'server') {
|
|
$scenario = $argv[2];
|
|
$listener = stream_socket_server('tcp://127.0.0.1:0', $error, $message);
|
|
if ($listener === false) exit(2);
|
|
stream_set_blocking($listener, false);
|
|
echo json_encode(['address' => stream_socket_get_name($listener, false)]) . "\n";
|
|
flush();
|
|
$clients = []; $generation = null; $generationAt = null; $terminalAt = null; $tailAt = null;
|
|
$earlySent = false; $taskSent = false; $done = false;
|
|
$metrics = ['stop_count' => 0, 'generation_open_on_stop' => false, 'title_disabled' => false,
|
|
'inputs_object' => false, 'generation_eof' => false, 'terminal_tail_sent' => false,
|
|
'closed_before_terminal_tail' => false];
|
|
$deadline = microtime(true) + 8;
|
|
while (!$done && microtime(true) < $deadline) {
|
|
while (($socket = @stream_socket_accept($listener, 0)) !== false) {
|
|
stream_set_blocking($socket, false);
|
|
$clients[(int) $socket] = ['socket' => $socket, 'buffer' => '', 'handled' => false];
|
|
}
|
|
foreach (array_keys($clients) as $key) {
|
|
$socket = $clients[$key]['socket'];
|
|
$chunk = @fread($socket, 65536);
|
|
if ($chunk === '' && feof($socket)) {
|
|
if ($generation === $key) {
|
|
$metrics['generation_eof'] = true;
|
|
$metrics['closed_before_terminal_tail'] = $tailAt !== null;
|
|
$done = true;
|
|
}
|
|
fclose($socket); unset($clients[$key]); continue;
|
|
}
|
|
if ($clients[$key]['handled']) continue;
|
|
$clients[$key]['buffer'] .= is_string($chunk) ? $chunk : '';
|
|
$parts = explode("\r\n\r\n", $clients[$key]['buffer'], 2);
|
|
if (count($parts) !== 2) continue;
|
|
preg_match('/Content-Length:\s*(\d+)/i', $parts[0], $length);
|
|
if (strlen($parts[1]) < (int) ($length[1] ?? 0)) continue;
|
|
$clients[$key]['handled'] = true;
|
|
$payload = json_decode($parts[1]);
|
|
if (preg_match('#^POST /v1/chat-messages/synthetic-drain-task/stop #', $parts[0])) {
|
|
++$metrics['stop_count'];
|
|
$metrics['generation_open_on_stop'] = $generation !== null && isset($clients[$generation]);
|
|
$metrics['user_matches'] = ($payload->user ?? '') === 'synthetic-drain-user';
|
|
$body = '{"result":"success"}';
|
|
fwrite($socket, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: "
|
|
. strlen($body) . "\r\nConnection: close\r\n\r\n" . $body);
|
|
fclose($socket); unset($clients[$key]);
|
|
if (in_array($scenario, ['late_metadata', 'terminal_tail', 'terminal_silent'], true)) $terminalAt = microtime(true) + 0.08;
|
|
} else {
|
|
$generation = $key; $generationAt = microtime(true);
|
|
$metrics['title_disabled'] = ($payload->auto_generate_name ?? null) === false;
|
|
$metrics['inputs_object'] = is_object($payload->inputs ?? null);
|
|
if ($scenario !== 'silent_headers') {
|
|
fwrite($socket, "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n");
|
|
}
|
|
if (in_array($scenario, ['no_terminal', 'terminal_tail', 'terminal_silent'], true)) {
|
|
fwrite($socket, "data: {\"event\":\"workflow_started\",\"task_id\":\"synthetic-drain-task\"}\n\n");
|
|
$taskSent = true;
|
|
}
|
|
}
|
|
}
|
|
if ($generation !== null && isset($clients[$generation])) {
|
|
$socket = $clients[$generation]['socket'];
|
|
if ($scenario === 'late_metadata' && !$earlySent && microtime(true) - $generationAt >= 0.12) {
|
|
// Cancellation has already been requested, but task metadata is
|
|
// still absent. This body must not short-circuit the grace.
|
|
fwrite($socket, "data: {\"event\":\"message\",\"answer\":\"discard-before-task\"}\n\n");
|
|
$earlySent = true;
|
|
}
|
|
if ($scenario === 'late_metadata' && !$taskSent && microtime(true) - $generationAt >= 0.20) {
|
|
fwrite($socket, "data: {\"event\":\"workflow_started\",\"task_id\":\"synthetic-drain-task\"}\n\n");
|
|
$taskSent = true;
|
|
}
|
|
if ($terminalAt !== null && microtime(true) >= $terminalAt) {
|
|
if (in_array($scenario, ['terminal_tail', 'terminal_silent'], true)) {
|
|
fwrite($socket, "data: {\"event\":\"workflow_finished\",\"task_id\":\"synthetic-drain-task\",\"data\":{\"status\":\"stopped\"}}\n\n");
|
|
$terminalAt = null;
|
|
if ($scenario === 'terminal_tail') $tailAt = microtime(true) + 0.20;
|
|
continue;
|
|
}
|
|
fwrite($socket, "data: {\"event\":\"message\",\"task_id\":\"synthetic-drain-task\",\"answer\":\"discard-tail\"}\n\n"
|
|
. "data: {\"event\":\"message_end\"}\n\n"
|
|
. "data: {\"event\":\"workflow_finished\",\"task_id\":\"synthetic-drain-task\",\"data\":{\"status\":\"stopped\"}}\n\n");
|
|
fclose($socket); unset($clients[$generation]); $done = true;
|
|
}
|
|
if ($tailAt !== null && microtime(true) >= $tailAt) {
|
|
$metrics['terminal_tail_sent'] = true;
|
|
fwrite($socket, "data: {\"event\":\"message_end\"}\n\n");
|
|
$tailAt = null;
|
|
fclose($socket); unset($clients[$generation]); $done = true;
|
|
}
|
|
}
|
|
usleep(1000);
|
|
}
|
|
foreach ($clients as $client) fclose($client['socket']);
|
|
fclose($listener);
|
|
echo json_encode($metrics) . "\n";
|
|
exit($done ? 0 : 3);
|
|
}
|
|
|
|
$directory = sys_get_temp_dir() . '/zyt-drain-' . bin2hex(random_bytes(8));
|
|
$rows = [];
|
|
try {
|
|
foreach (['late_metadata', 'terminal_tail', 'terminal_silent', 'no_terminal', 'silent_headers', 'silent_body'] as $index => $scenario) {
|
|
$process = proc_open([PHP_BINARY, __FILE__, 'server', $scenario],
|
|
[0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes);
|
|
drainExpect(is_resource($process), 'loopback server starts');
|
|
fclose($pipes[0]); stream_set_timeout($pipes[1], 10);
|
|
$ready = json_decode((string) fgets($pipes[1]), true);
|
|
drainExpect(isset($ready['address']), 'loopback server reports its ephemeral port');
|
|
$drainConfig = ['enable' => true, 'base_url' => 'http://' . $ready['address'] . '/v1/chat-messages',
|
|
'timeout' => 6, 'models' => ['qwen' => ['name' => 'synthetic', 'api_key' => 'loopback-only']]];
|
|
$requestId = sprintf('12345678-1234-4234-8234-%012d', $index);
|
|
$control = new AssistantRequestControl(7, 11, $requestId, $directory);
|
|
drainExpect($control->begin(), 'fresh synthetic control begins');
|
|
$transferAt = null; $cancelAt = null; $deltas = []; $terminals = [];
|
|
$abort = static function () use ($control, &$transferAt, &$cancelAt): bool {
|
|
if ($transferAt !== null && $cancelAt === null && hrtime(true) / 1e9 - $transferAt >= 0.08) {
|
|
$cancelAt = hrtime(true) / 1e9;
|
|
$control->signalCancel();
|
|
}
|
|
return $control->shouldAbort();
|
|
};
|
|
$result = DifyChatService::streamChat('qwen', [], '合成取消测试，无真实患者', 'synthetic-drain-user',
|
|
static function (string $text) use (&$deltas): bool { $deltas[] = $text; return true; }, $abort, [], [
|
|
'knowledge_mode' => 'evidence',
|
|
'on_progress' => static function (array $event) use (&$transferAt): void {
|
|
if ($transferAt === null) $transferAt = hrtime(true) / 1e9;
|
|
},
|
|
'on_task' => static fn (string $taskId) => $control->bindTask('qwen', 'synthetic-drain-user', $taskId),
|
|
'on_cancel_task' => static function (string $taskId, callable $stop) use ($control): array {
|
|
$control->signalCancel(); return $control->stopBoundTask($stop, $taskId);
|
|
},
|
|
'on_terminal' => static function (string $taskId, string $status) use ($control, &$terminals): void {
|
|
$terminals[] = $status; $control->markTerminal($taskId, $status);
|
|
},
|
|
'on_stop' => static fn (bool $accepted, string $taskId) => $control->markStop($accepted, $taskId),
|
|
'on_transfer_end' => static fn (string $taskId) => $control->clearTask($taskId !== '' ? $taskId : null),
|
|
]);
|
|
$cancelMs = (hrtime(true) / 1e9 - $cancelAt) * 1000;
|
|
$control->finish('cancelled');
|
|
$state = $control->publicState();
|
|
$peer = json_decode((string) fgets($pipes[1]), true);
|
|
$stderr = stream_get_contents($pipes[2]);
|
|
fclose($pipes[1]); fclose($pipes[2]);
|
|
drainExpect(proc_close($process) === 0 && $stderr === '', $scenario . ': loopback peer exits cleanly');
|
|
drainExpect(($result['error_code'] ?? '') === 'AI_ASSISTANT_CANCELLED' && $deltas === [],
|
|
$scenario . ': cancellation delivers no body and returns cancelled');
|
|
drainExpect($peer['title_disabled'] && $peer['inputs_object'], 'real wire disables titles and preserves input object');
|
|
drainExpect($cancelMs < 2900, $scenario . ': absolute cancellation budget bounds a silent upstream');
|
|
if (in_array($scenario, ['late_metadata', 'terminal_tail'], true)) {
|
|
drainExpect($peer['stop_count'] === 1 && $peer['generation_open_on_stop'] && $peer['user_matches'],
|
|
'known task stops once before the generation closes using its exact user');
|
|
drainExpect($terminals === ['stopped'] && $state['upstream_terminal_confirmed']
|
|
&& $state['upstream_terminal_status'] === 'stopped', 'discarded tails still expose workflow terminal metadata');
|
|
if ($scenario === 'terminal_tail') {
|
|
drainExpect($peer['terminal_tail_sent'] && !$peer['closed_before_terminal_tail'],
|
|
'workflow_finished preserves the connection through delayed message_end and natural EOF');
|
|
}
|
|
} elseif ($scenario === 'terminal_silent') {
|
|
drainExpect($peer['stop_count'] === 1 && $state['upstream_terminal_confirmed']
|
|
&& $peer['generation_eof'] && $cancelMs >= 2400,
|
|
'terminal followed by silence exits on the original budget without repeated stop');
|
|
} elseif ($scenario === 'no_terminal') {
|
|
drainExpect($peer['stop_count'] === 1 && $state['upstream_stop_confirmed']
|
|
&& !$state['upstream_terminal_confirmed'] && $peer['generation_eof'],
|
|
'stop acceptance without terminal metadata remains distinct after bounded draining');
|
|
drainExpect($cancelMs >= 2400, 'known task gets its bounded terminal observation window');
|
|
} else {
|
|
drainExpect($peer['stop_count'] === 0 && !$state['upstream_stop_confirmed']
|
|
&& !$state['upstream_terminal_confirmed'] && $peer['generation_eof'],
|
|
'unknown task closes transport without inventing upstream stop or terminal confirmation');
|
|
drainExpect($cancelMs >= 1900 && $cancelMs < 2300, 'unknown metadata grace is bounded to two seconds');
|
|
}
|
|
$rows[] = ['scenario' => $scenario, 'cancel_ms' => round($cancelMs, 2), 'stop_count' => $peer['stop_count']];
|
|
}
|
|
echo 'Diagnosis AI real loopback cancel drain / terminal / silent bounds: OK ' . json_encode($rows) . "\n";
|
|
} finally {
|
|
foreach (glob($directory . '/*') ?: [] as $path) unlink($path);
|
|
if (is_file($directory . '/.gc.lock')) unlink($directory . '/.gc.lock');
|
|
if (is_dir($directory)) rmdir($directory);
|
|
}
|
|
}
|