fix(loop): 堵失败聚集两裂缝——必填key被吞畸形走非流式重试 + RepeatGuard连续同错streak拦(bump 0.58.33)

扫近14天工具失败面板 top13 定位两条反复烧 token 的裂缝:

① 必填 key 被吞畸形(面板 #3 edit 缺 path,14次/13task/9用户,最广):
定层坐实是流式 wire 乱序(path 值碎片被吞进 old_str 尾部,独立 path 键消失),
跨 provider,非模型漏参。关键区别是 JSON parse 成功 → 既不命中
_malformed_tool_calls(只抓 parse 失败)、salvage 也救不了,一路漏到 executor
才报错反复烧。新增 _toolcalls_partial_args(非空 dict 且 0<缺失必填<全部必填)
接进 _stream_llm attempt 循环走非流式重试(服务端一次拼好绕开乱序),
_log_partial_args 落 usage_events(kind=tool_malformed,签名 missing required keys)
纳入面板可观测(此前伪装成普通 [Error] 完全是盲区)。

② 换参数撞同一堵墙(面板 #2 edit old_str not found,单task 9次/24h 7次):
模型微调 old_str 重试,精确 args 指纹每次不同 → RepeatGuard arg 判据不累计。
补第二道判据:按工具的连续同类错误 streak(_norm_err 抹平路径/数字后签名相同、
跨≥2 不同 args),SOFT 注入定向提示、HARD 拦截一次并重置到 SOFT(非永久封死),
任一成功立即清零。

测试 test_loop_repeat_guard.py +12(err-streak 5 + partial-args 7),20/20 通过;
相邻 salvage/toolfail_malformed 23/23 通过,未碰坏既有行为。additive 内部机制,
不碰对外 API/DB 契约。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

PROGRESS.md

@@ -23,6 +23,7 @@
 
 ### 2026-07
 
+- **07-17 / 0.58.33**:**堵工具失败聚集里两条反复烧 token 的裂缝**(扫近 14 天失败面板 top13 定位,`scripts/standalone` 复现口径)。**裂缝①「必填 key 被吞」畸形(面板 #3 `edit 缺少必填参数 ['path']`,14 次 / 13 task / 9 用户,最广)**:定层查一条原始 tool_call 坐实是**流式 wire 乱序**——某键的值碎片被瞬移拼进相邻字符串(实证:path 的 `.../gen_final_report_v2.py` 被吞进 `old_str` 尾部,独立的 `"path"` 键随之消失),跨 provider(`call_00_` 网关档 + `toolu_` Anthropic 系),非模型漏参。这类与既有 char-0 前缀畸形的关键区别是 **JSON parse 成功** → 既不命中 `_malformed_tool_calls`(只抓 parse 失败)、salvage 也救不了(parse-to-end/key 白名单对合法但错位无能),一路漏到 executor 才在语义层报错、回 `[Error]` 喂回模型 → 再拼再乱序反复烧。**修**:`_toolcalls_partial_args`(解析为非空 dict 且 `0<len(missing_required)<len(required)`,即至少一个必填在场、至少一个缺失=错位吞键特征;空 `{}`/必填全缺不算,仍交 executor+RepeatGuard)接进 `_stream_llm` 的 attempt 循环,和畸形/空响应同类:丢弃本轮走**非流式重试**(服务端一次拼好绕开流式 delta 乱序),耗尽仍缺才落回 executor 返「缺必填参数」(多为真漏参,不再空转);`_log_partial_args` 落 `usage_events`(kind=`tool_malformed`,error 签名固定 `missing required keys [...]`,和 char-0 型区分)——这类此前伪装成普通 messages `[Error]`、面板完全看不见,现纳入可观测。**裂缝②「换参数撞同一堵墙」(面板 #2 `edit old_str not found` 单 task 9 次 / 24h 7 次)**:模型每次微调 old_str 重试,精确 args 指纹每次不同 → `_RepeatGuard` 的 arg 判据不累计,只能等 `_STALL_LIMIT` 慢慢兜。**修**:`_RepeatGuard` 补第二道判据——按工具的**连续同类错误 streak**(`_norm_err` 抹平路径/数字后签名相同、跨 ≥2 个不同 args),SOFT 注入定向提示(先 read/grep 看确切内容再动手)、HARD 拦截一次并重置到 SOFT(非永久封死,换路后重试留活口);任一非错误结果立即清零该工具 streak。测试 `tests/test_loop_repeat_guard.py` +12(err-streak 5 + partial-args 7),全 23 相邻测试(salvage/toolfail_malformed)绿。**本机验证坑**:`core/loop.py` 顶层 `import litellm` 本机导入死卡(GitHub cost map),纯逻辑测试用 `sys.modules` 注入 litellm+子模块 stub 绕开跑真代码;失败面板扫描脚本因 import 链同样卡,改直连 psycopg 复刻 `scan_tool_failures` 口径查(结果与走 `core` 的交叉验证一致)。盲区:provider-wire 抖动本机复现不出、/verify 端到端跑不了,靠单测 + 生产灰度。additive 内部机制,不碰对外 API/DB 契约。剩 ①run_python 中文全角标点/未闭合字符串致 SyntaxError(面板最大头,产物质量,待单独立项)、③超时 19 次未动。相关 memory:deepseek 畸形 salvage 案 / 高轮数烧 token 三根因。
 - **07-15 / 0.58.32**:**空响应防御:provider 吐空不再被当正常收尾静默 done**(复盘 task 2a1bc25d "ppt3":unifyllm 网关对某档 Claude 偶发把 tool_use 漏成正文 / 直接吐空,回来的轮 `tool_calls=[]` 且正文空,`loop.py` 的 `if not tool_calls: done` 当"模型答完"收尾——run_status=idle、无报错、无终态失败,表现为"自己中断",只能人肉挖 DB 才发现;复测该 bug 现已消失/瞬态,但"空 tool_calls 直接 done"这个失败模式本身太隐蔽,任何 provider 未来吐一次空都会静默卡死,故加防御)。**两层**(对称既有畸形 salvage 链):① 自动重试——`_is_empty_response`(tc 空且 content 去空白为空;纯 tool_call 轮不误判)接进 `_stream_llm` 的 attempt 循环,和畸形同类:丢弃本轮走非流式重试(多数瞬态重发一次即好,用户无感),`_log_empty_response` stdout+`usage_events`(kind=`empty_response`,cost 0)双写留痕、DB 失败静默不阻塞;② 重试耗尽仍空——`run()` 收尾点分空-空分支:不发静默 `done`,改发**可见 warn**「模型返回空响应(已自动重试仍为空),已停止,回复『继续』可重试」+ done(复用 stall 熔断的 warn+自停话术,run_status 落 idle 可续),**不引入终态 error**(空响应可恢复,红 error 过重且与既有路径不对称)。③ 可观测——`core/toolfail.py` 加第四段扫描 kind=empty_response 聚成 tool=`(empty)`/kind=`empty` 的 cluster(sample=model_profile 看哪个网关档在吐空),即便自动救回也留痕、跨 task 系统性吐空才冒头(单次瞬态 <阈值不触发)。**刻意不做**内容嗅探式"narrated tool_call 特征→重试":marker 每次变、且与助手正常解释/贴代码的正文高度重叠,假阳性(把合法回答判坏反复重试)比它防的罕见静默停更糟——那类只靠面板留痕兜、不改热路径行为。测试:`tests/test_loop_empty_response.py`(_is_empty_response 四态含纯 tool_call 不误判 + 重试恢复/耗尽返空/正常文本不重试)、`tests/test_toolfail_malformed.py` +2(空响应聚集 / 单次瞬态不触发),顺手补回 `test_loop_malformed_retry.py` 因 salvage(0.58.24)落地后失修的 `loop.executor` 桩(空 schemas → salvage 返 False 走既有重试)。全 247 测试绿。盲区:provider-wire 抖动本机复现不出、/verify 端到端跑不了,靠单测+生产灰度。additive 内部机制,不碰对外 API/DB 契约。相关 memory:unifyllm tool_use 漏正文案 / deepseek 畸形 salvage 案。
 - **07-15 / 0.58.31**:**撤回 §8.11 delegate(0.58.30 曾整体落地,revert + force-push 抹除)**。delegate(隔离检索子循环:FilteredExecutor + 内存态子 Session + loop 拦截 + 降 flash)机制本身没错、对标 Claude Code subagent 也成立,但**触发信号没坐实就上了**:自评时回看 `diag_search_args.py`,motivating 案子 `document_search` 122 次呈"一批批不同材料体系并行搜"形态,更像**批量扇出而非结果驱动探索**——若属实其正解与 mp_search 同为**批量工具 `document_search_batch`**(便宜/可预测/可诊断),delegate 是过度设计;叠加子循环 transcript 不落盘(诊断驱动的功能反不可诊断)、20 轮上限对 122 次负载偏低、强制 flash 对难检索可能降质等未验证坑。故整体撤回,只留验证过的批量工具(0.58.28)。**重建前置收紧**:先用 diag 确认某检索 task 的 query 是真探索(依赖前序结果、无法一次列全)而非可枚举批量,是批量走批量工具、只有真探索才 delegate(DESIGN §8.11 记该教训)。撤法:`git reset --hard` 到 0.58.29 + `--force-with-lease` 推,delegate commit 从历史抹除(远程 tip 确认无他人后续提交才 force);保留本条 PROGRESS 记录"试过并撤回"免日后盲目重建。
 - **07-15 / 0.58.29**:**修对话正文文件锚点两处回归(含 `·` 的 pptx 丢 chip + 正文文件链接点击整页 404)**。① chip 丢失:`media.js` 的 `_TAIL_CLS` 把间隔号 `·`(U+00B7)当尾部排除字符,而它是文件名合法分隔符(`AI赋能陶瓷产业·…-图标增强版.pptx`)→ 在 `·` 处截断成目录不挂 chip。去掉 `·`(尾部标点另有 strip 兜底);node 验证 OLD→`[]`、NEW→两路径全出。② 链接 404:模型把相对路径写成 `[<rel>](<rel>)`,`renderMd` 出 `<a>`,但 `chat.js` 的 `#chat-stream` 点击委托无 `<a>` 拦截 → 原生跳转打到 API 不存在路由返 FastAPI 默认 `{"detail":"Not Found"}`。加拦截 `.msg .body a[href]`:外链开新标签,内部相对路径 `openFilePreview(rel)`(复用预览 modal,pptx 转 PDF),`#` 锚点走默认。纯前端修复,不碰对外契约。

core/__init__.py

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

core/loop.py

@@ -11,6 +11,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import re
 import threading
 import time
 from pathlib import Path
@@ -42,6 +43,21 @@ _PPTX_GUARD_TOOLS = ("shell", "run_python")
 _CANCELLED_TOOL_PLACEHOLDER = "[cancelled by user]"
 
 
+# 错误签名归一:同一类工具报错在不同参数/路径/数字下抹平,让「反复撞同一堵墙」
+# 能被识别成同一签名(如 old_str 每次微调但结果始终 "old_str not found in <path>")。
+# 与 core/toolfail._normalize 同源思路,此处保持 loop 自包含(不反向依赖 toolfail)。
+_ERR_PATH_RE = re.compile(r"(?:[A-Za-z]:)?(?:[/\\][\w.\-一-鿿*]+){2,}")
+_ERR_NUM_RE = re.compile(r"\d+")
+
+
+def _norm_err(result: str) -> str:
+    """取 [Error] 结果首行、抹平路径/数字,得到稳定的错误签名(截断 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]
+
+
 class _RepeatGuard:
     """检测「同名同参 + 无产出」的病理性重复调用,断掉死循环。
 
@@ -59,6 +75,13 @@ class _RepeatGuard:
     顺带堵掉 `_malformed_tool_calls` 的洞:大参数畸形退化成合法空 `{}` 时,executor 每次
     返回同一句「缺少必填参数」→ 走 dup 分支被这同一机制拦下,无需单独特判空 `{}`。
 
+    第二道判据(2026-07,失败面板 #2:edit `old_str not found` 单 task 反复撞墙):模型每次
+    **微调参数**(old_str 改一点)重试同一操作,精确 args 指纹每次不同 → 上面的 arg 判据不累计,
+    只能等 _STALL_LIMIT 慢慢兜。补一条**按工具 + 归一化错误签名的连续 streak**:同一工具连续返回
+    同一类错误(路径/数字抹平后相同)且跨 >= 2 个不同 args → 认定「换法子撞同一堵墙」,SOFT 注入
+    定向提示、HARD 拦截一次(拦后重置到 SOFT,非永久封死,给换路后的重试留活口)。任一非错误结果
+    (真推进了)立即清零该工具的 streak。
+
     状态活在单次 task run 内(AgentLoop 实例持有),不跨 task。
     """
 
@@ -68,6 +91,8 @@ class _RepeatGuard:
     def __init__(self) -> None:
         # key -> {"hashes": set[str], "unproductive": int, "n": int, "blocked": int}
         self._h: Dict[str, dict] = {}
+        # name -> {"esig": str, "count": int, "args": set[str]}:按工具的连续同类错误 streak
+        self._err_streak: Dict[str, dict] = {}
 
     @staticmethod
     def _key(name: str, args: Any) -> str:
@@ -112,8 +137,41 @@ class _RepeatGuard:
                 st["unproductive"] = 0
         st["hashes"].add(h)
         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
+                s["args"].add(self._key(name, args))
+            else:
+                self._err_streak[name] = {"esig": esig, "count": 1, "args": {self._key(name, args)}}
+        else:
+            self._err_streak.pop(name, None)  # 真推进了 → 清零该工具 streak
         return st["unproductive"], not (is_err or dup)
 
+    def err_streak(self, name: str) -> Tuple[int, int, str]:
+        """返回 (连续同类错误次数, 涉及的不同 args 数, 错误签名);无则 (0,0,'')。"""
+        s = self._err_streak.get(name)
+        if not s:
+            return 0, 0, ""
+        return s["count"], len(s["args"]), s["esig"]
+
+    def should_block_err(self, name: str) -> bool:
+        """执行前:该工具已连续 >= HARD 次同类错误且跨 >= 2 个不同 args → 拦截一次。"""
+        cnt, n_args, _ = self.err_streak(name)
+        return cnt >= self.HARD and n_args >= 2
+
+    def register_err_block(self, name: str) -> Tuple[int, str]:
+        """记一次 err-streak 拦截,返回 (拦截时的连续次数, 错误签名);拦后重置到 SOFT
+        (保留 esig/args)—— 非永久封死,让模型换路后的下一次重试还能进来。"""
+        s = self._err_streak.get(name)
+        if not s:
+            return 0, ""
+        cnt, esig = s["count"], s["esig"]
+        s["count"] = self.SOFT
+        return cnt, esig
+
 
 def _extract_delta_content(chunk: Any) -> Optional[str]:
     """从 stream chunk 提 delta.content(文本片段)。chunk 形态 litellm ModelResponseStream:
@@ -178,6 +236,89 @@ def _malformed_tool_calls(response: Any) -> List[str]:
     return bad
 
 
+def _toolcalls_partial_args(
+    response: Any, required_by_tool: Dict[str, List[str]]
+) -> List[Tuple[Any, str, List[str]]]:
+    """检出「JSON 能解析、但必填 key 被吞掉」的畸形 tool_call。
+
+    背景(2026-07,失败面板 #3:edit `缺少必填参数 ['path']` 跨 13 task/9 用户):流式
+    arguments delta 乱序把某个键的值碎片瞬移拼进相邻字符串(实证 [1]:path 的
+    `.../gen_final_report_v2.py` 被吞进 old_str 尾部),独立的 `"path"` 键随之消失。这类
+    与 char-0 前缀畸形的关键区别是 **JSON parse 成功** → 既不命中 `_malformed_tool_calls`
+    (只抓 parse 失败)、salvage 也救不了(parse-to-end/key 白名单对合法但错位无能),一路
+    漏到 executor 才在语义层报「缺必填参数」,回 [Error] 喂回模型 → 再拼再乱序,反复烧。
+
+    判据(窄,避免误伤模型真漏参):解析为**非空 dict** 且 **至少一个必填 key 在场**同时
+    **至少一个必填 key 缺失**(即 0 < len(missing) < len(required))。空 `{}` / 必填全缺
+    (纯垃圾/无关键)不算 —— 交给 executor + _RepeatGuard 现状处理。
+
+    返回 [(tc, name, missing_keys), ...]。
+    """
+    try:
+        msg = response.choices[0].message
+    except Exception:
+        return []
+    out: List[Tuple[Any, str, List[str]]] = []
+    for tc in (getattr(msg, "tool_calls", None) or []):
+        try:
+            name = tc.function.name
+            raw = (getattr(tc.function, "arguments", None) or "").strip()
+        except Exception:
+            continue
+        if not raw:
+            continue
+        try:
+            obj = json.loads(raw)
+        except (json.JSONDecodeError, ValueError):
+            continue  # parse 失败归 _malformed_tool_calls,不重复处理
+        if not isinstance(obj, dict) or not obj:
+            continue
+        required = required_by_tool.get(name) or []
+        if not required:
+            continue
+        missing = [k for k in required if k not in obj]
+        if 0 < len(missing) < len(required):
+            out.append((tc, name, missing))
+    return out
+
+
+def _log_partial_args(
+    task_id: Any, user_id: Any, model_profile: str,
+    partial: List[Tuple[Any, str, List[str]]], response: Any,
+) -> None:
+    """必填 key 被吞的畸形留痕:与 _log_malformed_args 对称,进 usage_events(kind=
+    tool_malformed),error 签名固定为 `missing required keys [...]` —— 在失败面板里和
+    char-0 型(`Expecting value`)、executor 的「缺必填参数」区分开,便于统计这条新裂缝。
+    任何一路失败都静默,绝不打断重试主路径。"""
+    try:
+        usage = _extract_usage_details(getattr(response, "usage", None))
+        for tc, name, missing in partial:
+            raw = (getattr(tc.function, "arguments", None) or "")
+            err = f"missing required keys {missing}"
+            print(
+                f"[malformed:partial] task={task_id} tool={name} len={len(raw)} "
+                f"{err} 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=name,
+                    arg_len=len(raw),
+                    error=err,
+                    head=raw[:300],
+                    tail=raw[-300:],
+                    tokens_in=usage["tokens_in"],
+                    tokens_out=usage["tokens_out"],
+                )
+            except Exception:
+                pass
+    except Exception:
+        pass
+
+
 def _is_empty_response(response: Any) -> bool:
     """检出「空响应」:assistant 轮既无 tool_calls 又无正文(去空白后为空)。
 
@@ -575,6 +716,11 @@ class AgentLoop:
             # 50% 进压缩区、85% 触发折叠(context_fold)。
             "context_limit_chars": int(self.caps.reliable_context * self._CHARS_PER_TOKEN),
         }
+        # 各工具必填参数(单一事实源:executor schema),供「必填 key 被吞」畸形检测。
+        required_by_tool = {
+            sc["function"]["name"]: (sc["function"].get("parameters") or {}).get("required") or []
+            for sc in self.executor.schemas()
+        }
         for attempt in range(self._MAX_MALFORMED_ATTEMPTS):
             use_nonstream = attempt > 0
             # 每个 attempt 重发 llm_start(stats 同一份):非流式重试完成前零 delta 事件,
@@ -609,6 +755,25 @@ class AgentLoop:
                         ),
                     })
                     continue
+                # 必填 key 被吞的畸形(parse 成功、salvage 无能):非流式重试(服务端一次拼好,
+                # 绕开流式 delta 乱序)。耗尽尝试仍缺 → 落下面 return,交 executor 返「缺必填参数」
+                # 给模型(多为模型真漏参,不再空转)。
+                partial = _toolcalls_partial_args(response, required_by_tool)
+                if partial:
+                    _log_partial_args(
+                        self.session.task_id, self.user_id,
+                        f"{self.caps.family}.{self.caps.variant}", partial, response,
+                    )
+                    names = ", ".join(f"{n}(missing={m})" for _, n, m in partial)
+                    self._emit({
+                        "type": "warn",
+                        "msg": (
+                            f"工具调用必填参数被吞 {names},丢弃本轮"
+                            f"{'重试' if use_nonstream else ',改非流式重试'}"
+                            f" ({attempt + 1}/{self._MAX_MALFORMED_ATTEMPTS})"
+                        ),
+                    })
+                    continue
                 return response, False
             # 先尝试就地抢救:畸形是 char-0 垃圾前缀 + 尾部完好 JSON(定层已证 provider-wire),
             # 全部畸形 tool_call 都能抠出干净 JSON 才改写并当轮继续,省掉一次非流式重试;
@@ -824,6 +989,26 @@ class AgentLoop:
             })
             return result, False
 
+        # err-streak 拦截:换着参数撞同一堵墙(如 edit 反复 old_str not found)。拦一次逼换路,
+        # 拦后 streak 重置到 SOFT(非永久封死)。
+        if self._repeat_guard.should_block_err(name):
+            cnt, esig = self._repeat_guard.register_err_block(name)
+            result = (
+                f"[已拦截重复调用] {name} 已连续 {cnt} 次撞同一个错误「{esig}」(每次只微调了参数)。"
+                "再这么试下去不会有新结果。换个做法:① 先 read 目标文件/用 grep 看确切内容"
+                "(old_str 必须逐字匹配,含空白与缩进);② 或换工具/换思路;③ 实在推进不了就停下来"
+                "如实告诉用户卡在哪。"
+            )
+            self._emit({"type": "warn", "msg": f"拦截撞墙调用 {name}(连续同错第 {cnt} 次)"})
+            self._emit({
+                "type": "tool_result",
+                "name": name,
+                "result": result,
+                "preview": result,
+                "truncated": False,
+            })
+            return result, False
+
         ctx = ExecCtx(
             user_id=self.user_id,
             task_id=self.session.task_id,
@@ -879,6 +1064,15 @@ class AgentLoop:
                 f"\n\n[重复调用警告] 你已用完全相同的参数调用 {name} {unproductive + 1} 次、结果没有变化。"
                 "再原样重调不会有新结果——换参数/换工具/换思路,或停下来向用户说明卡在哪。"
             )
+        else:
+            # err-streak SOFT 提示:参数每次不同但连撞同一个错(arg 判据不累计,单独提示)。
+            cnt, n_args, esig = self._repeat_guard.err_streak(name)
+            if cnt == _RepeatGuard.SOFT and n_args >= 2:
+                self._emit({"type": "warn", "msg": f"{name} 连续 {cnt} 次同错「{esig[:40]}」,已提示模型换路"})
+                result += (
+                    f"\n\n[撞墙警告] 你换着参数调用 {name} 已连续 {cnt} 次撞同一个错误「{esig}」。"
+                    "光微调参数没用——先 read/grep 看清目标的确切内容再动手,或换工具/换思路。"
+                )
 
         # 平台层产物机检(0.35.1 复发后落地):本步 shell/run_python 新产出的 .pptx
         # 若命中「整页贴图」伪导出特征,把 ERROR 注入 tool 结果逼模型当场返工。

tests/test_loop_repeat_guard.py

@@ -3,10 +3,11 @@ from __future__ import annotations
 import sys
 import unittest
 from pathlib import Path
+from types import SimpleNamespace
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
-from core.loop import _RepeatGuard  # noqa: E402
+from core.loop import _RepeatGuard, _toolcalls_partial_args  # noqa: E402
 
 
 def _simulate(guard: _RepeatGuard, name: str, args, results: list[str]) -> list[str]:
@@ -98,5 +99,126 @@ class TestRepeatGuard(unittest.TestCase):
         self.assertFalse(g.should_block("document_search", {"queries": ["b"]}))
 
 
+class TestErrStreak(unittest.TestCase):
+    """第二道判据:换着参数撞同一堵墙(如 edit 反复 old_str not found)。"""
+
+    def _run_varargs(self, g, name, errs):
+        """每次用不同 args 调用,结果为 errs[i]。返回是否曾被 err-block。"""
+        blocked = False
+        for i, r in enumerate(errs):
+            if g.should_block_err(name):
+                g.register_err_block(name)
+                blocked = True
+                continue
+            g.record(name, {"old_str": f"variant-{i}", "path": "/x/y.py"}, r)
+        return blocked
+
+    def test_varied_args_same_error_blocks(self):
+        """参数每次不同、错误签名相同(路径抹平)→ err-streak 累计到 HARD 后拦一次。"""
+        g = _RepeatGuard()
+        errs = [f"[Error] old_str not found in /proj/file{i}.py" for i in range(8)]
+        blocked = self._run_varargs(g, "edit", errs)
+        self.assertTrue(blocked)
+
+    def test_register_err_block_resets_to_soft(self):
+        """拦截后 streak 重置到 SOFT(非永久封死),留活口。"""
+        g = _RepeatGuard()
+        for i in range(_RepeatGuard.HARD + 1):
+            g.record("edit", {"old_str": f"v{i}", "path": "/a/b.py"}, "[Error] old_str not found in /a/b.py")
+        self.assertTrue(g.should_block_err("edit"))
+        cnt, esig = g.register_err_block("edit")
+        self.assertGreaterEqual(cnt, _RepeatGuard.HARD)
+        # 重置后不再处于拦截态
+        self.assertFalse(g.should_block_err("edit"))
+
+    def test_different_errors_not_blocked(self):
+        """每次撞的是不同错误 → 不算撞同一堵墙,不拦。"""
+        g = _RepeatGuard()
+        errs = [
+            "[Error] old_str not found in /a.py",
+            "[Error] old_str appears 3 times in /a.py",
+            "[Error] permission denied",
+            "[Error] file not found",
+        ] * 2
+        blocked = self._run_varargs(g, "edit", errs)
+        self.assertFalse(blocked)
+
+    def test_productive_result_resets_streak(self):
+        """撞墙几次后成功一次(非错误)→ streak 清零,不进拦截态。"""
+        g = _RepeatGuard()
+        for i in range(3):
+            g.record("edit", {"old_str": f"v{i}", "path": "/a.py"}, "[Error] old_str not found in /a.py")
+        g.record("edit", {"old_str": "vok", "path": "/a.py"}, "edited /a.py")  # 成功
+        cnt, n_args, _ = g.err_streak("edit")
+        self.assertEqual(cnt, 0)
+        self.assertFalse(g.should_block_err("edit"))
+
+    def test_single_arg_not_err_blocked(self):
+        """同一 args 反复撞同错 → 由 arg 判据管(err-block 要求 >=2 个不同 args)。"""
+        g = _RepeatGuard()
+        for _ in range(_RepeatGuard.HARD + 2):
+            g.record("edit", {"old_str": "same", "path": "/a.py"}, "[Error] old_str not found in /a.py")
+        cnt, n_args, _ = g.err_streak("edit")
+        self.assertGreaterEqual(cnt, _RepeatGuard.HARD)
+        self.assertEqual(n_args, 1)
+        self.assertFalse(g.should_block_err("edit"))  # 单 arg 不走 err-block
+
+
+def _resp_with_toolcalls(*calls):
+    """calls: (name, arguments_str) → 构造带 tool_calls 的假 response。"""
+    tcs = [
+        SimpleNamespace(function=SimpleNamespace(name=n, arguments=a))
+        for n, a in calls
+    ]
+    msg = SimpleNamespace(tool_calls=tcs)
+    return SimpleNamespace(choices=[SimpleNamespace(message=msg)])
+
+
+class TestPartialArgs(unittest.TestCase):
+    """③ 必填 key 被吞的畸形检测(parse 成功、salvage 无能)。"""
+
+    REQ = {"edit": ["path", "old_str", "new_str"], "write": ["path", "content"]}
+
+    def test_missing_path_detected(self):
+        """#3 实证形态:new_str/old_str 在、path 被吞 → 检出 missing=['path']。"""
+        r = _resp_with_toolcalls(("edit", '{"new_str": "b", "old_str": "a...gen.py"}'))
+        out = _toolcalls_partial_args(r, self.REQ)
+        self.assertEqual(len(out), 1)
+        _, name, missing = out[0]
+        self.assertEqual(name, "edit")
+        self.assertEqual(missing, ["path"])
+
+    def test_path_only_detected(self):
+        """#4 反向形态:只有 path、其余被挤掉 → 检出(0<2<3)。"""
+        r = _resp_with_toolcalls(("edit", '{"path": "/x/y.docx\\nTASK=1"}'))
+        out = _toolcalls_partial_args(r, self.REQ)
+        self.assertEqual(len(out), 1)
+        self.assertEqual(sorted(out[0][2]), ["new_str", "old_str"])
+
+    def test_full_args_not_flagged(self):
+        r = _resp_with_toolcalls(("edit", '{"path": "a", "old_str": "b", "new_str": "c"}'))
+        self.assertEqual(_toolcalls_partial_args(r, self.REQ), [])
+
+    def test_empty_obj_not_flagged(self):
+        """空 {} 不算(交给 executor + _RepeatGuard 现状)。"""
+        r = _resp_with_toolcalls(("edit", "{}"))
+        self.assertEqual(_toolcalls_partial_args(r, self.REQ), [])
+
+    def test_all_required_missing_not_flagged(self):
+        """必填全缺(纯无关键)→ 不算 partial(len(missing)==len(required))。"""
+        r = _resp_with_toolcalls(("edit", '{"foo": 1}'))
+        self.assertEqual(_toolcalls_partial_args(r, self.REQ), [])
+
+    def test_parse_fail_skipped(self):
+        """parse 失败归 _malformed_tool_calls,这里不重复处理。"""
+        r = _resp_with_toolcalls(("edit", 'garbage{"path":'))
+        self.assertEqual(_toolcalls_partial_args(r, self.REQ), [])
+
+    def test_write_missing_content(self):
+        r = _resp_with_toolcalls(("write", '{"path": "a.txt"}'))
+        out = _toolcalls_partial_args(r, self.REQ)
+        self.assertEqual(out[0][2], ["content"])
+
+
 if __name__ == "__main__":
     unittest.main()