| 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, "" |
|
|
| |
| self._recent_scores.append(threat_score) |
|
|
| |
| if threat_score < self.threshold: |
| self._recent_scores.clear() |
| self._recent_scores.append(threat_score) |
| self._triggered = False |
| return False, "" |
|
|
| |
| 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 |
