diff --git a/agentwatch/lattice/__init__.py b/agentwatch/lattice/__init__.py index 8f6cde85..2b64b0ce 100644 --- a/agentwatch/lattice/__init__.py +++ b/agentwatch/lattice/__init__.py @@ -11,6 +11,12 @@ ShadowFilesystem, ) +from agentwatch.lattice.miscalibration import ( + CalibrationEntry, + MiscalibrationDetector, + MiscalibrationResult, +) + __all__ = [ "CRITICAL_SYSTEM_PATHS", "FileAction", @@ -18,4 +24,7 @@ "MutationResult", "MutationType", "ShadowFilesystem", + "CalibrationEntry", + "MiscalibrationDetector", + "MiscalibrationResult", ] diff --git a/agentwatch/lattice/miscalibration.py b/agentwatch/lattice/miscalibration.py new file mode 100644 index 00000000..e1c9a32f --- /dev/null +++ b/agentwatch/lattice/miscalibration.py @@ -0,0 +1,82 @@ +"""Session-level confidence calibration using the Brier Score. + +Tracks an agent's stated confidence against actual outcomes to detect +systematic overconfidence or underconfidence. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +__all__ = [ + "CalibrationEntry", + "MiscalibrationDetector", + "MiscalibrationResult", +] + +@dataclass(frozen=True) +class CalibrationEntry: + """One prediction and its observed outcome.""" + + confidence: float + success: bool + + +@dataclass(frozen=True) +class MiscalibrationResult: + """Result returned by MiscalibrationDetector.""" + + brier_score: float + warning: bool + blocked: bool + sample_size: int + + +@dataclass +class MiscalibrationDetector: + """Detect session miscalibration using the Brier Score.""" + + warning_threshold: float = 0.15 + blocking_threshold: float = 0.25 + history: list[CalibrationEntry] = field(default_factory=list) + + def record(self, confidence: float, success: bool) -> None: + """Record one prediction and its observed outcome.""" + + if not 0.0 <= confidence <= 1.0: + raise ValueError( + f"confidence must be between 0.0 and 1.0, got {confidence}" + ) + + self.history.append( + CalibrationEntry( + confidence=confidence, + success=success, + ) + ) + + def brier_score(self) -> float: + """Return the session Brier Score.""" + + if not self.history: + return 0.0 + + total = 0.0 + + for entry in self.history: + actual = 1.0 if entry.success else 0.0 + total += (entry.confidence - actual) ** 2 + + return total / len(self.history) + + def evaluate(self) -> MiscalibrationResult: + """Evaluate the current session calibration.""" + + score = self.brier_score() + + return MiscalibrationResult( + brier_score=score, + warning=score > self.warning_threshold, + blocked=score > self.blocking_threshold, + sample_size=len(self.history), + ) \ No newline at end of file diff --git a/tests/test_miscalibration.py b/tests/test_miscalibration.py new file mode 100644 index 00000000..c0a8c18a --- /dev/null +++ b/tests/test_miscalibration.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import pytest + +from agentwatch.lattice.miscalibration import ( + MiscalibrationDetector, +) + + +def test_empty_history_has_zero_brier_score(): + detector = MiscalibrationDetector() + + assert detector.brier_score() == 0.0 + + result = detector.evaluate() + + assert result.brier_score == 0.0 + assert result.warning is False + assert result.blocked is False + assert result.sample_size == 0 + + +def test_record_adds_history_entry(): + detector = MiscalibrationDetector() + + detector.record(0.8, True) + + assert len(detector.history) == 1 + assert detector.history[0].confidence == 0.8 + assert detector.history[0].success is True + + +def test_invalid_confidence_raises_value_error(): + detector = MiscalibrationDetector() + + with pytest.raises(ValueError): + detector.record(1.2, True) + + with pytest.raises(ValueError): + detector.record(-0.1, False) + + +def test_good_calibration_does_not_trigger_warning(): + detector = MiscalibrationDetector() + + detector.record(1.0, True) + detector.record(0.0, False) + detector.record(0.9, True) + + result = detector.evaluate() + + assert result.warning is False + assert result.blocked is False + + +def test_warning_threshold_is_triggered(): + detector = MiscalibrationDetector() + + detector.record(0.5, True) + detector.record(0.5, False) + + result = detector.evaluate() + + assert result.warning is True + + +def test_blocking_threshold_is_triggered(): + detector = MiscalibrationDetector() + + detector.record(1.0, False) + detector.record(1.0, False) + + result = detector.evaluate() + + assert result.blocked is True + + +def test_sample_size_matches_history(): + detector = MiscalibrationDetector() + + detector.record(0.7, True) + detector.record(0.2, False) + detector.record(0.9, True) + + assert detector.evaluate().sample_size == 3 + +