# -*- coding: utf-8 -*- """P1-01 渲染窗口 400 条消息规模诊断（offscreen；临时 DB + 临时配置；不启动真实 LLM） 运行: QT_QPA_PLATFORM=offscreen python tests/diag_render_scale.py [N] N 默认 400（位置参数可覆盖） 验证项（manual 模式保证确定性；末尾追加 auto 模式抽查）： 1. 初始窗口：.message-wrapper == min(N, size) 且 ≤ size（默认 40）； 窗口 = 最新 size 条；load-older 可见、load-newer 隐藏； 2. 向上分页至头部：每页锚点误差 ≤ 2px（绝对顶部例外：scrollTop==0 且露出新页）； 全程窗口 ≤ size；到达头部后 hasMoreOlder=false； 全部 N 条消息可达（各步渲染 id 并集 == 链 id 集），无重复 id； 3. 自头部向下回翻 2 页：锚点稳定（≤2px）； 4. 规模报告：DOM 节点总数、页面高度、每页耗时采样（ms）； 5. auto 模式抽查：切 auto + 滚到顶部 → 自动补页发生。 调试铁律：本脚本自带总超时（QTimer 300s），bash 侧以 timeout=360 运行。 """ 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") os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = "--disable-gpu" # 铁律：临时 DB + 临时配置（含 render_window_mode/size），先于任何 UI import from tests._test_env import isolate, default_config # noqa: E402 N = int(sys.argv[1]) if len(sys.argv) > 1 else 400 _SIZE = 40 _PAGE = _SIZE // 2 # 页大小 = 半窗（与 main_window._rw_page_size 一致） _cfg = dict(default_config()) _cfg["render_window_mode"] = "manual" # 确定性诊断；auto 行为末尾单独抽查 _cfg["render_window_size"] = _SIZE _env = isolate("render_scale", config=_cfg) import core.db_manager as _dbm # noqa: E402 from core.db_manager import DBManager # noqa: E402 # ---------- 造数：N 条链（含附件用户消息 / assistant 时间线 / 一个分支兄弟） ---------- db = DBManager() sess = db.create_session(title=f"P1-01 scale {N}") sid = sess["id"] parent = None chain_ids = [] for i in range(N): role = "user" if i % 2 == 0 else "assistant" content = f"消息 {i} —— " + "这是一段用于撑起 DOM 高度的填充段落。" * 6 atts = None if role == "user" and i % 50 == 0: atts = json.dumps({"user_text": content, "attachments": [ {"type": "pdf", "mode": "text", "name": f"spec_{i}.pdf", "size_kb": 12, "pages": 3, "lines": 40, "content": "文本附件内容"} ]}, ensure_ascii=False) timeline = None if role == "assistant" and i % 75 == 3: timeline = json.dumps([ {"t": "think", "text": f"思考片段 {i}"}, {"t": "text", "text": f"时间线正文 {i}"}, ], ensure_ascii=False) msg_id = f"scale-{i}" db.add_message(session_id=sid, role=role, content=content, parent_id=parent, msg_id=msg_id, attachment_metadata=atts, timeline=timeline) parent = msg_id chain_ids.append(msg_id) # 分支兄弟：必须插在链中间（add_message 会把新消息设为叶子， # 若放在主链之后会抢走叶子、截断可见链） if i == 298 and N > 300: db.add_message(session_id=sid, role="assistant", content="分支兄弟回复", parent_id=parent, msg_id="scale-branch-sib") db.mark_session_has_messages(sid) from PyQt6.QtWidgets import QApplication # noqa: E402 from PyQt6.QtCore import QTimer # noqa: E402 from ui.views.main_window import MainWindow # noqa: E402 app = QApplication(sys.argv) results = [] def check(name, fn): try: fn() print(f" PASS {name}") results.append(True) except Exception as e: import traceback traceback.print_exc() print(f" FAIL {name}: {e}") results.append(False) window = None js = {"done": False} def _run_js(code, timeout_s=15): result = {"val": None, "done": False} def on_ret(val): if isinstance(val, str): try: val = json.loads(val) except Exception: pass result["val"] = val result["done"] = True if hasattr(window.browser, "execute_js_async"): window.browser.execute_js_async(code, on_ret) else: window.browser.page().runJavaScript(code, on_ret) t0 = time.time() while not result["done"] and time.time() - t0 < timeout_s: app.processEvents() time.sleep(0.02) assert result["done"], f"JS 执行超时: {code[:60]}" return result["val"] def wait_until(cond_js, timeout_s=30, desc=""): t0 = time.time() while time.time() - t0 < timeout_s: if _run_js(cond_js, timeout_s=5): return True time.sleep(0.1) raise AssertionError(f"等待超时: {desc}") page_metrics = {"pages": [], "heights": [], "dom_nodes": []} def try_load(): global window window = MainWindow() # offscreen 零视口（innerHeight=0）→ 给真实尺寸，滚动/锚点几何才有效 window.resize(1400, 950) window.show() for _ in range(10): app.processEvents() time.sleep(0.05) # 等真实视口 wait_until("window.innerHeight > 0 && window.innerWidth > 0", 30, "视口尺寸") # 等 JS 就绪 + 首轮（可能零视口）渲染完成 wait_until("window.jsReady === true", 30, "jsReady") wait_until(f"document.querySelectorAll('#chat-container .message-wrapper').length >= {_SIZE}", 60, "首轮渲染") # 真实视口下走完整 load_messages_to_web 路径重新窗口化加载 window.load_messages_to_web(window.current_session_id, show_loading=False) wait_until(f"rwState !== null && rwState.generation >= 2 && " f"document.querySelectorAll('#chat-container .message-wrapper').length >= {_SIZE}", 60, "视口重载") wait_until("(window.__rwPageRendering === false && document.getElementById('load-older') !== null) ? 1 : 0", 10, "初始窗口收尾") run_checks() def run_checks(): sel = "#chat-container .message-wrapper" # ---- 1) 初始窗口 ---- def initial(): cnt = _run_js(f"document.querySelectorAll('{sel}').length") assert cnt == _SIZE, f"初始窗口 {cnt} != {_SIZE}" st = _run_js("JSON.stringify({len: rwState.order.length, hidO: rwState.hiddenOlder, " "hidN: rwState.hiddenNewer, older: rwState.hasMoreOlder, newer: rwState.hasMoreNewer, " "mode: rwState.mode, size: rwState.size})") assert st["mode"] == "manual" and st["size"] == _SIZE, st assert st["len"] == _SIZE and st["hidO"] == N - _SIZE, st assert st["newer"] is False and st["older"] is True, st assert _run_js("document.getElementById('load-older').hidden") is False assert _run_js("document.getElementById('load-newer').hidden") is True # 窗口 = 最新 size 条 first = _run_js(f"document.querySelector('{sel}').id") assert first == chain_ids[N - _SIZE], first # 初始贴底 assert _run_js("isNearBottom()"), "初始窗口未对齐底部" check(f"初始窗口 = 最新 {_SIZE} 条（链长 {N}）", initial) # ---- 2a) 中部锚点保持（真实"窗口中部换页"路径，误差 ≤2px） ---- seen = set(chain_ids[-_SIZE:]) anchor_errors = [] def page_older_once(): """执行一次向上换页（调用方保证视口位置）；断言窗口上限与锚点。""" before = _run_js("JSON.stringify(rwCaptureAnchor() || {})") before = json.loads(before) if isinstance(before, str) else (before or {}) scroll_before = _run_js("window.scrollY") t0 = time.time() _run_js("rwRequestPage('older')") # 等待 pending 消费 + 批次渲染收尾 wait_until("rwState.pending === null && window.__rwPageRendering === false", 30, "换页完成") dt = (time.time() - t0) * 1000 page_metrics["pages"].append(dt) cnt = _run_js(f"document.querySelectorAll('{sel}').length") assert cnt <= _SIZE, f"窗口超限 {cnt}" if before.get("msgId"): el = _run_js(f"document.getElementById({json.dumps(before['msgId'])}) ? " f"document.getElementById({json.dumps(before['msgId'])}).getBoundingClientRect().top + window.scrollY : null") assert el is not None, f"锚点消息被裁剪: {before['msgId']}" if scroll_before <= 1: # 绝对顶部：停在 0 露出新页（需求定义的例外：不做锚点断言） assert _run_js("window.scrollY") <= 1, "绝对顶部未保持" anchor_errors.append(0.0) else: expect = scroll_before + (el - before["docTop"]) actual = _run_js("window.scrollY") err = abs(expect - actual) anchor_errors.append(err) assert err <= 2.0, f"锚点误差 {err}px (expect={expect} actual={actual})" else: anchor_errors.append(0.0) ids_now = _run_js(f"Array.from(document.querySelectorAll('{sel}')).map(e => e.id)") for i in ids_now: seen.add(i) page_metrics["dom_nodes"].append(_run_js("document.querySelectorAll('body *').length")) page_metrics["heights"].append(_run_js("document.body.scrollHeight")) def anchor_middle(): # 滚到窗口中部（35% 处）：既非绝对顶部也非贴底 → 走完整锚点数学 _run_js("window.scrollTo(0, Math.max(1, document.body.scrollHeight * 0.15)); 1") time.sleep(0.2) app.processEvents() scroll_before = _run_js("window.scrollY") assert scroll_before > 10, f"未到中部: {scroll_before}" assert not _run_js("isNearBottom()"), "中部位置不应贴底（防振荡前置条件）" page_older_once() st = _run_js("JSON.stringify({hidO: rwState.hiddenOlder, hidN: rwState.hiddenNewer})") assert st["hidN"] > 0, f"向上换页后应裁出较新隐藏区（无振荡）: {st}" assert st["hidO"] == N - _SIZE - _PAGE, f"窗口应上移 _PAGE 条: {st}" check("中部锚点保持（误差≤2px，无 newer 振荡）", anchor_middle) # ---- 2b) 自顶部循环向上分页至头部（真实"load older"点击路径） ---- def page_up_to_head(): _run_js("window.scrollTo(0, 0); 1") time.sleep(0.2) app.processEvents() steps = 0 while True: if not _run_js("rwState.hasMoreOlder"): break page_older_once() steps += 1 assert steps <= N // _SIZE + 10, "换页次数异常（疑似死循环）" first = _run_js(f"document.querySelector('{sel}').id") assert first == chain_ids[0], f"未到达头部: {first}" check("自顶部向上分页至头部（绝对顶部例外）", page_up_to_head) def head_state(): st = _run_js("JSON.stringify({len: rwState.order.length, hidO: rwState.hiddenOlder, " "hidN: rwState.hiddenNewer, older: rwState.hasMoreOlder, newer: rwState.hasMoreNewer})") assert st["hidO"] == 0 and st["older"] is False, st assert st["newer"] is True and st["hidN"] == N - _SIZE, st # 全链可达 + 无重复 all_ids = _run_js(f"Array.from(document.querySelectorAll('{sel}')).map(e => e.id)") assert len(all_ids) == len(set(all_ids)), "DOM 出现重复消息 id" assert len(seen) == N, f"可达 {len(seen)}/{N} 条" check("头部状态 + 全链可达无重复", head_state) # ---- 3) 自顶部向下回翻 2 页（窗口下移，视口留在顶部） ---- def page_down_two(): for k in range(2): _run_js("window.scrollTo(0, 0); 1") time.sleep(0.15) app.processEvents() _run_js("rwRequestPage('newer')") wait_until("rwState.pending === null && window.__rwPageRendering === false", 30, f"向下换页{k}完成") # 顶部回翻：锚点（最旧一条）随 trimHead 移除 → 视口留在顶部露出新页 assert _run_js("window.scrollY") <= 1, "顶部回翻应留在顶部" first = _run_js(f"document.querySelector('{sel}').id") assert first == chain_ids[_PAGE * (k + 1)], f"窗口未下移: {first}" st = _run_js("rwState.hasMoreOlder") assert st is True, "回翻后应仍有更旧消息" check("自顶部向下回翻 2 页（窗口下移）", page_down_two) # ---- 5) auto 模式抽查：顶部自动补页 ---- def auto_probe(): _run_js("window.scrollTo(0, 0); 1") time.sleep(0.15) app.processEvents() _run_js("rwState.mode = 'auto'; 1") hid_before = _run_js("rwState.hiddenOlder") _run_js("rwAutoCheck(); 1") # 直接触发自动补页判定（事件接线另行覆盖） # 给自动补页 12s（每页渲染约 0.5-2s） t0 = time.time() moved = False while time.time() - t0 < 12: app.processEvents() time.sleep(0.2) if _run_js("rwState.hiddenOlder") < hid_before: moved = True break assert moved, "auto 模式顶部未自动补页" check("auto 模式：顶部自动补页", auto_probe) # ---- 4) 规模报告 ---- nodes = page_metrics["dom_nodes"] heights = page_metrics["heights"] pages_ms = page_metrics["pages"] print("\n================ 规模报告 ================") print(f"链长 N={N} | 窗口 size={_SIZE}") print(f"换页数={len(pages_ms)} | 每页耗时 min/avg/max = " f"{min(pages_ms):.0f}/{(sum(pages_ms) / len(pages_ms)):.0f}/{max(pages_ms):.0f} ms") print(f"DOM 节点总数 min/max = {min(nodes)}/{max(nodes)}") print(f"页面高度 min/max = {min(heights)}/{max(heights)} px") if anchor_errors: print(f"锚点误差 min/max = {min(anchor_errors):.2f}/{max(anchor_errors):.2f} px") print("==========================================") def finish(ok=None): app.quit() QTimer.singleShot(500, try_load) QTimer.singleShot(300000, lambda: (print("[超时] 300s 总超时"), finish(False))) app.exec() try: window.close() except Exception: pass print(f"\n===== {'ALL PASS' if all(results) else 'HAS FAILURES'}: {sum(results)}/{len(results)} =====") # QtWebEngine 退出段错误规避：汇总已打印，直接退出 os._exit(0 if all(results) else 1)