fix(agent): 修复工具链恢复与产物发布

CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ## Unreleased
 
+- 提高工具参数异常时的自动恢复率，产物发布和工具健康分类也更加稳定准确。
+
 - 管理后台会异步汇总工具健康大数，并避免短时间内重复请求；工具异常统一按结构化事件统计，代理主动保护与真正故障分开展示，多人同时查看或自动刷新时不再反复扫描历史消息。
 
 - 单用户可同时运行的重型任务由 2 个提升到 3 个；任务等待执行容量时，对话会直接说明是当前用户、整机或宿主内存限制，获得槽位后自动继续。

PROGRESS.md

@@ -20,6 +20,8 @@
 ---
 ## 已完成关键能力
 
+- **09-03 / Unreleased / 工具链可靠性收敛**：tool arguments salvage 新增“至少两个完整且完全一致副本 + 尾部截断副本”的保守恢复，语义不一致仍拒绝。修复 artifacts 部分唯一索引谓词被参数化后 PostgreSQL 无法匹配 `ON CONFLICT` 的问题，并让输出中任意行首 `[GATE FAIL]` 均按质量门记录，避免 SVG/PPT 质检占用真实故障大数。无 schema/migration/API 变化，生产失败历史未改写。
+
 - **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；结构化事件口径由后续同日改动完成。

core/artifact_lifecycle.py

@@ -9,7 +9,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from uuid import UUID, uuid4
 
-from sqlalchemy import or_, select
+from sqlalchemy import or_, select, text
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from .artifacts import ARTIFACT_TRASH_DIR, ArtifactRef
@@ -85,7 +85,10 @@ def register_published_artifacts(
                 content_sha256=content_sha256,
             ).on_conflict_do_update(
                 index_elements=[Artifact.user_id, Artifact.current_path],
-                index_where=Artifact.status == "active",
+                # 必须与 0028 的部分唯一索引谓词按 SQL 字面量一致。若用
+                # ``Artifact.status == "active"``，SQLAlchemy 会编译成绑定参数，
+                # PostgreSQL 无法据此推断 ON CONFLICT 对应哪个部分索引。
+                index_where=text("status = 'active'"),
                 set_=update_values,
             ).returning(Artifact.artifact_id)
             artifact_id = session.execute(statement).scalar_one()

core/salvage.py

@@ -22,22 +22,45 @@ def salvage_tool_arguments(raw: str, allowed_keys: set) -> Optional[dict]:
       (a) parse-to-end 成功(json.loads 要求整个后缀是合法 JSON,尾部有残渣就失败);
       (b) 结果是非空 dict;
       (c) 顶层 key ⊆ 该工具 schema 的参数名。
-    双护栏缺一不可:(a) 挡尾部被截断的半个 JSON(截断则拒绝、回落既有重试,不冒险执行
-    残缺写入);(c) 挡垃圾前缀里恰好自洽的旁支 JSON(如模型正文里引用的 `{...}` 片段)。
+    双护栏缺一不可:(a) 挡尾部被截断的半个 JSON；唯一例外是 wire 已给出至少两个
+    完全一致的完整参数对象、随后又粘了截断副本，此时一致副本可作为相互校验；
+    (c) 挡垃圾前缀里恰好自洽的旁支 JSON(如模型正文里引用的 `{...}` 片段)。
     都不满足返回 None。纯函数、无副作用,便于单测(线上真前缀样本当夹具)。
     """
     # 只可能出现在 '{' 处;非 '{' 直接跳过,省下大部分 json.loads 尝试。
     start = 0
+    decoder = json.JSONDecoder()
+    repeated_candidates: list[dict] = []
     while True:
         i = raw.find("{", start)
         if i < 0:
-            return None
+            break
         try:
             obj = json.loads(raw[i:])
         except (json.JSONDecodeError, ValueError):
+            # 某些 wire 抖动会形成 ``{完整参数}{完整参数}{截断副本``。
+            # parse-to-end 无法处理，但 raw_decode 仍能取出前面的完整对象。
+            # 这里只收集候选，最终必须至少两个且完全一致才可执行。
+            try:
+                partial_obj, _end = decoder.raw_decode(raw, i)
+            except (json.JSONDecodeError, ValueError):
+                start = i + 1
+                continue
+            if (
+                isinstance(partial_obj, dict)
+                and partial_obj
+                and set(partial_obj.keys()) <= allowed_keys
+            ):
+                repeated_candidates.append(partial_obj)
             start = i + 1
             continue
         if isinstance(obj, dict) and obj and set(obj.keys()) <= allowed_keys:
             return obj
         # 解析成功但不是想要的 dict(如前缀里一段自洽 JSON):继续往后找,真 JSON 可能在更后面。
         start = i + 1
+    if (
+        len(repeated_candidates) >= 2
+        and all(obj == repeated_candidates[0] for obj in repeated_candidates[1:])
+    ):
+        return repeated_candidates[0]
+    return None

core/tool_failure.py

@@ -99,7 +99,11 @@ def shell_command_hint_from_arguments(arguments: Any) -> str:
 def failure_category(kind: str, signature_line: str, sample: str) -> str:
     if kind != "exit":
         return "failure"
-    if signature_line.startswith("[GATE FAIL]"):
+    # 组合导出脚本会在 gate 行之后继续打印修复提示/命令，因此 gate 不一定是
+    # classify_failure 选中的最后一条签名行。按完整输出中的行首标记识别。
+    if signature_line.startswith("[GATE FAIL]") or re.search(
+        r"(?m)^\[GATE FAIL\]", sample
+    ):
         return "quality_gate"
     if (
         ("[篇幅核算]" in sample or "[字数核算]" in sample)

tests/test_artifact_upsert_sql.py

@@ -0,0 +1,21 @@
+from sqlalchemy.dialects.postgresql import dialect, insert as pg_insert
+from sqlalchemy import text
+
+from core.storage.models import Artifact
+
+
+def test_artifact_upsert_partial_index_predicate_is_literal():
+    statement = pg_insert(Artifact).values(
+        user_id="00000000-0000-0000-0000-000000000001",
+        origin_task_id="00000000-0000-0000-0000-000000000002",
+        current_path="task/report.docx",
+        label="report",
+    ).on_conflict_do_update(
+        index_elements=[Artifact.user_id, Artifact.current_path],
+        index_where=text("status = 'active'"),
+        set_={"label": "report"},
+    )
+
+    sql = str(statement.compile(dialect=dialect()))
+    assert "WHERE status = 'active'" in sql
+    assert "WHERE status = %(" not in sql

tests/test_health_telemetry.py

@@ -71,6 +71,22 @@ class TestHealthTelemetry(unittest.TestCase):
         self.assertEqual(self.session.rows[0].kind, telemetry.KIND_QUALITY_GATE)
         self.assertEqual(self.session.rows[0].units["category"], "quality_gate")
 
+    def test_quality_gate_is_detected_before_trailing_command_hint(self):
+        ids = self._ids()
+        telemetry.record_tool_failure(
+            **ids,
+            model_profile="deepseek_v4.flash",
+            tool="shell",
+            content=(
+                "[stdout]\n[ERROR] slide.svg - Failed\n"
+                "[GATE FAIL] svg_quality_checker: 1 error\n"
+                "python scripts/svg_to_pptx.py \".\"\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 = {

tests/test_salvage.py

@@ -98,6 +98,21 @@ class TestSalvage(unittest.TestCase):
         raw = '前缀{"path": "/a", "content": "x"}尾部残渣'
         self.assertIsNone(salvage_tool_arguments(raw, WRITE_KEYS))
 
+    def test_repeated_identical_json_with_truncated_copy_is_salvaged(self):
+        """wire 重复同一参数且末尾又截断时，可用两个一致的完整副本恢复。"""
+        good = {"command": "rg -n 'needle' /workspace/report.md"}
+        encoded = json.dumps(good, ensure_ascii=False)
+        raw = encoded + encoded + encoded[:18]
+        self.assertEqual(salvage_tool_arguments(raw, SHELL_KEYS), good)
+
+    def test_distinct_complete_objects_with_trailing_garbage_are_rejected(self):
+        """存在两个不同的合法参数对象时语义不唯一，不能猜测执行哪一个。"""
+        first = json.dumps({"command": "first"})
+        second = json.dumps({"command": "second"})
+        self.assertIsNone(
+            salvage_tool_arguments(first + second + "{truncated", SHELL_KEYS)
+        )
+
     def test_clean_json_at_char0(self):
         """char-0 就是完好 JSON(理论上不会进 salvage,但函数应幂等返回它)。"""
         raw = '{"path": "/a", "content": "x"}'