feat(lattice): add miscalibration detector using Brier score

agentwatch/lattice/__init__.py

@@ -11,11 +11,20 @@
     ShadowFilesystem,
 )
 
+from agentwatch.lattice.miscalibration import (
+    CalibrationEntry,
+    MiscalibrationDetector,
+    MiscalibrationResult,
+)
+
 __all__ = [
     "CRITICAL_SYSTEM_PATHS",
     "FileAction",
     "FileOperation",
     "MutationResult",
     "MutationType",
     "ShadowFilesystem",
+    "CalibrationEntry",
+    "MiscalibrationDetector",
+    "MiscalibrationResult",
 ]

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),
+        )

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
+
+