diff --git a/tier2/gen-worker/tests/test_circuit_stuck_signature.py b/tier2/gen-worker/tests/test_circuit_stuck_signature.py new file mode 100644 index 00000000..c482e2f0 --- /dev/null +++ b/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 == {} diff --git a/tier2/gen-worker/worker/middleware.py b/tier2/gen-worker/worker/middleware.py index 8a293df2..851c02ee 100644 --- a/tier2/gen-worker/worker/middleware.py +++ b/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: