fix(consensus): replace non-reentrant Lock with RLock in CircuitBreaker (#332)

src/qwed_new/core/consensus_verifier.py

@@ -169,7 +169,7 @@ def __init__(
         self.success_threshold = success_threshold
 
         self._engines: Dict[str, EngineHealth] = {}
-        self._lock = threading.Lock()
+        self._lock = threading.RLock()
 
     def get_health(self, engine_name: str) -> EngineHealth:
         """
@@ -196,22 +196,22 @@ def is_available(self, engine_name: str) -> bool:
         Returns:
             bool: True if engine can accept requests.
         """
-        health = self.get_health(engine_name)
-
-        if health.state == EngineState.HEALTHY:
-            return True
-
-        if health.state == EngineState.OPEN:
-            # Check if recovery time has passed
-            if health.circuit_open_until and time.time() > health.circuit_open_until:
-                # Transition to half-open (allow test request)
-                with self._lock:
-                    health.state = EngineState.DEGRADED
+        with self._lock:
+            health = self.get_health(engine_name)
+
+            if health.state == EngineState.HEALTHY:
                 return True
-            return False
-
-        # DEGRADED state - allow requests
-        return True
+
+            if health.state == EngineState.OPEN:
+                # Check if recovery time has passed
+                if health.circuit_open_until and time.time() > health.circuit_open_until:
+                    # Transition to half-open (allow test request)
+                    health.state = EngineState.DEGRADED
+                    return True
+                return False
+
+            # DEGRADED state - allow requests
+            return True
 
     def record_success(self, engine_name: str, latency_ms: float):
         """
@@ -262,16 +262,22 @@ def get_all_health(self) -> Dict[str, Dict[str, Any]]:
         Returns:
             Dict mapping engine names to health statistics.
         """
-        return {
-            name: {
-                "state": health.state.value,
-                "failures": health.total_failures,
-                "calls": health.total_calls,
-                "avg_latency_ms": round(health.avg_latency_ms, 2),
-                "failure_rate": round(health.total_failures / max(health.total_calls, 1), 3)
+        with self._lock:
+            return {
+                name: {
+                    "state": health.state.value,
+                    "failures": health.total_failures,
+                    "calls": health.total_calls,
+                    "avg_latency_ms": round(health.avg_latency_ms, 2),
+                    "failure_rate": round(health.total_failures / max(health.total_calls, 1), 3)
+                }
+                for name, health in self._engines.items()
             }
-            for name, health in self._engines.items()
-        }
+
+    def reset(self):
+        """Reset all tracked engine health states in a thread-safe manner."""
+        with self._lock:
+            self._engines.clear()
 
 
 
@@ -962,7 +968,7 @@ def reset_circuit_breakers(self):
             >>> verifier.reset_circuit_breakers()
         """
         if self.circuit_breaker:
-            self.circuit_breaker._engines.clear()
+            self.circuit_breaker.reset()
 
     def to_verification_context(self, result: "DiagnosticResult", query: str, attestation_token: Optional[str] = None) -> "VerificationContextDocument":
         """Map a DiagnosticResult to a Verification Context v1.0 document."""

tests/test_circuit_breaker_deadlock_332.py

@@ -0,0 +1,112 @@
+"""Regression test for Issue #332: CircuitBreaker self-deadlock fix."""
+import threading
+
+from qwed_new.core.consensus_verifier import CircuitBreaker, ConsensusVerifier, EngineState
+
+
+def test_circuit_breaker_record_success_no_deadlock():
+    """record_success acquires lock and calls get_health() which re-enters lock."""
+    cb = CircuitBreaker()
+    # Prior to fix with non-reentrant Lock, this deadlocks on the first call
+    cb.record_success("SymPy", latency_ms=12.5)
+
+    health = cb.get_health("SymPy")
+    assert health.total_calls == 1
+    assert health.consecutive_failures == 0
+    assert health.avg_latency_ms == 12.5
+
+
+def test_circuit_breaker_record_failure_no_deadlock(monkeypatch):
+    """record_failure acquires lock and calls get_health() which re-enters lock."""
+    clock = [100.0]
+    monkeypatch.setattr(
+        "qwed_new.core.consensus_verifier.time.time",
+        lambda: clock[0],
+    )
+    cb = CircuitBreaker(failure_threshold=3, recovery_time_seconds=1.0)
+
+    # 1st failure
+    cb.record_failure("Z3")
+    assert cb.get_health("Z3").consecutive_failures == 1
+    assert cb.is_available("Z3") is True
+
+    # 2nd failure
+    cb.record_failure("Z3")
+    assert cb.get_health("Z3").consecutive_failures == 2
+    assert cb.is_available("Z3") is True
+
+    # 3rd failure -> trips OPEN
+    cb.record_failure("Z3")
+    assert cb.get_health("Z3").state == EngineState.OPEN
+    assert cb.is_available("Z3") is False
+
+    # Advance clock past recovery threshold deterministically without time.sleep
+    clock[0] = 102.0
+
+    # is_available transitions to DEGRADED
+    assert cb.is_available("Z3") is True
+    assert cb.get_health("Z3").state == EngineState.DEGRADED
+
+    # Success resets to HEALTHY
+    cb.record_success("Z3", latency_ms=5.0)
+    assert cb.get_health("Z3").state == EngineState.HEALTHY
+
+
+def test_circuit_breaker_get_all_health_thread_safe():
+    """get_all_health returns complete statistics without raising under concurrency."""
+    cb = CircuitBreaker()
+    cb.record_success("SymPy", 10.0)
+    cb.record_failure("Python")
+
+    stats = cb.get_all_health()
+    assert "SymPy" in stats
+    assert "Python" in stats
+    assert stats["SymPy"]["calls"] == 1
+    assert stats["Python"]["failures"] == 1
+
+
+def test_circuit_breaker_reset_thread_safe():
+    """reset() clears engine statistics under lock."""
+    cb = CircuitBreaker()
+    cb.record_success("SymPy", 10.0)
+    assert len(cb.get_all_health()) == 1
+
+    cb.reset()
+    assert len(cb.get_all_health()) == 0
+
+    verifier = ConsensusVerifier(enable_circuit_breaker=True)
+    verifier.circuit_breaker.record_success("Z3", 5.0)
+    assert len(verifier.get_engine_health()) == 1
+    verifier.reset_circuit_breakers()
+    assert len(verifier.get_engine_health()) == 0
+
+
+def test_circuit_breaker_concurrent_access():
+    """Verify concurrent threads recording success/failure do not deadlock."""
+    cb = CircuitBreaker(failure_threshold=5, recovery_time_seconds=1.0)
+    errors = []
+
+    def worker(engine: str):
+        try:
+            for _ in range(50):
+                cb.record_success(engine, 10.0)
+                cb.record_failure(engine)
+                cb.is_available(engine)
+                cb.get_all_health()
+                if engine.endswith("0"):
+                    cb.reset()
+        except Exception as e:
+            errors.append(e)
+
+    threads = [
+        threading.Thread(target=worker, args=(f"Engine-{i % 3}",))
+        for i in range(10)
+    ]
+
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=5.0)
+        assert not t.is_alive(), "Thread deadlocked in CircuitBreaker!"
+
+    assert not errors, f"Errors encountered during concurrent execution: {errors}"