feat(#270): require N consecutive successes in HALF_OPEN before closing circuit

app/ai-service/services/circuit_breaker.py

@@ -21,7 +21,10 @@ class CircuitBreaker:
     States:
     - CLOSED: Normal operation. Requests flow through.
     - OPEN: Service is failing. Requests fail-fast (return False/raise error).
-    - HALF_OPEN: Recovery window elapsed. Allow a request to test downstream health.
+    - HALF_OPEN: Recovery window elapsed. Probe requests are allowed through.
+      The breaker only returns to CLOSED after *success_threshold_in_half_open*
+      consecutive successes (default 2), preventing premature closure on a
+      single lucky probe.
 
     The breaker publishes Prometheus metrics on every state change:
       - CIRCUIT_STATE (Gauge): current state, encoded as 0/1/2.
@@ -31,13 +34,24 @@ class CircuitBreaker:
     exported values can never diverge from the underlying state.
     """
 
-    def __init__(self, name: str, failure_threshold: int = 3, recovery_timeout: float = 30.0):
+    def __init__(
+        self,
+        name: str,
+        failure_threshold: int = 3,
+        recovery_timeout: float = 30.0,
+        success_threshold_in_half_open: int = 2,
+    ):
         self.name = name
         self.failure_threshold = failure_threshold
         self.recovery_timeout = recovery_timeout
+        # Number of consecutive successes required in HALF_OPEN before
+        # the breaker transitions back to CLOSED.
+        self.success_threshold_in_half_open = success_threshold_in_half_open
 
         self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
         self.failure_count = 0
+        # Tracks consecutive successes accumulated while in HALF_OPEN.
+        self._half_open_success_count = 0
         self.last_state_change = time.time()
         self._lock = Lock()
 
@@ -48,7 +62,8 @@ def __init__(self, name: str, failure_threshold: int = 3, recovery_timeout: floa
     def allow_request(self) -> bool:
         """
         Check if a request is allowed to proceed.
-        If in OPEN state and recovery timeout has elapsed, transitions to HALF_OPEN.
+        If in OPEN state and recovery timeout has elapsed, transitions to HALF_OPEN
+        and resets the consecutive-success counter for that probe window.
         """
         with self._lock:
             now = time.time()
@@ -64,6 +79,7 @@ def allow_request(self) -> bool:
                         self.recovery_timeout,
                     )
                     self.state = "HALF_OPEN"
+                    self._half_open_success_count = 0
                     self.last_state_change = now
                     set_circuit_state(self.name, CIRCUIT_STATE_HALF_OPEN)
                     CIRCUIT_RECOVERY_TIME.labels(breaker_name=self.name).observe(recovery_seconds)
@@ -74,27 +90,46 @@ def allow_request(self) -> bool:
     def record_success(self) -> None:
         """
         Record a successful request.
-        If in HALF_OPEN, transitions back to CLOSED and resets failure count.
+
+        In HALF_OPEN: increments the consecutive-success counter.  The breaker
+        only transitions to CLOSED once *success_threshold_in_half_open*
+        consecutive successes have been recorded, preventing premature closure
+        on a single lucky probe.  Requests continue to be allowed through while
+        the threshold is being accumulated.
+
+        In CLOSED: resets the failure counter (defensive, keeps count accurate).
         """
         with self._lock:
             now = time.time()
             if self.state == "HALF_OPEN":
+                self._half_open_success_count += 1
                 logger.info(
-                    "Circuit breaker for provider '%s' transitioning from HALF_OPEN to CLOSED "
-                    "(successful probe request)",
+                    "Circuit breaker for provider '%s' probe success %d/%d in HALF_OPEN",
                     self.name,
+                    self._half_open_success_count,
+                    self.success_threshold_in_half_open,
                 )
-                self.state = "CLOSED"
-                self.failure_count = 0
-                self.last_state_change = now
-                set_circuit_state(self.name, CIRCUIT_STATE_CLOSED)
+                if self._half_open_success_count >= self.success_threshold_in_half_open:
+                    logger.info(
+                        "Circuit breaker for provider '%s' transitioning from HALF_OPEN to CLOSED "
+                        "(%d consecutive successes reached threshold %d)",
+                        self.name,
+                        self._half_open_success_count,
+                        self.success_threshold_in_half_open,
+                    )
+                    self.state = "CLOSED"
+                    self.failure_count = 0
+                    self._half_open_success_count = 0
+                    self.last_state_change = now
+                    set_circuit_state(self.name, CIRCUIT_STATE_CLOSED)
             elif self.state == "CLOSED":
                 self.failure_count = 0
 
     def record_failure(self) -> None:
         """
         Record a failed request.
         If in CLOSED and threshold is reached, or if in HALF_OPEN, transitions to OPEN.
+        Any accumulated HALF_OPEN consecutive-success count is reset on failure.
         """
         with self._lock:
             now = time.time()
@@ -110,5 +145,6 @@ def record_failure(self) -> None:
                     self.failure_threshold,
                 )
                 self.state = "OPEN"
+                self._half_open_success_count = 0
                 self.last_state_change = now
                 set_circuit_state(self.name, CIRCUIT_STATE_OPEN)

app/ai-service/tests/test_circuit_breaker.py

@@ -36,10 +36,17 @@ def test_circuit_breaker_basic_transitions():
     assert breaker.allow_request() is True
     assert breaker.state == "HALF_OPEN"
 
-    # 6. Success closes the circuit
+    # 6. Default success_threshold_in_half_open is 2, so first success
+    #    still leaves the breaker in HALF_OPEN.
+    breaker.record_success()
+    assert breaker.state == "HALF_OPEN"
+    assert breaker._half_open_success_count == 1
+
+    # 7. Second success meets the threshold → CLOSED
     breaker.record_success()
     assert breaker.state == "CLOSED"
     assert breaker.failure_count == 0
+    assert breaker._half_open_success_count == 0
 
 def test_circuit_breaker_half_open_failure():
     breaker = CircuitBreaker("test-provider", failure_threshold=2, recovery_timeout=0.1)
@@ -60,6 +67,154 @@ def test_circuit_breaker_half_open_failure():
     assert breaker.allow_request() is False
 
 
+# ---------------------------------------------------------------------------
+# Acceptance criteria for issue #270
+# ---------------------------------------------------------------------------
+
+class TestHalfOpenSuccessThreshold:
+    """
+    Configurable success_threshold_in_half_open: the breaker must accumulate
+    N consecutive successes in HALF_OPEN before returning to CLOSED.
+    """
+
+    def _open_then_half_open(self, breaker: CircuitBreaker) -> None:
+        """Helper: trip the breaker and advance it to HALF_OPEN."""
+        breaker.record_failure()
+        assert breaker.state == "OPEN"
+        time.sleep(0.12)
+        assert breaker.allow_request() is True
+        assert breaker.state == "HALF_OPEN"
+
+    def test_threshold_3_two_successes_remain_half_open(self):
+        """
+        AC: threshold=3, 2 successes in HALF_OPEN → still HALF_OPEN.
+        """
+        breaker = CircuitBreaker(
+            "ac-threshold-3-partial",
+            failure_threshold=1,
+            recovery_timeout=0.1,
+            success_threshold_in_half_open=3,
+        )
+        self._open_then_half_open(breaker)
+
+        breaker.record_success()
+        assert breaker.state == "HALF_OPEN", "1st success must not close the circuit (threshold=3)"
+        assert breaker._half_open_success_count == 1
+
+        breaker.record_success()
+        assert breaker.state == "HALF_OPEN", "2nd success must not close the circuit (threshold=3)"
+        assert breaker._half_open_success_count == 2
+
+    def test_threshold_3_third_success_closes(self):
+        """
+        AC: threshold=3, 3rd consecutive success in HALF_OPEN → CLOSED.
+        """
+        breaker = CircuitBreaker(
+            "ac-threshold-3-full",
+            failure_threshold=1,
+            recovery_timeout=0.1,
+            success_threshold_in_half_open=3,
+        )
+        self._open_then_half_open(breaker)
+
+        breaker.record_success()
+        breaker.record_success()
+        assert breaker.state == "HALF_OPEN"
+
+        breaker.record_success()
+        assert breaker.state == "CLOSED", "3rd success must close the circuit (threshold=3)"
+        assert breaker.failure_count == 0
+        assert breaker._half_open_success_count == 0
+
+    def test_default_threshold_is_2(self):
+        """
+        Default success_threshold_in_half_open=2: first success stays HALF_OPEN,
+        second closes.
+        """
+        breaker = CircuitBreaker(
+            "default-threshold",
+            failure_threshold=1,
+            recovery_timeout=0.1,
+        )
+        assert breaker.success_threshold_in_half_open == 2
+        self._open_then_half_open(breaker)
+
+        breaker.record_success()
+        assert breaker.state == "HALF_OPEN"
+
+        breaker.record_success()
+        assert breaker.state == "CLOSED"
+
+    def test_threshold_1_single_success_closes(self):
+        """
+        threshold=1 restores the previous single-success behaviour.
+        """
+        breaker = CircuitBreaker(
+            "threshold-1",
+            failure_threshold=1,
+            recovery_timeout=0.1,
+            success_threshold_in_half_open=1,
+        )
+        self._open_then_half_open(breaker)
+
+        breaker.record_success()
+        assert breaker.state == "CLOSED"
+
+    def test_failure_resets_success_count_and_reopens(self):
+        """
+        A failure mid-probe must reset the accumulated success count and
+        reopen the circuit, so a subsequent HALF_OPEN entry starts from zero.
+        """
+        breaker = CircuitBreaker(
+            "failure-resets-count",
+            failure_threshold=1,
+            recovery_timeout=0.1,
+            success_threshold_in_half_open=3,
+        )
+        self._open_then_half_open(breaker)
+
+        breaker.record_success()
+        assert breaker._half_open_success_count == 1
+
+        # Failure mid-probe: count must be reset, circuit reopens
+        breaker.record_failure()
+        assert breaker.state == "OPEN"
+        assert breaker._half_open_success_count == 0
+
+        # Re-enter HALF_OPEN: must start accumulating from zero, not from 1
+        time.sleep(0.12)
+        assert breaker.allow_request() is True
+        assert breaker._half_open_success_count == 0
+
+        breaker.record_success()
+        assert breaker._half_open_success_count == 1
+        assert breaker.state == "HALF_OPEN"
+
+    def test_allow_request_resets_count_on_reentry(self):
+        """
+        allow_request() must reset _half_open_success_count every time the
+        breaker transitions OPEN → HALF_OPEN, ensuring each probe window is
+        independent.
+        """
+        breaker = CircuitBreaker(
+            "reentry-reset",
+            failure_threshold=1,
+            recovery_timeout=0.1,
+            success_threshold_in_half_open=3,
+        )
+        # First probe window: 1 success then a failure
+        self._open_then_half_open(breaker)
+        breaker.record_success()
+        assert breaker._half_open_success_count == 1
+        breaker.record_failure()  # reopens
+
+        # Second probe window must start from 0
+        time.sleep(0.12)
+        breaker.allow_request()
+        assert breaker._half_open_success_count == 0
+        assert breaker.state == "HALF_OPEN"
+
+
 class TestHumanitarianVerificationServiceCircuitBreaker:
     def setup_method(self):
         self.service = HumanitarianVerificationService()
@@ -198,6 +353,9 @@ def test_success_closes_circuit_and_resets_state_gauge(self):
         breaker.record_failure()
         time.sleep(0.07)
         breaker.allow_request()  # -> HALF_OPEN
+        # Default success_threshold_in_half_open=2: need two successes to close
+        breaker.record_success()
+        assert breaker.state == "HALF_OPEN"  # still probing after 1st success
         breaker.record_success()
 
         assert _sample("circuit_breaker_state", {"breaker_name": "metrics-success"}) == 0