1import asyncio
2
3from doom_loop import DoomLoopConfig, DoomLoopDetector, RecoveryAction
4
5
6def make_detector():
7    return DoomLoopDetector(DoomLoopConfig(
8        warn_threshold_base=2, critical_threshold_base=3, adaptive_enabled=False,
9    ))
10
11
12def feed_failures(d, n=6):
13    for _ in range(n):
14        d.record_tool_call("run_tests", {})
15        d.record_tool_outcome("run_tests", {}, "FAILED: 3 errors")
16
17
18def test_no_loop_initially():
19    d = make_detector()
20    d.record_tool_call("read_file", {"path": "/a"})
21    d.record_tool_outcome("read_file", {"path": "/a"}, "file contents here")
22    res = d.detect()
23    assert not res.stuck
24
25
26def test_repeat_loop_detected():
27    d = make_detector()
28    feed_failures(d)
29    res = d.detect()
30    assert res.stuck
31
32
33def test_recovery_callback_fires():
34    d = make_detector()
35    fired = []
36
37    async def on_recover(result):
38        fired.append(result)
39        return True
40
41    # generic_repeat loops recover via SWITCH_TOOL by default
42    d.register_recovery_callback(RecoveryAction.SWITCH_TOOL, on_recover)
43    feed_failures(d)
44    res = d.detect()
45    assert res.stuck
46    ok = asyncio.run(d.execute_recovery(res))
47    assert len(fired) == 1
48    # the loop pattern is still in the call history, so post-recovery
49    # validation escalates and reports failure (by design)
50    assert ok is False
51