diff --git a/app/src/doctor_workstation/ui/dialogs/ai_consult.py b/app/src/doctor_workstation/ui/dialogs/ai_consult.py index 1718e5d0c..0d14109ad 100755 --- a/app/src/doctor_workstation/ui/dialogs/ai_consult.py +++ b/app/src/doctor_workstation/ui/dialogs/ai_consult.py @@ -2956,6 +2956,8 @@ def _queue_report_render(render: Callable[[], None]) -> None: class _ChatBubble(QWidget): + transfer_receipt_toggled = Signal() + def __init__( self, *, @@ -3054,19 +3056,65 @@ class _ChatBubble(QWidget): self._provenance = dict(value) if isinstance(value, Mapping) else {} def set_transfer_status(self, text: str) -> None: - """Keep actual input/submission status visible while answer text streams.""" + """Keep one collapsed receipt above every form of the AI answer.""" label = getattr(self, "transfer_status", None) if label is None: - label = QLabel(self._bubble_frame) + receipt = QWidget(self._bubble_frame) + receipt.setObjectName("AiConsultTransferReceipt") + layout = QVBoxLayout(receipt) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(7) + toggle = QPushButton("资料提交回执 · 展开详情", receipt) + toggle.setObjectName("AiConsultTransferToggle") + toggle.setCheckable(True) + toggle.setAutoDefault(False) + toggle.setIcon(icons.icon("chevron_right", "#65718A", 12)) + toggle.setIconSize(QSize(12, 12)) + toggle.setCursor(Qt.CursorShape.PointingHandCursor) + toggle.setAccessibleName("展开资料提交回执") + toggle.setStyleSheet( + "QPushButton {color:#65718A;font-size:12px;text-align:left;" + "background:transparent;border:0;padding:4px 0;}" + "QPushButton:hover {color:#4F63D9;}" + "QPushButton:focus {border-bottom:1px solid #4F63D9;}" + ) + layout.addWidget(toggle) + label = QLabel(receipt) label.setObjectName("AiConsultTransferStatus") label.setTextFormat(Qt.TextFormat.PlainText) label.setWordWrap(True) + label.setTextInteractionFlags( + Qt.TextInteractionFlag.TextSelectableByMouse + | Qt.TextInteractionFlag.TextSelectableByKeyboard + ) label.setStyleSheet("color:#65718A;font-size:12px;") - self._inner.insertWidget(1, label) + label.hide() + layout.addWidget(label) + self._inner.insertWidget(0, receipt) + self.transfer_receipt = receipt + self.transfer_toggle = toggle self.transfer_status = label - label.setText(text) - label.setVisible(bool(text)) + toggle.toggled.connect(self._toggle_transfer_status) + # Progress updates change only the receipt, never reparse the answer or + # reset the reader's expanded/collapsed choice. + if label.text() != text: + label.setText(text) + self.transfer_toggle.setToolTip(text.split("\n", 1)[0]) + self.transfer_receipt.setVisible(bool(text)) + + def _toggle_transfer_status(self, expanded: bool) -> None: + # Pause chat following before the layout grows, so opening a long + # receipt keeps the reader at the summary they clicked. + self.transfer_receipt_toggled.emit() + self.transfer_status.setVisible(expanded) + self.transfer_toggle.setText( + "资料提交回执 · 收起详情" if expanded else "资料提交回执 · 展开详情" + ) + self.transfer_toggle.setIcon(icons.icon("chevron_down" if expanded else "chevron_right", "#65718A", 12)) + self.transfer_toggle.setAccessibleName( + "收起资料提交回执" if expanded else "展开资料提交回执" + ) def set_cancel_status(self, text: str, *, has_text: bool) -> None: """Update cancellation without reparsing the received clinical text.""" @@ -4544,6 +4592,7 @@ class AiConsultDialog(QDialog): extra: QWidget | None = None, ) -> _ChatBubble: bubble = _ChatBubble(role=role, text=text, time_text=time_text, extra=extra) + bubble.transfer_receipt_toggled.connect(self._pause_chat_follow) self.chat_layout.insertWidget(self.chat_layout.count() - 1, bubble) bubble.setVisible(self._chat_item_matches_filter(bubble)) if self._follow_chat: @@ -4576,6 +4625,10 @@ class AiConsultDialog(QDialog): bar = self.chat_scroll.verticalScrollBar() self._follow_chat = value >= bar.maximum() - 4 + def _pause_chat_follow(self) -> None: + self._follow_chat = False + self._chat_scroll_timer.stop() + def _chat_scroll_range_changed(self, _minimum: int, _maximum: int) -> None: if self._follow_chat: self._request_chat_scroll() diff --git a/app/tests/test_ai_consult_transfer_receipt_ui.py b/app/tests/test_ai_consult_transfer_receipt_ui.py new file mode 100644 index 000000000..6d4653e32 --- /dev/null +++ b/app/tests/test_ai_consult_transfer_receipt_ui.py @@ -0,0 +1,230 @@ +"""Transfer receipts stay collapsed above the answer throughout its lifecycle.""" +from __future__ import annotations + +import os +from datetime import datetime +from types import SimpleNamespace + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +import pytest +from PySide6.QtCore import QEvent, Qt, QTimer +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication, QLabel + +from doctor_workstation.core import PermissionSet +from doctor_workstation.ui.dialogs import ai_consult as module +from doctor_workstation.ui.dialogs.ai_consult_charts import GlucoseChartData, GlucoseReading + +REPORT = """以下为合成测试资料的核对清单,所有结论需要结合原始记录及记录日期进一步确认。 + +1. 记录核对 +- 来源范围:保留现有资料,逐项确认填写情况与来源,不补写缺失的内容。 + +2. 后续核对 +- 记录日期:继续核对已有记录的完整性,保留最终唯一标记。 +""" + + +@pytest.fixture(scope="module") +def application(): + return QApplication.instance() or QApplication([]) + + +@pytest.fixture +def dialog(application, monkeypatch): + monkeypatch.setattr(module, "QThreadPool", SimpleNamespace( + globalInstance=lambda: SimpleNamespace(start=lambda _: None), + )) + instance = module.AiConsultDialog(object(), PermissionSet(["tcm.diagnosis/aiAssistant"])) + monkeypatch.setattr(instance, "_load_workspace", lambda _: None) + instance.open_for(diagnosis_id=501, patient_id=1501) + instance.resize(1280, 820) + instance.show() + yield instance + for timer in instance.findChildren(QTimer): + timer.stop() + instance.close() + instance.deleteLater() + application.sendPostedEvents(None, QEvent.Type.DeferredDelete) + application.processEvents() + + +def _manifest(): + statuses = ["submitted", "prepared", "submitting", "failed", "omitted"] + return { + "source_summary": {"diagnosis_count": 2, "blood_record_count": 3}, + "required_sections": [ + {"key": key, "status": statuses[index % len(statuses)], + "availability": "empty" if index == 1 else "present"} + for index, (key, _) in enumerate(module._REQUIRED_TRANSFER_SECTIONS) + ], + "items": [ + {"kind": "image", "label": "图片附件1", "status": "submitted"}, + {"kind": "document", "label": "文档附件2", "status": "skipped", "reason": "not_found"}, + {"kind": "audio", "label": "音频附件3", "status": "failed"}, + {"kind": "custom", "label": "附件4", "status": "omitted"}, + ], + "attachment_count": 4, + "submitted_attachment_count": 1, + "attachment_transfer": "uploaded_files", + "source_form": "evidence_summary", + "evidence_cache_hit": True, + } + + +@pytest.mark.parametrize("report", [False, True]) +def test_receipt_is_collapsed_above_report_chart_and_body(application, report): + bubble = module._ChatBubble(role="ai", text=REPORT if report else "核对合成记录。") + description = module._transfer_description(_manifest(), "本次回复已完成。") + bubble.set_transfer_status(description) + chart = GlucoseChartData(readings=(GlucoseReading(datetime(2026, 10, 20), fasting=6.2),)) + assert bubble.finalize_clinical_analysis(chart_data=chart) is report + bubble.resize(780, 700) + bubble.show() + application.processEvents() + assert bubble._inner.itemAt(0).widget() is bubble.transfer_receipt + assert bubble.transfer_receipt.isVisible() + assert bubble.transfer_status.isHidden() + assert not bubble.transfer_toggle.isChecked() + assert "展开详情" in bubble.transfer_toggle.text() + assert bubble.transfer_status.text() == description + assert len(bubble.findChildren(QLabel, "AiConsultTransferStatus")) == 1 + QTest.mouseClick(bubble.transfer_toggle, Qt.MouseButton.LeftButton) + assert bubble.transfer_status.isVisible() and bubble.transfer_toggle.isChecked() + assert "收起详情" in bubble.transfer_toggle.text() + assert bubble.transfer_status.textInteractionFlags() & Qt.TextInteractionFlag.TextSelectableByMouse + assert all(f"{label}:" in bubble.transfer_status.text() + for _, label in module._REQUIRED_TRANSFER_SECTIONS) + assert "无(已整理,待提交)" in bubble.transfer_status.text() + assert "血糖血压记录 3 条" in bubble.transfer_status.text() + assert "文件不存在,已跳过" in bubble.transfer_status.text() + assert "音频附件3(音频):提交失败" in bubble.transfer_status.text() + assert "附件4(附件):未提交" in bubble.transfer_status.text() + assert "本请求附件通道:1 项" in bubble.transfer_status.text() + assert "资料栏目提交清单" not in bubble.body.toPlainText() + QTest.mouseClick(bubble.transfer_toggle, Qt.MouseButton.LeftButton) + assert bubble.transfer_status.isHidden() + bubble.close() + bubble.deleteLater() + + +def test_receipt_progress_keeps_expansion_and_does_not_reparse_body(application, monkeypatch): + bubble = module._ChatBubble(role="ai", text="合成正文保持完整。") + first = module._transfer_description(_manifest(), "正在提交资料…") + bubble.set_transfer_status(first) + bubble.transfer_toggle.click() + bubble.transfer_status.setSelection(0, 4) + selected = bubble.transfer_status.selectedText() + monkeypatch.setattr(bubble.body, "set_payload", lambda *_: pytest.fail("receipt reparsed answer")) + for _ in range(20): + bubble.set_transfer_status(first) + assert bubble.transfer_status.selectedText() == selected + next_text = module._transfer_description(_manifest(), "已收到 AI 正文,正在生成…") + bubble.set_transfer_status(next_text) + assert bubble.transfer_toggle.isChecked() and not bubble.transfer_status.isHidden() + bubble.transfer_toggle.click() + bubble.set_transfer_status(module._transfer_description(_manifest(), "本次回复已完成。")) + assert not bubble.transfer_toggle.isChecked() and bubble.transfer_status.isHidden() + assert bubble.body.toPlainText() == "合成正文保持完整。" + bubble.deleteLater() + + +@pytest.mark.parametrize("expanded", [False, True]) +@pytest.mark.parametrize("ending", ["done", "error", "cancelled"]) +def test_stream_terminal_preserves_receipt_choice_and_skipped_files(dialog, expanded, ending): + dialog._ask("核对合成记录") + generation, stream = dialog._generation, dialog._stream_generation + bubble = dialog._stream_bubble + dialog._stream_event(generation, stream, {"event": "start", **_manifest()}) + assert bubble.transfer_status.isHidden() + if expanded: + bubble.transfer_toggle.click() + dialog._stream_event(generation, stream, {"event": "progress", "phase": "upstream", **_manifest()}) + for chunk in ("首段合成正文。\n\n", "最后唯一标记"): + dialog._stream_event(generation, stream, {"event": "delta", "text": chunk}) + dialog._flush_stream_chunks() + if ending == "done": + # A legacy terminal bulk receipt must not relabel the missing file. + stale = _manifest() + stale["items"] = [{**item, "status": "submitted"} for item in stale["items"]] + dialog._stream_event(generation, stream, {"event": "done", **stale}) + elif ending == "error": + dialog._stream_failed(generation, stream, RuntimeError("合成连接中断")) + else: + dialog._cancel_stream(reason="合成测试停止") + assert bubble.transfer_toggle.isChecked() is expanded + assert bubble.transfer_status.isHidden() is not expanded + assert bubble._inner.itemAt(0).widget() is bubble.transfer_receipt + assert "文件不存在,已跳过" in bubble.transfer_status.text() + assert "最后唯一标记" in bubble.body.toPlainText() + assert "资料栏目提交清单" not in bubble._raw_payload + assert "资料栏目提交清单" not in bubble.body.toPlainText() + assert len(bubble.findChildren(QLabel, "AiConsultTransferStatus")) == 1 + + +def test_empty_cancel_and_error_keep_collapsed_receipt(dialog): + for ending in ("error", "cancelled"): + dialog._ask("核对合成记录") + generation, stream = dialog._generation, dialog._stream_generation + bubble = dialog._stream_bubble + dialog._stream_event(generation, stream, {"event": "start", **_manifest()}) + if ending == "error": + dialog._stream_failed(generation, stream, RuntimeError("合成错误")) + else: + dialog._cancel_stream(reason="合成测试停止") + assert bubble.body.isHidden() + assert bubble.transfer_status.isHidden() + assert not bubble.transfer_toggle.isChecked() + bubble.transfer_toggle.click() + assert not bubble.transfer_status.isHidden() + assert "文件不存在,已跳过" in bubble.transfer_status.text() + + +@pytest.mark.parametrize("expanded", [False, True]) +def test_deferred_report_keeps_receipt_at_top_and_preserves_choice(dialog, expanded): + dialog._ask("核对合成记录") + generation, stream = dialog._generation, dialog._stream_generation + bubble = dialog._stream_bubble + dialog._stream_event(generation, stream, {"event": "start", **_manifest()}) + if expanded: + bubble.transfer_toggle.click() + text = REPORT + "\n\n" + "合成补充记录,核对资料完整性。" * 400 + dialog._stream_event(generation, stream, {"event": "delta", "text": text}) + dialog._stream_event(generation, stream, {"event": "done"}) + assert bubble._report_render_pending + for _ in range(200): + if not bubble._report_render_pending: + break + QTest.qWait(5) + assert not bubble._report_render_pending and bubble._clinical_panel is not None + assert bubble._inner.itemAt(0).widget() is bubble.transfer_receipt + assert bubble.transfer_toggle.isChecked() is expanded + assert bubble.transfer_status.isHidden() is not expanded + assert bubble._raw_payload == text + + +def test_opening_receipt_keeps_scroll_position_while_stream_continues(dialog, application): + text = "\n\n".join(["合成记录" * 30] * 35) + bubble = dialog._append_bubble("ai", "") + bubble.append_stream_text(text, text) + bubble.set_transfer_status(module._transfer_description(_manifest(), "正在生成…")) + for _ in range(6): + application.processEvents() + QTest.qWait(5) + bar = dialog.chat_scroll.verticalScrollBar() + bar.setValue(bar.maximum()) + before = bar.value() + assert before > 0 and dialog._follow_chat + bubble.transfer_toggle.click() + for _ in range(6): + application.processEvents() + QTest.qWait(5) + assert bar.value() == before + assert not dialog._follow_chat + bubble.append_stream_text("新增合成末行", str(bubble._raw_payload) + "新增合成末行") + bubble.set_transfer_status(module._transfer_description(_manifest(), "本次回复已完成。")) + for _ in range(6): + application.processEvents() + QTest.qWait(5) + assert bar.value() == before and not dialog._follow_chat diff --git a/server/app/adminapi/controller/tcm/DiagnosisController.php b/server/app/adminapi/controller/tcm/DiagnosisController.php index e232f4f9d..d8cc3d474 100755 --- a/server/app/adminapi/controller/tcm/DiagnosisController.php +++ b/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; }, diff --git a/server/app/adminapi/logic/tcm/DiagnosisAiLogic.php b/server/app/adminapi/logic/tcm/DiagnosisAiLogic.php index 183b8f8d4..de704e2c0 100755 --- a/server/app/adminapi/logic/tcm/DiagnosisAiLogic.php +++ b/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) { diff --git a/server/app/adminapi/service/AssistantSseTiming.php b/server/app/adminapi/service/AssistantSseTiming.php index 1de83df9a..07adc8a99 100755 --- a/server/app/adminapi/service/AssistantSseTiming.php +++ b/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) {} + } } diff --git a/server/app/common/service/DifyChatService.php b/server/app/common/service/DifyChatService.php index 5dc5088a2..c345022d3 100755 --- a/server/app/common/service/DifyChatService.php +++ b/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; } diff --git a/server/app/common/service/DifyRequiredAttachmentUpload.php b/server/app/common/service/DifyRequiredAttachmentUpload.php index ce575eac1..9af6eb436 100644 --- a/server/app/common/service/DifyRequiredAttachmentUpload.php +++ b/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); } } diff --git a/server/app/common/service/DifyStreamTiming.php b/server/app/common/service/DifyStreamTiming.php new file mode 100644 index 000000000..cb4058650 --- /dev/null +++ b/server/app/common/service/DifyStreamTiming.php @@ -0,0 +1,107 @@ +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]; + } +} diff --git a/server/tests/DiagnosisAiAssistantTimingTest.php b/server/tests/DiagnosisAiAssistantTimingTest.php index 30bf9f1ae..2ffa09601 100755 --- a/server/tests/DiagnosisAiAssistantTimingTest.php +++ b/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"; diff --git a/server/tests/DifyRequiredAttachmentUploadTest.php b/server/tests/DifyRequiredAttachmentUploadTest.php index 671c3ebfa..745e3752e 100644 --- a/server/tests/DifyRequiredAttachmentUploadTest.php +++ b/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); diff --git a/server/tests/DifyStreamTimingTest.php b/server/tests/DifyStreamTimingTest.php new file mode 100644 index 000000000..e50e7b7de --- /dev/null +++ b/server/tests/DifyStreamTimingTest.php @@ -0,0 +1,89 @@ +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";