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, _toolcalls_partial_args # noqa: E402 def _simulate(guard: _RepeatGuard, name: str, args, results: list[str]) -> list[str]: """模拟 loop 逐次调用:先 should_block,未拦才 record。返回每次的判定标记。 'BLOCK' = 被拦截未执行;否则返回 'exec(unprod=N)'。 """ out = [] for r in results: if guard.should_block(name, args): guard.register_block(name, args) out.append("BLOCK") continue unprod, _productive = guard.record(name, args, r) out.append(f"exec(unprod={unprod})") return out class TestRepeatGuard(unittest.TestCase): def test_identical_error_repeats_get_blocked(self): g = _RepeatGuard() trace = _simulate(g, "glob", {"path": "/home/ubuntu/zcbot"}, ["[Error] base path not found"] * 8) # 第一次执行无产出计 0,之后每次 +1;累计到 HARD 后拦截 self.assertIn("BLOCK", trace) self.assertTrue(g.should_block("glob", {"path": "/home/ubuntu/zcbot"})) # 拦截前最多放过 HARD 次无产出重复(共 HARD+1 次执行) n_exec = sum(1 for t in trace if t.startswith("exec")) self.assertEqual(n_exec, _RepeatGuard.HARD + 1) def test_empty_args_error_storm_blocked(self): """空 {} 缺参风暴:executor 每次返回同一句错误 → 被同一机制拦下(堵 malformed 洞)。""" g = _RepeatGuard() trace = _simulate(g, "shell", {}, ["[Error] 缺少必填参数 [command]"] * 7) self.assertIn("BLOCK", trace) def test_identical_nonerror_result_blocked(self): """同参且结果一字不差(非错误)也算无产出 → 拦截。""" g = _RepeatGuard() trace = _simulate(g, "read", {"path": "a.txt"}, ["same content"] * 8) self.assertIn("BLOCK", trace) def test_changing_results_never_blocked(self): """同参但每次结果不同(改脚本后重跑)= 有产出 → 永不拦,计数清零。""" g = _RepeatGuard() results = [f"[stdout]\nrun {i} output\n[exit 0]" for i in range(10)] trace = _simulate(g, "run_python", {"script_path": "x.py"}, results) self.assertNotIn("BLOCK", trace) self.assertFalse(g.should_block("run_python", {"script_path": "x.py"})) # 每次都是新结果,无产出计数恒为 0 self.assertTrue(all(t == "exec(unprod=0)" for t in trace)) def test_productive_result_resets_counter(self): """报错几次后拿到新结果(修好了)→ 计数清零,不会被先前的失败拖去拦截。""" g = _RepeatGuard() seq = ["[Error] x", "[Error] x", "[stdout]\nfixed!\n[exit 0]", "[stdout]\nfixed!\n[exit 0]"] _simulate(g, "shell", {"command": "make"}, seq) # 中途修好清零,不该进入 block 态 self.assertFalse(g.should_block("shell", {"command": "make"})) def test_soft_threshold_reached_before_hard(self): g = _RepeatGuard() unprods = [] for _ in range(_RepeatGuard.SOFT + 1): unprods.append(g.record("document_search", {"queries": ["x"]}, "(no documents found)")[0]) # 累计达到 SOFT(此时应注入软提示),但还没到 HARD 拦截 self.assertGreaterEqual(max(unprods), _RepeatGuard.SOFT) self.assertFalse(g.should_block("document_search", {"queries": ["x"]})) def test_record_returns_productive_signal(self): """record 第二个返回值喂全局无进展熔断:新非错结果=有产出,[Error]/重复=无产出。""" g = _RepeatGuard() # 首个新结果 → 有产出 _, p1 = g.record("read", {"path": "a"}, "[stdout] hello") self.assertTrue(p1) # 一字不差重复同一结果 → 无产出 _, p2 = g.record("read", {"path": "a"}, "[stdout] hello") self.assertFalse(p2) # 换出新结果 → 又有产出 _, p3 = g.record("read", {"path": "a"}, "[stdout] world") self.assertTrue(p3) # [Error] 开头 → 无产出(哪怕是该指纹首次) _, p4 = g.record("glob", {"path": "/nope"}, "[Error] not found") self.assertFalse(p4) def test_distinct_args_tracked_separately(self): g = _RepeatGuard() _simulate(g, "document_search", {"queries": ["a"]}, ["[Error] e"] * 8) # 不同参数互不影响 self.assertTrue(g.should_block("document_search", {"queries": ["a"]})) 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()