90 lines
6.2 KiB
PHP
90 lines
6.2 KiB
PHP
<?php
|
|
declare(strict_types=1);
|
|
require dirname(__DIR__) . '/vendor/autoload.php';
|
|
|
|
use app\common\service\DifyStreamTiming;
|
|
use app\common\service\DifyChatService;
|
|
use app\adminapi\service\AssistantSseTiming;
|
|
|
|
function streamTimingExpect(bool $ok, string $message): void { if (!$ok) throw new RuntimeException($message); }
|
|
$now = 1000.0;
|
|
$clock = static function () use (&$now): float { return $now; };
|
|
$stages = new DifyStreamTiming($clock);
|
|
$now += 10; $stages->headers(100);
|
|
$now += 10; $stages->headers(200); $stages->body(10);
|
|
$stages->frame(['event' => 'workflow_started', 'private' => 'secret-patient-body'], 'dify');
|
|
$now += 10;
|
|
$stages->frame(['event' => 'node_started', 'data' => ['node_id' => 'private-node-id', 'node_type' => 'knowledge-retrieval',
|
|
'title' => 'secret-patient-body', 'inputs' => ['api_key' => 'secret-api-key']]], 'dify');
|
|
$now += 15;
|
|
$stages->frame(['event' => 'node_finished', 'data' => ['node_id' => 'private-node-id', 'node_type' => 'knowledge-retrieval', 'elapsed_time' => 0.015]], 'dify');
|
|
$now += 5; $stages->frame(['event' => 'node_started', 'data' => ['node_id' => 'private-llm-id', 'node_type' => 'llm']], 'dify');
|
|
$now += 300;
|
|
$stages->frame(['event' => 'agent_thought', 'thought' => 'private-reasoning-body'], 'dify');
|
|
$stages->frame(['event' => 'message', 'answer' => ''], 'dify');
|
|
streamTimingExpect(!isset($stages->snapshot()['first_answer_ms']), 'thinking and empty messages never count as an answer');
|
|
$now += 200; $stages->frame(['event' => 'message', 'answer' => 'private-answer-body'], 'dify');
|
|
$now += 2; $stages->delivered();
|
|
$now += 100; $stages->frame(['event' => 'node_finished', 'data' => ['node_id' => 'private-llm-id', 'elapsed_time' => 0.602]], 'dify');
|
|
$stages->frame(['event' => 'message_end', 'metadata' => ['usage' => ['prompt_tokens' => 123, 'completion_tokens' => 45,
|
|
'total_tokens' => 168, 'reasoning_tokens' => -1, 'api_key' => 'secret-api-key']]], 'dify');
|
|
$stages->transport(['connect_ms' => 3.7, 'tls_ms' => 9, 'curl_errno' => -3, 'url' => 'private-url', 'api_key' => 'secret-api-key']);
|
|
$snapshot = $stages->snapshot();
|
|
streamTimingExpect($snapshot['first_header_ms'] === 10 && $snapshot['final_header_ms'] === 20
|
|
&& $snapshot['first_answer_ms'] === 550 && $snapshot['first_delivered_ms'] === 552,
|
|
'separate interim headers, final headers, upstream answer and accepted delivery');
|
|
streamTimingExpect($snapshot['nodes'][0]['type'] === 'knowledge-retrieval' && $snapshot['nodes'][0]['start_ms'] === 30
|
|
&& $snapshot['nodes'][0]['finish_ms'] === 45 && $snapshot['nodes'][0]['provider_elapsed_ms'] === 15
|
|
&& $snapshot['nodes'][1]['type'] === 'llm', 'pair node timing without retaining node IDs');
|
|
streamTimingExpect($snapshot['prompt_tokens'] === 123 && !isset($snapshot['reasoning_tokens']) && !isset($snapshot['curl_errno']),
|
|
'missing reasoning and unknown transport errors remain unknown');
|
|
$encoded = json_encode($snapshot);
|
|
foreach (['secret-patient-body', 'secret-api-key', 'private-node-id', 'private-llm-id', 'private-reasoning-body', 'private-answer-body', 'private-url'] as $private) {
|
|
streamTimingExpect(!str_contains($encoded, $private), 'safe snapshot strips arbitrary upstream data');
|
|
}
|
|
$unknown = new DifyStreamTiming($clock);
|
|
$unknown->frame(['event' => 'private-secret-event', 'data' => ['node_id' => 'private-id']], 'dify');
|
|
streamTimingExpect(isset($unknown->snapshot()['events']['other']) && !str_contains(json_encode($unknown->snapshot()), 'private-'),
|
|
'unknown provider strings never become log labels');
|
|
$openai = new DifyStreamTiming($clock);
|
|
$now += 10; $openai->frame(['choices' => [['delta' => ['reasoning_content' => 'private-thought']]]], 'openai');
|
|
streamTimingExpect(!isset($openai->snapshot()['first_answer_ms']), 'reasoning content is not a visible first token');
|
|
$now += 20; $openai->frame(['choices' => [['delta' => ['content' => 'private-answer']]]], 'openai');
|
|
streamTimingExpect($openai->snapshot()['first_answer_ms'] === 30, 'Chat Completions first content is timed');
|
|
|
|
// Exercise the real parser, so telemetry agrees with content the consumer sees.
|
|
$parser = new ReflectionClass(DifyChatService::class);
|
|
$state = $parser->getMethod('newStreamState')->invoke(null);
|
|
$partsTiming = new DifyStreamTiming($clock);
|
|
$state['timing'] = $partsTiming;
|
|
$received = '';
|
|
$accept = static function (string $delta) use (&$received): void { $received .= $delta; };
|
|
$consume = $parser->getMethod('consumeStreamFrame');
|
|
$now += 10;
|
|
$consume->invokeArgs(null, ['openai', 'data: ' . json_encode(['choices' => [['delta' => ['content' => [
|
|
['type' => 'reasoning', 'text' => 'private-thought'], ['type' => 'text', 'text' => ''],
|
|
]]]]]), &$state, $accept]);
|
|
streamTimingExpect($received === '' && !isset($partsTiming->snapshot()['first_answer_ms']), 'non-text parts and empty text never count');
|
|
$now += 20;
|
|
$consume->invokeArgs(null, ['openai', 'data: ' . json_encode(['choices' => [['delta' => ['content' => [
|
|
['type' => 'image_url', 'image_url' => ['url' => 'private-url']],
|
|
['type' => 'text', 'text' => 'private-answer'], ['type' => 'text', 'text' => '-suffix'],
|
|
]]]]]), &$state, $accept]);
|
|
$partsSnapshot = $partsTiming->snapshot();
|
|
streamTimingExpect($received === 'private-answer-suffix' && $partsSnapshot['first_answer_ms'] === 30
|
|
&& $partsSnapshot['first_delivered_ms'] === 30, 'text-part array first answer and delivery agree with the real parser');
|
|
streamTimingExpect(!str_contains(json_encode($partsSnapshot), 'private-'), 'text-part timing retains no content');
|
|
|
|
$app = new \think\App(dirname(__DIR__) . '/');
|
|
$directory = sys_get_temp_dir() . '/zyt-upstream-timing-' . bin2hex(random_bytes(8));
|
|
$app->config->set(['default' => 'timing', 'channels' => ['timing' => ['type' => 'File', 'path' => $directory,
|
|
'single' => 'timing', 'realtime_write' => false]]], 'log');
|
|
$logger = new AssistantSseTiming(); $logger->logUpstream($stages);
|
|
$file = $directory . '/timing.log';
|
|
streamTimingExpect(is_file($file), 'stage telemetry persists before SSE exit');
|
|
$saved = file_get_contents($file);
|
|
streamTimingExpect(str_contains($saved, $logger->id()) && str_contains($saved, 'first_answer_ms')
|
|
&& !str_contains($saved, 'private-') && !str_contains($saved, 'secret-'), 'timing logs correlate stages without patient data');
|
|
unlink($file); rmdir($directory);
|
|
echo "Upstream stream stages / retrieval and LLM / true first content / patient-free logs: OK\n";
|