# -*- coding: utf-8 -*- """E2E onscreen：真实可见窗口 + 真实 API + 流式过程中多点采样 DOM。 复现用户环境（非 offscreen，rAF 行为与真实窗口一致）。 运行: python tests/diag_live_onscreen.py （会在桌面弹出窗口） """ import json import os import sys import time sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) os.environ.pop("QT_QPA_PLATFORM", None) # onscreen 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 # noqa: E402 from ui.views.main_window import MainWindow # noqa: E402 app = QApplication(sys.argv) window = None results = [] SNAP_BUF_JS = ( "(function() {" " var w = document.getElementById('msg-e2e-text');" " if (!w) return 'no-wrapper';" " var segs = w.querySelectorAll('.md-segment');" " var out = [];" " for (var i = 0; i < segs.length; i++) {" " out.push('buf:' + ((segs[i].__buf || '').length)" " + '/dom:' + ((segs[i].textContent || '').length));" " }" " return out.join(' ') || 'no-segs';" "})()" ) SNAP_JS = ( "(function() {" " var w = document.getElementById('msg-e2e-text');" " if (!w) return 'no-wrapper';" " var segs = w.querySelectorAll('.md-segment');" " var n = 0;" " for (var i = 0; i < segs.length; i++)" " if ((segs[i].textContent || '').trim()) n++;" " return 'segs=' + segs.length + ' nonempty=' + n +" " ' chip=' + !!w.querySelector('.tool-chip') +" " ' streaming=' + w.classList.contains('streaming');" "})()" ) FINAL_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 || '').slice(0, 80); }); var chip = w.querySelector('.tool-chip'); return JSON.stringify({blocks: blocks, segTexts: segTexts, chip: !!chip, streaming: w.classList.contains('streaming')}); } catch (e) { return JSON.stringify({error: String(e)}); } })() """ def check(name, ok, detail=""): print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" [{detail}]" if detail and not ok else "")) results.append(ok) def step1(): global window window = MainWindow() window.show() QTimer.singleShot(3500, step2) def step2(): from core.llm_engine import AgentWorker import core.llm_engine as le db = window.db sess = db.create_session("E2E onscreen 测试") 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"], } real_stream = le.openai_stream def logging_stream(context, model, signal, max_tokens, tools=None): for kind, payload in real_stream(context, model, signal, max_tokens, tools): t = getattr(payload, "type", kind) if t in ("text_delta", "thinking_delta", "toolcall_delta"): txt = str(getattr(payload, "text", ""))[:25] print(f" [stream] {t} {txt!r}", flush=True) else: print(f" [stream] {kind}/{t}", 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-onscreen " "并告诉我输出内容"}], enable_tools=True) 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, flush=True), 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（窗口已可见，观察屏幕）...", flush=True) def snapshot(tag): window.browser.page().runJavaScript( SNAP_JS, lambda v, tag=tag: print(f" [snapshot {tag}] {v}", flush=True)) window.browser.page().runJavaScript( SNAP_BUF_JS, lambda v, tag=tag: print(f" [snapshot {tag} BUF] {v}", flush=True)) QTimer.singleShot(8000, lambda: snapshot("t+8s")) QTimer.singleShot(15000, lambda: snapshot("t+15s")) QTimer.singleShot(25000, lambda: snapshot("t+25s")) QTimer.singleShot(45000, step3) def step3(): 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" chip={d['chip']} streaming={d['streaming']}") check("live 正文段非空", any(t.strip() for t in d["segTexts"]), str(d["segTexts"])) check("最终回答含命令输出", any("hello-from-onscreen" in t for t in d["segTexts"]), str(d["segTexts"])) check("工具 chip 存在", d["chip"]) check("streaming 已收尾", not d["streaming"]) finish() window.browser.page().runJavaScript(FINAL_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()