feat(ops): 畸形 tool_call 进工具失败聚集 + 巡检邮件只发活跃聚集(bump 0.58.19)

畸形轮整轮丢弃、messages 无痕,admin「工具失败聚集」原本看不见这类失败
(toolfail 只扫 role=tool 报错);且巡检邮件去重集是内存态,每次部署清零,
高频部署期每次重启都把 7 天窗口内早已安静的存量聚集重发一遍。

- 新增 record_malformed_tool_call:丢弃畸形轮时写 usage_events 一行
  kind=tool_malformed(units={tool,len,err,head,tail,tokens_in/out}),
  cost_cny 恒 0 —— cost 全 kind 合计且随任务展示,provider 抖动不算用户
  花销;真实 token 快照进 units 供反推浪费。DB 写失败静默不阻塞重试
- core/toolfail.py 加第二段扫描(usage_events kind=tool_malformed),聚成
  kind=malformed 同构 cluster,admin 面板/巡检邮件零改动自动带上;聚合
  逻辑抽 _add 闭包双数据源复用
- 巡检邮件发信前加 count_24h>0 过滤:已安静的聚集重启后不再重发,还在
  烧的重启后再提醒一次(刻意保留)
- 新增 tests/test_toolfail_malformed.py 4 用例(mock session_scope);
  RUN.md env 巡检段 + 故障兜底同步;PROGRESS 更新

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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 第一个兜底)。

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_*;
+  # tool 错误消息 + 被丢弃的畸形 tool_call 参数(0.58.19,kind=malformed),同签名 >=5 次
-  # 未配邮箱或 SMTP 则只打日志 [toolfail] 行)。admin 页"工具失败"表可随时看低阈值全量。
+  # 且跨 >=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_<user_uuid>" /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,已知边缘 |

core/__init__.py

@@ -1,3 +1,3 @@
 # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
 # 改版本只动这一行。
-__version__ = "0.58.18"
+__version__ = "0.58.19"

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:
+def _log_malformed_args(
-    """畸形 arguments 的首尾片段 + JSON 报错位置打服务端日志(stdout)。
+    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 会崩),
+    stdout 片段过 ascii() 转义:日志消费端编码不可控(Windows dev 控制台 GBK 遇 emoji
-    转义后 grep '\\[malformed\\]' 拿到的内容可无损还原。日志失败静默 —— 留痕绝不能
+    会崩),转义后 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。"""
+        没有 chunk 级 cancel,content 也拿不到 delta —— 整段 text 一次性补 emit。
-        response = self.llm.chat(
+
+        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)

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",

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,

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:
+    def _add(
-            continue
+        tool_name: str, kind: str, sig_line: str, sample: str,
-        hit = _classify(content)
+        task_id: Any, user_id: Any, created_at: datetime,
-        if hit is None:
+    ) -> None:
-            continue
-        kind, sig_line = hit
         # 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():

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()

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()

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 则只打日志)。签名进程内去重 —— 重启/蓝绿切换后同签名会再推
+        # 邮箱/SMTP 则只打日志)。签名进程内去重,且只发近 24h 仍活跃的聚集
-        # 一次(聚集还在=问题还在,重提醒可接受)。ZCBOT_TOOLFAIL_SCAN_INTERVAL
+        # (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: