feat: make WorkflowContext thread-safe with a Lock

Author: baranylcn

gateframe/core/context.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import threading
 from datetime import datetime, timezone
 from typing import Any
 
@@ -66,59 +67,70 @@ def __init__(
     ) -> None:
         self.workflow_id = workflow_id
         self.escalation_threshold = escalation_threshold
-        self.confidence = initial_confidence
+        self._confidence = initial_confidence
         self._initial_confidence = initial_confidence
         self._soft_fail_penalty = soft_fail_penalty
         self._silent_fail_penalty = silent_fail_penalty
         self._history: list[StepRecord] = []
+        self._lock = threading.Lock()
+
+    @property
+    def confidence(self) -> float:
+        with self._lock:
+            return self._confidence
 
     @property
     def history(self) -> list[StepRecord]:
-        return list(self._history)
+        with self._lock:
+            return list(self._history)
 
     @property
     def step_count(self) -> int:
-        return len(self._history)
+        with self._lock:
+            return len(self._history)
 
     @property
     def threshold_breached(self) -> bool:
-        return self.confidence < self.escalation_threshold
+        with self._lock:
+            return self._confidence < self.escalation_threshold
 
     def update(self, result: ValidationResult) -> None:
-        confidence_before = self.confidence
         penalty = self._compute_penalty(result)
-        self.confidence = max(0.0, self.confidence - penalty)
-
-        record = StepRecord(
-            step_index=len(self._history),
-            contract_name=result.contract_name,
-            passed=result.passed,
-            confidence_before=confidence_before,
-            confidence_after=self.confidence,
-            penalty_applied=penalty,
-            failure_modes=[f.failure_mode for f in result.failures],
-        )
-        self._history.append(record)
+        with self._lock:
+            confidence_before = self._confidence
+            self._confidence = max(0.0, self._confidence - penalty)
+            confidence_after = self._confidence
+            record = StepRecord(
+                step_index=len(self._history),
+                contract_name=result.contract_name,
+                passed=result.passed,
+                confidence_before=confidence_before,
+                confidence_after=confidence_after,
+                penalty_applied=penalty,
+                failure_modes=[f.failure_mode for f in result.failures],
+            )
+            self._history.append(record)
 
         logger.info(
             "workflow_step",
             workflow_id=self.workflow_id,
             step_index=record.step_index,
             contract=result.contract_name,
             passed=result.passed,
-            confidence=self.confidence,
+            confidence=confidence_after,
             threshold_breached=self.threshold_breached,
         )
 
     def reset(self) -> None:
-        self.confidence = self._initial_confidence
-        self._history.clear()
+        with self._lock:
+            self._confidence = self._initial_confidence
+            self._history.clear()
 
         logger.info(
             "workflow_reset",
             workflow_id=self.workflow_id,
-            confidence=self.confidence,
-            threshold_breached=self.threshold_breached,
+            confidence=self._initial_confidence,
+            threshold_breached=self._initial_confidence < self.escalation_threshold,
         )
 
     def _compute_penalty(self, result: ValidationResult) -> float:
@@ -131,11 +143,12 @@ def _compute_penalty(self, result: ValidationResult) -> float:
         return penalty
 
     def to_dict(self) -> dict[str, Any]:
-        return {
-            "workflow_id": self.workflow_id,
-            "confidence": self.confidence,
-            "escalation_threshold": self.escalation_threshold,
-            "threshold_breached": self.threshold_breached,
-            "step_count": self.step_count,
-            "history": [r.to_dict() for r in self._history],
-        }
+        with self._lock:
+            return {
+                "workflow_id": self.workflow_id,
+                "confidence": self._confidence,
+                "escalation_threshold": self.escalation_threshold,
+                "threshold_breached": self._confidence < self.escalation_threshold,
+                "step_count": len(self._history),
+                "history": [r.to_dict() for r in self._history],
+            }

tests/core/test_context.py

@@ -1,3 +1,4 @@
+import threading
 from typing import Any
 
 import pytest
@@ -194,6 +195,54 @@ def test_reset_restores_initial_confidence_and_clears_history(self) -> None:
         assert ctx.threshold_breached is False
 
 
+    def test_concurrent_updates_do_not_corrupt_confidence(self) -> None:
+        ctx = WorkflowContext("wf_concurrent", soft_fail_penalty=0.01)
+        result = _make_result(passed=False, failures=[_soft_failure()])
+
+        def do_updates() -> None:
+            for _ in range(50):
+                ctx.update(result)
+
+        threads = [threading.Thread(target=do_updates) for _ in range(4)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert ctx.step_count == 200
+        assert 0.0 <= ctx.confidence <= 1.0
+
+    def test_concurrent_update_and_reset(self) -> None:
+        ctx = WorkflowContext("wf_reset_race")
+        result = _make_result(passed=False, failures=[_soft_failure()])
+
+        errors: list[Exception] = []
+
+        def do_updates() -> None:
+            try:
+                for _ in range(30):
+                    ctx.update(result)
+            except Exception as exc:  # noqa: BLE001
+                errors.append(exc)
+
+        def do_resets() -> None:
+            try:
+                for _ in range(10):
+                    ctx.reset()
+            except Exception as exc:  # noqa: BLE001
+                errors.append(exc)
+
+        threads = [threading.Thread(target=do_updates) for _ in range(3)]
+        threads += [threading.Thread(target=do_resets) for _ in range(2)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+
+        assert errors == []
+        assert 0.0 <= ctx.confidence <= 1.0
+
+
 class TestStepRecord:
     def test_to_dict(self) -> None:
         record = StepRecord(