# -*- coding: utf-8 -*- """失败轮次持久化（对照 pi：message_end 无条件入库 → 出错也写会话）+ 回放取舍。 核心诉求：agent 跑了 200 个工具后第 201 步出错，**已完成的工具结果不能白跑** —— 必须入库、必须能进下次上下文，这样"接着执行最后一次"才可能。 覆盖矩阵： T1 全空失败轮 → 入库留痕(is_ignored=1)，回放**跳过** T2 有工具/正文的失败轮 → 入库 + 回放（工具调用与结果成对）★核心 T3 轮内重试 → 只动内存、**不入库**（不会产生一堆错误行） T4 孤儿工具(ok=None) → 回放注入合成结果（对照 pi insertSyntheticToolResults） T5 叶子前进 → 不再"时光倒流"；parent_id 指向本轮提问 T6 压缩切点交互 → 切点之后照常回放；之前 → 出上下文 T7 旧库自动迁移 → 新列补齐，旧行 stop_reason=NULL 行为不变 T8 copy_session → 复制错误行时携带 stop_reason/error_message/is_ignored 运行: QT_QPA_PLATFORM=offscreen python tests/test_error_persist.py """ import os import sys import json import uuid import sqlite3 import tempfile 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" os.environ.setdefault("PYTHONIOENCODING", "utf-8") # 铁律：测试不得污染真实 data/chat_history.db import core.db_manager as _dbm # noqa: E402 _DB_TMP = os.path.join(tempfile.gettempdir(), f"haocode_test_errpersist_{os.getpid()}.db") if os.path.exists(_DB_TMP): os.remove(_DB_TMP) _dbm._DEFAULT_DB = _DB_TMP # 铁律：不得污染真实 data/config.json _CFG_TMP = os.path.join(tempfile.gettempdir(), f"haocode_test_errcfg_{os.getpid()}.json") with open(_CFG_TMP, "w", encoding="utf-8") as _f: _f.write('{"providers": {}}') os.environ["HAOCODE_CONFIG_FILE"] = _CFG_TMP from PyQt6.QtWidgets import QApplication # noqa: E402 from PyQt6.QtTest import QTest # noqa: E402 from ui.views.main_window import MainWindow # noqa: E402 from core.db_manager import DBManager # noqa: E402 from core.agent.recovery import AgentRunner # noqa: E402 from core.agent.types import (AgentConfig, AgentMessage, ModelConfig, # noqa: E402 RetryConfig) app = QApplication(sys.argv) ok = True def check(name, cond, extra=""): global ok print((" PASS " if cond else " FAIL ") + name + ("" if cond else f" {extra}"), flush=True) if not cond: ok = False def settle(ms=120): for _ in range(int(ms / 20) + 1): app.processEvents() QTest.qWait(20) window = MainWindow() window.show() settle(300) def mk_failed_turn(sid, body="", reasoning="", tools=None, err="连接失败: Connection error"): """模拟一次「已流出一部分 → 出错」的轮次，返回 (user_msg_id, stream_state)""" chain = window.db.get_message_chain(sid) parent = chain[-1]["id"] if chain else None st = { "msg_id": f"msg_err_{uuid.uuid4().hex[:10]}", "parent_id": parent, "previous_leaf_id": parent, "content": body, "reasoning": reasoning, "timeline": list(tools or []), "usage": {}, "tl_kind": "text", "worker": None, } window.current_session_id = sid window._active_streams[sid] = st window.on_error(sid, err) settle(120) return parent, st def tool_entry(cid, name, args, ok_flag, result): return {"t": "tool", "id": cid, "name": name, "args": args, "ok": ok_flag, "result": result} # ====================================================================== # T1 全空失败轮 → 入库留痕但回放跳过 # ====================================================================== sid1 = window.db.create_session("T1 全空失败")["id"] uid1 = window.db.add_message(session_id=sid1, role="user", content="开始吧", parent_id=None)["id"] mk_failed_turn(sid1, body="", reasoning="", tools=[]) chain1 = window.db.get_message_chain(sid1) last1 = chain1[-1] check("T1.1 全空失败轮也入库（对照 pi 无条件 append）", len(chain1) == 2 and last1["role"] == "assistant", f"chain={len(chain1)}") check("T1.2 标记 stop_reason=error + error_message", last1.get("stop_reason") == "error" and "Connection error" in (last1.get("error_message") or ""), f"{last1.get('stop_reason')} / {last1.get('error_message')}") check("T1.3 全空 → is_ignored=1（只在库里留痕）", int(last1.get("is_ignored") or 0) == 1) check("T1.4 正文含 ⚠️ 中断说明（供历史/UI 可见）", "⚠️" in (last1.get("content") or "") and "本轮中断" in (last1.get("content") or ""), (last1.get("content") or "")[:60]) ctx1 = window.build_api_context(sid1) check("T1.5 回放**跳过**全空错误行（避免空 assistant 触发服务商 400）", not any(m.get("role") == "assistant" for m in ctx1), str([m.get("role") for m in ctx1])) # ====================================================================== # T2 ★核心：200 个工具后第 201 步出错 → 工具结果必须进上下文 # ====================================================================== sid2 = window.db.create_session("T2 工具后出错")["id"] uid2 = window.db.add_message(session_id=sid2, role="user", content="帮我建 3 个目录", parent_id=None)["id"] tools2 = [tool_entry(f"c{i}", "bash", json.dumps({"command": f"mkdir d{i}"}), True, f"d{i} created") for i in range(1, 4)] tools2.append(tool_entry("c4", "bash", json.dumps({"command": "mkdir d4"}), None, "")) # 孤儿：开始了但没结果 mk_failed_turn(sid2, body="正在创建目录…", reasoning="先建目录", tools=tools2) chain2 = window.db.get_message_chain(sid2) last2 = chain2[-1] check("T2.1 有工具/正文的失败轮入库且 is_ignored=0（会回放）", last2["role"] == "assistant" and int(last2.get("is_ignored") or 0) == 0, f"ignored={last2.get('is_ignored')}") check("T2.2 timeline 完整保留 4 个工具条目", len([e for e in json.loads(last2["timeline"] or "[]") if e.get("t") == "tool"]) == 4, str([e.get("t") for e in json.loads(last2["timeline"] or "[]")])) check("T2.2b timeline 尾部多一条 text 条目（中断说明，回放时才看得到）", (json.loads(last2["timeline"] or "[]")[-1].get("t") == "text" and "⚠️" in json.loads(last2["timeline"] or "[]")[-1].get("text", "")), str(json.loads(last2["timeline"] or "[]")[-1])[:80]) check("T2.3 正文保留 + 尾部中断说明", "正在创建目录" in last2["content"] and "⚠️" in last2["content"]) ctx2 = window.build_api_context(sid2) tcs = [tc for m in ctx2 if m.get("role") == "assistant" for tc in (m.get("tool_calls") or [])] tool_msgs = [m for m in ctx2 if m.get("role") == "tool"] check("T2.4 ★回放里出现 4 个 tool_call（200 个工具的记录没丢）", len(tcs) == 4, f"tool_calls={len(tcs)}") check("T2.5 ★每个 tool_call 都有配对的 tool 结果（成对，服务商不会 400）", len(tool_msgs) == len(tcs) and {m["tool_call_id"] for m in tool_msgs} == {t["id"] for t in tcs}, f"tool_msgs={len(tool_msgs)}") check("T2.6 已完成的工具结果原文进上下文", any("d1 created" in (m.get("content") or "") for m in tool_msgs)) check("T2.7 错误说明文本也进上下文（模型知道上轮为何断）", any(m.get("role") == "assistant" and "⚠️" in (m.get("content") or "") for m in ctx2)) check("T2.8 顺序正确：assistant tool_calls 在 tool 结果之前", [m.get("role") for m in ctx2].index("assistant") < [m.get("role") for m in ctx2].index("tool")) # ====================================================================== # T3 轮内重试只动内存、不入库 # ====================================================================== class _FakeState: def __init__(self, msgs): self.messages = msgs class _FakeAgent: def __init__(self): self.config = AgentConfig( model=ModelConfig(name="fake"), retry=RetryConfig(max_attempts=3, base_delay_ms=1, factor=2.0), ) self.state = _FakeState([ AgentMessage(role="user", content="hi"), AgentMessage(role="assistant", stop_reason="error", error_message="connection error"), ]) fa = _FakeAgent() runner = AgentRunner(fa, summarize_fn=None) n_before = len(window.db.get_message_chain(sid2)) retried = runner._prepare_retry(fa.config.retry, reason="connection error") n_after = len(window.db.get_message_chain(sid2)) check("T3.1 _prepare_retry 生效（可重试错误）", retried is True) check("T3.2 重试只从**内存**移除坏消息（轮次未结束，不该入库）", len(fa.state.messages) == 1 and fa.state.messages[-1].role == "user", str([m.role for m in fa.state.messages])) check("T3.3 ★重试前后 DB 行数不变（不会堆一堆错误行）", n_before == n_after, f"{n_before} -> {n_after}") # 正常成功轮 → 不写 error 列 sid3 = window.db.create_session("T3 正常轮")["id"] uid3 = window.db.add_message(session_id=sid3, role="user", content="你好", parent_id=None)["id"] st3 = {"msg_id": "msg_ok_1", "parent_id": uid3, "previous_leaf_id": uid3, "content": "你好，我在。", "reasoning": "", "timeline": [], "usage": {}, "tl_kind": "text", "worker": None} window.current_session_id = sid3 window._active_streams[sid3] = st3 window.on_reply_finished(sid3) settle(150) ok_row = window.db.get_message_chain(sid3)[-1] check("T3.4 正常轮 stop_reason 为空（与错误行可区分）", not ok_row.get("stop_reason"), f"{ok_row.get('stop_reason')!r}") # ====================================================================== # T4 孤儿工具 → 合成结果（对照 pi insertSyntheticToolResults） # ====================================================================== sid4 = window.db.create_session("T4 孤儿工具")["id"] uid4 = window.db.add_message(session_id=sid4, role="user", content="跑个命令", parent_id=None)["id"] mk_failed_turn(sid4, body="", tools=[tool_entry("orphan1", "bash", "{}", None, "")]) ctx4 = window.build_api_context(sid4) tool4 = [m for m in ctx4 if m.get("role") == "tool"] check("T4.1 孤儿工具也有配对结果", len(tool4) == 1, str(len(tool4))) check("T4.2 孤儿结果是合成说明（不是空串，避免服务商拒绝）", tool4 and "未收到" in tool4[0]["content"], tool4[0]["content"] if tool4 else "") check("T4.3 孤儿工具的 tool_call 同时存在", any(tc["id"] == "orphan1" for m in ctx4 if m.get("role") == "assistant" for tc in (m.get("tool_calls") or []))) # ====================================================================== # T5 叶子前进（不再时光倒流） # ====================================================================== check("T5.1 会话叶子 = 错误行（叶子前进，不再回退到提问）", window.db.get_session_leaf(sid2) == last2["id"], f"leaf={window.db.get_session_leaf(sid2)} last={last2['id']}") check("T5.2 错误行 parent_id 指向本轮提问", last2["parent_id"] == uid2, f"{last2['parent_id']} vs {uid2}") check("T5.3 链上顺序 = [user, assistant(error)]", [m["role"] for m in chain2] == ["user", "assistant"], str([m["role"] for m in chain2])) check("T5.4 下次提问可接着链（叶子非空 → 可继续）", window.db.get_session_leaf(sid2) is not None) # ====================================================================== # T6 压缩切点交互 # ====================================================================== sid6 = window.db.create_session("T6 压缩交互")["id"] u6 = window.db.add_message(session_id=sid6, role="user", content="老问题", parent_id=None)["id"] mk_failed_turn(sid6, body="中途断了", tools=[]) err6 = window.db.get_message_chain(sid6)[-1] u6b = window.db.add_message(session_id=sid6, role="user", content="新问题", parent_id=err6["id"])["id"] a6 = window.db.add_message(session_id=sid6, role="assistant", content="新回答", parent_id=u6b)["id"] window.db.insert_compaction_mark(sid6, "【摘要】老问题与中断", err6["id"], u6b, json.dumps({"path": "test", "before": 1, "after": 1})) ctx6 = window.build_api_context(sid6) flat6 = json.dumps(ctx6, ensure_ascii=False) check("T6.1 切点之前的错误行 → 不进上下文（已出上下文）", "中途断了" not in flat6 and "⚠️" not in flat6) check("T6.2 摘要进上下文 + 切点之后照常", "【摘要】老问题与中断" in flat6 and "新回答" in flat6) # ====================================================================== # T7 旧库自动迁移（无新列 → 补齐；旧行 stop_reason=NULL） # ====================================================================== _OLD = os.path.join(tempfile.gettempdir(), f"haocode_old_schema_{os.getpid()}.db") if os.path.exists(_OLD): os.remove(_OLD) _c = sqlite3.connect(_OLD) _c.executescript(""" CREATE TABLE sessions (id TEXT PRIMARY KEY, title TEXT, created_at INTEGER, updated_at INTEGER); CREATE TABLE messages (id TEXT PRIMARY KEY, session_id TEXT, role TEXT, content TEXT, reasoning TEXT, is_ignored BOOLEAN, created_at INTEGER, attachment_metadata TEXT, parent_id TEXT, timeline TEXT, usage TEXT); """) _c.execute("INSERT INTO sessions VALUES ('s1','旧会话',1,1)") _c.execute("INSERT INTO messages (id,session_id,role,content,created_at,parent_id) " "VALUES ('m1','s1','user','你好',1,NULL)") _c.commit() _c.close() d_old = DBManager(db_path=_OLD) cols = [r[1] for r in d_old.get_connection().execute("PRAGMA table_info(messages)")] check("T7.1 旧库自动补上 stop_reason / error_message 列", "stop_reason" in cols and "error_message" in cols, str(cols)) row_old = d_old.get_message_chain("s1") check("T7.2 旧行读出来 stop_reason 为空（行为与升级前一致）", row_old and not row_old[-1].get("stop_reason"), str(row_old[-1].get("stop_reason") if row_old else "no row")) check("T7.3 旧库可正常写入错误行（新列可用）", d_old.add_message(session_id="s1", role="assistant", content="⚠️ 中断", parent_id="m1", stop_reason="error", error_message="boom")["id"] is not None) check("T7.4 旧库错误行可读回", d_old.get_message_chain("s1")[-1].get("error_message") == "boom") # ====================================================================== # T8 copy_session 携带新列 # ====================================================================== copy8 = window.db.copy_session(sid2) chain8 = window.db.get_message_chain(copy8["id"]) last8 = chain8[-1] check("T8.1 复制后错误行保留 stop_reason/error_message", last8.get("stop_reason") == "error" and "Connection error" in (last8.get("error_message") or ""), f"{last8.get('stop_reason')} / {last8.get('error_message')}") check("T8.2 复制后 timeline 工具条目一致", len(json.loads(last8["timeline"] or "[]")) == len(json.loads(last2["timeline"] or "[]"))) check("T8.3 复制后 is_ignored 一致", int(last8.get("is_ignored") or 0) == int(last2.get("is_ignored") or 0)) check("T8.4 复制后链条角色一致", [m["role"] for m in chain8] == [m["role"] for m in chain2]) # ====================================================================== # T9 ★数据安全：追加新列绝不得触发「旧数据链表化重构」 # _upgrade_schema 里 upgraded=True 会把每个会话的消息按时间拍平成线性链， # 直接毁掉树状分支（用户的 12 个分支点！）。新列迁移必须走旁路。 # ====================================================================== _BR = os.path.join(tempfile.gettempdir(), f"haocode_branch_{os.getpid()}.db") if os.path.exists(_BR): os.remove(_BR) _bc = sqlite3.connect(_BR) _bc.executescript(""" CREATE TABLE sessions (id TEXT PRIMARY KEY, title TEXT, created_at INTEGER, updated_at INTEGER, has_messages BOOLEAN DEFAULT 0, sort_order INTEGER DEFAULT 0, is_starred BOOLEAN DEFAULT 0, current_leaf_msg_id TEXT, mode TEXT); CREATE TABLE messages (id TEXT PRIMARY KEY, session_id TEXT, role TEXT, content TEXT, reasoning TEXT, is_ignored BOOLEAN, created_at INTEGER, attachment_metadata TEXT, parent_id TEXT, timeline TEXT, usage TEXT); """) _bc.execute("INSERT INTO sessions VALUES ('s9','分支会话',1,1,1,0,0,'m5',NULL)") # 链 u1(1) → a1(2) → u2(3) → {a2(4), a2b(5)} ← u2 下两个孩子 = 分支点 for _mid, _role, _ct, _par in [("m1", "user", 1, None), ("m2", "assistant", 2, "m1"), ("m3", "user", 3, "m2"), ("m4", "assistant", 4, "m3"), ("m5", "assistant", 5, "m3")]: _bc.execute("INSERT INTO messages (id,session_id,role,content,created_at,parent_id) " "VALUES (?,?,?,?,?,?)", (_mid, "s9", _role, _mid, _ct, _par)) _bc.commit() _bc.close() d_b = DBManager(db_path=_BR) _cols_b = [r[1] for r in d_b.get_connection().execute("PRAGMA table_info(messages)")] check("T9.1 旧库自动追加 stop_reason / error_message 列", "stop_reason" in _cols_b and "error_message" in _cols_b, str(_cols_b)) _rows_b = {r["id"]: r for r in d_b.get_connection().execute( "SELECT id, parent_id FROM messages WHERE session_id='s9'")} check("T9.2 ★分支结构未被拍平（a2b.parent 仍指向分支点 u2）", _rows_b["m5"]["parent_id"] == "m3", f"m5.parent={_rows_b['m5']['parent_id']}（拍平后会变成 m4）") check("T9.3 叶子节点未被改写", d_b.get_session_leaf("s9") == "m5", str(d_b.get_session_leaf("s9"))) check("T9.4 原有行内容未被改动", all(_rows_b[k]["parent_id"] == v for k, v in [("m2", "m1"), ("m3", "m2"), ("m4", "m3")]), str({k: _rows_b[k]["parent_id"] for k in _rows_b})) print("\n===== " + ("ALL PASS" if ok else "HAS FAILURES") + " =====", flush=True) for _p in (_DB_TMP, _CFG_TMP, _OLD, _BR): try: if os.path.exists(_p): os.remove(_p) except Exception: pass sys.exit(0 if ok else 1)