From 2c6091db03063f78c2213a859ca021433cbd5f5d Mon Sep 17 00:00:00 2001 From: caoqianming Date: Thu, 3 Sep 2026 10:06:37 +0800 Subject: [PATCH] feat(observability): record structured health events --- CHANGELOG.md | 2 +- DESIGN.md | 2 +- PROGRESS.md | 6 +- RUN.md | 6 +- core/context_fold.py | 11 +- core/loop.py | 82 +++++- core/storage/__init__.py | 10 + core/storage/models.py | 16 ++ core/storage/telemetry.py | 131 ++++++++- core/tool_failure.py | 135 +++++++++ core/toolfail.py | 258 +++++------------- .../20260903_1400_0040_tool_health_indexes.py | 41 +++ tests/test_context_fold.py | 12 +- tests/test_health_telemetry.py | 104 +++++++ tests/test_loop_persisted_turn.py | 45 ++- tests/test_static_vendor.py | 3 + tests/test_storage_migration.py | 21 ++ tests/test_toolfail_malformed.py | 86 ++++-- web/static/js/admin.js | 20 +- 19 files changed, 760 insertions(+), 231 deletions(-) create mode 100644 core/tool_failure.py create mode 100644 db/migrations/versions/20260903_1400_0040_tool_health_indexes.py create mode 100644 tests/test_health_telemetry.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b2ecaeb..c2ea0fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ ## Unreleased -- 管理后台会异步汇总工具健康大数，并避免短时间内重复请求；多人同时查看或自动刷新时不再反复扫描历史记录。 +- 管理后台会异步汇总工具健康大数，并避免短时间内重复请求；工具异常统一按结构化事件统计，代理主动保护与真正故障分开展示，多人同时查看或自动刷新时不再反复扫描历史消息。 - 单用户可同时运行的重型任务由 2 个提升到 3 个；任务等待执行容量时，对话会直接说明是当前用户、整机或宿主内存限制，获得槽位后自动继续。 diff --git a/DESIGN.md b/DESIGN.md index b6b32cd..bbef21c 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -331,7 +331,7 @@ 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 或重试策略。 +**工具健康事件与信号分层(2026-09-03)**:`usage_events` 是工具健康的唯一事实源，不再从 `messages.payload` 回扫或补算；普通 tool result 在落消息后仅当分类为失败时写 `tool_failure`（质量门写 `quality_gate`），并在写入时完成稳定签名、shell 命令类别和类别分类。切换前仅存在于 messages 的普通失败历史有意不迁移、不纳入新口径。畸形参数、抢救成功、run 终态错误沿用 `tool_malformed/tool_salvaged/run_error/empty_response`；主动停止、重复调用硬拦、上下文折叠失败分别记 `run_stopped/agent_guard/context_fold_failure`。这些控制事件统一 `category=agent_control`，与 `category=failure|quality_gate` 分栏，管理页“工具健康”大数只计真实失败，避免把保护机制触发误报为平台故障。7 天窗口继续保留取证，默认判断对象是近 24h 活跃信号；单 task、跨 task、质量门、控制事件和已安静历史分别展示。provider wire 健康仍按 `model_profile+tool` 对 `tool_salvaged/tool_malformed` 计算 24h/窗口抢救率，不混入失败数。读侧合并为一次带 `(kind, created_at)` 部分索引的事件查询；Admin 首次进入即加载，之后 90 秒节流并由服务端短时单飞缓存兜底。 ### 8.5 定时任务(✅ 2026-06-18) diff --git a/PROGRESS.md b/PROGRESS.md index 9bef23d..3999e3a 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -2,7 +2,7 @@ > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。 -最后更新:2026-09-03（Unreleased：工具健康按需刷新与短时缓存） +最后更新:2026-09-03（Unreleased：工具健康统一事件口径） --- @@ -20,7 +20,9 @@ --- ## 已完成关键能力 -- **09-03 / Unreleased / 工具健康监控请求收敛**：Admin 首次进入时异步汇总工具健康大数，之后固定按 90 秒节流刷新并对进行中请求单飞，不再每 10 秒重复扫描近 7 天记录。服务端按查询参数增加 90 秒进程内单飞缓存，多页面或多管理员并发时共享一次扫描，日巡检继续直读不受缓存影响。未新增表、字段、migration 或外部配置，结构化普通工具失败事件因双写会与 messages 历史口径重复而留待独立迁移。 +- **09-03 / Unreleased / 工具健康统一写入 usage_events**：普通工具失败在 tool result 落消息后结构化写 `tool_failure`，并新增 `run_stopped`、`agent_guard`、`context_fold_failure`、`quality_gate` 四类任务异常事件；工具健康聚合彻底移除 messages JSONB 回扫，只用一次 usage_events 查询，切换前的普通失败历史不迁移。Admin 大数只计真实 failure，代理主动停止/重复保护/上下文异常改为独立控制区，质量门继续单列。新增 0040 migration，为健康 kind 时间窗和普通失败 message 去重补部分索引。 + +- **09-03 / Unreleased / 工具健康监控请求收敛**：Admin 首次进入时异步汇总工具健康大数，之后固定按 90 秒节流刷新并对进行中请求单飞，不再每 10 秒重复扫描近 7 天记录。服务端按查询参数增加 90 秒进程内单飞缓存，多页面或多管理员并发时共享一次扫描，日巡检继续直读不受缓存影响。本提交本身未新增表、字段、migration；结构化事件口径由后续同日改动完成。 - **09-03 / Unreleased / 管理后台紧凑总览与分组工作区**：Admin 由纵向长页改为顶部常驻的两级总览和“运行、用量、用户、资源、集成、质量”六个页签，四项运行状态使用主卡、四项运营指标收进紧凑指标栏；点击总览可切换页签并精确定位同口径详情，任务、容量、用户活跃、Token/成本和存储均补齐详情指标。页签 hash 支持刷新与前进后退，桌面端页签吸顶，移动端顶栏操作收进菜单、运营指标与页签可横向滑动；1440px/390px 浏览器渲染、六页签与深链交互检查通过，无 API、schema、migration 或运行方式变化。 diff --git a/RUN.md b/RUN.md index b14e923..6acda9c 100644 --- a/RUN.md +++ b/RUN.md @@ -74,13 +74,13 @@ # SMTP_PASSWORD=<授权码/应用专用密码,非登录密码> # SMTP_FROM=you@qq.com # 可选,默认取 SMTP_USER # SMTP_FROM_NAME=总院科研辅助智能体 # 可选,发件人显示名,默"总院科研辅助智能体"(不暴露内部代号) - # 工具失败聚集巡检(0.58.11,反 9dcae061"挂 90 天无人知"):web 进程每天扫近 7 天 - # tool 错误消息 + 被丢弃的畸形 tool_call 参数(0.58.19,kind=malformed),同签名 >=5 次 + # 工具失败聚集巡检(0.58.11,2026-09-03 改为只扫 usage_events):web 进程每天扫近 7 天 + # 结构化工具/运行异常事件（含畸形 tool_call 参数）,同签名 >=5 次 # 且跨 >=2 task 判聚集 → 发下面邮箱(复用上面 SMTP_*;未配邮箱或 SMTP 则只打日志 # [toolfail] 行)。**只发近 24h 仍活跃的聚集**(0.58.19):去重集是内存态、每次部署清零, # 不加活跃过滤的话高频部署期每次重启都会把 7 天窗口内早已安静的存量聚集重发一遍。 # admin 页仍取低阈值全量,分区展示:近24h跨任务系统性故障 / 单任务反复失败 / - # 按设计拦截的质量门 / 已安静历史；修复后看对应项进入“已安静”。同区的“工具 + # 代理控制与上下文异常 / 按设计拦截的质量门 / 已安静历史；修复后看对应项进入“已安静”。同区的“工具 # 调用链路健康”按模型档+工具展示 tool_salvaged/tool_malformed 的24h/7d抢救率。 # ZCBOT_DEVELOPER_EMAIL=dev@example.com # ZCBOT_TOOLFAIL_SCAN_INTERVAL=86400 # 秒,默 86400;<=0 关掉巡检 diff --git a/core/context_fold.py b/core/context_fold.py index 6b5b75a..40aa574 100644 --- a/core/context_fold.py +++ b/core/context_fold.py @@ -35,7 +35,7 @@ from .context import ( prepare_messages_with_stats, ) from .llm_transport import extract_usage_details -from .storage import session_scope +from .storage import record_context_fold_failure, session_scope from .storage.models import Task from .storage.usage import record_chat_usage @@ -180,6 +180,15 @@ def maybe_fold( choices = getattr(response, "choices", None) or [] summary = ((choices[0].message.content if choices else "") or "").strip() if not summary: + try: + record_context_fold_failure( + task_id=session.task_id, + user_id=user_id, + model_profile=target_profile, + reason="empty_summary", + ) + except Exception: + pass emit({"type": "warn", "msg": "context fold: 摘要调用返回空,本轮跳过折叠"}) return None diff --git a/core/loop.py b/core/loop.py index 7a21a4a..48e4ae7 100644 --- a/core/loop.py +++ b/core/loop.py @@ -46,8 +46,13 @@ from .llm_transport import ( from .salvage import salvage_tool_arguments from .session import Session from .storage import ( + record_agent_guard, record_chat_usage, + record_context_fold_failure, + record_quality_gate, + record_run_stopped, record_salvaged_tool_call, + record_tool_failure, ) from .task_actions import DeferredTaskActions @@ -384,7 +389,7 @@ class AgentLoop: result, productive, artifacts = self._execute_tool_call(tc) self._remember_artifacts(artifacts) step_productive = step_productive or productive - self.session.append( + message_id = self.session.append( { "role": "tool", "tool_call_id": tc.id, @@ -392,6 +397,20 @@ class AgentLoop: "content": result, } ) + if message_id is not None: + try: + record_tool_failure( + task_id=self.session.task_id, + user_id=self.user_id, + message_id=message_id, + model_profile=model_profile_of(self.caps), + tool=tc.function.name, + content=result, + arguments=tc.function.arguments, + ) + except Exception: + # 可观测性留痕失败不能打断主对话。 + pass # ask_user:本步调用了人工选择工具 → 提前结束本轮,等用户点选项 / 文字讨论, # 不回灌 LLM。选项已随该 tool_call 的 arguments 流给前端渲染成选项卡;tool 结果 @@ -407,6 +426,15 @@ class AgentLoop: else: self._stall += 1 if self._stall >= self._STALL_LIMIT: + try: + record_run_stopped( + task_id=self.session.task_id, + user_id=self.user_id, + model_profile=model_profile_of(self.caps), + reason="no_progress", + ) + except Exception: + pass self._emit({ "type": "warn", "msg": ( @@ -418,6 +446,15 @@ class AgentLoop: return "[stopped: no progress]" # 跑满 backstop:不是出错,是单轮自主步数到顶。明确提示可续跑,别静默停。 + try: + record_run_stopped( + task_id=self.session.task_id, + user_id=self.user_id, + model_profile=model_profile_of(self.caps), + reason="max_iterations", + ) + except Exception: + pass self._emit({ "type": "warn", "msg": ( @@ -497,6 +534,16 @@ class AgentLoop: user_id=self.user_id, emit=self._emit, ) except Exception as e: + try: + record_context_fold_failure( + task_id=self.session.task_id, + user_id=self.user_id, + model_profile=model_profile_of(self.caps), + reason="exception", + detail=f"{type(e).__name__}: {e}", + ) + except Exception: + pass self._emit({ "type": "warn", "msg": f"context fold failed: {type(e).__name__}: {e};本轮跳过折叠", @@ -842,6 +889,17 @@ class AgentLoop: """ if self._repeat_guard.should_block(name, args): n, _blocked = self._repeat_guard.register_block(name, args) + try: + record_agent_guard( + task_id=self.session.task_id, + user_id=self.user_id, + model_profile=model_profile_of(self.caps), + tool=name, + guard="same_arguments", + count=n, + ) + except Exception: + pass result = ( f"[已拦截重复调用] {name} 用完全相同的参数已调用 {n} 次且结果始终未变,本次未执行。" "这通常意味着思路卡死:① 换不同的参数或方法;② 读一下相关文件/报错重新定位;" @@ -853,6 +911,18 @@ class AgentLoop: if self._repeat_guard.should_block_err(name): cnt, esig = self._repeat_guard.register_err_block(name) + try: + record_agent_guard( + task_id=self.session.task_id, + user_id=self.user_id, + model_profile=model_profile_of(self.caps), + tool=name, + guard="same_error", + count=cnt, + signature=esig, + ) + except Exception: + pass result = ( f"[已拦截重复调用] {name} 已连续 {cnt} 次撞同一个错误「{esig}」(每次只微调了参数)。" "再这么试下去不会有新结果。换个做法:① 先 read 目标文件/用 grep 看确切内容" @@ -945,6 +1015,16 @@ class AgentLoop: except Exception: guard_msg = None # 机检自身故障绝不拖垮工具链路 if guard_msg: + try: + record_quality_gate( + task_id=self.session.task_id, + user_id=self.user_id, + model_profile=model_profile_of(self.caps), + tool=name, + gate="pptx_full_page_image", + ) + except Exception: + pass result += "\n\n" + guard_msg self._emit({ "type": "warn", diff --git a/core/storage/__init__.py b/core/storage/__init__.py index 813b43a..72cc761 100644 --- a/core/storage/__init__.py +++ b/core/storage/__init__.py @@ -14,9 +14,14 @@ from .engine import ( ) from .telemetry import ( record_empty_response, + record_agent_guard, + record_context_fold_failure, record_malformed_tool_call, + record_quality_gate, record_run_error, + record_run_stopped, record_salvaged_tool_call, + record_tool_failure, ) from .usage import record_chat_usage from .utils import ( @@ -35,10 +40,15 @@ __all__ = [ "get_engine", "get_task", "record_chat_usage", + "record_agent_guard", + "record_context_fold_failure", "record_empty_response", "record_malformed_tool_call", + "record_quality_gate", "record_run_error", + "record_run_stopped", "record_salvaged_tool_call", + "record_tool_failure", "session_scope", "update_task", "upsert_task", diff --git a/core/storage/models.py b/core/storage/models.py index 0f1642a..3172197 100644 --- a/core/storage/models.py +++ b/core/storage/models.py @@ -366,6 +366,22 @@ class UsageEvent(Base): __tablename__ = "usage_events" __table_args__ = ( Index("ix_usage_created_brin", "created_at", postgresql_using="brin"), + Index( + "ix_usage_health_kind_created", + "kind", + "created_at", + postgresql_where=text( + "kind IN ('tool_failure', 'tool_malformed', 'tool_salvaged', " + "'run_error', 'empty_response', 'run_stopped', 'agent_guard', " + "'context_fold_failure', 'quality_gate')" + ), + ), + Index( + "ux_usage_tool_failure_message", + "message_id", + unique=True, + postgresql_where=text("kind IN ('tool_failure', 'quality_gate')"), + ), ) event_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True, default=uuid4) diff --git a/core/storage/telemetry.py b/core/storage/telemetry.py index 8a98d2b..8696e11 100644 --- a/core/storage/telemetry.py +++ b/core/storage/telemetry.py @@ -1,6 +1,6 @@ """失败埋点(telemetry)—— 与计费(usage.py)分家的另一类 usage_events 写入。 -judged by kind:这里的四类 cost 恒 0、tokens 不入 kind=chat 汇总,语义是**可观测性 +这里的异常类 kind 均 cost 恒 0、tokens 不入 kind=chat 汇总,语义是**可观测性 留痕**而非用户花销 —— 它们是 admin「工具失败聚集」面板与巡检邮件(core/toolfail.py) 的唯一持久数据源。此前与真计费(chat/image/video/vision)混住 usage.py,污染 「usage=计费」的心智模型(架构审查数据层 Top4),2026-07-23 拆出。 @@ -11,6 +11,7 @@ kind 常量是 loop/llm_transport(写)与 toolfail(读 SQL)之间的契约单一 from __future__ import annotations from decimal import Decimal +from typing import Any, Optional from uuid import UUID from .engine import session_scope @@ -19,8 +20,136 @@ from .models import UsageEvent # ── kind 契约常量(写侧本模块 / 读侧 core/toolfail.py 共用)── KIND_TOOL_MALFORMED = "tool_malformed" KIND_TOOL_SALVAGED = "tool_salvaged" +KIND_TOOL_FAILURE = "tool_failure" KIND_EMPTY_RESPONSE = "empty_response" KIND_RUN_ERROR = "run_error" +KIND_RUN_STOPPED = "run_stopped" +KIND_AGENT_GUARD = "agent_guard" +KIND_CONTEXT_FOLD_FAILURE = "context_fold_failure" +KIND_QUALITY_GATE = "quality_gate" + + +def _record_health_event( + *, + kind: str, + task_id: UUID, + user_id: UUID, + model_profile: str, + units: dict[str, Any], + message_id: Optional[UUID] = None, +) -> None: + with session_scope() as s: + s.add(UsageEvent( + user_id=user_id, + task_id=task_id, + message_id=message_id, + kind=kind, + model_profile=model_profile, + units=units, + cost_cny=Decimal("0"), + )) + + +def record_tool_failure( + *, + task_id: UUID, + user_id: UUID, + message_id: UUID, + model_profile: str, + tool: str, + content: str, + arguments=None, +) -> bool: + """结构化记录普通工具失败；正常结果不写入并返回 False。""" + from core.tool_failure import structured_failure + + units = structured_failure(tool, content, arguments=arguments) + if units is None: + return False + event_kind = KIND_QUALITY_GATE if units["category"] == "quality_gate" else KIND_TOOL_FAILURE + _record_health_event( + kind=event_kind, + task_id=task_id, + user_id=user_id, + message_id=message_id, + model_profile=model_profile, + units=units, + ) + return True + + +def record_run_stopped( + *, task_id: UUID, user_id: UUID, model_profile: str, reason: str, +) -> None: + _record_health_event( + kind=KIND_RUN_STOPPED, + task_id=task_id, + user_id=user_id, + model_profile=model_profile, + units={ + "tool": "(run)", "failure_kind": "stopped", "signature": reason, + "category": "agent_control", "sample": reason, + }, + ) + + +def record_agent_guard( + *, + task_id: UUID, + user_id: UUID, + model_profile: str, + tool: str, + guard: str, + count: int, + signature: str = "", +) -> None: + from core.tool_failure import normalize_failure_signature + + stable = normalize_failure_signature(f"{guard}: {signature}" if signature else guard) + _record_health_event( + kind=KIND_AGENT_GUARD, + task_id=task_id, + user_id=user_id, + model_profile=model_profile, + units={ + "tool": tool, "failure_kind": "guard", "signature": stable, + "category": "agent_control", "sample": f"{guard}, count={count}", + "count": int(count), + }, + ) + + +def record_context_fold_failure( + *, task_id: UUID, user_id: UUID, model_profile: str, reason: str, detail: str = "", +) -> None: + from core.tool_failure import normalize_failure_signature + + _record_health_event( + kind=KIND_CONTEXT_FOLD_FAILURE, + task_id=task_id, + user_id=user_id, + model_profile=model_profile, + units={ + "tool": "(context)", "failure_kind": "context", "signature": reason, + "category": "agent_control", + "sample": normalize_failure_signature(detail) if detail else reason, + }, + ) + + +def record_quality_gate( + *, task_id: UUID, user_id: UUID, model_profile: str, tool: str, gate: str, +) -> None: + _record_health_event( + kind=KIND_QUALITY_GATE, + task_id=task_id, + user_id=user_id, + model_profile=model_profile, + units={ + "tool": tool, "failure_kind": "gate", "signature": gate, + "category": "quality_gate", "sample": gate, + }, + ) def record_malformed_tool_call( diff --git a/core/tool_failure.py b/core/tool_failure.py new file mode 100644 index 0000000..3c14507 --- /dev/null +++ b/core/tool_failure.py @@ -0,0 +1,135 @@ +"""工具结果失败分类的单一事实源。""" +from __future__ import annotations + +import json +import re +from typing import Any, Optional, Tuple + + +_RE_UUID = re.compile( + r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-" + r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}" +) +_RE_HEX = re.compile(r"0x[0-9a-fA-F]+") +_RE_PATH = re.compile(r"(?:[A-Za-z]:)?(?:[/\\][\w.\-一-鿿*]+){2,}") +_RE_NUM = re.compile(r"\d+") +_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_failure_signature(value: str) -> str: + value = _RE_UUID.sub("", value) + value = _RE_HEX.sub("", value) + value = _RE_PATH.sub("", value) + value = _RE_NUM.sub("N", value) + return _RE_WS.sub(" ", value).strip()[:120] + + +def classify_failure(content: str) -> Optional[Tuple[str, str]]: + """返回 ``(kind, 原始签名行)``；正常工具结果返回 None。""" + head = content.lstrip() + if "command timed out" in content: + return "timeout", "command timed out" + if head.startswith("[Error"): + return "error", head.splitlines()[0] + match = _EXIT_TAIL.search(content) + if match and match.group(1) != "0": + lines = [ + line.strip() + for line in content.splitlines()[:-1] + if line.strip() and line.strip() not in _STREAM_MARKS + ] + return "exit", (lines[-1] if lines else f"exit {match.group(1)}") + return None + + +def is_bare_exit_signature(value: str) -> bool: + return _BARE_EXIT_RE.match(value) is not None + + +def _command_hint(command: str) -> str: + command = command.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" + match = re.search(r"(?:^|&&|;|\|)\s*([a-z0-9_.-]+)", command) + return match.group(1) if match else "" + + +def shell_command_hint(prior_payload: Any, tool_call_id: str) -> str: + """从历史 assistant tool_call 中提取稳定的 shell 命令类别。""" + 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 + function = call.get("function") or {} + if function.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 "" + return shell_command_hint_from_arguments( + (candidates[0].get("function") or {}).get("arguments") + ) + + +def shell_command_hint_from_arguments(arguments: Any) -> str: + try: + args = json.loads(arguments) if isinstance(arguments, str) else arguments + except (TypeError, ValueError): + return "" + if not isinstance(args, dict): + return "" + return _command_hint(str(args.get("command") or "")) + + +def failure_category(kind: str, signature_line: str, sample: str) -> str: + if kind != "exit": + return "failure" + if signature_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 structured_failure( + tool: str, + content: str, + *, + arguments: Any = None, +) -> Optional[dict[str, str]]: + """把工具结果转成可直接持久化的稳定失败事件。""" + hit = classify_failure(content) + if hit is None: + return None + kind, signature_line = hit + if tool == "shell" and kind == "exit" and _BARE_EXIT_RE.match(signature_line): + hint = shell_command_hint_from_arguments(arguments) + if hint: + signature_line = f"{signature_line} ({hint})" + return { + "tool": tool or "?", + "failure_kind": kind, + "signature": normalize_failure_signature(signature_line), + "category": failure_category(kind, signature_line, content), + "sample": content[:300], + } diff --git a/core/toolfail.py b/core/toolfail.py index 4140079..724f130 100644 --- a/core/toolfail.py +++ b/core/toolfail.py @@ -2,11 +2,12 @@ 背景(2026-07,task 9dcae061 终案的结构性教训):mermaid 渲染在生产挂了 90 天 0 成功(67 次超时 + 26 次 launch fail、烧掉数十万 token),没有任何机制发现, -靠人工扫 DB 才挖出来。本模块把「失败聚集」变成信号:扫 messages 里 role=tool -的错误结果,按 (工具名 + 归一化错误签名) 聚合,超阈值即算聚集。 +靠人工扫 DB 才挖出来。本模块把「失败聚集」变成信号:统一扫描 usage_events +中的结构化异常事件,按 (工具名 + 归一化错误签名) 聚合,超阈值即算聚集。 -纯只读查询、无新表无状态;告警通道由调用方决定(web/app.py 的巡检 loop 发 -开发者邮箱,admin API 直接返给前端表格)。 +普通工具失败从 2026-09-03 起在 tool result 落 messages 后同步写 tool_failure； +切换前只存在于 messages 的历史故障不再回扫。纯只读查询、无派生状态；告警通道 +由调用方决定(web/app.py 的巡检 loop 发开发者邮箱,admin API 直接返给前端表格)。 第二数据源(0.58.19):被丢弃的畸形 tool_call 参数(kind=malformed)——这类失败 整轮不入 messages(防投毒级联),loop 落 usage_events(kind=tool_malformed), @@ -31,7 +32,6 @@ tool_use 漏成正文后丢空)。loop 落 usage_events(kind=empty_response),在 """ from __future__ import annotations -import json import math import re import threading @@ -43,11 +43,20 @@ from sqlalchemy import text from core.storage import session_scope from core.storage.telemetry import ( + KIND_AGENT_GUARD, + KIND_CONTEXT_FOLD_FAILURE, KIND_EMPTY_RESPONSE, + KIND_QUALITY_GATE, KIND_RUN_ERROR, + KIND_RUN_STOPPED, + KIND_TOOL_FAILURE, KIND_TOOL_MALFORMED, KIND_TOOL_SALVAGED, ) +from core.tool_failure import ( + failure_category as _failure_category, + normalize_failure_signature as _normalize, +) _CACHE_TTL_SECONDS = 90.0 @@ -64,111 +73,12 @@ def _clear_tool_health_cache() -> None: with _wire_cache_lock: _wire_cache.clear() -# 签名归一:同一类错误在不同 task/参数下的差异(路径/数字/uuid/十六进制)抹平, -# 让 "figures/a.png doesn't exist" 和 "figures/b.png doesn't exist" 聚成一条。 -_RE_UUID = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") -_RE_HEX = re.compile(r"0x[0-9a-fA-F]+") -_RE_PATH = re.compile(r"(?:[A-Za-z]:)?(?:[/\\][\w.\-一-鿿*]+){2,}") -_RE_NUM = re.compile(r"\d+") -_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: - s = _RE_UUID.sub("", s) - s = _RE_HEX.sub("", s) - s = _RE_PATH.sub("", s) - s = _RE_NUM.sub("N", s) - s = _RE_WS.sub(" ", s).strip() - return s[:120] - - -def _classify(content: str) -> Optional[Tuple[str, str]]: - """返回 (kind, 原始签名行) 或 None(不算失败)。""" - head = content.lstrip() - # 超时判定在前:超时结果形如 "[Error] command timed out after 30s", - # 让它归 timeout 而不是被 [Error 前缀截走(kind 对排查方向有指示意义) - if "command timed out" in content: - return "timeout", "command timed out" - if head.startswith("[Error"): - return "error", head.splitlines()[0] - m = _EXIT_TAIL.search(content) - if m and m.group(1) != "0": - # 签名取 [exit N] 前最后一行有实际内容的输出(通常是真正的报错行) - lines = [ - ln.strip() for ln in content.splitlines()[:-1] - if ln.strip() and ln.strip() not in _STREAM_MARKS - ] - return "exit", (lines[-1] if lines else f"exit {m.group(1)}") - 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, min_tasks: int = 2, ) -> List[Dict[str, Any]]: - """扫近 `days` 天的 tool 错误消息,返回超阈值的聚集。 + """扫近 `days` 天的结构化异常事件,返回超阈值的聚集。 阈值语义:同签名 >= min_count 次 且 跨 >= min_tasks 个 task —— 单 task 内 模型试错几次就自愈的正常噪音不触发;跨 task 复现的才是平台性问题。 @@ -183,62 +93,21 @@ def scan_tool_failures( cutoff = now - timedelta(days=days) n_buckets = max(1, math.ceil(days)) with session_scope() as s: - rows = s.execute( + event_rows = s.execute( text( - "select m.task_id, t.user_id, m.created_at, " - " m.payload->>'name' as tool_name, " - " 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%' " - " or m.payload->>'content' like '%command timed out%' " - " or m.payload->>'content' like '%[exit %')" - ), - {"cutoff": cutoff}, - ).fetchall() - # 第二段:被丢弃的畸形 tool_call 参数(kind=tool_malformed)。这类失败整轮 - # 不入 messages(防投毒),llm_transport.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 " + "select kind, task_id, user_id, created_at, model_profile, " + " units->>'tool' as tool_name, units->>'err' as err, " + " units->>'head' as head, units->>'tail' as tail, " + " units->>'failure_kind' as failure_kind, " + " units->>'signature' as signature, " + " units->>'category' as category, " + " units->>'sample' as sample " "from usage_events " - f"where kind = '{KIND_TOOL_MALFORMED}' and created_at >= :cutoff" - ), - {"cutoff": cutoff}, - ).fetchall() - # 第三段:run 级终态错误(kind=run)。LLM 层抛异常时整轮无 tool 消息, - # tasks.run_error 只留最后一次,usage_events(kind=run_error)才是完整留痕。 - rrows = s.execute( - text( - "select task_id, user_id, created_at, " - " units->>'err' as err " - "from usage_events " - f"where kind = '{KIND_RUN_ERROR}' and created_at >= :cutoff" - ), - {"cutoff": cutoff}, - ).fetchall() - # 第四段:provider 吐空(kind=empty_response)。assistant 轮既无 tool_calls 又无正文, - # 会被 run loop 当正常收尾静默 done;loop 落 usage_events(kind=empty_response)是唯一 - # 留痕。tool 名固定 "(empty)",签名固定,sample=model_profile —— 看是哪个网关档在吐空。 - erows = s.execute( - text( - "select task_id, user_id, created_at, model_profile " - "from usage_events " - f"where kind = '{KIND_EMPTY_RESPONSE}' and created_at >= :cutoff" + f"where kind in ('{KIND_TOOL_FAILURE}', '{KIND_TOOL_MALFORMED}', " + f"'{KIND_RUN_ERROR}', '{KIND_EMPTY_RESPONSE}', '{KIND_RUN_STOPPED}', " + f"'{KIND_AGENT_GUARD}', '{KIND_CONTEXT_FOLD_FAILURE}', " + f"'{KIND_QUALITY_GATE}') " + " and created_at >= :cutoff" ), {"cutoff": cutoff}, ).fetchall() @@ -248,10 +117,15 @@ def scan_tool_failures( def _add( tool_name: str, kind: str, sig_line: str, sample: str, task_id: Any, user_id: Any, created_at: datetime, + *, category: Optional[str] = None, signature_normalized: bool = False, ) -> 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)) + key = ( + tool_name or "?", + sig_line if signature_normalized else _normalize(sig_line), + ) + resolved_category = category or _failure_category(kind, sig_line, sample) c = agg.get(key) if c is None: c = agg[key] = { @@ -259,7 +133,7 @@ def scan_tool_failures( "signature": key[1], "kind": kind, # 分类必须看未截断的完整结果；sample 只保留 300 字给前端悬浮。 - "category": _failure_category(kind, sig_line, sample), + "category": resolved_category, "count": 0, "tasks": set(), "users": set(), @@ -268,7 +142,7 @@ def scan_tool_failures( "sample": sample[:300], "daily": [0] * n_buckets, } - elif _failure_category(kind, sig_line, sample) == "quality_gate": + elif resolved_category == "quality_gate": c["category"] = "quality_gate" c["count"] += 1 c["tasks"].add(task_id) @@ -282,38 +156,34 @@ def scan_tool_failures( c["last_at"] = ts c["sample"] = sample[:300] - 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: - _add( - tool_name, "malformed", err or "?", - f"{head or ''} … {tail or ''}", - task_id, user_id, created_at, - ) - - for task_id, user_id, created_at, err in rrows: - _add("(run)", "run", err or "?", err or "", task_id, user_id, created_at) - - for task_id, user_id, created_at, model_profile in erows: - _add( - "(empty)", "empty", "provider returned empty response", - model_profile or "?", task_id, user_id, created_at, - ) + for ( + event_kind, task_id, user_id, created_at, model_profile, tool_name, + err, head, tail, failure_kind, signature, category, sample, + ) in event_rows: + if event_kind in { + KIND_TOOL_FAILURE, + KIND_RUN_STOPPED, + KIND_AGENT_GUARD, + KIND_CONTEXT_FOLD_FAILURE, + KIND_QUALITY_GATE, + }: + _add( + tool_name, failure_kind or "failure", signature or "?", sample or "", + task_id, user_id, created_at, + category=category or "failure", signature_normalized=True, + ) + elif event_kind == KIND_TOOL_MALFORMED: + _add( + tool_name, "malformed", err or "?", f"{head or ''} … {tail or ''}", + task_id, user_id, created_at, + ) + elif event_kind == KIND_RUN_ERROR: + _add("(run)", "run", err or "?", err or "", task_id, user_id, created_at) + elif event_kind == KIND_EMPTY_RESPONSE: + _add( + "(empty)", "empty", "provider returned empty response", + model_profile or "?", task_id, user_id, created_at, + ) out = [] for c in agg.values(): diff --git a/db/migrations/versions/20260903_1400_0040_tool_health_indexes.py b/db/migrations/versions/20260903_1400_0040_tool_health_indexes.py new file mode 100644 index 0000000..ed6d7cc --- /dev/null +++ b/db/migrations/versions/20260903_1400_0040_tool_health_indexes.py @@ -0,0 +1,41 @@ +"""Add indexes for structured tool health events. + +Revision ID: 0040 +Revises: 0039 +Create Date: 2026-09-03 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +revision: str = "0040" +down_revision: Union[str, None] = "0039" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_index( + "ix_usage_health_kind_created", + "usage_events", + ["kind", "created_at"], + postgresql_where=sa.text( + "kind IN ('tool_failure', 'tool_malformed', 'tool_salvaged', " + "'run_error', 'empty_response', 'run_stopped', 'agent_guard', " + "'context_fold_failure', 'quality_gate')" + ), + ) + op.create_index( + "ux_usage_tool_failure_message", + "usage_events", + ["message_id"], + unique=True, + postgresql_where=sa.text("kind IN ('tool_failure', 'quality_gate')"), + ) + + +def downgrade() -> None: + op.drop_index("ux_usage_tool_failure_message", table_name="usage_events") + op.drop_index("ix_usage_health_kind_created", table_name="usage_events") diff --git a/tests/test_context_fold.py b/tests/test_context_fold.py index 4d27486..8912a44 100644 --- a/tests/test_context_fold.py +++ b/tests/test_context_fold.py @@ -218,12 +218,20 @@ class MaybeFoldTests(unittest.TestCase): sess = self._session() before = list(sess.messages) events = [] - with patch.object(cf, "persist_fold") as persist: + user_id = uuid4() + with patch.object(cf, "persist_fold") as persist, \ + patch.object(cf, "record_context_fold_failure") as failure: result = cf.maybe_fold( - sess, _FakeLLM(" "), _FAKE_CAPS, user_id=uuid4(), emit=events.append + sess, _FakeLLM(" "), _FAKE_CAPS, user_id=user_id, emit=events.append ) self.assertIsNone(result) persist.assert_not_called() + failure.assert_called_once_with( + task_id=sess.task_id, + user_id=user_id, + model_profile="deepseek_v4.flash", + reason="empty_summary", + ) self.assertEqual(sess.messages, before) # 内存零污染 self.assertTrue(any(e.get("type") == "warn" for e in events)) diff --git a/tests/test_health_telemetry.py b/tests/test_health_telemetry.py new file mode 100644 index 0000000..3ce7400 --- /dev/null +++ b/tests/test_health_telemetry.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import sys +import unittest +from contextlib import contextmanager +from pathlib import Path +from uuid import uuid4 +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import core.storage.telemetry as telemetry # noqa: E402 + + +class _CaptureSession: + def __init__(self): + self.rows = [] + + def add(self, row): + self.rows.append(row) + + +class TestHealthTelemetry(unittest.TestCase): + def setUp(self): + self.session = _CaptureSession() + + @contextmanager + def fake_scope(): + yield self.session + + self.scope_patch = patch.object(telemetry, "session_scope", fake_scope) + self.scope_patch.start() + + def tearDown(self): + self.scope_patch.stop() + + def _ids(self): + return {"task_id": uuid4(), "user_id": uuid4(), "message_id": uuid4()} + + def test_normal_tool_result_does_not_write(self): + ids = self._ids() + written = telemetry.record_tool_failure( + **ids, model_profile="test.fast", tool="read", content="ok", + ) + self.assertFalse(written) + self.assertEqual(self.session.rows, []) + + def test_tool_failure_is_structured(self): + ids = self._ids() + written = telemetry.record_tool_failure( + **ids, + model_profile="test.fast", + tool="shell", + content="[exit 1]", + arguments={"command": "rg needle docs"}, + ) + self.assertTrue(written) + row = self.session.rows[0] + self.assertEqual(row.kind, telemetry.KIND_TOOL_FAILURE) + self.assertEqual(row.units["signature"], "exit N (search/no match)") + self.assertEqual(row.units["category"], "failure") + + def test_quality_gate_uses_own_kind(self): + ids = self._ids() + telemetry.record_tool_failure( + **ids, + model_profile="test.fast", + tool="shell", + content="[GATE FAIL] svg_quality_checker: overlap\n[exit 1]", + ) + self.assertEqual(self.session.rows[0].kind, telemetry.KIND_QUALITY_GATE) + self.assertEqual(self.session.rows[0].units["category"], "quality_gate") + + def test_control_events_are_structured(self): + ids = self._ids() + common = { + "task_id": ids["task_id"], "user_id": ids["user_id"], + "model_profile": "test.fast", + } + telemetry.record_run_stopped(**common, reason="max_iterations") + telemetry.record_agent_guard( + **common, tool="edit", guard="same_error", count=4, + signature="old_str not found", + ) + telemetry.record_context_fold_failure( + **common, reason="empty_summary", + ) + telemetry.record_quality_gate( + **common, tool="shell", gate="pptx_full_page_image", + ) + self.assertEqual( + [row.kind for row in self.session.rows], + [ + telemetry.KIND_RUN_STOPPED, + telemetry.KIND_AGENT_GUARD, + telemetry.KIND_CONTEXT_FOLD_FAILURE, + telemetry.KIND_QUALITY_GATE, + ], + ) + self.assertEqual(self.session.rows[0].units["category"], "agent_control") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_loop_persisted_turn.py b/tests/test_loop_persisted_turn.py index c100d31..e100e77 100644 --- a/tests/test_loop_persisted_turn.py +++ b/tests/test_loop_persisted_turn.py @@ -13,11 +13,12 @@ from core.loop import AgentLoop class _Session: def __init__(self, messages=None): + self.task_id = uuid4() self.messages = list(messages or []) self.appended = [] self.append_artifacts = [] - def append(self, message, *, artifact_refs=None): + def append(self, message, *, artifact_refs=None, **_kwargs): self.messages.append(message) self.appended.append(message) self.append_artifacts.append(artifact_refs) @@ -101,6 +102,48 @@ class PersistedTurnTests(unittest.TestCase): "version": 1, "scope": "working_dir", "path": "report.pdf", }]) + def test_failed_tool_result_records_health_event(self) -> None: + session = _Session([{"role": "user", "content": "执行"}]) + executor = MagicMock() + executor.call_tool.return_value = SimpleNamespace( + content="[Error] command failed", artifacts=(), + ) + loop = AgentLoop( + llm=MagicMock(), executor=executor, session=session, + capabilities=SimpleNamespace( + max_iterations=2, family="test", variant="model", + input_cny_per_mtoken=0, output_cny_per_mtoken=0, + cache_hit_cny_per_mtoken=0, pricing={}, + ), + user_id=uuid4(), working_dir=Path("."), + ) + loop._maybe_fold_context = MagicMock() + tool_call = SimpleNamespace( + id="call-1", + function=SimpleNamespace(name="shell", arguments='{"command":"false"}'), + ) + responses = [ + SimpleNamespace( + choices=[SimpleNamespace(message=SimpleNamespace( + content="", tool_calls=[tool_call], + ))], usage=None, + ), + SimpleNamespace( + choices=[SimpleNamespace(message=SimpleNamespace( + content="已说明失败", tool_calls=None, + ))], usage=None, + ), + ] + loop._stream_llm = MagicMock(side_effect=[(r, False) for r in responses]) + with patch("core.loop.record_chat_usage"), \ + patch("core.loop.record_tool_failure") as failure: + result = loop.run_persisted_turn() + + self.assertEqual(result, "已说明失败") + failure.assert_called_once() + self.assertEqual(failure.call_args.kwargs["tool"], "shell") + self.assertEqual(failure.call_args.kwargs["content"], "[Error] command failed") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_static_vendor.py b/tests/test_static_vendor.py index 8544241..598d633 100644 --- a/tests/test_static_vendor.py +++ b/tests/test_static_vendor.py @@ -149,6 +149,9 @@ class StaticVendorTests(unittest.TestCase): self.assertIn("if (toolFailuresRequest) return toolFailuresRequest", admin_js) self.assertIn("loadToolFailures();", admin_js) self.assertNotIn('activeTab === "quality"', admin_js) + self.assertIn('item.category === "failure"', admin_js) + self.assertIn('item.category === "agent_control"', admin_js) + self.assertIn("代理控制与上下文异常（近 24h）", admin_js) self.assertIn('id="capacity-drawer" class="capacity-drawer"', html) self.assertIn(".capacity-drawer-panel {", html) diff --git a/tests/test_storage_migration.py b/tests/test_storage_migration.py index a90e79c..c3a4742 100644 --- a/tests/test_storage_migration.py +++ b/tests/test_storage_migration.py @@ -32,6 +32,27 @@ class StorageMigrationTests(unittest.TestCase): self.assertIn("ix_scheduled_jobs_due_active", rendered) self.assertIn("USING brin", rendered) + def test_0040_upgrade_compiles_health_event_indexes(self) -> None: + statements: list[str] = [] + + def capture(sql, *multiparams, **params): + statements.append(str(sql.compile(dialect=postgresql.dialect()))) + + engine = create_mock_engine("postgresql+psycopg://", capture) + operations = Operations(MigrationContext.configure(engine.connect())) + migration = importlib.import_module( + "db.migrations.versions.20260903_1400_0040_tool_health_indexes" + ) + with patch.object(migration, "op", operations): + migration.upgrade() + + rendered = "\n".join(statements) + self.assertIn("ix_usage_health_kind_created", rendered) + self.assertIn("ux_usage_tool_failure_message", rendered) + self.assertIn("tool_failure", rendered) + self.assertIn("agent_guard", rendered) + self.assertIn("context_fold_failure", rendered) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_toolfail_malformed.py b/tests/test_toolfail_malformed.py index a22db64..08cf2a4 100644 --- a/tests/test_toolfail_malformed.py +++ b/tests/test_toolfail_malformed.py @@ -11,36 +11,67 @@ from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import core.toolfail as tf # noqa: E402 +from core.storage.telemetry import ( # noqa: E402 + KIND_AGENT_GUARD, + KIND_EMPTY_RESPONSE, + KIND_RUN_ERROR, + KIND_TOOL_FAILURE, + KIND_TOOL_MALFORMED, +) +from core.tool_failure import structured_failure # noqa: E402 NOW = datetime.now(timezone.utc) class _FakeSession: - """按 SQL 关键字分流四段查询:messages / tool_malformed / run_error / empty_response。""" + """工具健康只允许一次 usage_events 查询。""" - def __init__(self, msg_rows, malformed_rows, run_rows, empty_rows): - self._msg_rows = msg_rows - self._malformed_rows = malformed_rows - self._run_rows = run_rows - self._empty_rows = empty_rows + def __init__(self, rows): + self._rows = rows + self.calls = 0 def execute(self, clause, params=None): sql = str(clause) - if "tool_malformed" in sql: - rows = self._malformed_rows - elif "run_error" in sql: - rows = self._run_rows - elif "empty_response" in sql: - rows = self._empty_rows - else: - rows = self._msg_rows - return SimpleNamespace(fetchall=lambda: rows) + self.calls += 1 + if "from usage_events" not in sql or "from messages" in sql: + raise AssertionError(sql) + return SimpleNamespace(fetchall=lambda: self._rows) + + +def _structured_row(row): + task_id, user_id, created_at, tool, content = row[:5] + arguments = None + if len(row) > 6 and isinstance(row[6], dict): + calls = row[6].get("tool_calls") or [] + if calls: + arguments = (calls[0].get("function") or {}).get("arguments") + units = structured_failure(tool, content, arguments=arguments) + assert units is not None + return ( + KIND_TOOL_FAILURE, task_id, user_id, created_at, None, tool, + None, None, None, units["failure_kind"], units["signature"], + units["category"], units["sample"], + ) def _scan(msg_rows, malformed_rows, run_rows=None, empty_rows=None, **kw): + event_rows = [_structured_row(row) for row in msg_rows] + event_rows.extend(( + KIND_TOOL_MALFORMED, task_id, user_id, created_at, None, tool, + err, head, tail, None, None, None, None, + ) for task_id, user_id, created_at, tool, err, head, tail in malformed_rows) + event_rows.extend(( + KIND_RUN_ERROR, task_id, user_id, created_at, None, None, + err, None, None, None, None, None, None, + ) for task_id, user_id, created_at, err in (run_rows or [])) + event_rows.extend(( + KIND_EMPTY_RESPONSE, task_id, user_id, created_at, model_profile, None, + None, None, None, None, None, None, None, + ) for task_id, user_id, created_at, model_profile in (empty_rows or [])) + @contextmanager def fake_scope(): - yield _FakeSession(msg_rows, malformed_rows, run_rows or [], empty_rows or []) + yield _FakeSession(event_rows) with patch.object(tf, "session_scope", fake_scope): return tf.scan_tool_failures(**kw) @@ -73,8 +104,8 @@ class TestToolfailMalformed(unittest.TestCase): 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。""" + def test_merges_structured_and_malformed_clusters(self): + """同表不同 kind 并存互不干扰:普通失败与 malformed 各出一组。""" msg_rows = [ ("t1", "u1", NOW, "glob", "[Error] base path not found"), ("t2", "u2", NOW, "glob", "[Error] base path not found"), @@ -184,6 +215,25 @@ class TestToolfailEmptyResponse(unittest.TestCase): self.assertEqual(out, []) +class TestStructuredHealthEvents(unittest.TestCase): + def test_agent_control_is_separate_from_operational_failure(self): + rows = [( + KIND_AGENT_GUARD, f"t{i}", "u1", NOW, "deepseek_v4.pro", "edit", + None, None, None, "guard", "same_error: old_str not found", + "agent_control", "same_error, count=4", + ) for i in range(2)] + + @contextmanager + def fake_scope(): + yield _FakeSession(rows) + + with patch.object(tf, "session_scope", fake_scope): + out = tf.scan_tool_failures(days=1, min_count=2, min_tasks=2) + self.assertEqual(len(out), 1) + self.assertEqual(out[0]["category"], "agent_control") + self.assertEqual(out[0]["kind"], "guard") + + class TestToolWireHealth(unittest.TestCase): def test_aggregates_recovery_rates_and_totals(self): rows = [ diff --git a/web/static/js/admin.js b/web/static/js/admin.js index 2e3ac77..4a774a1 100644 --- a/web/static/js/admin.js +++ b/web/static/js/admin.js @@ -188,10 +188,12 @@ function renderOpsSummary() { const failureRows = (toolFailuresData && toolFailuresData.clusters) || []; const activeFailures = failureRows.filter(item => (item.count_24h || 0) > 0); - const operationalFailures = activeFailures.filter(item => item.category !== "quality_gate"); + const operationalFailures = activeFailures.filter(item => item.category === "failure"); const failureCount = operationalFailures.reduce((sum, item) => sum + (Number(item.count_24h) || 0), 0); - const systemicCount = activeFailures.filter(item => item.category !== "quality_gate" && item.task_count >= 2).length; - const localCount = activeFailures.filter(item => item.category !== "quality_gate" && item.task_count < 2).length; + const systemicCount = operationalFailures.filter(item => item.task_count >= 2).length; + const localCount = operationalFailures.filter(item => item.task_count < 2).length; + const controlCount = activeFailures.filter(item => item.category === "agent_control") + .reduce((sum, item) => sum + (Number(item.count_24h) || 0), 0); const gateCount = activeFailures.filter(item => item.category === "quality_gate") .reduce((sum, item) => sum + (Number(item.count_24h) || 0), 0); const malformed = Number(((toolWireData || {}).total || {}).malformed_24h) || 0; @@ -238,7 +240,7 @@ function renderOpsSummary() { sub: toolFailuresData ? (systemicCount || malformed ? `系统性聚集 ${systemicCount} · 链路残余 ${malformed}` : "未发现系统性工具异常") : "正在汇总工具调用状态", - meta: toolFailuresData ? [`单任务 ${localCount}`, `质量门 ${gateCount}`] : [], + meta: toolFailuresData ? [`单任务 ${localCount}`, `控制 ${controlCount} · 质量门 ${gateCount}`] : [], badge: toolFailuresData ? (toolTone ? { text: systemicCount || malformed ? "需处理" : "有记录", tone: toolTone } : { text: "正常", tone: "ok" }) : null, @@ -1169,8 +1171,9 @@ function wireHealthHTML(d) { 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 systemic = active.filter(c => c.category === "failure" && c.task_count >= 2); + const taskLocal = active.filter(c => c.category === "failure" && c.task_count < 2); + const controls = active.filter(c => c.category === "agent_control"); const gates = active.filter(c => c.category === "quality_gate"); const quiet = rows.filter(c => !(c.count_24h || 0)); const days = d.days || 7; @@ -1190,6 +1193,11 @@ function renderToolFailures(d, wire = null) { gates, "近 24 小时无质量门拦截记录", ) + + toolFailureTable( + "代理控制与上下文异常（近 24h）", + controls, + "近 24 小时无代理控制或上下文异常", + ) + toolFailureTable( `已安静历史（近 ${days} 天窗口，近 24h 为 0）`, quiet,