# -*- coding: utf-8 -*- """E2E live 测试：真实 AgentWorker + 假流式 + 真实 MainWindow 信号链路。 验证 live 时正文 md-segment 是否有内容（用户报告的 bug）。 运行: QT_QPA_PLATFORM=offscreen python tests/diag_live_text.py """ import json import os import sys import time sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") os.environ.setdefault("HAOCODE_RENDER", "software") from PyQt6.QtWidgets import QApplication # noqa: E402 from PyQt6.QtCore import QTimer # noqa: E402 from core.agent.types import (AssistantMessageEvent, AgentMessage, ToolCall) # noqa: E402 from core.agent import stream_fn as sf # noqa: E402 from core.agent.tools import default_tools # noqa: E402 from ui.views.main_window import MainWindow # noqa: E402 app = QApplication(sys.argv) window = None results = [] CALL_STATE = {"n": 0} def check(name, ok, detail=""): print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" [{detail}]" if detail and not ok else "")) results.append(ok) # ---- 假 stream_fn：第 1 轮 思考+文本+bash 工具调用；第 2 轮 最终回答 ---- def fake_stream(context, model, signal, max_tokens, tools=None): CALL_STATE["n"] += 1 n = CALL_STATE["n"] if n == 1: yield ("event", AssistantMessageEvent(type="thinking_delta", text="我先看看")) yield ("event", AssistantMessageEvent(type="thinking_delta", text="目录。")) yield ("event", AssistantMessageEvent(type="text_delta", text="我来执行命令")) yield ("event", AssistantMessageEvent(type="toolcall_delta", tool_call_index=0, tool_call_field="id", tool_call_delta="call-e2e-1")) yield ("event", AssistantMessageEvent(type="toolcall_delta", tool_call_index=0, tool_call_field="name", tool_call_delta="bash")) yield ("event", AssistantMessageEvent(type="toolcall_delta", tool_call_index=0, tool_call_field="arguments", tool_call_delta='{"command": "echo e2e-ok"}')) time.sleep(0.3) yield ("final", AgentMessage( role="assistant", stop_reason="stop", tool_calls=[ToolCall(id="call-e2e-1", name="bash", raw_arguments='{"command": "echo e2e-ok"}')])) else: for tok in ["最终", "回答", "：任务", "完成。"]: yield ("event", AssistantMessageEvent(type="text_delta", text=tok)) time.sleep(0.05) yield ("final", AgentMessage(role="assistant", stop_reason="stop")) def step1(): global window window = MainWindow() QTimer.singleShot(3500, step2) def step2(): from core.llm_engine import AgentWorker db = window.db sess = db.create_session("E2E 正文测试") sid = sess["id"] mid = "msg-e2e-text" user_row = db.add_message(sid, "user", "测试问题", None) window.current_session_id = sid window.chat_bridge.create_message(mid, "assistant", "", "E2E") window._active_streams[sid] = { "msg_id": mid, "content": "", "reasoning": "", "timeline": [], "tl_kind": None, "parent_id": user_row["id"], "branch_info": None, "worker": None, "previous_leaf_id": user_row["id"], } # 真实 API + 流事件日志（定位 text_delta 是否到达） import core.llm_engine as le real_stream = le.openai_stream def logging_stream(context, model, signal, max_tokens, tools=None): n_ev = 0 for kind, payload in real_stream(context, model, signal, max_tokens, tools): t = getattr(payload, "type", kind) txt = str(getattr(payload, "text", ""))[:30] if t not in ("text_delta", "thinking_delta", "toolcall_delta"): print(f" [stream] {kind}/{t}", flush=True) else: n_ev += 1 if n_ev <= 6 or n_ev % 20 == 0: print(f" [stream] {t} {txt!r}", flush=True) yield kind, payload le.openai_stream = logging_stream worker = AgentWorker(provider_name=window.current_provider, model_name=window.current_model, openai_messages=[{"role": "user", "content": "运行命令 echo hello-from-diag 并告诉我输出"}], enable_tools=True) # 与 send_message 相同的信号连接 worker.reasoning_received.connect(lambda t: window.on_reasoning_received(sid, t)) worker.chunk_received.connect(lambda t: window.on_chunk_received(sid, t)) worker.tool_execution_started.connect( lambda cid, name, args: window._on_tool_started(sid, cid, name, args)) worker.tool_execution_updated.connect( lambda cid, text: window._on_tool_updated(sid, cid, text)) worker.tool_execution_finished.connect( lambda cid, name, ok, text: window._on_tool_finished(sid, cid, name, ok, text)) worker.error_occurred.connect(lambda err: (print(' [worker error]', err), window.on_error(sid, err))) worker.finished.connect(lambda: window.on_reply_finished(sid)) window._active_streams[sid]["worker"] = worker worker.start() print("worker started, 等待流结束...") QTimer.singleShot(40000, step3) def step3(): # live DOM 检查（不重载 DB） js = """ (function() { try { var w = document.getElementById('msg-e2e-text'); if (!w) return JSON.stringify({error: 'no wrapper'}); var tl = w.querySelector('.reply-content'); var blocks = Array.prototype.map.call(tl.children, function(el) { return el.className.split(' ')[0]; }); var segs = w.querySelectorAll('.md-segment'); var segTexts = Array.prototype.map.call(segs, function(x) { return x.textContent; }); var thinks = w.querySelectorAll('.think-content'); var thinkTexts = Array.prototype.map.call(thinks, function(x) { return x.textContent; }); var chip = w.querySelector('.tool-chip'); return JSON.stringify({blocks: blocks, segTexts: segTexts, thinkTexts: thinkTexts, chip: !!chip, streaming: w.classList.contains('streaming')}); } catch (e) { return JSON.stringify({error: String(e)}); } })() """ def got(res): d = json.loads(str(res)) if "error" in d: check("E2E DOM", False, d["error"]) finish() return print(f" blocks = {d['blocks']}") print(f" segTexts = {d['segTexts']}") print(f" thinkTexts = {d['thinkTexts']}") print(f" chip = {d['chip']} streaming={d['streaming']}") check("live 正文段有内容（至少一段非空）", any((t or "").strip() for t in d["segTexts"]), str(d["segTexts"])) check("live 最终回答含命令输出", any("hello-from-diag" in (t or "") for t in d["segTexts"]), str(d["segTexts"])) check("live 思考段有内容（或模型未思考）", True, str(d["thinkTexts"])[:80]) check("live 工具 chip 存在", d["chip"]) check("streaming 已收尾", not d["streaming"]) # 时间线入库检查 st = window._active_streams.get(sid := window.current_session_id) tl = window.db.get_session(window.current_session_id) if False else None finish() window.browser.page().runJavaScript(js, got) def finish(): try: window.db.delete_session(window.current_session_id) except Exception: pass print(f"\n===== {'ALL PASS' if all(results) else 'HAS FAILURES'}: " f"{sum(results)}/{len(results)} =====") app.quit() QTimer.singleShot(500, step1) app.exec()