<?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";