更新

server/app/adminapi/controller/tcm/DiagnosisController.php

@@ -969,7 +969,8 @@ class DiagnosisController extends BaseAdminController
                 $profile = (string) $metadata['model_key'];
                 $emit = $this->openAssistantSse();
                 $sseStartedAt = hrtime(true);
-                $started = $emit('start', $metadata + ['request_id' => $requestId, 'message' => '已连接，正在整理资料']);
+                $started = $emit('start', $metadata + ['request_id' => $requestId,
+                    'trace_id' => $timing->id(), 'message' => '已连接，正在整理资料']);
                 $timing->log('sse_start', ['status' => $started ? 'ok' : 'disconnected']);
                 return true;
             },

server/app/adminapi/logic/tcm/DiagnosisAiLogic.php

@@ -533,6 +533,13 @@ class DiagnosisAiLogic extends BaseLogic
         $retryMs = 0;
         $status = 'error';
         $upstreamStartedAt = hrtime(true);
+        if ($timing !== null) {
+            $priorObserver = $options['on_upstream_timing'] ?? null;
+            $options['on_upstream_timing'] = static function (\app\common\service\DifyStreamTiming $stages) use ($timing, $priorObserver): void {
+                $timing->logUpstream($stages);
+                if (is_callable($priorObserver)) $priorObserver($stages);
+            };
+        }
         $forwardDelta = static function (string $delta) use (&$deliveredDelta, &$deltaCount, &$deltaBytes, $onDelta) {
             $accepted = $onDelta($delta);
             if ($accepted !== false) {

server/app/adminapi/service/AssistantSseTiming.php

@@ -4,6 +4,7 @@ declare(strict_types=1);
 
 namespace app\adminapi\service;
 
+use app\common\service\DifyStreamTiming;
 use think\facade\Log;
 
 /** A request-local, patient-free timing record for the diagnosis assistant SSE path. */
@@ -91,4 +92,15 @@ final class AssistantSseTiming
             // Telemetry must never interrupt the clinical request.
         }
     }
+
+    /** The typed collector exposes only bounded numeric timing and enum metadata. */
+    public function logUpstream(DifyStreamTiming $upstream): void
+    {
+        try {
+            Log::write('diagnosis ai upstream stages ' . json_encode([
+                'trace_id' => $this->traceId, 'request_elapsed_ms' => self::elapsedMs($this->startedAt),
+                'timing' => $upstream->snapshot(),
+            ]), 'info');
+        } catch (\Throwable $ignored) {}
+    }
 }

server/app/common/service/DifyChatService.php

@@ -1350,6 +1350,8 @@ class DifyChatService
 
         $buffer = '';
         $state = self::newStreamState();
+        $upstreamTiming = is_callable($options['on_upstream_timing'] ?? null) ? new DifyStreamTiming() : null;
+        $state['timing'] = $upstreamTiming;
         $state['on_task'] = $options['on_task'] ?? null;
         $state['on_terminal'] = $options['on_terminal'] ?? null;
         $notify = self::transferProgressNotifier($options);
@@ -1416,9 +1418,10 @@ class DifyChatService
         $responseCode = 0;
         $errorBody = '';
         $errorBodyOverflow = false;
-        $header = static function ($handle, string $line) use (&$responseCode, &$errorBody, &$errorBodyOverflow, $markSubmitted): int {
+        $header = static function ($handle, string $line) use (&$responseCode, &$errorBody, &$errorBodyOverflow, $markSubmitted, $upstreamTiming): int {
             if (preg_match('/^HTTP\/\S+\s+(\d{3})(?:\s|$)/i', trim($line), $matches) === 1) {
                 $responseCode = (int) $matches[1];
+                $upstreamTiming?->headers($responseCode);
                 $errorBody = '';
                 $errorBodyOverflow = false;
                 if ($responseCode >= 200 && $responseCode < 300) $markSubmitted();
@@ -1432,9 +1435,11 @@ class DifyChatService
             &$responseCode,
             &$errorBody,
             &$errorBodyOverflow,
+            $upstreamTiming,
             $forward,
             $abort
         ): int {
+            $upstreamTiming?->body(strlen($chunk));
             if ($responseCode < 200 || $responseCode >= 300) {
                 // Keep a bounded JSON error only until its safe code/reason has
                 // been extracted. Never feed HTTP error bodies to the SSE parser.
@@ -1499,6 +1504,15 @@ class DifyChatService
             // A callback failure may escape libcurl before the regular cancel
             // poll. Make the coordinated stop attempt before closing this handle.
             try { $abort(); } catch (\Throwable $ignored) {}
+            if ($upstreamTiming !== null) {
+                try {
+                    $upstreamTiming->transport(['connect_ms' => curl_getinfo($ch, CURLINFO_CONNECT_TIME) * 1000,
+                        'tls_ms' => curl_getinfo($ch, CURLINFO_APPCONNECT_TIME) * 1000,
+                        'header_ttfb_ms' => curl_getinfo($ch, CURLINFO_STARTTRANSFER_TIME) * 1000,
+                        'transfer_ms' => curl_getinfo($ch, CURLINFO_TOTAL_TIME) * 1000,
+                        'http_status' => $httpCode, 'curl_errno' => $errno]);
+                } catch (\Throwable $ignored) {}
+            }
             curl_close($ch);
         }
 
@@ -1522,6 +1536,9 @@ class DifyChatService
             $state['upstream_reason'] = $metadata['upstream_reason'];
         }
         unset($errorBody);
+        if ($upstreamTiming !== null) {
+            try { ($options['on_upstream_timing'])($upstreamTiming); } catch (\Throwable $ignored) {}
+        }
 
         return [
             'errno' => $errno,
@@ -1686,6 +1703,7 @@ class DifyChatService
                 }
             }
             $event = strtolower((string) ($decoded['event'] ?? ''));
+            if (($state['timing'] ?? null) instanceof DifyStreamTiming) $state['timing']->frame($decoded, $protocol);
             if ($event === 'workflow_finished') {
                 $rawStatus = $decoded['data']['status'] ?? '';
                 $status = is_string($rawStatus) ? strtolower($rawStatus) : '';
@@ -1720,6 +1738,7 @@ class DifyChatService
             $state['message_id'] = (string) ($decoded['message_id'] ?? $state['message_id']);
         } else {
             if ($state['finished'] || $state['upstream_error']) return;
+            if (($state['timing'] ?? null) instanceof DifyStreamTiming) $state['timing']->frame($decoded, $protocol);
             if (is_array($decoded['error'] ?? null)) {
                 $state['upstream_error'] = true;
                 $metadata = self::upstreamErrorMetadata($decoded);
@@ -1745,6 +1764,7 @@ class DifyChatService
             return;
         }
         if ($state['discard_deltas']) return;
+        if (($state['timing'] ?? null) instanceof DifyStreamTiming) $state['timing']->delivered();
         $state['content'] .= $delta;
         $state['emitted'] = true;
     }

server/app/common/service/DifyRequiredAttachmentUpload.php

@@ -42,6 +42,7 @@ final class DifyRequiredAttachmentUpload
         array $options = []
     ): array {
         $temporary = [];
+        $multi = null;
         $skipped = [];
         $index = 0;
         $total = 0;
@@ -78,6 +79,11 @@ final class DifyRequiredAttachmentUpload
                 }
             }
             unset($item);
+            // Keep libcurl's connection cache only for this preparation. Each
+            // transfer still owns a fresh easy handle, so source downloads never
+            // inherit the Dify upload headers or request body.
+            $multi = curl_multi_init();
+            if ($multi === false) self::fail('CURL_UNAVAILABLE');
             $prepared = [];
             $uploadedIndices = [];
             $cache = [];
@@ -99,7 +105,7 @@ final class DifyRequiredAttachmentUpload
                     if (isset($item['path'])) {
                         self::copyLocal($item['path'], $path, $maxBytes, $deadline, $shouldAbort);
                     } else {
-                        $exists = self::download($item['url'], $item['parts'], $path, $maxBytes, $deadline, $shouldAbort, $options);
+                        $exists = self::download($item['url'], $item['parts'], $path, $maxBytes, $deadline, $shouldAbort, $options, $multi);
                     }
                     self::check($deadline, $shouldAbort);
                     if (!$exists) {
@@ -113,7 +119,7 @@ final class DifyRequiredAttachmentUpload
                                 'audio/wav' => 'wav', 'audio/mpeg' => 'mp3', 'text/plain' => 'txt'][$mime] ?? 'bin';
                         }
                         $cache[$cacheKey] = self::upload($endpoint, $apiKey, $user, $path, $mime,
-                            'attachment-' . $index . '.' . $extension, $deadline, $shouldAbort, $options['ca_file']);
+                            'attachment-' . $index . '.' . $extension, $deadline, $shouldAbort, $options['ca_file'], $multi);
                     }
                 }
                 if ($cache[$cacheKey] === null) {
@@ -155,6 +161,7 @@ final class DifyRequiredAttachmentUpload
                 'failed_index' => $index, 'attachment_count' => $total, 'skipped' => $skipped];
         } finally {
             foreach ($temporary as $path) if (is_file($path)) @unlink($path);
+            if ($multi !== null && $multi !== false) curl_multi_close($multi);
         }
     }
 
@@ -340,7 +347,7 @@ final class DifyRequiredAttachmentUpload
         }
     }
 
-    private static function download(string $url, array $parts, string $path, int $maxBytes, float $deadline, ?callable $shouldAbort, array $options): bool
+    private static function download(string $url, array $parts, string $path, int $maxBytes, float $deadline, ?callable $shouldAbort, array $options, $multi): bool
     {
         self::check($deadline, $shouldAbort);
         $host = trim($parts['host'], '[]');
@@ -391,7 +398,7 @@ final class DifyRequiredAttachmentUpload
                 return $written;
             };
             if (!curl_setopt_array($ch, $curlOptions)) self::fail('REQUIRED_FILES_DOWNLOAD_FAILED');
-            $result = self::execute($ch, $deadline, $shouldAbort);
+            $result = self::execute($multi, $ch, $deadline, $shouldAbort);
             if ($writeError !== '') self::fail($writeError);
             if ($result['errno'] === CURLE_OPERATION_TIMEDOUT) self::fail('REQUIRED_FILES_TIMEOUT');
             if ($result['errno'] !== CURLE_OK) self::fail('REQUIRED_FILES_DOWNLOAD_FAILED');
@@ -426,7 +433,7 @@ final class DifyRequiredAttachmentUpload
             && ($remaining === 0 || (ord($packed[$bytes]) & (255 << (8 - $remaining))) === (ord($base[$bytes]) & (255 << (8 - $remaining))));
     }
 
-    private static function upload(string $endpoint, string $apiKey, string $user, string $path, string $mime, string $name, float $deadline, ?callable $shouldAbort, string $caFile): string
+    private static function upload(string $endpoint, string $apiKey, string $user, string $path, string $mime, string $name, float $deadline, ?callable $shouldAbort, string $caFile, $multi): string
     {
         self::check($deadline, $shouldAbort);
         $ch = curl_init();
@@ -444,7 +451,7 @@ final class DifyRequiredAttachmentUpload
                 return strlen($chunk);
             };
             if (!curl_setopt_array($ch, $options)) self::fail('REQUIRED_FILES_UPLOAD_FAILED');
-            $result = self::execute($ch, $deadline, $shouldAbort);
+            $result = self::execute($multi, $ch, $deadline, $shouldAbort);
             if ($result['errno'] === CURLE_OPERATION_TIMEDOUT) self::fail('REQUIRED_FILES_TIMEOUT');
             if ($result['errno'] !== CURLE_OK || $tooLarge || !in_array($result['status'], [200, 201], true)) self::fail('REQUIRED_FILES_UPLOAD_FAILED');
             $decoded = json_decode($body, true);
@@ -475,10 +482,9 @@ final class DifyRequiredAttachmentUpload
         return $options;
     }
 
-    /** curl_multi polling observes cancellation even while no headers/body arrive. */
-    private static function execute($ch, float $deadline, ?callable $shouldAbort): array
+    /** Request-local pool retains connections; polling also observes silent cancellation. */
+    private static function execute($multi, $ch, float $deadline, ?callable $shouldAbort): array
     {
-        $multi = curl_multi_init();
         $added = false;
         try {
             if (curl_multi_add_handle($multi, $ch) !== CURLM_OK) self::fail('REQUIRED_FILES_PREPARE_FAILED');
@@ -495,7 +501,6 @@ final class DifyRequiredAttachmentUpload
             return ['errno' => $errno, 'status' => (int) curl_getinfo($ch, CURLINFO_HTTP_CODE)];
         } finally {
             if ($added) curl_multi_remove_handle($multi, $ch);
-            curl_multi_close($multi);
         }
     }
 

server/app/common/service/DifyStreamTiming.php

@@ -0,0 +1,107 @@
+<?php
+declare(strict_types=1);
+
+namespace app\common\service;
+
+/** Request-local numeric telemetry. Raw events, prompts, titles, URLs and IDs never leave it. */
+final class DifyStreamTiming
+{
+    private const EVENTS = ['workflow_started', 'workflow_finished', 'node_started', 'node_finished',
+        'message', 'agent_message', 'agent_thought', 'message_end', 'error', 'ping', 'choices'];
+    private const TYPES = ['start', 'llm', 'knowledge-retrieval', 'tool', 'code', 'template-transform',
+        'if-else', 'answer', 'end', 'agent', 'http-request', 'iteration', 'question-classifier',
+        'variable-aggregator', 'parameter-extractor'];
+    private \Closure $clock;
+    private float $started;
+    private array $metrics = [];
+    private array $events = [];
+    private array $nodes = [];
+    private array $nodePositions = [];
+
+    public function __construct(?callable $clock = null)
+    {
+        $this->clock = $clock === null ? static fn (): float => hrtime(true) / 1e6 : \Closure::fromCallable($clock);
+        $this->started = ($this->clock)();
+    }
+
+    private function elapsed(): int { return max(0, (int) floor(($this->clock)() - $this->started)); }
+
+    public function headers(int $status): void
+    {
+        $this->metrics['first_header_ms'] ??= $this->elapsed();
+        if ($status >= 200 && $status <= 599) $this->metrics['final_header_ms'] ??= $this->elapsed();
+    }
+
+    public function body(int $bytes): void
+    {
+        if ($bytes <= 0) return;
+        $this->metrics['first_body_ms'] ??= $this->elapsed();
+    }
+
+    public function frame(array $decoded, string $protocol): void
+    {
+        $this->metrics['first_frame_ms'] ??= $this->elapsed();
+        $name = $protocol === 'dify' ? ($decoded['event'] ?? 'other') : (isset($decoded['choices']) ? 'choices' : 'other');
+        if (!is_string($name) || !in_array($name, self::EVENTS, true)) $name = 'other';
+        $now = $this->elapsed();
+        $this->events[$name] ??= ['first_ms' => $now, 'last_ms' => $now, 'count' => 0];
+        ++$this->events[$name]['count']; $this->events[$name]['last_ms'] = $now;
+        if (in_array($name, ['message', 'agent_message', 'choices'], true)) {
+            $answer = $protocol === 'dify' ? ($decoded['answer'] ?? '') : ($decoded['choices'][0]['delta']['content'] ?? '');
+            $visible = is_string($answer) && $answer !== '';
+            // Chat Completions also supports text-part arrays. Match the parser's
+            // accepted shape without retaining or concatenating clinical content.
+            if ($protocol !== 'dify' && is_array($answer)) {
+                foreach ($answer as $part) {
+                    if (is_array($part) && ($part['type'] ?? '') === 'text'
+                        && is_string($part['text'] ?? null) && $part['text'] !== '') { $visible = true; break; }
+                }
+            }
+            if ($visible) $this->metrics['first_answer_ms'] ??= $now;
+        }
+        if (in_array($name, ['node_started', 'node_finished'], true)) {
+            $data = is_array($decoded['data'] ?? null) ? $decoded['data'] : [];
+            $id = $data['id'] ?? $data['node_id'] ?? '';
+            if (is_string($id) && $id !== '' && strlen($id) <= 512) {
+                $key = hash('sha256', $id);
+                if (!isset($this->nodePositions[$key]) && count($this->nodes) >= 128) return;
+                if (!isset($this->nodePositions[$key])) {
+                    $this->nodePositions[$key] = count($this->nodes);
+                    $type = $data['node_type'] ?? 'other';
+                    $this->nodes[] = ['type' => is_string($type) && in_array($type, self::TYPES, true) ? $type : 'other'];
+                }
+                $position = $this->nodePositions[$key];
+                $field = $name === 'node_started' ? 'start_ms' : 'finish_ms';
+                $this->nodes[$position][$field] ??= $now;
+                $elapsed = $data['elapsed_time'] ?? null;
+                if (is_numeric($elapsed) && is_finite((float) $elapsed) && (float) $elapsed >= 0) {
+                    $this->nodes[$position]['provider_elapsed_ms'] = (int) min(3600000, (float) $elapsed * 1000);
+                }
+            }
+        }
+        $usage = $name === 'message_end' ? ($decoded['metadata']['usage'] ?? []) : ($decoded['usage'] ?? []);
+        foreach (['prompt_tokens', 'completion_tokens', 'total_tokens', 'reasoning_tokens'] as $key) {
+            $value = is_array($usage) ? ($usage[$key] ?? null) : null;
+            if (is_numeric($value) && is_finite((float) $value) && (float) $value >= 0) {
+                $this->metrics[$key] = (int) min(1000000000, (float) $value);
+            }
+        }
+    }
+
+    public function delivered(): void { $this->metrics['first_delivered_ms'] ??= $this->elapsed(); }
+
+    public function transport(array $values): void
+    {
+        foreach (['connect_ms', 'tls_ms', 'header_ttfb_ms', 'transfer_ms', 'http_status', 'curl_errno'] as $key) {
+            $value = $values[$key] ?? null;
+            if ((is_float($value) || is_int($value)) && is_finite((float) $value) && $value >= 0) {
+                $this->metrics[$key] = (int) min(3600000, $value);
+            }
+        }
+    }
+
+    public function snapshot(): array
+    {
+        return ['duration_ms' => $this->elapsed()] + $this->metrics + ['events' => $this->events, 'nodes' => $this->nodes];
+    }
+}

server/tests/DiagnosisAiAssistantTimingTest.php

@@ -91,5 +91,6 @@ timingExpect(is_string($controller) && is_string($logic), 'instrumented source i
 timingExpect(str_contains($controller, "log('first_delta'") && str_contains($controller, "log('sse_done'"), 'SSE first delta and completion are timed');
 timingExpect(str_contains($logic, "log('prepare'") && str_contains($logic, "log('upstream_done'"), 'prepare and upstream are timed');
 timingExpect(str_contains($logic, 'compaction_attempted') && str_contains($logic, 'retry_attempted'), 'compression and compatibility retry are recorded');
+timingExpect(str_contains($controller, "'trace_id' => \$timing->id()"), 'SSE start exposes only the random trace id for timing correlation');
 
 echo "Diagnosis AI assistant timing: OK\n";

server/tests/DifyRequiredAttachmentUploadTest.php

@@ -18,6 +18,56 @@ namespace app\common\service {
         return [['ip' => '198.18.0.1']];
     }
 
+    /** Weak keys observe lifetime without keeping a connection pool alive in the test. */
+    function attachmentTransportId(object $handle): int
+    {
+        $handles = $GLOBALS['attachmentTransportHandles'];
+        if (!isset($handles[$handle])) $handles[$handle] = ++$GLOBALS['attachmentTransportSequence'];
+        return $handles[$handle];
+    }
+
+    function curl_init()
+    {
+        $handle = \curl_init();
+        if ($handle !== false) $GLOBALS['attachmentTransportEvents'][] = ['kind' => 'easy_init', 'easy' => attachmentTransportId($handle)];
+        return $handle;
+    }
+
+    function curl_close($handle): void
+    {
+        $GLOBALS['attachmentTransportEvents'][] = ['kind' => 'easy_close', 'easy' => attachmentTransportId($handle)];
+        \curl_close($handle);
+    }
+
+    function curl_multi_init()
+    {
+        $multi = \curl_multi_init();
+        if ($multi !== false) $GLOBALS['attachmentTransportEvents'][] = ['kind' => 'pool_init', 'pool' => attachmentTransportId($multi)];
+        return $multi;
+    }
+
+    function curl_multi_add_handle($multi, $handle): int
+    {
+        $result = !empty($GLOBALS['attachmentFailMultiAdd']) ? CURLM_BAD_HANDLE : \curl_multi_add_handle($multi, $handle);
+        $GLOBALS['attachmentTransportEvents'][] = ['kind' => 'add', 'pool' => attachmentTransportId($multi),
+            'easy' => attachmentTransportId($handle), 'result' => $result];
+        return $result;
+    }
+
+    function curl_multi_remove_handle($multi, $handle): int
+    {
+        $result = \curl_multi_remove_handle($multi, $handle);
+        $GLOBALS['attachmentTransportEvents'][] = ['kind' => 'remove', 'pool' => attachmentTransportId($multi),
+            'easy' => attachmentTransportId($handle), 'result' => $result];
+        return $result;
+    }
+
+    function curl_multi_close($multi): void
+    {
+        $GLOBALS['attachmentTransportEvents'][] = ['kind' => 'pool_close', 'pool' => attachmentTransportId($multi)];
+        \curl_multi_close($multi);
+    }
+
     /** Route synthetic public storage to loopback AFTER recording real security options. */
     function curl_setopt_array($handle, array $options): bool
     {
@@ -41,6 +91,58 @@ namespace {
         return $GLOBALS['attachmentSyntheticEnv'][$name] ?? $default;
     }
 
+    // PHP's built-in HTTP server closes connections. This second fixture keeps a
+    // real HTTP/1.1 socket alive so tests verify libcurl reuse, not just pool counts.
+    if (PHP_SAPI === 'cli' && ($argv[1] ?? '') === '--keepalive-fixture') {
+        $fixtureRoot = (string) getenv('DIFY_ATTACHMENT_FIXTURE_ROOT');
+        $fixturePort = (int) ($argv[2] ?? 0);
+        $server = stream_socket_server('tcp://127.0.0.1:' . $fixturePort, $errno, $error);
+        if (!is_resource($server)) exit(2);
+        $connection = 0;
+        while (true) {
+            $client = @stream_socket_accept($server, 10);
+            if (!is_resource($client)) continue;
+            ++$connection;
+            stream_set_timeout($client, 3);
+            while (!feof($client)) {
+                $requestLine = fgets($client);
+                if ($requestLine === false || trim($requestLine) === '') break;
+                $headers = [];
+                while (($line = fgets($client)) !== false && trim($line) !== '') {
+                    $split = strpos($line, ':');
+                    if ($split !== false) $headers[strtolower(substr($line, 0, $split))] = trim(substr($line, $split + 1));
+                }
+                $length = (int) ($headers['content-length'] ?? 0);
+                if ($length < 0 || $length > 1024 * 1024 || isset($headers['transfer-encoding'])) break;
+                $received = 0;
+                while ($received < $length && !feof($client)) {
+                    $chunk = fread($client, min(65536, $length - $received));
+                    if ($chunk === false || $chunk === '') break;
+                    $received += strlen($chunk);
+                }
+                if ($received !== $length) break;
+                $upload = str_starts_with($requestLine, 'POST ');
+                file_put_contents($fixtureRoot . '/keepalive-requests.jsonl', json_encode([
+                    'kind' => $upload ? 'upload' : 'download', 'connection' => $connection,
+                    'authorization' => $headers['authorization'] ?? '',
+                ]) . "\n", FILE_APPEND | LOCK_EX);
+                $body = $upload ? json_encode(['id' => bin2hex(random_bytes(16))])
+                    : file_get_contents($fixtureRoot . '/public/uploads/first.png');
+                $response = "HTTP/1.1 " . ($upload ? '201 Created' : '200 OK') . "\r\n"
+                    . 'Content-Type: ' . ($upload ? 'application/json' : 'image/png') . "\r\n"
+                    . 'Content-Length: ' . strlen($body) . "\r\nConnection: keep-alive\r\n\r\n" . $body;
+                $sent = 0;
+                while ($sent < strlen($response)) {
+                    $written = fwrite($client, substr($response, $sent));
+                    if ($written === false || $written === 0) break;
+                    $sent += $written;
+                }
+                if ($sent !== strlen($response)) break;
+            }
+            fclose($client);
+        }
+    }
+
     // This same file is a loopback-only fixture router. It never loads application config.
     if (PHP_SAPI === 'cli-server') {
         $root = (string) getenv('DIFY_ATTACHMENT_FIXTURE_ROOT');
@@ -102,6 +204,35 @@ namespace {
             $message . ' (' . ($result['error_code'] ?? 'no error') . ')');
         attachmentExpect(!str_contains(json_encode($result), 'fixture-private-path-url-key-body'), 'upstream body never escapes');
     }
+    function attachmentTransportClosed(array $events): void
+    {
+        $pools = []; $closedPools = []; $created = []; $closed = []; $added = []; $active = [];
+        foreach ($events as $event) {
+            $kind = $event['kind'];
+            if ($kind === 'pool_init') { $pools[] = $event['pool']; $active[$event['pool']] = []; }
+            if ($kind === 'easy_init') $created[] = $event['easy'];
+            if ($kind === 'easy_close') {
+                foreach ($active as $handles) attachmentExpect(!isset($handles[$event['easy']]), 'easy handle is removed before close');
+                $closed[] = $event['easy'];
+            }
+            if ($kind === 'add' && $event['result'] === CURLM_OK) {
+                attachmentExpect(!isset($added[$event['easy']]), 'every transfer uses a fresh easy handle');
+                $added[$event['easy']] = true;
+                $active[$event['pool']][$event['easy']] = true;
+                attachmentExpect(count($active[$event['pool']]) === 1, 'attachment transfers remain sequential');
+            }
+            if ($kind === 'remove') {
+                attachmentExpect($event['result'] === CURLM_OK && isset($active[$event['pool']][$event['easy']]), 'remove exactly the added easy handle');
+                unset($active[$event['pool']][$event['easy']]);
+            }
+            if ($kind === 'pool_close') {
+                attachmentExpect(($active[$event['pool']] ?? null) === [], 'pool closes after every easy handle is detached');
+                $closedPools[] = $event['pool'];
+            }
+        }
+        attachmentExpect(count($pools) <= 1 && $pools === $closedPools, 'one request-local pool is closed on every terminal path');
+        attachmentExpect($created === $closed, 'all transfer easy handles close on every terminal path');
+    }
     function attachmentRemoveTree(string $path, string $workspace): void
     {
         $resolved = realpath($path); $root = realpath($workspace);
@@ -139,6 +270,10 @@ namespace {
     $attachmentFixturePort = $port;
     $attachmentCurlOptions = [];
     $attachmentNativeDnsCalls = [];
+    $attachmentTransportHandles = new WeakMap();
+    $attachmentTransportSequence = 0;
+    $attachmentTransportEvents = [];
+    $attachmentFailMultiAdd = false;
     $process = proc_open([PHP_BINARY, '-S', '127.0.0.1:' . $port, __FILE__],
         [['pipe', 'r'], ['file', $root . '/fixture-output.log', 'ab'], ['file', $root . '/fixture-error.log', 'ab']],
         $pipes, $root, ['DIFY_ATTACHMENT_FIXTURE_ROOT' => $root,
@@ -173,8 +308,14 @@ namespace {
         $clean = static function () use ($root): void {
             attachmentExpect((glob($root . '/private/uploads/required-*') ?: []) === [], 'temporary files are removed after every terminal outcome');
         };
-        $run = static function (array $input, array $extra = [], ?callable $abort = null, string $user = 'synthetic-user', ?string $target = null, int $timeout = 5) use ($endpoint, $options): array {
-            return DifyRequiredAttachmentUpload::prepare($target ?? $endpoint, 'fixture-only-key', $user, $input, $timeout, $abort, array_replace($options, $extra));
+        $run = static function (array $input, array $extra = [], ?callable $abort = null, string $user = 'synthetic-user', ?string $target = null, int $timeout = 5) use ($endpoint, $options, &$attachmentTransportEvents, $attachmentTransportHandles): array {
+            $eventOffset = count($attachmentTransportEvents);
+            try {
+                return DifyRequiredAttachmentUpload::prepare($target ?? $endpoint, 'fixture-only-key', $user, $input, $timeout, $abort, array_replace($options, $extra));
+            } finally {
+                attachmentTransportClosed(array_slice($attachmentTransportEvents, $eventOffset));
+                attachmentExpect(count($attachmentTransportHandles) === 0, 'request retains no easy handles or pools after prepare returns');
+            }
         };
 
         $clear(); $events = [];
@@ -199,6 +340,66 @@ namespace {
         }
         $clean();
 
+        $keepaliveProcess = null;
+        $originalFixturePort = $attachmentFixturePort;
+        try {
+            $socket = stream_socket_server('tcp://127.0.0.1:0', $errno, $error);
+            attachmentExpect(is_resource($socket), 'allocate keep-alive fixture port');
+            $address = stream_socket_get_name($socket, false);
+            fclose($socket);
+            $keepalivePort = (int) substr($address, strrpos($address, ':') + 1);
+            $keepaliveProcess = proc_open([PHP_BINARY, __FILE__, '--keepalive-fixture', (string) $keepalivePort],
+                [['pipe', 'r'], ['file', $root . '/keepalive-output.log', 'ab'], ['file', $root . '/keepalive-error.log', 'ab']],
+                $keepalivePipes, $root, ['DIFY_ATTACHMENT_FIXTURE_ROOT' => $root,
+                    'SystemRoot' => (string) getenv('SystemRoot'), 'TEMP' => (string) getenv('TEMP'), 'TMP' => (string) getenv('TMP')], ['bypass_shell' => true]);
+            attachmentExpect(is_resource($keepaliveProcess), 'start keep-alive fixture');
+            if (isset($keepalivePipes[0]) && is_resource($keepalivePipes[0])) fclose($keepalivePipes[0]);
+            $ready = false;
+            for ($attempt = 0; $attempt < 150; ++$attempt) {
+                $probe = @fsockopen('127.0.0.1', $keepalivePort, $errno, $error, 0.02);
+                if (is_resource($probe)) { fclose($probe); $ready = true; break; }
+                usleep(20000);
+            }
+            attachmentExpect($ready, 'keep-alive fixture becomes ready');
+            $attachmentFixturePort = $keepalivePort;
+            $keepaliveEndpoint = 'http://127.0.0.1:' . $keepalivePort . '/v1/chat-messages';
+            $eventOffset = count($attachmentTransportEvents);
+            $reused = $run([
+                ['type' => 'image', 'url' => 'https://storage.synthetic.test/first.png'],
+                ['type' => 'image', 'url' => 'https://storage.synthetic.test/second.png'],
+            ], [], null, 'synthetic-user', $keepaliveEndpoint);
+            attachmentExpect(!empty($reused['ok']) && count($reused['files']) === 2, 'keep-alive path retains both complete image uploads');
+            $reuseEvents = array_slice($attachmentTransportEvents, $eventOffset);
+            $reusePools = array_values(array_filter($reuseEvents, static fn (array $event): bool => $event['kind'] === 'pool_init'));
+            attachmentExpect(count($reusePools) === 1
+                && count(array_filter($reuseEvents, static fn (array $event): bool => $event['kind'] === 'easy_init')) === 4,
+                'two sequential downloads and uploads share one pool with four fresh easy handles');
+            $keepaliveRequests = array_map(static fn (string $line): array => json_decode($line, true),
+                file($root . '/keepalive-requests.jsonl', FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES));
+            attachmentExpect(array_column($keepaliveRequests, 'kind') === ['download', 'upload', 'download', 'upload']
+                && count(array_unique(array_column($keepaliveRequests, 'connection'))) === 1,
+                'libcurl really reuses one HTTP connection across all four transfers');
+            attachmentExpect(array_column($keepaliveRequests, 'authorization') === ['', 'Bearer fixture-only-key', '', 'Bearer fixture-only-key'],
+                'fresh download handles never inherit upload authorization on a reused connection');
+
+            $eventOffset = count($attachmentTransportEvents);
+            $nextPreparation = $run(array_slice($files, 0, 2), [], null, 'synthetic-user', $keepaliveEndpoint);
+            attachmentExpect(!empty($nextPreparation['ok']), 'next preparation uploads independently');
+            $nextPools = array_values(array_filter(array_slice($attachmentTransportEvents, $eventOffset),
+                static fn (array $event): bool => $event['kind'] === 'pool_init'));
+            $keepaliveRequests = array_map(static fn (string $line): array => json_decode($line, true),
+                file($root . '/keepalive-requests.jsonl', FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES));
+            attachmentExpect(count($nextPools) === 1 && $nextPools[0]['pool'] !== $reusePools[0]['pool']
+                && count($keepaliveRequests) === 6
+                && $keepaliveRequests[4]['connection'] !== $keepaliveRequests[0]['connection']
+                && $keepaliveRequests[4]['connection'] === $keepaliveRequests[5]['connection'],
+                'even same-user next preparation gets a new pool and connection, then reuses it locally');
+            $clean();
+        } finally {
+            $attachmentFixturePort = $originalFixturePort;
+            if (is_resource($keepaliveProcess)) { proc_terminate($keepaliveProcess); proc_close($keepaliveProcess); }
+        }
+
         $clear();
         $duplicates = $run([$files[0], $files[0]]);
         attachmentExpect(!empty($duplicates['ok']) && count($duplicates['files']) === 2 && count($requests()) === 1
@@ -284,11 +485,14 @@ namespace {
             && array_column($requests(), 'kind') === ['upload', 'upload'], 'local missing-first and missing-middle sources preserve every remaining source without downloads');
         $clean();
         $clear();
+        $eventOffset = count($attachmentTransportEvents);
         $allLocalMissing = $run([$source('https://site.synthetic.test/uploads/missing.png')[0],
             $source('https://site.synthetic.test/uploads/nested/missing.png')[0]]);
         attachmentExpect(!empty($allLocalMissing['ok']) && $allLocalMissing['files'] === []
             && $allLocalMissing['uploaded_indices'] === [] && array_column($allLocalMissing['skipped'], 'index') === [1, 2]
             && $requests() === [], 'all missing local files return a safe empty upload list for text-only chat');
+        attachmentExpect(array_column(array_slice($attachmentTransportEvents, $eventOffset), 'kind') === ['pool_init', 'pool_close'],
+            'all missing local sources close the request pool without creating any easy handle');
         $videoEvents = [];
         $videoMixed = $run([['type' => 'video', 'url' => 'https://untrusted.invalid/record.mp4'],
             $source('https://site.synthetic.test/uploads/missing.png')[0], $files[0]],
@@ -577,11 +781,28 @@ namespace {
         attachmentExpect(microtime(true) - $start < 1.4 && count($requests()) === 2, 'upload does not receive a fresh budget after downloading');
         $clean(); usleep(700000);
         $clear();
+        $eventOffset = count($attachmentTransportEvents);
+        $emptyPreparation = $run([]);
+        attachmentExpect(!empty($emptyPreparation['ok']) && $emptyPreparation['files'] === []
+            && count($attachmentTransportEvents) === $eventOffset, 'empty manifest needs no connection pool or transfer handles');
+        $attachmentFailMultiAdd = true;
+        try {
+            attachmentError($run([$files[0]]), 'REQUIRED_FILES_PREPARE_FAILED', 'failed pool attachment still closes its fresh easy handle and pool');
+            attachmentExpect($requests() === [], 'failed multi add opens no HTTP request');
+        } finally { $attachmentFailMultiAdd = false; }
+        $clean();
+        $clear();
+        attachmentError($run([$files[0]], ['on_upload' => static function (array $event): void {
+            if ($event['status'] === 'uploaded') throw new RuntimeException('fixture-private-path-url-key-body');
+        }]), 'REQUIRED_FILES_PREPARE_FAILED', 'observer failure after upload still closes idle pooled connections and exposes no partial IDs');
+        attachmentExpect(array_column($requests(), 'kind') === ['upload'], 'observer failure does not retry or continue uploads');
+        $clean();
+        $clear();
         attachmentError($run([$files[0]], ['temp_root' => $root . '/public/unsafe-temp']), 'CONFIG_INVALID', 'temporary files cannot be created in public');
         attachmentExpect($requests() === [], 'unsafe temp configuration opens no connection');
         attachmentError($run([['type' => 'image', 'transfer_method' => 'local_file', 'url' => $files[0]['url']]]), 'REQUIRED_FILES_INVALID', 'callers cannot inject upload IDs from other requests');
         $clean();
-        echo "Dify required attachment upload: binary multipart, confirmed-missing skips and original indices, origins/paths/DNS/pinning/redirects/limits, isolation, cancellation/deadline, cleanup: OK\n";
+        echo "Dify required attachment upload: binary multipart, confirmed-missing skips and original indices, origins/paths/DNS/pinning/redirects/limits, request-local connection reuse/isolation, cancellation/deadline, cleanup: OK\n";
     } finally {
         if (is_resource($process)) { proc_terminate($process); proc_close($process); }
         attachmentRemoveTree($root, $root);

server/tests/DifyStreamTimingTest.php

@@ -0,0 +1,89 @@
+<?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";