Coverage for src / harnessutils / models / velocity.py: 100%

1"""Conversation velocity tracking for predictive overflow detection.""" 

2 

3from dataclasses import dataclass, field 

4 

5 

6@dataclass 

7class ConversationVelocity: 

8 """Track token growth velocity for predictive overflow detection. 

9 

10 Tracks recent token deltas to predict future growth and trigger 

11 compaction before overflow occurs. 

12 """ 

13 

14 turn_deltas: list[int] = field(default_factory=list) 

15 window_size: int = 10 # Track last N turns 

16 

17 @property 

18 def avg_growth(self) -> float: 

19 """Calculate average tokens per turn. 

20 

21 Returns: 

22 Average token growth across tracked turns 

23 """ 

24 if not self.turn_deltas: 

25 return 0.0 

26 

27 return sum(self.turn_deltas) / len(self.turn_deltas) 

28 

29 @property 

30 def growth_trend(self) -> float: 

31 """Calculate acceleration (recent vs older growth). 

32 

33 Compares recent half vs older half to detect acceleration. 

34 

35 Returns: 

36 Positive = accelerating, Negative = decelerating, 0 = steady 

37 """ 

38 if len(self.turn_deltas) < 4: 

39 return 0.0 # Need at least 4 turns for trend 

40 

41 mid = len(self.turn_deltas) // 2 

42 recent_avg = sum(self.turn_deltas[mid:]) / len(self.turn_deltas[mid:]) 

43 older_avg = sum(self.turn_deltas[:mid]) / len(self.turn_deltas[:mid]) 

44 

45 return recent_avg - older_avg 

46 

47 def predict_tokens_ahead(self, turns: int) -> int: 

48 """Predict token count N turns ahead. 

49 

50 Uses linear projection with acceleration boost. 

51 

52 Args: 

53 turns: Number of turns to project ahead 

54 

55 Returns: 

56 Predicted token count increase 

57 """ 

58 if not self.turn_deltas: 

59 return 0 

60 

61 base_prediction = self.avg_growth * turns 

62 

63 # Add acceleration boost if accelerating 

64 if self.growth_trend > 0: 

65 acceleration_boost = self.growth_trend * turns * 0.5 

66 base_prediction += acceleration_boost 

67 

68 return int(base_prediction) 

69 

70 def add_delta(self, tokens: int) -> None: 

71 """Add a new token delta. 

72 

73 Args: 

74 tokens: Token count for this turn 

75 """ 

76 self.turn_deltas.append(tokens) 

77 

78 # Keep only last window_size deltas 

79 if len(self.turn_deltas) > self.window_size: 

80 self.turn_deltas.pop(0) 

81 

82 def to_dict(self) -> dict: 

83 """Convert to dictionary for storage. 

84 

85 Returns: 

86 Dictionary representation 

87 """ 

88 return { 

89 "turn_deltas": self.turn_deltas, 

90 "window_size": self.window_size, 

91 } 

92 

93 @classmethod 

94 def from_dict(cls, data: dict) -> "ConversationVelocity": 

95 """Create from dictionary. 

96 

97 Args: 

98 data: Dictionary representation 

99 

100 Returns: 

101 ConversationVelocity instance 

102 """ 

103 return cls( 

104 turn_deltas=data.get("turn_deltas", []), 

105 window_size=data.get("window_size", 10), 

106 )