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

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

2 

3from dataclasses import dataclass, field 

4from typing import Any 

5 

6 

7@dataclass 

8class ConversationVelocity: 

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

10 

11 Tracks recent token deltas to predict future growth and trigger 

12 compaction before overflow occurs. 

13 """ 

14 

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

16 window_size: int = 10 # Track last N turns 

17 

18 @property 

19 def avg_growth(self) -> float: 

20 """Calculate average tokens per turn. 

21 

22 Returns: 

23 Average token growth across tracked turns 

24 """ 

25 if not self.turn_deltas: 

26 return 0.0 

27 

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

29 

30 @property 

31 def growth_trend(self) -> float: 

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

33 

34 Compares recent half vs older half to detect acceleration. 

35 

36 Returns: 

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

38 """ 

39 if len(self.turn_deltas) < 4: 

40 return 0.0 # Need at least 4 turns for trend 

41 

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

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

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

45 

46 return recent_avg - older_avg 

47 

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

49 """Predict token count N turns ahead. 

50 

51 Uses linear projection with acceleration boost. 

52 

53 Args: 

54 turns: Number of turns to project ahead 

55 

56 Returns: 

57 Predicted token count increase 

58 """ 

59 if not self.turn_deltas: 

60 return 0 

61 

62 base_prediction = self.avg_growth * turns 

63 

64 # Add acceleration boost if accelerating 

65 if self.growth_trend > 0: 

66 acceleration_boost = self.growth_trend * turns * 0.5 

67 base_prediction += acceleration_boost 

68 

69 return int(base_prediction) 

70 

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

72 """Add a new token delta. 

73 

74 Args: 

75 tokens: Token count for this turn 

76 """ 

77 self.turn_deltas.append(tokens) 

78 

79 # Keep only last window_size deltas 

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

81 self.turn_deltas.pop(0) 

82 

83 def to_dict(self) -> dict[str, Any]: 

84 """Convert to dictionary for storage. 

85 

86 Returns: 

87 Dictionary representation 

88 """ 

89 return { 

90 "turn_deltas": self.turn_deltas, 

91 "window_size": self.window_size, 

92 } 

93 

94 @classmethod 

95 def from_dict(cls, data: dict[str, Any]) -> "ConversationVelocity": 

96 """Create from dictionary. 

97 

98 Args: 

99 data: Dictionary representation 

100 

101 Returns: 

102 ConversationVelocity instance 

103 """ 

104 return cls( 

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

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

107 )