from dataclasses import dataclass from typing import Optional from collections import deque from ..config import settings @dataclass class KillSwitchResult: triggered: bool reason: str consecutive_detections: int threat_score: float class KillSwitch: """ Active prevention kill-switch. Triggers TERMINATE_CALL when threat score > 70% sustained over two consecutive sliding windows (500ms). """ def __init__( self, threshold: float = settings.threat_threshold, consecutive_windows: int = settings.consecutive_windows, ): self.threshold = threshold self.consecutive_windows = consecutive_windows self._recent_scores: deque = deque(maxlen=consecutive_windows) self._triggered = False self._trigger_count = 0 self.enabled = True def check(self, threat_score: float) -> tuple[bool, str]: """ Check if kill-switch should trigger given the current threat score. Rules: - Threat score > threshold - Sustained over consecutive_windows windows - Score must be above threshold in CURRENT window too (reset on drop) Returns (triggered, reason) """ if not self.enabled: return False, "" # Record current score self._recent_scores.append(threat_score) # If current window drops below threshold, reset the streak if threat_score < self.threshold: self._recent_scores.clear() self._recent_scores.append(threat_score) self._triggered = False return False, "" # Check if we have enough consecutive windows above threshold if len(self._recent_scores) >= self.consecutive_windows: all_above = all(s > self.threshold for s in self._recent_scores) if all_above and not self._triggered: self._triggered = True self._trigger_count += 1 reason = "SYNTHETIC_IMPERSONATION" return True, reason return False, "" @property def triggered(self) -> bool: return self._triggered @property def trigger_count(self) -> int: return self._trigger_count def reset(self) -> None: self._recent_scores.clear() self._triggered = False def arm(self) -> None: self.enabled = True self._triggered = False def disarm(self) -> None: self.enabled = False self._triggered = False