Coverage for narrative_harm_classifier/classifier/tracking/tracker.py: 98%

1""" 

2classifier/tracking/tracker.py — Escalation-chain scoring. 

3 

4Computes trend and risk level from simple, explainable arithmetic over a 

5rolling window of observations (first-half vs second-half average severity 

6delta), rather than a learned model — consistent with the rest of this 

7project's rationale-driven design. 

8""" 

9 

10from narrative_harm_classifier.core.models import ClassifyRequest 

11from narrative_harm_classifier.classifier.rules.engine import ClassificationEngine 

12from narrative_harm_classifier.classifier.provenance import GENESIS_HASH, record_hash as compute_record_hash 

13from narrative_harm_classifier.classifier.tracking.models import ( 

14 ChainVerification, 

15 Observation, 

16 SourceProfile, 

17 SeverityLevel, 

18 severity_for_mechanism, 

19 risk_level_for, 

20) 

21from narrative_harm_classifier.classifier.tracking.store import TrackingStore 

22 

23# Minimum average-severity delta (second half vs first half) to call a trend 

24# escalating/de-escalating rather than stable. Chosen to require at least 

25# roughly one severity-level shift on average across the window. 

26TREND_THRESHOLD = 0.25 

27 

28DEFAULT_WINDOW = 20 

29 

30 

31class EscalationTracker: 

32 def __init__(self, engine: ClassificationEngine, store: TrackingStore): 

33 self.engine = engine 

34 self.store = store 

35 

36 def observe(self, source_id: str, request: ClassifyRequest) -> Observation: 

37 result = self.engine.classify(request) 

38 severity = severity_for_mechanism(result.harm_mechanism) 

39 observed_at = result.classified_at 

40 

41 prev_hash = self.store.last_record_hash(source_id) 

42 this_record_hash = compute_record_hash( 

43 prev_hash=prev_hash, 

44 source_id=source_id, 

45 text_excerpt=result.text_excerpt, 

46 is_harmful=result.is_harmful, 

47 harm_mechanism=result.harm_mechanism, 

48 confidence=result.confidence, 

49 observed_at=observed_at, 

50 content_hash_value=result.content_hash, 

51 ) 

52 

53 obs = Observation( 

54 source_id=source_id, 

55 text_excerpt=result.text_excerpt, 

56 is_harmful=result.is_harmful, 

57 harm_category=result.harm_category, 

58 harm_mechanism=result.harm_mechanism, 

59 confidence=result.confidence, 

60 severity=severity, 

61 observed_at=observed_at, 

62 content_hash=result.content_hash, 

63 prev_hash=prev_hash, 

64 record_hash=this_record_hash, 

65 ) 

66 return self.store.add_observation(obs) 

67 

68 def verify_chain(self, source_id: str) -> ChainVerification: 

69 """ 

70 Recompute the hash chain for a source's full observation history and 

71 confirm every record's stored record_hash matches what it should be 

72 given the previous record's hash — detects tampering with any 

73 historical record, though it does not prevent it. 

74 """ 

75 history = self.store.history(source_id) 

76 if not history: 

77 return ChainVerification(source_id=source_id, observation_count=0, intact=True) 

78 

79 expected_prev = GENESIS_HASH 

80 for obs in history: 

81 expected_hash = compute_record_hash( 

82 prev_hash=expected_prev, 

83 source_id=obs.source_id, 

84 text_excerpt=obs.text_excerpt, 

85 is_harmful=obs.is_harmful, 

86 harm_mechanism=obs.harm_mechanism, 

87 confidence=obs.confidence, 

88 observed_at=obs.observed_at, 

89 content_hash_value=obs.content_hash, 

90 ) 

91 if obs.prev_hash != expected_prev or obs.record_hash != expected_hash: 

92 return ChainVerification( 

93 source_id=source_id, 

94 observation_count=len(history), 

95 intact=False, 

96 first_broken_id=obs.id, 

97 ) 

98 expected_prev = obs.record_hash 

99 

100 return ChainVerification(source_id=source_id, observation_count=len(history), intact=True) 

101 

102 def profile(self, source_id: str, window: int = DEFAULT_WINDOW) -> SourceProfile: 

103 history = self.store.history(source_id, limit=window) 

104 

105 if not history: 

106 return SourceProfile( 

107 source_id=source_id, 

108 observation_count=0, 

109 current_severity=SeverityLevel.NONE, 

110 rolling_avg_severity=0.0, 

111 trend="insufficient_data", 

112 risk_level="low", 

113 history=[], 

114 ) 

115 

116 severities = [int(o.severity) for o in history] 

117 current_severity = history[-1].severity 

118 rolling_avg = sum(severities) / len(severities) 

119 

120 trend = self._compute_trend(severities) 

121 risk_level = risk_level_for(current_severity, trend) 

122 

123 return SourceProfile( 

124 source_id=source_id, 

125 observation_count=len(history), 

126 current_severity=current_severity, 

127 rolling_avg_severity=round(rolling_avg, 3), 

128 trend=trend, 

129 risk_level=risk_level, 

130 history=history, 

131 ) 

132 

133 def list_profiles(self, window: int = DEFAULT_WINDOW) -> list[SourceProfile]: 

134 profiles = [self.profile(sid, window=window) for sid in self.store.list_source_ids()] 

135 profiles.sort(key=lambda p: (p.current_severity, p.rolling_avg_severity), reverse=True) 

136 return profiles 

137 

138 @staticmethod 

139 def _compute_trend(severities: list[int]) -> str: 

140 if len(severities) < 2: 

141 return "insufficient_data" 

142 

143 midpoint = len(severities) // 2 

144 first_half = severities[:midpoint] if midpoint > 0 else severities[:1] 

145 second_half = severities[midpoint:] 

146 

147 first_avg = sum(first_half) / len(first_half) 

148 second_avg = sum(second_half) / len(second_half) 

149 delta = second_avg - first_avg 

150 

151 if delta >= TREND_THRESHOLD: 

152 return "escalating" 

153 if delta <= -TREND_THRESHOLD: 

154 return "de-escalating" 

155 return "stable"