Coverage for little_loops / fsm / stall_detector.py: 38%

1"""Stall detector for repeated `(state, exit_code, verdict)` triples. 

2 

3Detects deterministic FSM stalls where the same state produces the 

4same exit code and verdict across consecutive iterations, allowing 

5loops to abort or route to a recovery state instead of burning their 

6iteration budget on a stuck transition. 

7 

8Treats timeout-driven errors identically to deterministic "no" 

9verdicts: `(state, exit_code=124, verdict="error")` is a stall just 

10as much as `(state, exit_code=124, verdict="no")` — see BUG-1640. 

11""" 

12 

13from __future__ import annotations 

14 

15from collections import deque 

16from dataclasses import dataclass 

17 

18 

19@dataclass(frozen=True) 

20class Stall: 

21 """A detected stall: the repeating triple and its consecutive count.""" 

22 

23 triple: tuple[str, int, str] 

24 count: int 

25 

26 

27class StallDetector: 

28 """Track consecutive identical `(state, exit_code, verdict)` triples. 

29 

30 `record()` appends a triple per FSM transition. `check()` returns a 

31 `Stall` when the last `window` recorded triples are all identical, 

32 else None. 

33 """ 

34 

35 def __init__(self, window: int) -> None: 

36 if window < 1: 

37 raise ValueError(f"window must be >= 1, got {window}") 

38 self._window: int = window 

39 self._recent: deque[tuple[str, int, str]] = deque(maxlen=window) 

40 self._last_fingerprints: dict[str, tuple[object, ...] | None] = {} 

41 

42 def record( 

43 self, 

44 state: str, 

45 exit_code: int, 

46 verdict: str, 

47 fingerprint: tuple[object, ...] | None = None, 

48 ) -> None: 

49 """Append a transition triple to the rolling window. 

50 

51 If *fingerprint* is provided and differs from the previous fingerprint 

52 recorded for *state*, the window is reset before appending — meaning 

53 observable file-level progress between two visits to the same 

54 eval-bearing state prevents a false-positive stall fire. 

55 """ 

56 if fingerprint is not None: 

57 prev = self._last_fingerprints.get(state) 

58 if prev is not None and fingerprint != prev: 

59 self.reset() 

60 self._last_fingerprints[state] = fingerprint 

61 self._recent.append((state, exit_code, verdict)) 

62 

63 def check(self) -> Stall | None: 

64 """Return a Stall if the last `window` triples are all identical.""" 

65 if len(self._recent) < self._window: 

66 return None 

67 first = self._recent[0] 

68 for triple in self._recent: 

69 if triple != first: 

70 return None 

71 return Stall(triple=first, count=self._window) 

72 

73 def reset(self) -> None: 

74 """Clear the rolling window and fingerprint cache.""" 

75 self._recent.clear() 

76 self._last_fingerprints.clear()