<?php

declare(strict_types=1);

// All cURL, config and logs below are in-memory stubs. No socket, application
// bootstrap, credentials, database, upstream provider or model is accessed.
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, array $context = []): void { self::$lines[] = $message; }
    }
}
namespace app\common\service {
    function config(string $key): array { return $GLOBALS['privateWireConfig']; }
    function curl_init(): object { return (object) ['options' => [], 'errno' => 0, 'status' => 200]; }
    function curl_setopt_array(object $handle, array $options): bool {
        $handle->options = $options;
        if (!str_ends_with($options[CURLOPT_URL], '/stop')) $GLOBALS['privateWireBodies'][] = $options[CURLOPT_POSTFIELDS];
        return true;
    }
    function curl_exec(object $handle): bool|string {
        if (str_ends_with($handle->options[CURLOPT_URL], '/stop')) { ++$GLOBALS['privateWireStops']; return '{"result":"success"}'; }
        $scenario = array_shift($GLOBALS['privateWireScenarios']);
        if (!is_array($scenario)) throw new \RuntimeException('Unexpected synthetic transfer');
        $handle->status = $scenario['status'] ?? 200;
        ($handle->options[CURLOPT_HEADERFUNCTION])($handle, 'HTTP/1.1 ' . $handle->status . " OK\r\n");
        foreach ($scenario['chunks'] as $chunk) {
            if (($handle->options[CURLOPT_WRITEFUNCTION])($handle, $chunk) !== strlen($chunk)) {
                $handle->errno = CURLE_WRITE_ERROR; return false;
            }
        }
        return true;
    }
    function curl_errno(object $handle): int { return $handle->errno; }
    function curl_getinfo(object $handle, int $option): int|float { return $option === CURLINFO_HTTP_CODE ? $handle->status : 0; }
    function curl_close(object $handle): void { ++$GLOBALS['privateWireClosed']; }
}
namespace {
    require dirname(__DIR__) . '/vendor/autoload.php';
    require dirname(__DIR__, 2) . '/app/artifacts/ai_concurrency/private_tcm_r25_wire_audit.php';
    use app\common\service\DifyChatService;
    use app\adminapi\service\AssistantRequestControl;
    use think\facade\Log;

    $checks = 0; $privateWireBodies = []; $privateWireScenarios = []; $privateWireClosed = 0; $privateWireStops = 0;
    $privateWireConfig = ['enable' => true, 'base_url' => 'https://wire.synthetic.invalid/v1/chat-messages',
        'timeout' => 90, 'models' => ['qwen' => ['name' => 'synthetic-model', 'api_key' => 'private-never-sent-key']]];
    $expect = static function (bool $value, string $message) use (&$checks): void {
        ++$checks; if (!$value) throw new RuntimeException($message);
    };
    $frame = static fn (array $event): string => 'data: ' . json_encode($event, JSON_THROW_ON_ERROR) . "\n\n";
    $scenario = static function (string $task = 'synthetic-task', string $run = 'synthetic-run') use ($frame): array {
        return ['chunks' => [
            $frame(['event' => 'node_started', 'task_id' => $task]),
            $frame(['event' => 'workflow_started', 'task_id' => $task, 'workflow_run_id' => $run, 'data' => ['id' => $run]]),
            $frame(['event' => 'workflow_started', 'task_id' => $task, 'workflow_run_id' => $run, 'data' => ['id' => $run]]),
            $frame(['event' => 'workflow_started', 'task_id' => 'foreign-task', 'workflow_run_id' => 'foreign-run']),
            $frame(['event' => 'workflow_started', 'task_id' => $task, 'workflow_run_id' => 'changed-run']),
            $frame(['event' => 'message', 'task_id' => $task, 'answer' => 'synthetic-answer']),
            $frame(['event' => 'message_end', 'task_id' => $task]),
            $frame(['event' => 'workflow_finished', 'task_id' => $task, 'workflow_run_id' => $run,
                'data' => ['id' => $run, 'status' => 'succeeded']]),
        ]];
    };
    $invoke = static fn (string $method, array $args): mixed => (new ReflectionMethod(DifyChatService::class, $method))->invokeArgs(null, $args);
    $directory = sys_get_temp_dir() . '/tcm-r25-wire-test-' . bin2hex(random_bytes(8));
    mkdir($directory, 0700);
    $requestNumber = 0;
    $audit = static function () use (&$requestNumber, $directory): array {
        $id = sprintf('00000000-0000-4000-8000-%012d', ++$requestNumber);
        return [new PrivateTcmR25WireAudit($directory, $id, 'synthetic_test_' . $requestNumber, 'qwen'),
            $directory . '/' . hash('sha256', $id) . '.private.jsonl', $id];
    };
    $records = static function (string $path): array {
        return array_map(static fn (string $line): array => json_decode($line, true, 32, JSON_THROW_ON_ERROR),
            file($path, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES));
    };
    $cleanState = ['transport_closed' => true, 'upstream_terminal_confirmed' => true, 'upstream_stop_confirmed' => true,
        'cancel_requested' => false, 'timeout_reached' => false, 'upstream_pending_task_count' => 0];
    try {
        foreach (["plain-query", "中文 / path \\ \"quoted\"\nline", "invalid-utf8-\xff", "initial\n<REPORT_CONTINUATION_DATA>\n{}\n</REPORT_CONTINUATION_DATA>"] as $query) {
            [$trace, $path] = $audit();
            $captured = []; $controlTasks = []; $terminal = []; $seenBodyCount = count($privateWireBodies);
            $privateWireScenarios = [$scenario()];
            $reply = DifyChatService::streamChat('qwen', ['clinical' => "private-input-text/中文"], $query, 'private-user', static fn (): bool => true,
                null, [], ['knowledge_mode' => 'evidence', 'on_private_wire_request' => static function (array $wire) use ($trace, &$captured, $expect, $seenBodyCount): ?callable {
                    $expect(count($GLOBALS['privateWireBodies']) === $seenBodyCount + 1, 'wire observer runs after actual POSTFIELDS assignment');
                    $captured[] = $wire; return $trace->observe($wire);
                }, 'on_task' => static function (string $task, string $run = '') use (&$controlTasks): void { $controlTasks[] = [$task, $run]; },
                    'on_terminal' => static function (string $task, string $status, string $run = '') use (&$terminal): void { $terminal[] = [$task, $status, $run]; }]);
            $summary = $trace->finish($cleanState);
            $body = $privateWireBodies[array_key_last($privateWireBodies)];
            $wire = json_decode($body, false, 512, JSON_THROW_ON_ERROR);
            $expectedInputs = json_encode($wire->inputs, JSON_UNESCAPED_UNICODE | JSON_INVALID_UTF8_SUBSTITUTE | JSON_THROW_ON_ERROR);
            $expect($captured[0]['body_sha256'] === hash('sha256', $body) && $captured[0]['body_bytes'] === strlen($body), 'exact assigned serialized body fingerprint');
            $expect($captured[0]['query_sha256'] === hash('sha256', $wire->query) && $captured[0]['query_bytes'] === strlen($wire->query), 'actual post-serialization query fingerprint including UTF-8 substitution');
            $expect($captured[0]['inputs_sha256'] === hash('sha256', $expectedInputs) && $captured[0]['inputs_bytes'] === strlen($expectedInputs), 'actual serialized inputs fingerprint');
            $expect(!empty($reply['ok']) && $controlTasks === [['synthetic-task', ''], ['synthetic-task', 'synthetic-run']]
                && $terminal === [['synthetic-task', 'succeeded', 'synthetic-run']], 'duplicates and foreign metadata preserve normal identity and terminal behavior');
            $rows = $records($path);
            $expect($summary['persisted'] && $summary['audit_sha256'] === hash_file('sha256', $path)
                && $summary['original_task_count'] === 1 && $summary['original_run_count'] === 1, 'audit persists one original task and immutable enrichment');
            $expect($rows[1]['report_stage'] === (str_contains($wire->query, "\n<REPORT_CONTINUATION_DATA>\n") ? 2 : 1)
                && $rows[1]['stage_attempt'] === 1 && $rows[2]['generation'] === $rows[1]['generation']
                && $rows[3]['generation'] === $rows[1]['generation'], 'stage derives from actual serialized query and identity uses captured generation directly');
            $privateText = file_get_contents($path);
            foreach (['private-input-text', 'private-user', 'private-never-sent-key', 'synthetic-answer', 'foreign-task', 'changed-run'] as $secret)
                $expect(!str_contains($privateText, $secret), 'private audit excludes text/credentials and foreign identity');
            $expect(!str_contains(json_encode([$reply, $summary, Log::$lines]), 'synthetic-task')
                && !str_contains(json_encode([$reply, $summary, Log::$lines]), 'synthetic-run'), 'safe results/logs contain no original identity');
        }

        [$trace, $path] = $audit();
        $captured = []; $offset = count($privateWireBodies);
        $privateWireScenarios = [['status' => 400, 'chunks' => ['{"code":"invalid_inputs","message":"private-provider-error"}']], $scenario('retry-task', 'retry-run')];
        $reply = DifyChatService::streamChat('qwen', ['required' => 'private-input'], 'full-retry-query/中文', 'private-user', static fn (): bool => true,
            null, [], ['knowledge_mode' => 'evidence', 'on_private_wire_request' => static function (array $wire) use ($trace, &$captured): ?callable { $captured[] = $wire; return $trace->observe($wire); }]);
        $summary = $trace->finish($cleanState); $rows = $records($path);
        $serialized = array_values(array_filter($rows, static fn (array $row): bool => $row['event'] === 'serialized_request'));
        $identities = array_values(array_filter($rows, static fn (array $row): bool => $row['event'] === 'original_identity'));
        $expect(!empty($reply['ok']) && count($captured) === 2 && $summary['serialized_request_count'] === 2, 'actual inputs_invalid compatibility retry is separately audited');
        $expect($captured[0]['query_sha256'] === $captured[1]['query_sha256'] && $captured[0]['inputs_sha256'] !== $captured[1]['inputs_sha256']
            && $captured[1]['inputs_sha256'] === hash('sha256', '{}') && $captured[1]['inputs_bytes'] === 2, 'retry retains full query but distinguishes exact empty-object inputs');
        $expect(array_column($serialized, 'stage_attempt') === [1, 2] && array_column($serialized, 'generation') === [1, 2]
            && array_unique(array_column($identities, 'generation')) === [2], 'retry run links only to its own serialized request without neighbor inference');

        // The real private control releases its task/run after confirmed cleanup;
        // the independent audit remains readable with its original pair intact.
        [$trace, $path, $requestId] = $audit();
        $control = new AssistantRequestControl(991, 11, $requestId, $directory . '/controls');
        $expect($control->begin(), 'synthetic request control begins');
        $privateWireScenarios = [$scenario('cleanup-task', 'cleanup-run')];
        $reply = DifyChatService::streamChat('qwen', [], 'cleanup-query', 'private-user', static fn (): bool => true,
            null, [], ['knowledge_mode' => 'evidence', 'on_private_wire_request' => [$trace, 'observe'],
                'on_task' => static fn (string $task, string $run = '') => $control->bindTask('qwen', 'private-user', $task, $run),
                'on_terminal' => static fn (string $task, string $status, string $run = '') => $control->markTerminal($task, $status, $run),
                'on_transfer_end' => static fn (string $task) => $control->clearTask($task)]);
        $control->finish('done'); $summary = $trace->finish($control->publicState());
        $rows = $records($path); $last = $rows[array_key_last($rows)];
        $controlPath = $directory . '/controls/' . hash('sha256', '991:11:' . $requestId) . '.json';
        $expect(!empty($reply['ok']) && $control->publicState()['upstream_pending_task_count'] === 0
            && $last['event'] === 'after_control_cleanup' && $last['state']['upstream_pending_task_count'] === 0
            && $summary['audit_sha256'] === hash_file('sha256', $path), 'successful real control cleanup leaves independent immutable private trace');
        $expect(str_contains(file_get_contents($path), 'cleanup-task') && str_contains(file_get_contents($path), 'cleanup-run'), 'original exact pair retained after cleanup');
        $expect(!str_contains(file_get_contents($controlPath), 'cleanup-task') && !str_contains(file_get_contents($controlPath), 'cleanup-run'), 'successful control actually released the original identity');
        $expect($rows[0]['request_id'] === $requestId && $rows[0]['request_label'] === 'synthetic_test_' . $requestNumber,
            'private header retains exact own label-to-request association');

        [$trace, $path] = $audit();
        foreach (['initial-stage', "initial-stage\n<REPORT_CONTINUATION_DATA>\n{}\n</REPORT_CONTINUATION_DATA>"] as $number => $query) {
            $privateWireScenarios = [$scenario('stage-task-' . $number, 'stage-run-' . $number)];
            $reply = DifyChatService::streamChat('qwen', [], $query, 'private-user', static fn (): bool => true, null, [],
                ['knowledge_mode' => 'evidence', 'on_private_wire_request' => [$trace, 'observe']]);
            $expect(!empty($reply['ok']), 'initial and continuation keep normal delivery');
        }
        $summary = $trace->finish($cleanState); $rows = $records($path);
        $serialized = array_values(array_filter($rows, static fn (array $row): bool => $row['event'] === 'serialized_request'));
        $identities = array_values(array_filter($rows, static fn (array $row): bool => $row['event'] === 'original_identity' && $row['original_workflow_run_id'] !== ''));
        $expect(array_column($serialized, 'report_stage') === [1, 2] && array_column($serialized, 'stage_attempt') === [1, 1]
            && array_column($identities, 'original_workflow_run_id') === ['stage-run-0', 'stage-run-1']
            && array_column($identities, 'generation') === [1, 2], 'one request retains each original run with its own explicit report stage and attempt');

        // The closure is immutable even if independently called out of order.
        [$trace, $path, $requestId] = $audit();
        $sample = $captured[0];
        $first = $trace->observe($sample); $second = $trace->observe($sample);
        $second('second-task', ''); $first('first-task', '');
        $second('second-task', 'second-run'); $first('first-task', 'first-run');
        $first('foreign-task', 'foreign-run'); $first('first-task', 'changed-run');
        $first('first-task', 'first-run'); $first("bad\ntask", 'first-run');
        $first('first-task', '');
        $summary = $trace->finish($cleanState); $before = file_get_contents($path);
        $first('first-task', 'later-run');
        $duplicate = new PrivateTcmR25WireAudit($directory, $requestId, 'different_label', 'qwen');
        $expect(!$duplicate->finish($cleanState)['persisted'] && file_get_contents($path) === $before, 'exclusive create and finalization forbid overwrite or replay');
        $identities = array_values(array_filter($records($path), static fn (array $row): bool => $row['event'] === 'original_identity'));
        $expect(array_column($identities, 'generation') === [2, 1, 2, 1] && $identities[2]['original_workflow_run_id'] === 'second-run'
            && $identities[3]['original_workflow_run_id'] === 'first-run' && $summary['original_run_count'] === 2, 'out-of-order callbacks retain captured generation and immutable same-task runs');
        $expect(!str_contains($before, 'foreign-task') && !str_contains($before, 'changed-run') && !str_contains($before, 'bad'), 'wrong IDs and duplicate enrichment do not poison audit');

        foreach (['request_throw', 'identity_throw', 'return_invalid', 'disabled'] as $mode) {
            $privateWireScenarios = [$scenario()]; $observed = 0; $controlTasks = [];
            $options = ['knowledge_mode' => 'evidence', 'on_task' => static function (string $task, string $run = '') use (&$controlTasks): void { $controlTasks[] = [$task, $run]; }];
            if ($mode !== 'disabled') $options['on_private_wire_request'] = static function (array $wire) use ($mode, &$observed): mixed {
                ++$observed;
                if ($mode === 'request_throw') throw new RuntimeException('private-observer-failure');
                if ($mode === 'return_invalid') return new stdClass();
                return static function () use (&$observed): void { ++$observed; throw new RuntimeException('private-identity-failure'); };
            };
            $started = hrtime(true);
            $reply = DifyChatService::streamChat('qwen', [], 'same-query', 'private-user', static fn (): bool => true, null, [], $options);
            $expect(!empty($reply['ok']) && count($controlTasks) === 2 && (hrtime(true) - $started) / 1e6 < 1000,
                'private observer failure is bounded and leaves original control/delivery behavior intact');
            $expect($observed === ($mode === 'disabled' ? 0 : ($mode === 'identity_throw' ? 3 : 1)), 'private observer has no retry or duplicate enrichment callbacks');
            $expect(!str_contains(json_encode([$reply, Log::$lines]), 'private-observer-failure')
                && !str_contains(json_encode([$reply, Log::$lines]), 'private-identity-failure'), 'observer exceptions never enter result/log');
        }

        [$trace, $path] = $audit();
        $privateWireScenarios = [$scenario()];
        $result = $invoke('sendStreamRequest', ['dify', 'https://wire.synthetic.invalid/chat-messages', ['inputs' => (object) [], 'query' => 'bounded'], 'private-key', 90,
            static fn (): bool => true, null, ['on_private_wire_request' => [$trace, 'observe'],
                'on_task' => static function (): void { throw new RuntimeException('private-control-persist-failure'); }]]);
        $summary = $trace->finish([]);
        $expect($result['callback_error'] && !$result['upstream_terminal_confirmed'] && $summary['original_task_count'] === 1,
            'private tracing cannot clear the existing fail-closed control observer latch');
        $expect(!str_contains(json_encode([$result, $summary, Log::$lines]), 'private-control-persist-failure'), 'control callback exception text stays private');

        [$trace, $path] = $audit();
        for ($index = 0; $index < 16; ++$index) $expect(is_callable($trace->observe($sample)), 'finite per-request generation capacity');
        $expect($trace->observe($sample) === null && !$trace->finish([])['persisted'] && filesize($path) <= 65536, 'capacity overflow disables audit without unbounded writes');
        [$trace, $path] = $audit();
        $expect($trace->observe($sample + ['raw_query' => 'must-never-persist']) === null && !$trace->finish([])['persisted']
            && !str_contains(file_get_contents($path), 'must-never-persist'), 'closed metadata allowlist rejects raw text injection');

        $payload = ['model' => 'synthetic', 'messages' => [['role' => 'user', 'content' => 'private-openai-text']], 'stream' => true];
        $privateWireScenarios = [['chunks' => ['data: {"choices":[{"delta":{"content":"answer"}}]}' . "\n\n", "data: [DONE]\n\n"]]];
        $meta = null;
        $invoke('sendStreamRequest', ['openai', 'https://wire.synthetic.invalid/chat/completions', $payload, 'private-key', 90, static fn (): bool => true,
            null, ['on_private_wire_request' => static function (array $wire) use (&$meta): void { $meta = $wire; }]]);
        $expect($meta['protocol'] === 'openai' && $meta['query_sha256'] === null && $meta['inputs_sha256'] === null
            && $meta['body_sha256'] === hash('sha256', $privateWireBodies[array_key_last($privateWireBodies)]), 'alternate protocol truthfully marks absent query/inputs and retains exact body digest');
        $observed = 0; $beforeBodies = count($privateWireBodies);
        $expired = $invoke('sendStreamRequest', ['dify', 'https://wire.synthetic.invalid/chat-messages', ['query' => 'never-submit'], 'private-key', 90,
            static fn (): bool => true, null, ['request_deadline_ns' => hrtime(true) - 1,
                'on_private_wire_request' => static function () use (&$observed): void { ++$observed; }]]);
        $expect($expired['errno'] === CURLE_OPERATION_TIMEDOUT && $observed === 0 && count($privateWireBodies) === $beforeBodies,
            'expired ordinary deadline performs no serialization, private observation or transfer');
        $expect($privateWireScenarios === [], 'all and only declared stub transfers consumed');
        echo "Dify private wire audit: {$checks} assertions passed (in-memory transport, private synthetic files, no network).\n";
    } finally {
        // Only this test's freshly-created, resolved temporary directory is removed.
        $real = realpath($directory); $base = realpath(sys_get_temp_dir());
        if ($real === false || $base === false || !str_starts_with(str_replace('\\', '/', $real), rtrim(str_replace('\\', '/', $base), '/') . '/tcm-r25-wire-test-'))
            throw new RuntimeException('Synthetic cleanup boundary');
        $files = new RecursiveIteratorIterator(new RecursiveDirectoryIterator($real, FilesystemIterator::SKIP_DOTS), RecursiveIteratorIterator::CHILD_FIRST);
        foreach ($files as $file) { if ($file->isLink()) throw new RuntimeException('Synthetic cleanup symlink'); if ($file->isDir()) rmdir($file->getPathname()); else unlink($file->getPathname()); }
        rmdir($real);
    }
}