180 lines
13 KiB
PHP
180 lines
13 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');
|
|
$vision = new DifyStreamTiming($clock);
|
|
$vision->frame(['event' => 'node_finished', 'data' => ['id' => 'private-vision-node', 'node_type' => 'llm',
|
|
'inputs' => ['#files#' => [['type' => 'image', 'url' => 'private-source'], ['type' => 'document']]],
|
|
'process_data' => ['prompts' => [['role' => 'user', 'text' => 'private-prompt', 'files' => [
|
|
['type' => 'image', 'data' => 'private-image', 'detail' => 'high'],
|
|
['type' => 'image', 'data' => 'private-image', 'detail' => 'high'],
|
|
]]]]]], 'dify');
|
|
$visionNode = $vision->snapshot()['nodes'][0];
|
|
streamTimingExpect($visionNode['input_image_count'] === 1 && $visionNode['model_image_part_count'] === 2
|
|
&& $visionNode['model_high_detail_image_part_count'] === 2, 'detect expanded image input without retaining images');
|
|
streamTimingExpect(!str_contains(json_encode($vision->snapshot()), 'private-'), 'vision counts never retain prompts or sources');
|
|
$notExposed = new DifyStreamTiming($clock);
|
|
$notExposed->frame(['event' => 'node_finished', 'data' => ['id' => 'opaque', 'node_type' => 'llm']], 'dify');
|
|
streamTimingExpect(!isset($notExposed->snapshot()['nodes'][0]['model_image_part_count']), 'unexposed processing is unknown, not zero images');
|
|
foreach (['list-operator', 'document-extractor'] as $type) {
|
|
$notExposed->frame(['event' => 'node_finished', 'data' => ['id' => $type, 'node_type' => $type]], 'dify');
|
|
}
|
|
streamTimingExpect(array_column($notExposed->snapshot()['nodes'], 'type') === ['llm', 'list-operator', 'document-extractor'],
|
|
'document preparation timings have fixed safe node 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');
|
|
$openai->frame(['event' => 'node_retry', 'data' => ['retry_index' => 2, 'error' => 'private-provider-error']], 'dify');
|
|
$openai->frame(['choices' => [], 'usage' => ['completion_tokens_details' => ['reasoning_tokens' => 1234]]], 'openai');
|
|
$diagnostic = $openai->snapshot();
|
|
streamTimingExpect($diagnostic['reasoning_frame_count'] === 1 && $diagnostic['reasoning_tokens'] === 1234
|
|
&& $diagnostic['max_node_retry_index'] === 2 && isset($diagnostic['events']['node_retry']), 'reasoning and internal retry signals retain numeric diagnostic metadata only');
|
|
streamTimingExpect(!str_contains(json_encode($diagnostic), 'private-'), 'reasoning/retry diagnostics never retain content or provider errors');
|
|
|
|
$cleanup = new DifyStreamTiming($clock);
|
|
$now += 5; $cleanup->interrupted('private-abort-reason');
|
|
$now += 5; $cleanup->interrupted('timeout');
|
|
$cleanup->stopIoBegin();
|
|
$now += 7; $cleanup->stopIoEnd(false);
|
|
$now += 3; $cleanup->stopIoBegin();
|
|
$now += 7; $cleanup->stopIoEnd(true);
|
|
$now += 5; $cleanup->stopAccepted('coordinator');
|
|
$cleanup->workflowTerminal('private-terminal');
|
|
streamTimingExpect(!isset($cleanup->snapshot()['workflow_terminal_ms']), 'unknown workflow status cannot become terminal proof');
|
|
$cleanup->workflowTerminal('stopped');
|
|
$now += 5; $cleanup->workflowTerminal('succeeded');
|
|
$cleanup->transportReturned('natural_eof');
|
|
$now += 1; $cleanup->transportClosed();
|
|
$cleanupSnapshot = $cleanup->snapshot();
|
|
streamTimingExpect($cleanupSnapshot['interruption_detected_ms'] === 5 && $cleanupSnapshot['interruption_trigger'] === 'other'
|
|
&& $cleanupSnapshot['stop_io_begin_ms'] === 10 && $cleanupSnapshot['last_stop_io_begin_ms'] === 20
|
|
&& $cleanupSnapshot['stop_io_end_ms'] === 17 && $cleanupSnapshot['last_stop_io_end_ms'] === 27
|
|
&& $cleanupSnapshot['stop_io_attempt_count'] === 2 && $cleanupSnapshot['stop_io_completed_count'] === 2
|
|
&& $cleanupSnapshot['stop_io_accepted_count'] === 1 && $cleanupSnapshot['stop_accepted_ms'] === 27
|
|
&& $cleanupSnapshot['stop_accepted_source'] === 'local_io' && $cleanupSnapshot['workflow_terminal_ms'] === 32
|
|
&& $cleanupSnapshot['workflow_terminal_status'] === 'stopped' && $cleanupSnapshot['natural_eof_ms'] === 37
|
|
&& $cleanupSnapshot['transport_closed_ms'] === 38, 'bounded first/last I/O timeline separates attempts, acceptance, terminal, EOF and close');
|
|
streamTimingExpect(!str_contains(json_encode($cleanupSnapshot), 'private-'), 'cleanup keeps only numeric timestamps/counts and fixed safe enums');
|
|
$now -= 20;
|
|
streamTimingExpect($cleanup->snapshot()['duration_ms'] === 38, 'injected clock regression cannot move timestamps backwards');
|
|
$now += 1000000000;
|
|
streamTimingExpect($cleanup->snapshot()['duration_ms'] === 3600000, 'timestamps saturate at a bounded one-hour telemetry horizon');
|
|
$metricsProperty = new ReflectionProperty(DifyStreamTiming::class, 'metrics');
|
|
$boundedMetrics = $metricsProperty->getValue($cleanup);
|
|
$boundedMetrics['stop_io_attempt_count'] = 1000000000;
|
|
$metricsProperty->setValue($cleanup, $boundedMetrics);
|
|
$cleanup->stopIoBegin();
|
|
streamTimingExpect($cleanup->snapshot()['stop_io_attempt_count'] === 1000000000, 'actual I/O counter saturates without unbounded growth');
|
|
$unknownExit = new DifyStreamTiming($clock);
|
|
$unknownExit->transportReturned('private-transport-error');
|
|
$unknownExit->stopAccepted('private-peer-name');
|
|
streamTimingExpect($unknownExit->snapshot()['transport_exit'] === 'other'
|
|
&& $unknownExit->snapshot()['stop_accepted_source'] === 'other'
|
|
&& !isset($unknownExit->snapshot()['natural_eof_ms'])
|
|
&& !str_contains(json_encode($unknownExit->snapshot()), 'private-'), 'unknown source/outcome labels never leak or fabricate EOF');
|
|
|
|
// 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');
|
|
|
|
$terminalState = $parser->getMethod('newStreamState')->invoke(null);
|
|
$terminalTiming = new DifyStreamTiming($clock);
|
|
$terminalState['timing'] = $terminalTiming;
|
|
foreach ([
|
|
['event' => 'workflow_finished', 'data' => ['status' => 'stopped']],
|
|
['event' => 'workflow_started', 'task_id' => 'private-bound-task'],
|
|
['event' => 'workflow_finished', 'task_id' => 'private-bound-task', 'data' => ['status' => 'running']],
|
|
['event' => 'workflow_finished', 'task_id' => 'private-foreign-task', 'data' => ['status' => 'stopped']],
|
|
['event' => 'message_end'],
|
|
] as $frame) {
|
|
$consume->invokeArgs(null, ['dify', 'data: ' . json_encode($frame), &$terminalState, $accept]);
|
|
}
|
|
streamTimingExpect(!isset($terminalTiming->snapshot()['workflow_terminal_ms']), 'unbound/foreign/invalid terminal and message_end do not manufacture terminal timing');
|
|
$now += 5;
|
|
$consume->invokeArgs(null, ['dify', 'data: ' . json_encode(['event' => 'workflow_finished',
|
|
'task_id' => 'private-bound-task', 'data' => ['status' => 'cancelled', 'error' => 'private-provider-error']]), &$terminalState, $accept]);
|
|
$now += 5;
|
|
$consume->invokeArgs(null, ['dify', 'data: ' . json_encode(['event' => 'workflow_finished',
|
|
'task_id' => 'private-bound-task', 'data' => ['status' => 'succeeded']]), &$terminalState, $accept]);
|
|
streamTimingExpect($terminalTiming->snapshot()['workflow_terminal_ms'] === 5
|
|
&& $terminalTiming->snapshot()['workflow_terminal_status'] === 'stopped'
|
|
&& !str_contains(json_encode($terminalTiming->snapshot()), 'private-'), 'first allowed real workflow terminal remains immutable after message_end');
|
|
|
|
$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";
|