188 lines
7.7 KiB
Python
188 lines
7.7 KiB
Python
# -*- 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()
|