优化工具失败观测与重复保护

CHANGELOG.md

@@ -5,6 +5,10 @@
 > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
 > 工程口径的完整记录见 `PROGRESS.md` / git log。
 
+## 0.60.4 — 2026-07-29
+
+- 管理后台的工具失败信息现在会区分近期跨任务故障、单任务反复失败、按设计拦截的质量检查和已安静历史，并显示工具调用参数自动抢救率，排查时更容易判断问题是否仍在发生。
+
 ## 0.60.3 — 2026-07-28
 
 - 网页、微信、企业微信和定时任务现在采用一致的消息接收机制：消息会在开始处理前可靠保存，服务恰好在后台任务启动前重启时也不会静默丢失输入。

DESIGN.md

@@ -290,6 +290,8 @@ scheduled_jobs(§8.5)   channel_bindings(§8.7,判别列+JSONB)
 
 **优雅 drain 已是单实例上限**(SIGTERM 拒新 run + 等收尾);先撞的瓶颈是线程池(每活跃 run 占 1 线程)。落地排序:① 轻量监控(显式 executor + 60s `[stats]` 周期日志——要历史峰值不是快照);② 按数据决策扩容;③ `--reload` 缩 503 窗。**不做监控界面**:运维健康是少数标量,日志够;业务分析走 DB SQL。界面阶梯:日志 → /v1/stats → Grafana → 只读 dashboard,现停第一级。无感换版已由蓝绿落地(0.41)、broker 外置 Redis 已实施(0.42,均见 §7.0);扩容路径:调大线程池 → 稳态双实例分流(broker 已外置,纯 nginx 配置动作)。
 
+**工具失败聚集的信号分层(2026-07-29)**:7 天窗口继续保留取证，但管理端默认判断对象是“近 24h 仍活跃且跨 ≥2 task”的系统性故障；单 task 反复试错、按设计非零退出的质量门、近 24h 已归零的历史尾巴分别展示，避免正常迭代挤占故障榜首。API 保留低阈值全量 `clusters`，只加 `category=failure|quality_gate`，分区属于前端读侧语义，不删除既有字段。空输出 shell 非零退出从相邻 assistant tool_call 提取稳定命令类别（如 search/no match、dependency probe），只改善签名，不持久化第二份命令事实。RepeatGuard 只把 `run_python` 的 traceback + 非零退出纳入同错 streak；不泛化到 shell，避免 grep 未命中和质量门复检被误拦。provider wire 健康另走只读派生端点，按 `model_profile+tool` 对已有 `tool_salvaged/tool_malformed` 事件计算 24h/窗口抢救率；它是失败数的分母与趋势解释，不混进 cluster，不新增表/索引，也不反向改变并行调用、salvage 或重试策略。
+
 ### 8.5 定时任务(✅ 2026-06-18)
 
 **核心洞察**:job 本体 = cron+tz + 一句 prompt + 会话模式;守护循环只负责到点把带标记的 prompt 喂进**现成 agent 主管线**,不造第二套执行路径。**"发邮件"不是字段是 agent 动作**——加任何投递能力不改 schema。业界四源(OpenClaw/Autobot/Claude Code/geta)模式收敛佐证。

PROGRESS.md

@@ -2,7 +2,7 @@
 
 > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。
 
-最后更新:2026-07-28(全入口 run 生命周期收口,bump 0.60.3)
+最后更新:2026-07-29(工具失败信号分层 + wire 健康观测,bump 0.60.4)
 
 ---
 
@@ -21,6 +21,10 @@
 
 ## 已完成关键能力
 
+### 2026-07-29
+
+- **07-29 / 0.60.4 / 工具失败信号分层 + run_python 同错保护 + wire 健康观测**:管理端保留近7天低阈值全量事实，默认把近24h活跃数据拆成跨任务系统性故障、单任务反复失败、按设计非零退出的质量门，已归零记录沉入历史区；空输出 shell 关联最近 assistant tool_call 补 `search/no match` / `dependency probe` 等稳定类别。RepeatGuard 仅将 `run_python` traceback+非零退出纳入同错 streak，普通 shell exit 1 保持原语义。新增只读 `/v1/admin/tool-wire-health`，按 model_profile+tool 对既有 `tool_salvaged/tool_malformed` 计算24h/7d抢救率，无新表、migration、索引或执行策略变化。全量387测试通过（17项按环境跳过）。
+
 ### 2026-07-28
 
 - **07-28 / 0.60.3 / Web、渠道、定时任务统一 run 生命周期**:新增 `web/run_lifecycle.py`，收口 task 行锁/忙碌判断、user 消息与 running 原子提交、broker/inflight 登记及调度失败转 error；网页、微信/企微、定时任务全部消费已持久化轮次，消除渠道与 scheduler 原有的 worker 启动前丢输入窗口。Web 模型降级/自动标题、渠道回复、定时结果统计继续留在各自模块，CLI 保留旧 `run(message)`，无新表、migration、队列或 run 实体。全量 379 测试通过（17 项按环境跳过）。

RUN.md

@@ -78,7 +78,9 @@
   # 且跨 >=2 task 判聚集 → 发下面邮箱(复用上面 SMTP_*;未配邮箱或 SMTP 则只打日志
   # [toolfail] 行)。**只发近 24h 仍活跃的聚集**(0.58.19):去重集是内存态、每次部署清零,
   # 不加活跃过滤的话高频部署期每次重启都会把 7 天窗口内早已安静的存量聚集重发一遍。
-  # admin 页"工具失败"表可随时看低阈值全量(含已安静的灰行)。
+  # admin 页仍取低阈值全量,分区展示:近24h跨任务系统性故障 / 单任务反复失败 /
+  # 按设计拦截的质量门 / 已安静历史；修复后看对应项进入“已安静”。同区的“工具
+  # 调用链路健康”按模型档+工具展示 tool_salvaged/tool_malformed 的24h/7d抢救率。
   # ZCBOT_DEVELOPER_EMAIL=dev@example.com
   # ZCBOT_TOOLFAIL_SCAN_INTERVAL=86400   # 秒,默 86400;<=0 关掉巡检
   # broker 外置(DESIGN §7.0,蓝绿双实例跨进程 SSE/cancel):可选。设了 → event/cancel

core/__init__.py

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

core/loop.py

@@ -59,16 +59,40 @@ _CANCELLED_TOOL_PLACEHOLDER = "[cancelled by user]"
 # 与 core/toolfail._normalize 同源思路,此处保持 loop 自包含(不反向依赖 toolfail)。
 _ERR_PATH_RE = re.compile(r"(?:[A-Za-z]:)?(?:[/\\][\w.\-一-鿿*]+){2,}")
 _ERR_NUM_RE = re.compile(r"\d+")
+_NONZERO_EXIT_RE = re.compile(r"\[exit ([1-9]\d*)\]\s*$")
 
 
-def _norm_err(result: str) -> str:
+def _norm_err_line(line: str) -> str:
-    """取 [Error] 结果首行、抹平路径/数字,得到稳定的错误签名(截断 120)。"""
+    """抹平错误行里的路径/数字，得到稳定签名（截断 120）。"""
-    line = (result.lstrip().splitlines() or [""])[0]
     line = _ERR_PATH_RE.sub("<path>", line)
     line = _ERR_NUM_RE.sub("N", line)
     return line.strip()[:120]
 
 
+def _tool_error_signature(name: str, result: str) -> Optional[str]:
+    """返回重复守卫使用的错误签名；非错误返回 None。
+
+    通用工具维持既有 ``[Error]`` 契约。额外只识别 ``run_python`` 的
+    ``Traceback ... [exit N]``，使 Python 运行期同错能进入 err-streak；普通 shell
+    的 grep 未命中、质量门 exit 1 不在此扩面，避免把正常检查迭代当成撞墙。
+    """
+    head = result.lstrip()
+    if head.startswith("[Error"):
+        return _norm_err_line((head.splitlines() or [""])[0])
+    if (
+        name != "run_python"
+        or "Traceback (most recent call last):" not in result
+        or _NONZERO_EXIT_RE.search(result) is None
+    ):
+        return None
+    lines = [
+        line.strip()
+        for line in result.splitlines()[:-1]
+        if line.strip() and line.strip() not in ("[stdout]", "[stderr]")
+    ]
+    return _norm_err_line(lines[-1]) if lines else None
+
+
 class _RepeatGuard:
     """检测「同名同参 + 无产出」的病理性重复调用,断掉死循环。
 
@@ -79,8 +103,8 @@ class _RepeatGuard:
     命门是只惩罚「无产出」重复,绝不误伤正常迭代:
     - 同参但**每次结果不同**(改了脚本后重跑 run_python、修 bug 后重跑构建)→ 有产出,
       计数清零,永不拦。
-    - 同参且**结果是 `[Error]` 或与之前某次一字不差**(空 `{}` 缺参、反复撞同一个错)
+    - 同参且**结果是 `[Error]`、`run_python` traceback 非零退出，或与之前某次
-      → 无产出,累计。
+      一字不差**(空 `{}` 缺参、反复撞同一个错)→ 无产出,累计。
     累计 >= SOFT 注入软提示(模型当轮就看到);>= HARD 直接拦截不执行,逼它换路。
 
     顺带堵掉 `llm_transport.malformed_tool_calls` 的洞:大参数畸形退化成合法空 `{}` 时,executor 每次
@@ -138,7 +162,8 @@ class _RepeatGuard:
         """
         st = self._state(name, args)
         h = hashlib.sha1(result.encode("utf-8", "replace")).hexdigest()
-        is_err = result.lstrip().startswith("[Error")
+        esig = _tool_error_signature(name, result)
+        is_err = esig is not None
         dup = h in st["hashes"]
         if st["n"] >= 1:
             if is_err or dup:
@@ -150,7 +175,6 @@ class _RepeatGuard:
         st["n"] += 1
         # 第二道判据:按工具的连续同类错误 streak(跨不同 args 撞同一堵墙)
         if is_err:
-            esig = _norm_err(result)
             s = self._err_streak.get(name)
             if s and s["esig"] == esig:
                 s["count"] += 1

core/toolfail.py

@@ -31,6 +31,7 @@ tool_use 漏成正文后丢空)。loop 落 usage_events(kind=empty_response),在
 """
 from __future__ import annotations
 
+import json
 import math
 import re
 from datetime import datetime, timedelta, timezone
@@ -43,6 +44,7 @@ from core.storage.telemetry import (
     KIND_EMPTY_RESPONSE,
     KIND_RUN_ERROR,
     KIND_TOOL_MALFORMED,
+    KIND_TOOL_SALVAGED,
 )
 
 # 签名归一:同一类错误在不同 task/参数下的差异(路径/数字/uuid/十六进制)抹平,
@@ -55,6 +57,7 @@ _RE_WS = re.compile(r"\s+")
 
 _EXIT_TAIL = re.compile(r"\[exit (\d+)\]\s*$")
 _STREAM_MARKS = ("[stdout]", "[stderr]")
+_BARE_EXIT_RE = re.compile(r"^exit \d+$")
 
 
 def _normalize(s: str) -> str:
@@ -86,6 +89,63 @@ def _classify(content: str) -> Optional[Tuple[str, str]]:
     return None
 
 
+def _shell_command_hint(prior_payload: Any, tool_call_id: str) -> str:
+    """从相邻 assistant tool_call 提取空输出 shell 的命令类别。
+
+    shell 仅返回 ``[exit 1]`` 时，原签名会把 grep 未命中、依赖探测和真正的命令失败
+    全揉成 ``exit N``。messages 已保存 tool_call_id，可关联最近 assistant 精确找到
+    arguments；这里只返回稳定类别，不把完整命令或路径带进聚集签名。
+    """
+    if not isinstance(prior_payload, dict):
+        return ""
+    calls = prior_payload.get("tool_calls") or []
+    candidates = []
+    for call in calls:
+        if not isinstance(call, dict):
+            continue
+        fn = call.get("function") or {}
+        if fn.get("name") != "shell":
+            continue
+        if tool_call_id and call.get("id") == tool_call_id:
+            candidates = [call]
+            break
+        candidates.append(call)
+    if len(candidates) != 1:
+        return ""
+    raw = (candidates[0].get("function") or {}).get("arguments") or {}
+    try:
+        args = json.loads(raw) if isinstance(raw, str) else raw
+    except (TypeError, ValueError):
+        return ""
+    command = str((args or {}).get("command") or "").strip().lower()
+    if not command:
+        return ""
+    if re.search(r"\b(?:apt|dpkg|pip)\b.*\b(?:list|show)\b", command):
+        return "dependency probe"
+    if re.search(r"(?:^|[;&|]\s*)(?:which|whereis|command\s+-v)\b", command):
+        return "dependency probe"
+    if re.search(r"(?:^|[;&|]\s*)(?:grep|rg)\b", command):
+        return "search/no match"
+    m = re.search(r"(?:^|&&|;|\|)\s*([a-z0-9_.-]+)", command)
+    return m.group(1) if m else ""
+
+
+def _failure_category(kind: str, sig_line: str, sample: str) -> str:
+    """区分产品/内容质量门与平台工具故障；返回值作为 API 加法字段。"""
+    if kind != "exit":
+        return "failure"
+    if sig_line.startswith("[GATE FAIL]"):
+        return "quality_gate"
+    if (
+        ("[篇幅核算]" in sample or "[字数核算]" in sample)
+        and re.search(r"\[WARN\]\s*\d+\s*项超出\s*/\s*\d+\s*项不足", sample)
+    ):
+        return "quality_gate"
+    if "[质量检查]" in sample and "[WARN] 共发现" in sample:
+        return "quality_gate"
+    return "failure"
+
+
 def scan_tool_failures(
     days: float = 7,
     min_count: int = 5,
@@ -110,8 +170,16 @@ def scan_tool_failures(
             text(
                 "select m.task_id, t.user_id, m.created_at, "
                 "       m.payload->>'name' as tool_name, "
-                "       m.payload->>'content' as content "
+                "       m.payload->>'content' as content, "
+                "       m.payload->>'tool_call_id' as tool_call_id, "
+                "       p.payload as prior_payload "
                 "from messages m join tasks t on t.task_id = m.task_id "
+                "left join lateral ("
+                "  select pm.payload from messages pm "
+                "  where pm.task_id = m.task_id and pm.idx < m.idx "
+                "    and pm.payload->>'role' = 'assistant' "
+                "  order by pm.idx desc limit 1"
+                ") p on true "
                 "where m.created_at >= :cutoff "
                 "  and m.payload->>'role' = 'tool' "
                 "  and (m.payload->>'content' like '[Error%' "
@@ -173,6 +241,8 @@ def scan_tool_failures(
                 "tool": key[0],
                 "signature": key[1],
                 "kind": kind,
+                # 分类必须看未截断的完整结果；sample 只保留 300 字给前端悬浮。
+                "category": _failure_category(kind, sig_line, sample),
                 "count": 0,
                 "tasks": set(),
                 "users": set(),
@@ -181,6 +251,8 @@ def scan_tool_failures(
                 "sample": sample[:300],
                 "daily": [0] * n_buckets,
             }
+        elif _failure_category(kind, sig_line, sample) == "quality_gate":
+            c["category"] = "quality_gate"
         c["count"] += 1
         c["tasks"].add(task_id)
         c["users"].add(user_id)
@@ -193,13 +265,21 @@ def scan_tool_failures(
             c["last_at"] = ts
             c["sample"] = sample[:300]
 
-    for task_id, user_id, created_at, tool_name, content in rows:
+    for row in rows:
+        # 兼容旧测试夹具的 5 列消息行；生产查询额外带 tool_call_id/prior_payload。
+        task_id, user_id, created_at, tool_name, content = row[:5]
+        tool_call_id = row[5] if len(row) > 5 else ""
+        prior_payload = row[6] if len(row) > 6 else None
         if not content:
             continue
         hit = _classify(content)
         if hit is None:
             continue
         kind, sig_line = hit
+        if tool_name == "shell" and kind == "exit" and _BARE_EXIT_RE.match(sig_line):
+            hint = _shell_command_hint(prior_payload, tool_call_id or "")
+            if hint:
+                sig_line = f"{sig_line} ({hint})"
         _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:
@@ -226,6 +306,7 @@ def scan_tool_failures(
             "tool": c["tool"],
             "signature": c["signature"],
             "kind": c["kind"],
+            "category": c["category"],
             "count": c["count"],
             "count_24h": c["daily"][-1],
             "daily": c["daily"],
@@ -240,6 +321,91 @@ def scan_tool_failures(
     return out
 
 
+def scan_tool_wire_health(days: int = 7) -> Dict[str, Any]:
+    """聚合 provider 工具参数损坏的抢救/残余比例；纯只读、无派生状态。
+
+    ``tool_salvaged`` 与 ``tool_malformed`` 是同一类 wire 损坏的两个结局，二者之和
+    才是观测分母。按 model_profile + tool 展示近 24h 与窗口总量，避免只看 malformed
+    绝对数时把调用量上涨误判为 provider 恶化。
+    """
+    days = min(90, max(1, int(days)))
+    now = datetime.now(timezone.utc)
+    cutoff = now - timedelta(days=days)
+    cutoff_24h = now - timedelta(hours=24)
+    with session_scope() as s:
+        rows = s.execute(
+            text(
+                "select model_profile, coalesce(units->>'tool', '?') as tool, "
+                f" count(*) filter (where kind = '{KIND_TOOL_SALVAGED}') as salvaged, "
+                f" count(*) filter (where kind = '{KIND_TOOL_MALFORMED}') as malformed, "
+                f" count(*) filter (where kind = '{KIND_TOOL_SALVAGED}' "
+                "   and created_at >= :cutoff_24h) as salvaged_24h, "
+                f" count(*) filter (where kind = '{KIND_TOOL_MALFORMED}' "
+                "   and created_at >= :cutoff_24h) as malformed_24h, "
+                " max(created_at) as last_at "
+                "from usage_events "
+                f"where kind in ('{KIND_TOOL_SALVAGED}', '{KIND_TOOL_MALFORMED}') "
+                "  and created_at >= :cutoff "
+                "group by model_profile, coalesce(units->>'tool', '?')"
+            ),
+            {"cutoff": cutoff, "cutoff_24h": cutoff_24h},
+        ).fetchall()
+
+    out = []
+    totals = {
+        "salvaged": 0,
+        "malformed": 0,
+        "salvaged_24h": 0,
+        "malformed_24h": 0,
+    }
+
+    def _rate(saved: int, residual: int) -> Optional[float]:
+        total = saved + residual
+        return round(saved / total * 100, 1) if total else None
+
+    for (
+        model_profile, tool, salvaged, malformed,
+        salvaged_24h, malformed_24h, last_at,
+    ) in rows:
+        saved = int(salvaged or 0)
+        residual = int(malformed or 0)
+        saved_24h = int(salvaged_24h or 0)
+        residual_24h = int(malformed_24h or 0)
+        for key, value in (
+            ("salvaged", saved),
+            ("malformed", residual),
+            ("salvaged_24h", saved_24h),
+            ("malformed_24h", residual_24h),
+        ):
+            totals[key] += value
+        out.append({
+            "model_profile": model_profile or "?",
+            "tool": tool or "?",
+            "salvaged": saved,
+            "malformed": residual,
+            "recovery_rate": _rate(saved, residual),
+            "salvaged_24h": saved_24h,
+            "malformed_24h": residual_24h,
+            "recovery_rate_24h": _rate(saved_24h, residual_24h),
+            "last_at": last_at.isoformat() if last_at else None,
+        })
+
+    out.sort(
+        key=lambda x: (
+            x["malformed_24h"],
+            x["salvaged_24h"] + x["malformed_24h"],
+            x["malformed"],
+            x["salvaged"] + x["malformed"],
+        ),
+        reverse=True,
+    )
+    totals["recovery_rate"] = _rate(totals["salvaged"], totals["malformed"])
+    totals["recovery_rate_24h"] = _rate(
+        totals["salvaged_24h"], totals["malformed_24h"]
+    )
+    return {"days": days, "rows": out, "total": totals}
+
+
 # ── provider 级致命错误即时告警 ──
 # 命中判据:错误文案含余额/配额/认证类关键词 —— 这类错误不是单任务偶发,而是该
 # provider 上所有 run 全挂(如 Zai 余额不足),等日巡检的 5 次/2 task 阈值太慢。

tests/test_loop_repeat_guard.py

@@ -164,6 +164,45 @@ class TestErrStreak(unittest.TestCase):
         self.assertEqual(n_args, 1)
         self.assertFalse(g.should_block_err("edit"))  # 单 arg 不走 err-block
 
+    def test_run_python_varied_args_same_traceback_blocks(self):
+        """生成代码虽每次不同，但反复撞同一 Python 运行期异常时应进入 err-streak。"""
+        g = _RepeatGuard()
+        errors = []
+        for line in (125, 126, 140, 141, 142, 143):
+            errors.append(
+                "[stderr]\nTraceback (most recent call last):\n"
+                f'  File "/workspace/x.py", line {line}, in <module>\n'
+                "    cell.value = label\n"
+                "AttributeError: 'MergedCell' object attribute 'value' is read-only\n"
+                "[exit 1]"
+            )
+        blocked = self._run_varargs(g, "run_python", errors)
+        self.assertTrue(blocked)
+
+    def test_run_python_success_resets_traceback_streak(self):
+        g = _RepeatGuard()
+        err = (
+            "[stderr]\nTraceback (most recent call last):\n"
+            '  File "/workspace/x.py", line 1, in <module>\n'
+            "ValueError: bad value\n[exit 1]"
+        )
+        g.record("run_python", {"code": "a"}, err)
+        g.record("run_python", {"code": "b"}, err)
+        g.record("run_python", {"code": "fixed"}, "[stdout]\nOK\n[exit 0]")
+        self.assertEqual(g.err_streak("run_python"), (0, 0, ""))
+
+    def test_shell_nonzero_exit_not_treated_as_err_streak(self):
+        """grep 未命中、质量门等普通 shell exit 1 仍可正常迭代。"""
+        g = _RepeatGuard()
+        for i in range(_RepeatGuard.HARD + 2):
+            g.record(
+                "shell",
+                {"command": f"grep needle file-{i}.md"},
+                "[exit 1]",
+            )
+        self.assertEqual(g.err_streak("shell"), (0, 0, ""))
+        self.assertFalse(g.should_block_err("shell"))
+
 
 def _resp_with_toolcalls(*calls):
     """calls: (name, arguments_str) → 构造带 tool_calls 的假 response。"""

tests/test_toolfail_malformed.py

@@ -99,6 +99,47 @@ class TestToolfailMalformed(unittest.TestCase):
         self.assertEqual(sum(c["daily"]), 2)
         self.assertEqual(c["daily"][-4], 2)  # 尾桶=近24h,往前退 3 桶
 
+    def test_quality_gate_is_additively_categorized(self):
+        """按设计拦截内容质量的 exit 1 单列，不冒充平台工具故障。"""
+        sample = (
+            "[质量检查] type=review\n" + ("检查明细\n" * 80)
+            + "[篇幅核算] type=review lang=zh\n"
+            "[WARN] 0 项超出 / 1 项不足 (含摘要/正文)。回头调整。\n[exit 1]"
+        )
+        rows = [
+            ("t1", "u1", NOW, "shell", sample),
+            ("t1", "u1", NOW, "shell", sample),
+        ]
+        out = _scan(rows, [], days=1, min_count=2, min_tasks=1)
+        self.assertEqual(out[0]["category"], "quality_gate")
+
+    def test_regular_failure_category_preserved(self):
+        rows = [
+            ("t1", "u1", NOW, "glob", "[Error] base path not found"),
+            ("t2", "u2", NOW, "glob", "[Error] base path not found"),
+        ]
+        out = _scan(rows, [], days=1, min_count=2, min_tasks=2)
+        self.assertEqual(out[0]["category"], "failure")
+
+    def test_bare_shell_exit_gets_command_hint(self):
+        """空输出 grep 未命中不再和其他 exit 1 混成无意义的 ``exit N``。"""
+        prior = {
+            "role": "assistant",
+            "tool_calls": [{
+                "id": "call-1",
+                "function": {
+                    "name": "shell",
+                    "arguments": '{"command":"cd /workspace/x && grep -n needle a.md"}',
+                },
+            }],
+        }
+        rows = [
+            ("t1", "u1", NOW, "shell", "[exit 1]", "call-1", prior),
+            ("t2", "u2", NOW, "shell", "[exit 1]", "call-1", prior),
+        ]
+        out = _scan(rows, [], days=1, min_count=2, min_tasks=2)
+        self.assertEqual(out[0]["signature"], "exit N (search/no match)")
+
 
 class TestToolfailRunError(unittest.TestCase):
     def test_run_error_rows_cluster(self):
@@ -143,6 +184,48 @@ class TestToolfailEmptyResponse(unittest.TestCase):
         self.assertEqual(out, [])
 
 
+class TestToolWireHealth(unittest.TestCase):
+    def test_aggregates_recovery_rates_and_totals(self):
+        rows = [
+            ("deepseek_v4.flash", "write", 10, 2, 4, 1, NOW),
+            ("deepseek_v4.flash", "edit", 3, 0, 0, 0, NOW - timedelta(hours=2)),
+        ]
+
+        class FakeWireSession:
+            def execute(self, clause, params=None):
+                return SimpleNamespace(fetchall=lambda: rows)
+
+        @contextmanager
+        def fake_scope():
+            yield FakeWireSession()
+
+        with patch.object(tf, "session_scope", fake_scope):
+            out = tf.scan_tool_wire_health(days=7)
+
+        self.assertEqual(out["days"], 7)
+        self.assertEqual(out["rows"][0]["tool"], "write")  # 24h 残余活跃在前
+        self.assertEqual(out["rows"][0]["recovery_rate"], 83.3)
+        self.assertEqual(out["rows"][0]["recovery_rate_24h"], 80.0)
+        self.assertEqual(out["total"]["salvaged"], 13)
+        self.assertEqual(out["total"]["malformed"], 2)
+        self.assertEqual(out["total"]["recovery_rate"], 86.7)
+
+    def test_empty_window_has_null_rate(self):
+        class FakeWireSession:
+            def execute(self, clause, params=None):
+                return SimpleNamespace(fetchall=lambda: [])
+
+        @contextmanager
+        def fake_scope():
+            yield FakeWireSession()
+
+        with patch.object(tf, "session_scope", fake_scope):
+            out = tf.scan_tool_wire_health(days=7)
+        self.assertEqual(out["rows"], [])
+        self.assertIsNone(out["total"]["recovery_rate"])
+        self.assertIsNone(out["total"]["recovery_rate_24h"])
+
+
 class TestProviderCriticalAlert(unittest.TestCase):
     def setUp(self):
         tf._alerted_at.clear()

tests/test_web_routes_nodb.py

@@ -100,6 +100,7 @@ class AuthGateTests(unittest.TestCase):
         ("POST", "/v1/tasks"),
         ("POST", "/v1/asr/transcribe"),
         ("GET", "/v1/admin/overview"),
+        ("GET", "/v1/admin/tool-wire-health"),
     ]
 
     def test_protected_endpoints_401_without_token(self):

web/admin.py

@@ -256,6 +256,14 @@ def register_admin_routes(app: FastAPI, require_admin) -> None:
             ),
         }
 
+    @app.get("/v1/admin/tool-wire-health", tags=["admin"])
+    def admin_tool_wire_health(
+        days: int = 7, user_id: UUID = Depends(require_admin),
+    ):
+        """工具参数 wire 损坏的抢救率，按模型档+工具聚合；admin-only、纯只读。"""
+        from core.toolfail import scan_tool_wire_health
+        return scan_tool_wire_health(days=days)
+
     @app.get("/v1/admin/tiers", tags=["admin"])
     def admin_tiers(user_id: UUID = Depends(require_admin)):
         """模型档位定义 + 全模型目录。admin-only。

web/static/js/admin.js

@@ -299,9 +299,8 @@ function trendBar(daily) {
   return daily.map(v => v === 0 ? "·"
     : blocks[Math.min(blocks.length - 1, Math.ceil(v / max * blocks.length) - 1)]).join("");
 }
-function renderToolFailures(d) {
+function toolFailureTable(title, rows, emptyText, quiet = false) {
-  const rows = d.clusters || [];
+  const body = rows.map(c => `<tr${quiet ? ` class="quiet"` : ""}>`
-  const body = rows.map(c => `<tr${c.count_24h ? "" : ` class="quiet"`}>`
     + `<td>${escapeHtml(c.tool)}</td>`
     + `<td>${escapeHtml(c.kind)}</td>`
     + `<td class="email" title="${escapeHtml(c.sample || "")}">${escapeHtml(c.signature)}</td>`
@@ -312,12 +311,75 @@ function renderToolFailures(d) {
     + `<td class="num">${c.user_count}</td>`
     + `<td>${c.last_at ? fmtTime(c.last_at) : "—"}</td>`
     + `</tr>`).join("")
-    || `<tr><td colspan="9" class="empty">近 ${d.days || 7} 天无失败聚集</td></tr>`;
+    || `<tr><td colspan="9" class="empty">${escapeHtml(emptyText)}</td></tr>`;
-  $("s-toolfail").innerHTML = `<div class="card"><h2>工具失败聚集（近 ${d.days || 7} 天，同签名 ≥3 次；近 24h 活跃在前，灰行 = 已安静）</h2>`
+  return `<div class="card"><h2>${escapeHtml(title)}</h2>`
     + `<div class="scroll-x"><table>`
     + `<thead><tr><th>工具</th><th>类型</th><th>签名（悬浮看样例）</th><th>次数</th><th>近24h</th><th>趋势</th><th>任务数</th><th>用户数</th><th>最近</th></tr></thead>`
     + `<tbody>${body}</tbody></table></div></div>`;
 }
+function rateText(v) {
+  return v == null ? "—" : `${Number(v).toFixed(1)}%`;
+}
+function wireHealthHTML(d) {
+  if (!d || !Array.isArray(d.rows)) {
+    return `<div class="card"><h2>工具调用链路健康</h2>`
+      + `<div class="empty">链路健康数据暂不可用，失败聚集仍可正常查看</div></div>`;
+  }
+  const rows = d.rows || [];
+  const total = d.total || {};
+  const body = rows.map(r => `<tr>`
+    + `<td>${escapeHtml(r.model_profile || "?")}</td>`
+    + `<td>${escapeHtml(r.tool || "?")}</td>`
+    + `<td class="num">${r.salvaged_24h || 0}</td>`
+    + `<td class="num${r.malformed_24h ? " hot" : ""}">${r.malformed_24h || 0}</td>`
+    + `<td class="num">${rateText(r.recovery_rate_24h)}</td>`
+    + `<td class="num">${r.salvaged || 0}</td>`
+    + `<td class="num${r.malformed ? " hot" : ""}">${r.malformed || 0}</td>`
+    + `<td class="num">${rateText(r.recovery_rate)}</td>`
+    + `<td>${r.last_at ? fmtTime(r.last_at) : "—"}</td>`
+    + `</tr>`).join("")
+    || `<tr><td colspan="9" class="empty">近 ${d.days || 7} 天无工具参数损坏事件</td></tr>`;
+  const summary = `近24h 抢救 ${total.salvaged_24h || 0} / 残余 ${total.malformed_24h || 0}`
+    + ` / 抢救率 ${rateText(total.recovery_rate_24h)}；`
+    + `近${d.days || 7}天 抢救 ${total.salvaged || 0} / 残余 ${total.malformed || 0}`
+    + ` / 抢救率 ${rateText(total.recovery_rate)}`;
+  return `<div class="card"><h2>工具调用链路健康（${escapeHtml(summary)}）</h2>`
+    + `<div class="scroll-x"><table>`
+    + `<thead><tr><th>模型档</th><th>工具</th><th>24h抢救</th><th>24h残余</th>`
+    + `<th>24h抢救率</th><th>窗口抢救</th><th>窗口残余</th><th>窗口抢救率</th><th>最近</th></tr></thead>`
+    + `<tbody>${body}</tbody></table></div></div>`;
+}
+function renderToolFailures(d, wire = null) {
+  const rows = d.clusters || [];
+  const active = rows.filter(c => (c.count_24h || 0) > 0);
+  const systemic = active.filter(c => c.category !== "quality_gate" && c.task_count >= 2);
+  const taskLocal = active.filter(c => c.category !== "quality_gate" && c.task_count < 2);
+  const gates = active.filter(c => c.category === "quality_gate");
+  const quiet = rows.filter(c => !(c.count_24h || 0));
+  const days = d.days || 7;
+  $("s-toolfail").innerHTML = wireHealthHTML(wire)
+    + toolFailureTable(
+      "当前系统性工具故障（近 24h 活跃、跨 ≥2 个任务）",
+      systemic,
+      "近 24 小时无跨任务系统性工具故障",
+    )
+    + toolFailureTable(
+      "单任务反复失败（近 24h）",
+      taskLocal,
+      "近 24 小时无单任务反复失败",
+    )
+    + toolFailureTable(
+      "质量门记录（近 24h，按设计拦截不合规产物）",
+      gates,
+      "近 24 小时无质量门拦截记录",
+    )
+    + toolFailureTable(
+      `已安静历史（近 ${days} 天窗口，近 24h 为 0）`,
+      quiet,
+      `近 ${days} 天无已安静聚集`,
+      true,
+    );
+}
 
 function pagerHTML(prefix, page, maxPage, from, to, total) {
   return `<div class="pager">`
@@ -471,8 +533,15 @@ async function loadStorage(page) {
 
 async function loadToolFailures() {
   try {
-    const d = await apiGet("/v1/admin/tool-failures?days=7&min_count=3&min_tasks=1");
+    const [failures, wire] = await Promise.allSettled([
-    renderToolFailures(d);
+      apiGet("/v1/admin/tool-failures?days=7&min_count=3&min_tasks=1"),
+      apiGet("/v1/admin/tool-wire-health?days=7"),
+    ]);
+    if (failures.status !== "fulfilled") throw failures.reason;
+    renderToolFailures(
+      failures.value,
+      wire.status === "fulfilled" ? wire.value : null,
+    );
   } catch (e) { /* 同上 */ }
 }