<?php

declare(strict_types=1);

namespace think\facade {
    final class Log {
        public static array $lines = [];
        public static function write(string $message, string $level): void { self::$lines[] = $message; }
        public static function warning(string $message): void { self::$lines[] = $message; }
    }
}

namespace app\common\service {
    require __DIR__ . '/fixtures/CancellableCurlMultiStub.php';

    // These namespaced cURL functions and clock are deterministic in-process
    // fixtures. No socket, provider, runtime control or database is contacted.
    function hrtime(bool $asNumber = false): int|array {
        $nanoseconds = (int) (($GLOBALS['runIdentityNow'] ?? 0) * 1000000);
        return $asNumber ? $nanoseconds : [intdiv($nanoseconds, 1000000000), $nanoseconds % 1000000000];
    }
    function curl_init(): \stdClass { return (object) ['errno' => 0]; }
    function curl_setopt_array(\stdClass $handle, array $options): bool { $handle->options = $options; return true; }
    function curl_exec(\stdClass $handle): string|bool {
        $options = $handle->options;
        if (str_ends_with($options[CURLOPT_URL], '/stop')) {
            \runIdentityExpect(!$GLOBALS['runIdentityClosed'], 'stop precedes generation transport close');
            \runIdentityExpect(str_ends_with($options[CURLOPT_URL], '/private-task/stop')
                && json_decode($options[CURLOPT_POSTFIELDS], true) === ['user' => 'private-user'],
                'stop retains the original task and user');
            ++$GLOBALS['runIdentityStopIo'];
            $GLOBALS['runIdentityNow'] += 7;
            return '{"result":"success"}';
        }
        $header = $options[CURLOPT_HEADERFUNCTION];
        $write = $options[CURLOPT_WRITEFUNCTION];
        $progressOption = defined('CURLOPT_XFERINFOFUNCTION') ? CURLOPT_XFERINFOFUNCTION : CURLOPT_PROGRESSFUNCTION;
        $progress = $options[$progressOption];
        $header($handle, "HTTP/1.1 200 OK\r\n");
        $mode = $GLOBALS['runIdentityMode'];
        if (!in_array($mode, ['late_unknown', 'initial_observer_failure'], true)) {
            $write($handle, \runIdentityFrame(['event' => 'node_started', 'task_id' => 'private-task']));
        }
        $GLOBALS['runIdentityCancelled'] = !in_array($mode, ['observer_failure', 'initial_observer_failure'], true);
        $GLOBALS['runIdentityNow'] += 10;
        $progress($handle, 0, 0, 1, 1);
        $GLOBALS['runIdentityNow'] += $mode === 'absolute_bound' ? 2400 : 100;
        $start = \runIdentityFrame(['event' => 'workflow_started', 'task_id' => 'private-task',
            'workflow_run_id' => 'private-run', 'data' => ['id' => 'private-run']]);
        if (in_array($mode, ['observer_failure', 'initial_observer_failure'], true)) {
            // A whole terminal in the same received chunk still cannot bypass
            // failure to persist the preceding identity enrichment.
            $start .= \runIdentityFrame(['event' => 'workflow_finished', 'task_id' => 'private-task',
                'workflow_run_id' => 'private-run', 'data' => ['id' => 'private-run', 'status' => 'succeeded']]);
        }
        if ($write($handle, $start) === 0) { $handle->errno = CURLE_WRITE_ERROR; return false; }
        if ($mode === 'absolute_bound') {
            $GLOBALS['runIdentityNow'] = $GLOBALS['runIdentityInterruptedAt'] + 2501;
            \runIdentityExpect($progress($handle, 0, 0, 1, 1) !== 0, 'late enrichment cannot restart the absolute drain bound');
            $handle->errno = CURLE_ABORTED_BY_CALLBACK;
            return false;
        }
        $write($handle, \runIdentityFrame(['event' => 'message', 'task_id' => 'private-task', 'answer' => 'private-late-text']));
        $write($handle, \runIdentityFrame(['event' => 'message_end']) . "data: [DONE]\r\n\r\n");
        if ($mode !== 'accepted_only') {
            $terminal = \runIdentityFrame(['event' => 'workflow_finished', 'task_id' => 'private-task',
                'workflow_run_id' => 'private-run', 'data' => ['id' => 'private-run', 'status' => 'cancelled',
                    'error' => 'private-provider-error']]);
            // Natural EOF flushes a final unterminated terminal frame.
            $write($handle, rtrim($terminal, "\r\n"));
        }
        return true;
    }
    function curl_errno(\stdClass $handle): int { return $handle->errno; }
    function curl_getinfo(\stdClass $handle, int $option): int|float { return $option === CURLINFO_HTTP_CODE ? 200 : 0.001; }
    function curl_close(\stdClass $handle): void {
        if (!str_ends_with($handle->options[CURLOPT_URL], '/stop')) $GLOBALS['runIdentityClosed'] = true;
    }
}

namespace {
    require dirname(__DIR__) . '/vendor/autoload.php';

    use app\common\service\DifyChatService;
    use app\common\service\DifyStreamTiming;
    use think\facade\Log;

    $runIdentityAssertions = 0;
    function runIdentityExpect(bool $condition, string $message): void {
        ++$GLOBALS['runIdentityAssertions'];
        if (!$condition) throw new RuntimeException($message);
    }
    function runIdentityPrivate(string $method, array $arguments = []): mixed {
        return (new ReflectionClass(DifyChatService::class))->getMethod($method)->invokeArgs(null, $arguments);
    }
    function runIdentityFrame(array $event, string $separator = "\r\n\r\n"): string {
        return 'data: ' . json_encode($event, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) . $separator;
    }
    function runIdentityParse(array $events, ?array $chunks = null, array $overrides = []): array {
        $state = runIdentityPrivate('newStreamState');
        $timing = new DifyStreamTiming();
        $tasks = []; $terminals = []; $deltas = [];
        $state['timing'] = $timing;
        $state['on_task'] = static function (string $task, string $run = '') use (&$tasks): void { $tasks[] = [$task, $run]; };
        $state['on_terminal'] = static function (string $task, string $status, string $run = '') use (&$terminals): void {
            $terminals[] = [$task, $status, $run];
        };
        $state = array_replace($state, $overrides);
        $buffer = '';
        $receive = static function (string $delta) use (&$deltas): bool { $deltas[] = $delta; return true; };
        $chunks ??= array_map('runIdentityFrame', $events);
        foreach ($chunks as $chunk) runIdentityPrivate('consumeStreamBytes', ['dify', &$buffer, $chunk, &$state, $receive]);
        runIdentityPrivate('consumeStreamBytes', ['dify', &$buffer, '', &$state, $receive, true]);
        return compact('state', 'tasks', 'terminals', 'deltas', 'timing');
    }
    $start = ['event' => 'workflow_started', 'task_id' => 'private-task', 'workflow_run_id' => 'private-run',
        'data' => ['id' => 'private-run', 'workflow_id' => 'private-definition']];
    $terminal = ['event' => 'workflow_finished', 'task_id' => 'private-task', 'workflow_run_id' => 'private-run',
        'data' => ['id' => 'private-run', 'status' => 'cancelled']];
    $wire = ": ping\r\n\r\n" . runIdentityFrame(['event' => 'node_started', 'task_id' => 'private-task'])
        . runIdentityFrame($start) . runIdentityFrame($start)
        . runIdentityFrame(['event' => 'message', 'answer' => '<think>private-plan</think>诊断正文'])
        . runIdentityFrame(['event' => 'message_end', 'message_id' => 'public-message'])
        . "data: [DONE]\r\n\r\n" . runIdentityFrame($terminal, '');
    for ($split = 0; $split <= strlen($wire); ++$split) {
        $parsed = runIdentityParse([], [substr($wire, 0, $split), substr($wire, $split)]);
        runIdentityExpect($parsed['tasks'] === [['private-task', ''], ['private-task', 'private-run']]
            && $parsed['terminals'] === [['private-task', 'stopped', 'private-run']],
            'every byte split retains one original task announcement and one immutable run enrichment');
        runIdentityExpect($parsed['state']['content'] === '诊断正文' && $parsed['state']['finished']
            && $parsed['state']['upstream_terminal_confirmed'], 'fragmented CRLF, UTF-8, reasoning and final terminal keep exact delivery');
    }
    $parsed = runIdentityParse([], str_split($wire, 1));
    runIdentityExpect(count($parsed['tasks']) === 2 && count($parsed['terminals']) === 1, 'single-byte fragmentation is idempotent');

    foreach (['top', 'nested', 'both'] as $startShape) {
        foreach (['top', 'nested', 'both'] as $endShape) {
            $shapedStart = $start; $shapedEnd = $terminal;
            if ($startShape === 'top') unset($shapedStart['data']['id']);
            if ($startShape === 'nested') unset($shapedStart['workflow_run_id']);
            if ($endShape === 'top') unset($shapedEnd['data']['id']);
            if ($endShape === 'nested') unset($shapedEnd['workflow_run_id']);
            $parsed = runIdentityParse([$shapedStart, $shapedEnd, $shapedEnd]);
            runIdentityExpect($parsed['state']['workflow_run_id'] === 'private-run'
                && $parsed['terminals'] === [['private-task', 'stopped', 'private-run']], 'start/finish run aliases agree and duplicate terminals are idempotent');
        }
    }
    $invalidIds = [null, 1, false, [], new stdClass(), '', str_repeat('a', 129), 'bad/path', 'https://invalid.test/run',
        "private-run\n", "private-run\r", ' private-run', 'private-run ', 'private.run', '运行'];
    foreach ($invalidIds as $invalid) {
        foreach (['top', 'nested', 'good_top_bad_nested', 'bad_top_good_nested'] as $shape) {
            $badEnd = $terminal;
            if ($shape === 'top') { $badEnd['workflow_run_id'] = $invalid; unset($badEnd['data']['id']); }
            elseif ($shape === 'nested') { $badEnd['data']['id'] = $invalid; unset($badEnd['workflow_run_id']); }
            elseif ($shape === 'good_top_bad_nested') $badEnd['data']['id'] = $invalid;
            else $badEnd['workflow_run_id'] = $invalid;
            $parsed = runIdentityParse([$start, $badEnd]);
            runIdentityExpect(!$parsed['state']['upstream_terminal_confirmed'] && $parsed['terminals'] === [], 'malformed terminal aliases cannot confirm a run-bound task');
            $badStart = $badEnd; $badStart['event'] = 'workflow_started';
            $parsed = runIdentityParse([$badStart, ['event' => 'workflow_finished', 'task_id' => 'private-task', 'data' => ['status' => 'succeeded']]]);
            runIdentityExpect($parsed['state']['workflow_run_id'] === '' && !$parsed['state']['upstream_terminal_confirmed'], 'malformed start aliases cannot downgrade into task-only terminal proof');
        }
        $badTaskEnd = $terminal; $badTaskEnd['task_id'] = $invalid;
        $parsed = runIdentityParse([$start, $badTaskEnd]);
        runIdentityExpect($parsed['terminals'] === [] && !$parsed['state']['upstream_terminal_confirmed'], 'terminal requires an explicit valid matching task');
        $badTaskStart = $start; $badTaskStart['task_id'] = $invalid;
        $parsed = runIdentityParse([$badTaskStart, $terminal]);
        runIdentityExpect($parsed['tasks'] === [] && $parsed['state']['workflow_run_id'] === ''
            && !$parsed['state']['upstream_terminal_confirmed'], 'invalid start task never binds run and terminal never supplies missing task binding');
    }
    foreach (['task_id', 'workflow_run_id', 'all_run_fields'] as $missing) {
        $missingEnd = $terminal;
        if ($missing === 'all_run_fields') { unset($missingEnd['workflow_run_id'], $missingEnd['data']['id']); }
        elseif ($missing === 'workflow_run_id') { unset($missingEnd['workflow_run_id'], $missingEnd['data']['id']); }
        else unset($missingEnd['task_id']);
        runIdentityExpect(!runIdentityParse([$start, $missingEnd])['state']['upstream_terminal_confirmed'], 'missing terminal identity cannot reuse the stored identity');
    }
    $noTaskStart = $start; unset($noTaskStart['task_id']);
    $parsed = runIdentityParse([$noTaskStart, ['event' => 'message', 'task_id' => 'private-task'], $terminal]);
    runIdentityExpect($parsed['state']['workflow_run_id'] === '' && $parsed['terminals'] === [], 'start with absent task cannot bind a run indirectly through a later task frame');
    $parsed = runIdentityParse([$terminal]);
    runIdentityExpect($parsed['state']['task_id'] === '' && $parsed['tasks'] === [] && $parsed['terminals'] === [], 'terminal-only stream establishes no original identity');

    $foreignStart = $start; $foreignStart['task_id'] = 'foreign-task'; $foreignStart['workflow_run_id'] = 'foreign-run'; $foreignStart['data']['id'] = 'foreign-run';
    $conflictStart = $start; $conflictStart['workflow_run_id'] = 'foreign-run'; $conflictStart['data']['id'] = 'foreign-run';
    $conflictEnd = $terminal; $conflictEnd['workflow_run_id'] = 'foreign-run'; $conflictEnd['data']['id'] = 'foreign-run';
    $aliasConflict = $terminal; $aliasConflict['data']['id'] = 'foreign-run';
    $foreignEnd = $terminal; $foreignEnd['task_id'] = 'foreign-task';
    $parsed = runIdentityParse([$start, $foreignStart, $conflictStart, $foreignEnd, $conflictEnd, $aliasConflict,
        ['event' => 'message', 'task_id' => 'foreign-task', 'answer' => 'foreign-text']]);
    runIdentityExpect($parsed['state']['task_id'] === 'private-task' && $parsed['state']['workflow_run_id'] === 'private-run'
        && count($parsed['tasks']) === 2 && $parsed['terminals'] === [] && $parsed['deltas'] === [], 'foreign/conflicting start, terminal and text cannot replace or confirm the original pair');
    $parsed = runIdentityParse([$start, $conflictStart, $conflictEnd, $terminal]);
    runIdentityExpect($parsed['terminals'] === [['private-task', 'stopped', 'private-run']], 'later exact original terminal can still prove the immutable original pair');
    $mismatchedStart = $start; $mismatchedStart['data']['id'] = 'foreign-run';
    runIdentityExpect(runIdentityParse([$mismatchedStart])['state']['workflow_run_id'] === '', 'start aliases must agree');
    $notArrayStart = $start; $notArrayStart['data'] = 'private-malformed';
    runIdentityExpect(runIdentityParse([$notArrayStart, $terminal])['terminals'] === [], 'malformed start data does not hide behind a valid top-level run');
    $notArrayEnd = $terminal; $notArrayEnd['data'] = 'private-malformed';
    runIdentityExpect(runIdentityParse([$start, $notArrayEnd])['terminals'] === [], 'malformed terminal data cannot confirm');

    $legacyStart = ['event' => 'workflow_started', 'task_id' => 'private-task', 'data' => ['workflow_id' => 'private-definition']];
    $legacyEnd = ['event' => 'workflow_finished', 'task_id' => 'private-task', 'data' => ['status' => 'succeeded']];
    $parsed = runIdentityParse([$legacyStart, $legacyEnd]);
    runIdentityExpect($parsed['state']['workflow_run_id'] === '' && $parsed['tasks'] === [['private-task', '']]
        && $parsed['terminals'] === [['private-task', 'succeeded', '']], 'definition workflow_id never becomes run identity; explicit task-only legacy stream remains compatible');
    $parsed = runIdentityParse([$legacyStart, $legacyEnd, $start, $terminal]);
    runIdentityExpect($parsed['state']['workflow_run_id'] === '' && $parsed['tasks'] === [['private-task', '']]
        && $parsed['terminals'] === [['private-task', 'succeeded', '']], 'start after a proved legacy terminal cannot reopen or enrich a released task');
    foreach ($invalidIds as $invalid) {
        $badLegacy = $legacyEnd; $badLegacy['task_id'] = $invalid;
        runIdentityExpect(runIdentityParse([$legacyStart, $badLegacy])['terminals'] === [], 'legacy terminal never infers an invalid task');
    }
    $badLegacy = $legacyEnd; unset($badLegacy['task_id']);
    runIdentityExpect(runIdentityParse([$legacyStart, $badLegacy])['terminals'] === [], 'legacy terminal never infers an absent task');
    $parsed = runIdentityParse([
        ['event' => 'message', 'task_id' => 'private-task', 'workflow_run_id' => 'private-untrusted', 'answer' => '正文'],
        ['event' => 'node_started', 'task_id' => 'private-task', 'data' => ['id' => 'private-node']], $terminal]);
    runIdentityExpect($parsed['state']['workflow_run_id'] === '' && $parsed['terminals'] === [], 'message/node IDs and terminal run cannot invent the original workflow_started pair');
    foreach (['succeeded', 'failed', 'stopped', 'cancelled', 'SUCCEEDED', 'running', 'unknown', '', null, ['succeeded']] as $status) {
        $statusEnd = $terminal; $statusEnd['data']['status'] = $status;
        $parsed = runIdentityParse([$start, $statusEnd]);
        $expected = is_string($status) && in_array(strtolower($status), ['succeeded', 'failed', 'stopped', 'cancelled'], true);
        runIdentityExpect($parsed['state']['upstream_terminal_confirmed'] === $expected, 'only terminal status whitelist confirms');
        if ($status === 'cancelled') runIdentityExpect($parsed['state']['upstream_terminal_status'] === 'stopped', 'cancelled normalizes to stopped');
    }
    $parsed = runIdentityParse([$start, ['event' => 'message_end']], null, ['discard_deltas' => true, 'client_aborted' => true]);
    runIdentityExpect($parsed['state']['finished'] && !$parsed['state']['upstream_terminal_confirmed'], 'message_end plus EOF is separate from real workflow terminal');
    $parsed = runIdentityParse([], [runIdentityFrame($legacyStart), "data: [DONE]\r\n\r\n"]);
    runIdentityExpect($parsed['state']['finished'] && !$parsed['state']['upstream_terminal_confirmed'], 'DONE plus EOF is separate from real workflow terminal');
    $parsed = runIdentityParse([['event' => 'message_end'], $start,
        ['event' => 'message', 'answer' => 'private-late-text'], $terminal], null,
        ['discard_deltas' => true, 'client_aborted' => true]);
    runIdentityExpect($parsed['state']['client_aborted'] && $parsed['deltas'] === [] && $parsed['state']['content'] === ''
        && $parsed['terminals'] === [['private-task', 'stopped', 'private-run']], 'late start and exact terminal survive completion/cancellation drain while text stays discarded');
    $maxStart = $start; $maxStart['task_id'] = str_repeat('t', 128); $maxStart['workflow_run_id'] = str_repeat('r', 128); $maxStart['data']['id'] = str_repeat('r', 128);
    $maxEnd = $maxStart; $maxEnd['event'] = 'workflow_finished'; $maxEnd['data']['status'] = 'succeeded';
    runIdentityExpect(runIdentityParse([$maxStart, $maxEnd])['state']['upstream_terminal_confirmed'], 'bounded IDs accept exactly the maximum length');

    foreach (['initial', 'enrichment', 'terminal'] as $failure) {
        $observedTasks = []; $observedTerminals = [];
        $parsed = runIdentityParse([$start, $start, $conflictStart, $foreignStart, $terminal, $terminal], null, [
            'on_task' => static function (string $task, string $run = '') use ($failure, &$observedTasks): void {
                $observedTasks[] = [$task, $run];
                if (($failure === 'initial' && $run === '') || ($failure === 'enrichment' && $run !== '')) throw new RuntimeException('private-observer-error');
            },
            'on_terminal' => static function () use ($failure, &$observedTerminals): void {
                $observedTerminals[] = true;
                if ($failure === 'terminal') throw new RuntimeException('private-terminal-error');
            },
        ]);
        runIdentityExpect($observedTasks === [['private-task', ''], ['private-task', 'private-run']],
            'initial observer failure still receives one original run enrichment; repeated/foreign/conflicting starts never rebind or retry');
        runIdentityExpect(count($observedTerminals) === ($failure === 'terminal' ? 1 : 0),
            'failed identity observer never reaches terminal callback and failed terminal observer is not retried');
        runIdentityExpect($parsed['state']['callback_error'] && !$parsed['state']['upstream_terminal_confirmed']
            && $parsed['state']['task_id'] === 'private-task' && $parsed['state']['workflow_run_id'] === 'private-run'
            && !isset($parsed['timing']->snapshot()['workflow_terminal_ms']), 'observer exception fails closed, retains known pair and invents no terminal timing');
    }

    foreach (['late_known', 'late_unknown', 'accepted_only', 'absolute_bound', 'observer_failure', 'initial_observer_failure'] as $mode) {
        $GLOBALS['runIdentityMode'] = $mode; $GLOBALS['runIdentityNow'] = 1000;
        $GLOBALS['runIdentityCancelled'] = false; $GLOBALS['runIdentityClosed'] = false;
        $GLOBALS['runIdentityStopIo'] = 0; $GLOBALS['runIdentityInterruptedAt'] = null;
        $tasks = []; $terminals = []; $deltas = []; $safeTiming = []; $accepted = false; $stopNotification = null;
        $result = runIdentityPrivate('sendStreamRequest', ['dify', 'https://private.invalid/chat-messages',
            ['user' => 'private-user', 'query' => 'private-query'], 'private-api-key', 5,
            static function (string $delta) use (&$deltas): bool { $deltas[] = $delta; return true; },
            static fn (): bool => $GLOBALS['runIdentityCancelled'], [
                'on_task' => static function (string $task, string $run = '') use (&$tasks, &$accepted, $mode): void {
                    $tasks[] = [$task, $run];
                    if ($run !== '' && $mode === 'late_known') runIdentityExpect($accepted, 'same-task run enrichment preserves prior accepted stop');
                    if ($run !== '' && $mode === 'observer_failure') throw new RuntimeException('private-observer-error');
                    if ($mode === 'initial_observer_failure') throw new RuntimeException('UPSTREAM_TASK_HANDOFF_BLOCKED');
                },
                'on_cancel_task' => static function (string $task, callable $stop) use (&$accepted): array {
                    $GLOBALS['runIdentityInterruptedAt'] ??= $GLOBALS['runIdentityNow'];
                    if (!$accepted) $accepted = $stop();
                    return ['accepted' => $accepted];
                },
                'on_terminal' => static function (string $task, string $status, string $run = '') use (&$terminals): void { $terminals[] = [$task, $status, $run]; },
                'on_stop' => static function (bool $accepted, string $task) use (&$stopNotification): void { $stopNotification = $accepted; },
                'on_upstream_timing' => static function (DifyStreamTiming $timing) use (&$safeTiming): void { $safeTiming = $timing->snapshot(); },
            ]]);
        runIdentityExpect($GLOBALS['runIdentityClosed'] && $GLOBALS['runIdentityStopIo'] === 1
            && $stopNotification === true && $deltas === [] && $result['content'] === '', 'stubbed transport stops exact task once, closes, and discards cancelled text');
        runIdentityExpect($tasks === [['private-task', ''], ['private-task', 'private-run']], 'transport captures late original pair and enriches exactly once');
        $shouldConfirm = in_array($mode, ['late_known', 'late_unknown'], true);
        runIdentityExpect($result['upstream_terminal_confirmed'] === $shouldConfirm
            && count($terminals) === ($shouldConfirm ? 1 : 0), 'accepted stop, EOF, local abort or observer failure cannot impersonate exact terminal');
        if ($mode === 'absolute_bound') runIdentityExpect($safeTiming['duration_ms'] <= 2511
            && $safeTiming['transport_exit'] === 'local_abort' && !isset($safeTiming['natural_eof_ms']), 'run enrichment adds no new cleanup window');
        if (in_array($mode, ['observer_failure', 'initial_observer_failure'], true)) {
            runIdentityExpect($result['callback_error'], 'initial/enrichment observer failure stops known task and keeps terminal unconfirmed');
        }
        $formatted = runIdentityPrivate('formatStreamResponse', [$result, microtime(true)]);
        runIdentityPrivate('logUpstreamFailure', [['protocol' => 'dify', 'url' => 'https://private.invalid/chat-messages'],
            $result, 'private-query', [], $formatted]);
        runIdentityExpect(!str_contains((string) json_encode([$result, $formatted, $safeTiming, Log::$lines]), 'private-')
            && !array_key_exists('task_id', $result) && !array_key_exists('workflow_run_id', $result), 'private task/run/user/host/key/body/error never enter reply or log/timing allowlists');
    }

    echo "Dify workflow run identity: {$runIdentityAssertions} assertions passed (pure parser and in-process transport; no network).\n";
}