sorrow404null
a7412824e0
Import the pre-repair source tree as the history baseline. Runtime data (data/), virtualenvs, bytecode caches and logs are gitignored so local secrets and user state stay out of the repo.
369 lines
18 KiB
Python
369 lines
18 KiB
Python
# -*- 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)
|