merge: stuck熔断签名富化+工具失败即时观测(80009误熔断+归因盲区双修) (cutover S2)

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

tier2/gen-worker/tests/test_circuit_stuck_signature.py

@@ -0,0 +1,124 @@
+"""stuck 熔断签名富化单测(80009 事故回归):同错反复才判死圈,异类失败不误熔断,签名带工具名+错误头。
+跑:PYTHONPATH=tier2/gen-worker cheap-worker/.venv/bin/python -m pytest tier2/gen-worker/tests/test_circuit_stuck_signature.py -v
+"""
+import asyncio
+
+import pytest
+
+from worker.middleware import CircuitBreakerMiddleware, Tier2CircuitBreak
+
+
+# ── 假事件:熔断按 type(evt).__name__ 分派,类名必须与 AgentScope 事件同名 ──
+
+class ToolCallStartEvent:
+    def __init__(self, tcid: str, name: str) -> None:
+        self.tool_call_id = tcid
+        self.tool_call_name = name
+
+
+class ToolResultTextDeltaEvent:
+    def __init__(self, tcid: str, delta: str) -> None:
+        self.tool_call_id = tcid
+        self.delta = delta
+
+
+class ToolResultEndEvent:
+    def __init__(self, tcid: str, state: str) -> None:
+        self.tool_call_id = tcid
+        self.state = state  # use_enum_values=True → 真实事件里也是字符串值
+
+
+def _breaker(threshold: int = 4) -> CircuitBreakerMiddleware:
+    # 其余闸放宽到打不着,只测 stuck;pricing_params 传入避免测试内打网关取价。
+    return CircuitBreakerMiddleware(
+        max_tool_calls=999, max_model_calls=999,
+        wall_timeout_s=9999.0, step_timeout_s=9999.0,
+        stuck_repeat_threshold=threshold,
+        pricing_params={"pricing": {}, "qpu": 1, "usd_rate": 0},
+    )
+
+
+def _fail_seq(tcid: str, name: str, text: str) -> list:
+    """一次完整失败:call start → 结果文本 → 结果 error 终态。"""
+    return [
+        ToolCallStartEvent(tcid, name),
+        ToolResultTextDeltaEvent(tcid, text),
+        ToolResultEndEvent(tcid, "error"),
+    ]
+
+
+def _ok_seq(tcid: str, name: str) -> list:
+    return [ToolCallStartEvent(tcid, name), ToolResultEndEvent(tcid, "success")]
+
+
+def _drive(b: CircuitBreakerMiddleware, events: list) -> None:
+    async def handler(**kwargs):
+        for e in events:
+            yield e
+
+    async def run():
+        async for _ in b.on_reply(None, {}, handler):
+            pass
+
+    asyncio.run(run())
+
+
+def test_same_error_repeated_trips_with_rich_signature():
+    # 同工具同错 4 连败 → 熔断,且签名/原因带工具名与归一化错误头(不再是裸 "error")。
+    b = _breaker(threshold=4)
+    events = []
+    for i in range(4):
+        events += _fail_seq(f"call_{i:08x}", "build", "esbuild: Cannot find module 'phaser'")
+    with pytest.raises(Tier2CircuitBreak):
+        _drive(b, events)
+    assert b.tripped is not None and b.tripped.get("kind") == "stuck"
+    assert "build:error:" in b.tripped.get("reason", ""), "熔断原因必须带工具名签名"
+
+
+def test_signature_normalizes_varying_ids_and_numbers():
+    # 同类错误只差 id/行号/端口 → 归一化后仍同签名,照样判死圈。
+    b = _breaker(threshold=4)
+    events = []
+    for i in range(4):
+        events += _fail_seq(
+            f"call_{i:08x}", "run_gates",
+            f"端口 {9200 + i} 被占;session=deadbeef{i:07x} 起跑失败",
+        )
+    with pytest.raises(Tier2CircuitBreak):
+        _drive(b, events)
+    assert "run_gates:error:" in b.tripped.get("reason", "")
+
+
+def test_heterogeneous_failures_do_not_trip():
+    # 异类失败交替 4 次(工具名/错误都不同)→ 不判死圈(交 step_cap/¥ 闸兜底;80009 误熔断回归)。
+    b = _breaker(threshold=4)
+    events = []
+    events += _fail_seq("c1", "build", "esbuild missing")
+    events += _fail_seq("c2", "run_gates", "port busy")
+    events += _fail_seq("c3", "build", "esbuild missing")
+    events += _fail_seq("c4", "run_gates", "port busy")
+    _drive(b, events)  # 不应抛
+    assert b.tripped is None
+
+
+def test_success_resets_streak():
+    # 3 败 + 1 成 + 3 败 → 连续性被打断,不熔断。
+    b = _breaker(threshold=4)
+    events = []
+    for i in range(3):
+        events += _fail_seq(f"a{i}", "build", "same err")
+    events += _ok_seq("ok1", "read_file")
+    for i in range(3):
+        events += _fail_seq(f"b{i}", "build", "same err")
+    _drive(b, events)
+    assert b.tripped is None
+
+
+def test_tracking_dicts_bounded_and_drained():
+    # End 事件弹出映射:正常流转后无泄漏;in-flight 上限受 _SIG_TRACK_CAP 约束(此处只验清空)。
+    b = _breaker(threshold=99)
+    events = []
+    for i in range(6):
+        events += _fail_seq(f"x{i}", "build", "boom")
+    _drive(b, events)
+    assert b._tool_names == {} and b._result_heads == {}

tier2/gen-worker/worker/middleware.py

@@ -42,6 +42,7 @@ trace 接线(H1/H2,Tier2TraceMiddleware,本模块第二个 middleware;U3 升级
   再兜一层(_safe_ingest)——trace 失败绝不中断生成,只告警;相位钩子里下层真异常照常上抛(不吞)。
 """
 
+import re
 import time
 from datetime import datetime
 from typing import Optional
@@ -176,6 +177,14 @@ DEFAULT_RMB_HARD_LIMIT = genconfig.get("budget", "rmb_hard_limit", 50.0)  # 富
 #   finish 是收尾链本体,均放行。
 DEFAULT_SOFT_STOP_BLOCKED_TOOLS = frozenset({"write_file", "write_source", "scaffold_init"})
 
+# stuck 失败签名归一化:长 hex(≥8 位,如 tool_call_id/session id)与数字串 → "#",
+#   使「同类错误、仅路径/id/行号不同」仍判为同签名(设计意图=「反复同错」死圈;80009 事故实证:
+#   旧实现签名恒为状态值 "error",4 次异类临时错也误熔断,且不留工具名/错误文本 → 生产盲区)。
+_SIG_NORM_RE = re.compile(r"[0-9a-fA-F]{8,}|\d+")
+# 失败签名/观测缓存上限:in-flight tool_call 映射与结果文本头的字典大小、单条文本头长度。
+_SIG_TRACK_CAP = 64
+_SIG_HEAD_CAP = 160
+
 
 class CircuitBreakerMiddleware(MiddlewareBase):
     """四道熔断 + 软刹 + ¥ 累进硬闸(on_reply / on_model_call onion + on_system_prompt transformer)。
@@ -289,6 +298,11 @@ class CircuitBreakerMiddleware(MiddlewareBase):
         # 卡死探测:记最近一次工具失败签名 + 连续重复计数。
         self._last_fail_sig: str | None = None
         self._fail_repeat = 0
+        # 卡死探测辅助(签名富化):tool_call_id → 工具名(ToolCall/ToolResultStart 记,ToolResultEnd 弹出);
+        #   tool_call_id → 结果文本头(ToolResultTextDeltaEvent 累积,单条截 _SIG_HEAD_CAP)。
+        #   两 dict 均有界(_SIG_TRACK_CAP,防漏 End 事件泄漏);越界丢新条目,签名退化为 "?"+状态,不失效。
+        self._tool_names: dict[str, str] = {}
+        self._result_heads: dict[str, str] = {}
         # ¥ 累进硬闸:累计已花 ¥(每次模型调用后按实测 usage 折算累加)+ 当前是否走金额闸(取到活价)。
         self.spent_rmb = 0.0
         self._rmb_gate_active = self._pricing_params is not None  # 有计费参数才按金额拦,否则降级次数闸
@@ -351,15 +365,39 @@ class CircuitBreakerMiddleware(MiddlewareBase):
                 if self.tool_calls > self.max_tool_calls:
                     self._trip("step_cap", f"工具调用步数超硬顶(>{self.max_tool_calls})")
 
+            # ── ③ stuck 辅助:记 tool_call_id → 工具名 / 结果文本头(签名富化素材)──
+            if evt_type in ("ToolCallStartEvent", "ToolResultStartEvent"):
+                _tcid = getattr(evt, "tool_call_id", "")
+                _tname = getattr(evt, "tool_call_name", "") or "?"
+                if _tcid and (_tcid in self._tool_names or len(self._tool_names) < _SIG_TRACK_CAP):
+                    self._tool_names[_tcid] = _tname
+            if evt_type == "ToolResultTextDeltaEvent":
+                _tcid = getattr(evt, "tool_call_id", "")
+                if _tcid and (_tcid in self._result_heads or len(self._result_heads) < _SIG_TRACK_CAP):
+                    _head = self._result_heads.get(_tcid, "")
+                    if len(_head) < _SIG_HEAD_CAP:
+                        self._result_heads[_tcid] = (_head + str(getattr(evt, "delta", "")))[:_SIG_HEAD_CAP]
+
             # ── ③ stuck:连续同一失败签名 → 死圈 ──
             if evt_type == "ToolResultEndEvent":
                 # state 是 ToolResultState(use_enum_values=True → 多为字符串值);失败态计入签名。
                 state = getattr(evt, "state", None)
                 state_val = getattr(state, "value", state)
                 tool_call_id = getattr(evt, "tool_call_id", "")
+                tool_name = self._tool_names.pop(tool_call_id, "?")
+                head_raw = self._result_heads.pop(tool_call_id, "")
                 if str(state_val).lower() in ("error", "denied", "interrupted"):
-                    # 失败签名 = 状态值(tool_call_id 每次不同,不入签名以便识别「同类失败反复」)。
-                    sig = str(state_val).lower()
+                    # 失败签名 = 工具名 + 状态 + 归一化错误头(数字/长hex → "#"):按设计意图只把
+                    #   「同类失败反复」判死圈;异类失败连发交 step_cap/¥ 闸/timeout 兜底,不再误熔断。
+                    #   (80009 事故:旧签名恒为 "error",异类 4 连败也熔断,且事后无从知道是哪个工具。)
+                    norm_head = _SIG_NORM_RE.sub("#", head_raw)[:80]
+                    sig = f"{tool_name}:{str(state_val).lower()}:{norm_head}"
+                    # 生产观测:工具失败即时打印(名字+错误头),盲修空转/死圈归因不再是黑箱。
+                    print(
+                        f"[tier2-circuit] 工具失败 name={tool_name} state={state_val}"
+                        f" id={tool_call_id} head={head_raw[:120]!r}",
+                        flush=True,
+                    )
                     if sig == self._last_fail_sig:
                         self._fail_repeat += 1
                     else: