diff --git a/PROGRESS.md b/PROGRESS.md index 8bae949..1ecc6ed 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -23,6 +23,7 @@ ### 2026-07 +- **07-13 / 0.58.19**:**畸形 tool_call 进 admin「工具失败聚集」+ 巡检邮件只发活跃聚集**(0.58.18 的观测盲区收尾 + 用户报巡检邮件太频):① 畸形留痕落库:`_log_malformed_args` 在 stdout 外加写 `usage_events` 一行 `kind=tool_malformed`(新 `record_malformed_tool_call`,units={tool,len,err,head,tail,tokens_in/out}),**cost_cny 恒 0**——cost 统计全 kind 合计且随任务展示,provider 抖动的浪费不算用户花销;真实 token 快照进 units 供反推浪费(token 汇总只算 kind=chat,不污染)。DB 写失败静默,绝不阻塞重试主路径。② `core/toolfail.py` 加第二段扫描(usage_events kind=tool_malformed),聚成 kind=malformed 的同构 cluster(signature=归一化 JSON 报错——行列号数字被 `_normalize` 抹平自动聚拢,sample=head…tail),admin 面板/巡检邮件零改动自动带上;聚合逻辑抽 `_add` 闭包双数据源复用。③ 巡检邮件降频:签名去重集是内存态、每次部署清零,高频部署期每次重启把 7 天窗口内早已安静的存量聚集重发一遍——发信前加 `count_24h>0` 过滤,已安静的不再重发,还在烧的重启后再提醒一次(刻意保留)。新增 `tests/test_toolfail_malformed.py` 4 用例(mock session_scope 不碰 DB:聚集/签名归一/双源并存/历史分桶);RUN.md env 巡检段 + 故障兜底 909 行同步。 - **07-13 / 0.58.18**:**畸形 tool_call 首败即降级非流式 + 损坏原文留痕**(用户报 task 716ed3be 芜湖零碳园区方案逐章 write 反复「工具调用参数损坏」1/3→3/3):DB 实测 deepseek-v4-pro 也踩 v4-flash 老病(流式 delta 错位粘坏大参数 arguments),且**同轮流式重 roll 失败强相关**——3~4k 字符中文 write 连挂 3 次全畸形、每章拖 3 分钟+,最后总靠非流式兜底救场,3 次流式重试纯烧 token(每次一整段 2k+ token 生成,不记账但 provider 照计费)。改法(`core/loop.py`):① `_MAX_MALFORMED_RETRIES=3`(3 流式+1 非流式)→ `_MAX_MALFORMED_ATTEMPTS=3`(1 流式+2 非流式),首败即降级;顺手修掉旧版 warn 计数可显示 "4/3" 的 off-by-one。② 新增 `_log_malformed_args`:丢弃前把损坏 arguments 首尾各 300 字符(ascii() 转义防 GBK 控制台崩)+ JSON 报错位置打 stdout `[malformed]` 标签——畸形轮不入库,这是唯一留痕,供定性 provider delta 错位 vs 本地拼接 bug 及报 case 取证(toolfail 扫描器看不见此类失败:无 tool 消息)。新增 `tests/test_loop_malformed_retry.py` 4 用例(object.__new__ 绕构造器打桩双通道):干净流式不重试/首败即非流式/耗尽返最后 response 交 executor 返错/中途 cancel 短路。RUN.md 故障兜底 909 行同步(v4-pro 也中招 + `[malformed]` 日志取证口径)。 - **07-13 / 0.58.17**:**历史消息渲染丢图片预览/chip 两根因修复**(用户报"对话里看不到预览图,有时也不挂 chip"):① 确定性回归(0.12.16 `82feece` 并发化引入):`selectTask` 的 `Promise.all` 里 `loadMessages()` 内部就调 `renderMessages()`,而 `state.taskMeta = meta` 要等 Promise.all 整体完成——首屏渲染时 taskMeta 必然是上个 task 的/null → `_workingDirName` 算出的 wd 错/空 → `extractArtifactRels` 返 `[]`,chip+inline 图整体丢失(直播 SSE 期间可见、切走再切回/刷新就"消失"的体感即此)。修法:`loadMessages` 加 `{render}` 开关,selectTask 传 `render:false` 保持并发取数,`renderMessages` 延后到 taskMeta 赋值后(零额外 RTT,只等 meta/messages 较慢者)。② 中文语境抽取正则漏挂:media.js lead 边界原要求空格/英文标点,"已保存到gpt生图/xxx.png"路径前紧贴汉字不命中;且路径后紧贴"，请查看"时 tail 把中文标点连后文吞进文件名产生 404 坏 chip。修法:lead 改"行首或任意非 `[A-Za-z0-9._-]`"(汉字算边界,仍防 mygpt生图 前缀误挂),tail 排除集抽成 `_TAIL_CLS` 常量并补常见中文标点(主规则+legacy figures/videos 双处)。node 离线回归 12 用例全绿(CJK 前邻/后邻标点/前缀防护/中文文件名/legacy 不重复等)。 - **07-13 / 0.58.16**:**有 gpt_image 权限的用户生图默认 GPT 生图**:此前 `/v1/image_models` 的 `is_default` 恒标"档位过滤后第一个"(yaml 文件名排序豆包在前 → 人人默认 seedream)。改法:`web/app.py` 新增 `_default_image_variant(allowed)`——档位集合里有 `gpt_image`(pro 档 / admin)→ 默认它,否则仍第一个(seedream);`/v1/image_models` 的 `is_default` 与 `_resolve_image_model` 的空串解析(POST messages 时客户端没显式选 image_model)共用该 helper,保证下拉默认选中项和实际起 run 挂的图像工具永远一致。用户可在顶栏下拉自行切回豆包 Seedream(选中值随消息显式发送,不受默认影响)。前端零改动(chat.js 本就选 `is_default` 项);无 user_id 的内部路径(定时任务/messenger)行为不变(空串 → agent_builder yaml 第一个兜底)。 diff --git a/RUN.md b/RUN.md index 779b143..e7055ea 100644 --- a/RUN.md +++ b/RUN.md @@ -70,8 +70,11 @@ # SMTP_FROM=you@qq.com # 可选,默认取 SMTP_USER # SMTP_FROM_NAME=总院科研辅助智能体 # 可选,发件人显示名,默"总院科研辅助智能体"(不暴露内部代号) # 工具失败聚集巡检(0.58.11,反 9dcae061"挂 90 天无人知"):web 进程每天扫近 7 天 - # tool 错误消息,同签名 >=5 次且跨 >=2 task 判聚集 → 发下面邮箱(复用上面 SMTP_*; - # 未配邮箱或 SMTP 则只打日志 [toolfail] 行)。admin 页"工具失败"表可随时看低阈值全量。 + # tool 错误消息 + 被丢弃的畸形 tool_call 参数(0.58.19,kind=malformed),同签名 >=5 次 + # 且跨 >=2 task 判聚集 → 发下面邮箱(复用上面 SMTP_*;未配邮箱或 SMTP 则只打日志 + # [toolfail] 行)。**只发近 24h 仍活跃的聚集**(0.58.19):去重集是内存态、每次部署清零, + # 不加活跃过滤的话高频部署期每次重启都会把 7 天窗口内早已安静的存量聚集重发一遍。 + # admin 页"工具失败"表可随时看低阈值全量(含已安静的灰行)。 # ZCBOT_DEVELOPER_EMAIL=dev@example.com # ZCBOT_TOOLFAIL_SCAN_INTERVAL=86400 # 秒,默 86400;<=0 关掉巡检 # broker 外置(DESIGN §7.0,蓝绿双实例跨进程 SSE/cancel):可选。设了 → event/cancel @@ -906,7 +909,7 @@ sudo xfs_quota -x -c "limit -p bhard=10g zcbot_" /opt | matplotlib / mermaid 出的 PNG 里中文全是方块(豆腐块 □) | sandbox 镜像缺中文字体。Dockerfile 已加 `fonts-noto-cjk fonts-wqy-microhei` + `fc-cache`,但**改了 Dockerfile 必须重 build 镜像 + 清旧容器**才生效(旧容器仍跑老镜像):`docker build -t zcbot-sandbox:latest -f deploy/sandbox/Dockerfile --build-arg HOST_UID=$(id -u) --build-arg HOST_GID=$(id -g) .` → `docker rm -f $(docker ps -aq -f label=zcbot.product=sandbox)` → `systemctl restart zcbot`。验证:`docker run --rm zcbot-sandbox:latest fc-list :lang=zh` 应列出 Noto/WenQuanYi | | dev.html 显示 "load failed" 立刻回登录页 | token 过期或 JWT_SECRET 服务端变了。已自动跳登录页,按上次 tab 重登 | | dev.html 顶栏出现"连接断开,重连中…(N/3)" | SSE 流被切(`--reload` 重启 / nginx 切换 / 网络抖)。客户端自动重连,1s/2s/4s 退避;新进程已 reaper 标 error 则立即收 done + 卡片末尾"请重发"提示;若服务端还活着会继续看后续 delta(断开期间的丢失,broker 不持久化) | -| 对话里偶发 `[Error] invalid JSON arguments` / warn「工具调用参数损坏…重试」 | deepseek v4 系(flash/**pro 也踩过**,2026-07 task 716ed3be)**大参数工具调用(大 write/run_python,≈3K+ 字符中文长文)偶发把内容碎片错位粘进 arguments 或退化成空 `{}`**(上游流式抖动,同轮流式重 roll 连挂强相关)。`core/loop.py` 已自动兜底:畸形参数丢弃整轮,**首败即降级非流式重试**(≤3 次总尝试),非流式历史上从未再畸形。每次丢弃会打服务端日志 `[malformed] task=... tool=... head=... tail=...`(损坏原文唯一留痕,journalctl 里 grep `\[malformed\]` 取证/报 provider case)。仍频繁撞 → 引导模型把大文件拆小分块写 | +| 对话里偶发 `[Error] invalid JSON arguments` / warn「工具调用参数损坏…重试」 | deepseek v4 系(flash/**pro 也踩过**,2026-07 task 716ed3be)**大参数工具调用(大 write/run_python,≈3K+ 字符中文长文)偶发把内容碎片错位粘进 arguments 或退化成空 `{}`**(上游流式抖动,同轮流式重 roll 连挂强相关)。`core/loop.py` 已自动兜底:畸形参数丢弃整轮,**首败即降级非流式重试**(≤3 次总尝试),非流式历史上从未再畸形。每次丢弃会留痕两路:服务端日志 `[malformed] task=... tool=... head=... tail=...`(journalctl 里 grep `\[malformed\]` 取证/报 provider case)+ `usage_events` 一行 `kind=tool_malformed`(cost=0,admin 页「工具失败」表以 kind=malformed 聚集展示,可按模型分组看哪个 provider 在坏)。仍频繁撞 → 引导模型把大文件拆小分块写 | | 蓝绿脚本报 `旧单实例服务 zcbot.service 还在跑` | 还没做蓝绿 bootstrap(旧服务占 8765 与 `zcbot@blue` 冲突)。走 RUN.md「蓝绿部署」bootstrap 段:`disable --now zcbot` 后起 `zcbot@blue` | | 蓝绿 wait_health 报 `instance 不是 blue/green` | 实例 env 没生效:`/etc/systemd/system/zcbot@.service` 未装 / 没 `daemon-reload` / `deploy/instances/<色>.env` 缺失。`systemctl cat zcbot@blue` 核对 `EnvironmentFile` 两行都在 | | 蓝绿切换后用户直播不动、但任务其实在跑 | 用户恰在切换窗口刷新页面,run 还在旧实例上而 SSE 连到了新实例(broker 进程内不跨实例)。run 照常跑完落库,等旧实例日志安静后让用户刷新看历史;非 bug,已知边缘 | diff --git a/core/__init__.py b/core/__init__.py index 14fb0e7..41986f1 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,3 +1,3 @@ # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # 改版本只动这一行。 -__version__ = "0.58.18" +__version__ = "0.58.19" diff --git a/core/loop.py b/core/loop.py index a517ded..b2c3dc0 100644 --- a/core/loop.py +++ b/core/loop.py @@ -11,6 +11,7 @@ from __future__ import annotations import hashlib import json +import threading import time from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple @@ -25,7 +26,7 @@ from .context_fold import maybe_fold from .executor import ExecCtx, Executor from .llm import LLM from .session import Session -from .storage import record_chat_usage +from .storage import record_chat_usage, record_malformed_tool_call from . import pptx_guard # 产物机检只挂能落盘的执行类工具(fs 写工具不适合造 pptx,机检无意义) @@ -171,17 +172,21 @@ def _malformed_tool_calls(response: Any) -> List[str]: return bad -def _log_malformed_args(task_id: Any, response: Any) -> None: - """畸形 arguments 的首尾片段 + JSON 报错位置打服务端日志(stdout)。 +def _log_malformed_args( + task_id: Any, user_id: Any, model_profile: str, response: Any, +) -> None: + """畸形 arguments 的首尾片段 + JSON 报错位置留痕:stdout + usage_events 双写。 畸形轮不 append/不记账,这里是唯一留痕 —— 用于事后定性损坏形态(provider 流式 delta 错位 vs 本地 stream_chunk_builder 拼接 bug),以及向 provider 报 case 取证。 - 片段过 ascii() 转义:日志消费端编码不可控(Windows dev 控制台 GBK 遇 emoji 会崩), - 转义后 grep '\\[malformed\\]' 拿到的内容可无损还原。日志失败静默 —— 留痕绝不能 - 反过来打断重试主路径。 + stdout 片段过 ascii() 转义:日志消费端编码不可控(Windows dev 控制台 GBK 遇 emoji + 会崩),转义后 grep '\\[malformed\\]' 拿到的内容可无损还原。DB 行(kind=tool_malformed, + cost=0)喂 admin「工具失败聚集」面板 + 巡检邮件(core/toolfail.py),units 里 + 快照该轮真实 token 供估算浪费。任何一路失败都静默 —— 留痕绝不能反过来打断重试主路径。 """ try: msg = response.choices[0].message + usage = _extract_usage_details(getattr(response, "usage", None)) for tc in (getattr(msg, "tool_calls", None) or []): raw = (getattr(tc.function, "arguments", None) or "").strip() if not raw: @@ -195,6 +200,21 @@ def _log_malformed_args(task_id: Any, response: Any) -> None: f"head={ascii(raw[:300])} tail={ascii(raw[-300:])}", flush=True, ) + try: + record_malformed_tool_call( + task_id=task_id, + user_id=user_id, + model_profile=model_profile, + tool=tc.function.name, + arg_len=len(raw), + error=str(e), + head=raw[:300], + tail=raw[-300:], + tokens_in=usage["tokens_in"], + tokens_out=usage["tokens_out"], + ) + except Exception: + pass # DB 不可用(如单测无 DB)不影响 stdout 留痕与重试 except Exception: pass @@ -463,7 +483,8 @@ class AgentLoop: 返回 (response, cancelled_mid_stream): - 正常完结 → (response, False);response shape 与非流式 completion() 等价 (choices[0].message + usage) - - 中途 cancel → (None, True);已收 chunk 丢弃,内层 generator 在 finally 关闭底层连接 + - 中途 cancel → (None, True);已收 chunk 丢弃,内层 generator 在 finally 关闭底层连接。 + 非流式重试期间 cancel 同样 (None, True)(线程级 poll,见 _nonstream_once) 畸形重试:deepseek v4 系(flash/pro 均实测踩过)大参数工具调用偶发把内容碎片 错位粘进 arguments,拼回后 JSON 解析失败。这种损坏一旦入库会被每轮重发、诱导 @@ -480,17 +501,24 @@ class AgentLoop: self.session.messages, compact_threshold_chars=compact_threshold, ) - self._emit({ + llm_start_event = { "type": "llm_start", **{f"context_{k}": v for k, v in context_stats.items()}, # 窗口总容量(chars 口径,= reliable_context×2.5):前端压缩指示环算占用比; # 50% 进压缩区、85% 触发折叠(context_fold)。 "context_limit_chars": int(self.caps.reliable_context * self._CHARS_PER_TOKEN), - }) + } for attempt in range(self._MAX_MALFORMED_ATTEMPTS): use_nonstream = attempt > 0 + # 每个 attempt 重发 llm_start(stats 同一份):非流式重试完成前零 delta 事件, + # 而 warn 事件会让前端把当前文字段定稿关闭 —— 不重发的话「思考中 · Ns」占位段 + # 没人重建,页面静止到重试完成,与卡死无法区分。 + self._emit(dict(llm_start_event)) if use_nonstream: response = self._nonstream_once(llm_messages) + if response is None: + # 非流式重试期间用户点了停止(线程级 poll,见 _nonstream_once) + return None, True else: response, cancelled = self._collect_stream_once(llm_messages) if cancelled: @@ -499,7 +527,10 @@ class AgentLoop: bad = _malformed_tool_calls(response) if not bad: return response, False - _log_malformed_args(self.session.task_id, response) + _log_malformed_args( + self.session.task_id, self.user_id, + f"{self.caps.family}.{self.caps.variant}", response, + ) self._emit({ "type": "warn", "msg": ( @@ -552,14 +583,39 @@ class AgentLoop: response = litellm.stream_chunk_builder(chunks, messages=llm_messages) return response, False - def _nonstream_once(self, llm_messages: List[dict]) -> Any: + _NONSTREAM_CANCEL_POLL_SECS = 0.5 + + def _nonstream_once(self, llm_messages: List[dict]) -> Optional[Any]: """非流式兜底:provider 服务端一次性拼好 tool_calls,绕开流式 delta 错位。 - 没有 chunk 级 cancel,content 也拿不到 delta —— 整段 text 一次性补 emit。""" - response = self.llm.chat( - messages=llm_messages, - tools=self.executor.schemas(), - reasoning_effort=self.caps.default_reasoning_effort or None, - ) + 没有 chunk 级 cancel,content 也拿不到 delta —— 整段 text 一次性补 emit。 + + cancel 响应:非流式一次生成可达分钟级(重试轮常是大参数 write),同步阻塞会让 + 「停止」按钮等到调用整个返回才生效。故调用放后台 daemon 线程,主线程按拍 poll + cancel:命中即返回 None(线程弃养自行跑完,响应到达后被丢弃,不入库不记账 —— + 与流式 cancel 丢弃已收 chunk 的语义一致)。""" + box: Dict[str, Any] = {} + done = threading.Event() + + def _call() -> None: + try: + box["resp"] = self.llm.chat( + messages=llm_messages, + tools=self.executor.schemas(), + reasoning_effort=self.caps.default_reasoning_effort or None, + ) + except BaseException as e: # noqa: BLE001 — 原样转抛回主线程 + box["exc"] = e + finally: + done.set() + + worker = threading.Thread(target=_call, daemon=True, name="llm-nonstream-retry") + worker.start() + while not done.wait(self._NONSTREAM_CANCEL_POLL_SECS): + if self._is_cancelled(): + return None + if "exc" in box: + raise box["exc"] + response = box["resp"] try: msg = response.choices[0].message rc = getattr(msg, "reasoning_content", None) diff --git a/core/storage/__init__.py b/core/storage/__init__.py index e1ab4f5..4a1f228 100644 --- a/core/storage/__init__.py +++ b/core/storage/__init__.py @@ -12,7 +12,7 @@ from .engine import ( get_engine, session_scope, ) -from .usage import record_chat_usage +from .usage import record_chat_usage, record_malformed_tool_call from .utils import ( NoSubtaskError, check_no_subtask, @@ -29,6 +29,7 @@ __all__ = [ "get_engine", "get_task", "record_chat_usage", + "record_malformed_tool_call", "session_scope", "update_task", "upsert_task", diff --git a/core/storage/usage.py b/core/storage/usage.py index 33fffc6..41dcdf5 100644 --- a/core/storage/usage.py +++ b/core/storage/usage.py @@ -149,6 +149,47 @@ def record_chat_usage( return cost_cny +def record_malformed_tool_call( + *, + task_id: UUID, + user_id: UUID, + model_profile: str, + tool: str, + arg_len: int, + error: str, + head: str, + tail: str, + tokens_in: int = 0, + tokens_out: int = 0, +) -> None: + """记一次被丢弃的畸形 tool_call(kind=tool_malformed)。 + + 这不是记账而是留痕:畸形轮整轮丢弃、messages 无痕,此行是 admin「工具失败聚集」 + 面板与巡检邮件唯一的数据源(core/toolfail.py 第二段扫描按 units->>'tool'/'err' + 聚合)。cost_cny 恒 0 —— cost 统计全 kind 合计且随任务展示,provider 抖动的浪费 + 不该算成用户花销;该轮真实 token 快照进 units,将来要算浪费成本可从 units 反推 + (token 汇总只算 kind=chat,此处 tokens_in/out 不会混进 token 统计)。 + """ + with session_scope() as s: + s.add(UsageEvent( + user_id=user_id, + task_id=task_id, + message_id=None, # 畸形轮不入 messages,无可关联 + kind="tool_malformed", + model_profile=model_profile, + units={ + "tool": tool, + "len": int(arg_len), + "err": str(error)[:200], + "head": head, + "tail": tail, + "tokens_in": int(tokens_in), + "tokens_out": int(tokens_out), + }, + cost_cny=Decimal("0"), + )) + + def record_image_usage( *, task_id: UUID, diff --git a/core/toolfail.py b/core/toolfail.py index 7d05e0f..323ceff 100644 --- a/core/toolfail.py +++ b/core/toolfail.py @@ -8,6 +8,10 @@ 纯只读查询、无新表无状态;告警通道由调用方决定(web/app.py 的巡检 loop 发 开发者邮箱,admin API 直接返给前端表格)。 +第二数据源(0.58.19):被丢弃的畸形 tool_call 参数(kind=malformed)——这类失败 +整轮不入 messages(防投毒级联),loop 落 usage_events(kind=tool_malformed), +在此并入同一聚合口径(signature=归一化 JSON 报错,sample=损坏参数首尾片段)。 + 失败判定(tool content 的三类标记,形态见 executor_docker/_host): - `[Error` 开头 —— 执行器/工具层报错([Error]、[Error executing ...]) - `command timed out` —— shell/run_python 超时 @@ -85,7 +89,6 @@ def scan_tool_failures( now = datetime.now(timezone.utc) cutoff = now - timedelta(days=days) n_buckets = max(1, math.ceil(days)) - rows = [] with session_scope() as s: rows = s.execute( text( @@ -101,15 +104,28 @@ def scan_tool_failures( ), {"cutoff": cutoff}, ).fetchall() + # 第二段:被丢弃的畸形 tool_call 参数(kind=tool_malformed)。这类失败整轮 + # 不入 messages(防投毒),loop._log_malformed_args 落在 usage_events, + # 是它们进面板/巡检邮件的唯一路径。 + mrows = s.execute( + text( + "select task_id, user_id, created_at, " + " units->>'tool' as tool_name, " + " units->>'err' as err, " + " units->>'head' as head, " + " units->>'tail' as tail " + "from usage_events " + "where kind = 'tool_malformed' and created_at >= :cutoff" + ), + {"cutoff": cutoff}, + ).fetchall() agg: Dict[Tuple[str, str], Dict[str, Any]] = {} - for task_id, user_id, created_at, tool_name, content in rows: - if not content: - continue - hit = _classify(content) - if hit is None: - continue - kind, sig_line = hit + + def _add( + tool_name: str, kind: str, sig_line: str, sample: str, + task_id: Any, user_id: Any, created_at: datetime, + ) -> None: # DB 列若是 naive timestamp(存 UTC),补 tzinfo 才能和 now 做减法 ts = created_at if created_at.tzinfo else created_at.replace(tzinfo=timezone.utc) key = (tool_name or "?", _normalize(sig_line)) @@ -124,7 +140,7 @@ def scan_tool_failures( "users": set(), "first_at": ts, "last_at": ts, - "sample": content[:300], + "sample": sample[:300], "daily": [0] * n_buckets, } c["count"] += 1 @@ -137,7 +153,23 @@ def scan_tool_failures( c["first_at"] = ts if ts > c["last_at"]: c["last_at"] = ts - c["sample"] = content[:300] + c["sample"] = sample[:300] + + for task_id, user_id, created_at, tool_name, content in rows: + if not content: + continue + hit = _classify(content) + if hit is None: + continue + kind, sig_line = hit + _add(tool_name, kind, sig_line, content, task_id, user_id, created_at) + + for task_id, user_id, created_at, tool_name, err, head, tail in mrows: + _add( + tool_name, "malformed", err or "?", + f"{head or ''} … {tail or ''}", + task_id, user_id, created_at, + ) out = [] for c in agg.values(): diff --git a/tests/test_loop_malformed_retry.py b/tests/test_loop_malformed_retry.py index 8695373..d398e6f 100644 --- a/tests/test_loop_malformed_retry.py +++ b/tests/test_loop_malformed_retry.py @@ -1,6 +1,8 @@ from __future__ import annotations import sys +import threading +import time import unittest from pathlib import Path from types import SimpleNamespace @@ -24,8 +26,9 @@ BAD = '].cells[1].merge({"path": "a.md"' # 流式 delta 错位的典型形态 def _make_loop(stream_results, nonstream_results): """绕开构造器搭 AgentLoop:只装 _stream_llm 用到的属性,打桩两条取流路径。""" loop = object.__new__(AgentLoop) - loop.caps = SimpleNamespace(reliable_context=64_000) + loop.caps = SimpleNamespace(reliable_context=64_000, family="test", variant="t") loop.session = SimpleNamespace(messages=[], task_id="test-task") + loop.user_id = "test-user" # 无 DB 环境:_log_malformed_args 的落库路径应静默跳过 loop.events = [] loop._emit = loop.events.append calls = {"stream": 0, "nonstream": 0} @@ -61,6 +64,9 @@ class TestMalformedRetry(unittest.TestCase): warns = [e for e in loop.events if e.get("type") == "warn"] self.assertEqual(len(warns), 1) self.assertIn("非流式", warns[0]["msg"]) + # 每个 attempt 重发 llm_start:warn 关掉前端占位段后,靠它重建「思考中」跳秒 + starts = [e for e in loop.events if e.get("type") == "llm_start"] + self.assertEqual(len(starts), 2) def test_all_attempts_exhausted_returns_last(self): """全部尝试耗尽仍畸形 → 返回最后一次 response(交 executor 返错),不死循环。""" @@ -81,6 +87,52 @@ class TestMalformedRetry(unittest.TestCase): self.assertIsNone(resp) self.assertEqual(calls["nonstream"], 0) + def test_cancel_during_nonstream_retry(self): + """非流式重试期间 cancel(_nonstream_once 返回 None)→ (None, True),不再续试。""" + loop, calls = _make_loop([_resp(BAD)], [None, _resp(GOOD)]) + resp, cancelled = loop._stream_llm() + self.assertTrue(cancelled) + self.assertIsNone(resp) + self.assertEqual(calls, {"stream": 1, "nonstream": 1}) + + def test_nonstream_once_polls_cancel(self): + """真 _nonstream_once:llm.chat 阻塞时点停止,应在 poll 拍内返回 None, + 不等调用整个返回(此前同步阻塞,分钟级生成期间停止按钮无效)。""" + loop = object.__new__(AgentLoop) + loop.caps = SimpleNamespace(default_reasoning_effort=None) + loop.executor = SimpleNamespace(schemas=lambda: []) + loop.cancel_check = lambda: True + block = threading.Event() + + def slow_chat(**kwargs): + block.wait(30) + return _resp(GOOD) + + loop.llm = SimpleNamespace(chat=slow_chat) + t0 = time.monotonic() + resp = loop._nonstream_once([]) + elapsed = time.monotonic() - t0 + block.set() # 释放弃养线程,别让它拖住测试进程退出 + self.assertIsNone(resp) + self.assertLess(elapsed, 5) + + def test_nonstream_once_returns_response_when_not_cancelled(self): + """不取消时行为与原同步版等价:拿到 response 并补 emit 整段 text。""" + loop = object.__new__(AgentLoop) + loop.caps = SimpleNamespace(default_reasoning_effort=None) + loop.executor = SimpleNamespace(schemas=lambda: []) + loop.cancel_check = None + loop.events = [] + loop._emit = loop.events.append + good = _resp(GOOD) + good.choices[0].message.content = "hello" + loop.llm = SimpleNamespace(chat=lambda **kwargs: good) + resp = loop._nonstream_once([]) + self.assertIs(resp, good) + texts = [e for e in loop.events if e.get("type") == "text"] + self.assertEqual(len(texts), 1) + self.assertEqual(texts[0]["delta"], "hello") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_toolfail_malformed.py b/tests/test_toolfail_malformed.py new file mode 100644 index 0000000..5a1ab9f --- /dev/null +++ b/tests/test_toolfail_malformed.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import sys +import unittest +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import core.toolfail as tf # noqa: E402 + +NOW = datetime.now(timezone.utc) + + +class _FakeSession: + """按 SQL 关键字分流两段查询:messages 段与 usage_events(tool_malformed)段。""" + + def __init__(self, msg_rows, malformed_rows): + self._msg_rows = msg_rows + self._malformed_rows = malformed_rows + + def execute(self, clause, params=None): + sql = str(clause) + rows = self._malformed_rows if "tool_malformed" in sql else self._msg_rows + return SimpleNamespace(fetchall=lambda: rows) + + +def _scan(msg_rows, malformed_rows, **kw): + @contextmanager + def fake_scope(): + yield _FakeSession(msg_rows, malformed_rows) + + with patch.object(tf, "session_scope", fake_scope): + return tf.scan_tool_failures(**kw) + + +class TestToolfailMalformed(unittest.TestCase): + def test_malformed_rows_cluster(self): + """usage_events 的 tool_malformed 行聚成 kind=malformed 的 cluster。""" + err = "Expecting value: line 1 column 1 (char 0)" + rows = [ + ("t1", "u1", NOW - timedelta(hours=1), "write", err, "].cells[1]", '"}'), + ("t2", "u1", NOW - timedelta(hours=2), "write", err, "].merge(", '"}'), + ] + out = _scan([], rows, days=7, min_count=2, min_tasks=2) + self.assertEqual(len(out), 1) + c = out[0] + self.assertEqual(c["tool"], "write") + self.assertEqual(c["kind"], "malformed") + self.assertEqual(c["count"], 2) + self.assertEqual(c["count_24h"], 2) + self.assertEqual(c["task_count"], 2) + self.assertIn("…", c["sample"]) # sample = head … tail + + def test_err_signature_normalized(self): + """报错里的行号/列号数字被归一,不同位置的同类报错聚成一条。""" + rows = [ + ("t1", "u1", NOW, "write", "Expecting ',' delimiter: line 1 column 57 (char 56)", "h", "t"), + ("t2", "u2", NOW, "write", "Expecting ',' delimiter: line 3 column 9 (char 88)", "h", "t"), + ] + out = _scan([], rows, days=7, min_count=2, min_tasks=2) + self.assertEqual(len(out), 1) + + def test_merges_with_message_clusters(self): + """两段数据源并存互不干扰:messages 报错与 malformed 各出各的 cluster。""" + msg_rows = [ + ("t1", "u1", NOW, "glob", "[Error] base path not found"), + ("t2", "u2", NOW, "glob", "[Error] base path not found"), + ] + mal_rows = [ + ("t3", "u1", NOW, "write", "Expecting value", "h", "t"), + ("t4", "u2", NOW, "write", "Expecting value", "h", "t"), + ] + out = _scan(msg_rows, mal_rows, days=7, min_count=2, min_tasks=2) + kinds = sorted(c["kind"] for c in out) + self.assertEqual(kinds, ["error", "malformed"]) + + def test_quiet_cluster_daily_bucket(self): + """3 天前的畸形记录:count_24h=0(已安静),落在对应历史桶。""" + rows = [ + ("t1", "u1", NOW - timedelta(days=3, hours=1), "write", "Expecting value", "h", "t"), + ("t2", "u2", NOW - timedelta(days=3, hours=2), "write", "Expecting value", "h", "t"), + ] + out = _scan([], rows, days=7, min_count=2, min_tasks=2) + c = out[0] + self.assertEqual(c["count_24h"], 0) + self.assertEqual(sum(c["daily"]), 2) + self.assertEqual(c["daily"][-4], 2) # 尾桶=近24h,往前退 3 桶 + + +if __name__ == "__main__": + unittest.main() diff --git a/web/app.py b/web/app.py index 36566bc..d6f6b4f 100644 --- a/web/app.py +++ b/web/app.py @@ -977,8 +977,10 @@ def create_app() -> FastAPI: # ── 工具失败聚集巡检(0.58.11)── 反 9dcae061 教训(mmdc 在生产挂 90 天 # 0 成功无人知,靠人工扫 DB 才发现):每天扫近 7 天 tool 错误消息,同签名 # >=5 次且跨 >=2 task 判聚集,发 ZCBOT_DEVELOPER_EMAIL(复用 SMTP_*;未配 - # 邮箱/SMTP 则只打日志)。签名进程内去重 —— 重启/蓝绿切换后同签名会再推 - # 一次(聚集还在=问题还在,重提醒可接受)。ZCBOT_TOOLFAIL_SCAN_INTERVAL + # 邮箱/SMTP 则只打日志)。签名进程内去重,且只发近 24h 仍活跃的聚集 + # (count_24h>0)—— 去重集是内存态,重启/蓝绿清零,高频部署期若不加活跃 + # 过滤,每次部署都会把 7 天窗口内早已安静的存量聚集重发一遍(0.58.19 实改; + # 还在烧的重启后再提醒一次是刻意保留的)。ZCBOT_TOOLFAIL_SCAN_INTERVAL # 秒,默 86400,<=0 整体关掉。 dev_email = os.getenv("ZCBOT_DEVELOPER_EMAIL", "").strip() toolfail_interval = int(os.getenv("ZCBOT_TOOLFAIL_SCAN_INTERVAL", "").strip() or 86400) @@ -992,7 +994,8 @@ def create_app() -> FastAPI: try: clusters = await loop.run_in_executor(None, scan_tool_failures) fresh = [c for c in clusters - if (c["tool"], c["signature"]) not in alerted] + if c.get("count_24h") + and (c["tool"], c["signature"]) not in alerted] if fresh: alerted.update((c["tool"], c["signature"]) for c in fresh) for c in fresh: