refactor: add runtime state helper

Boris-code · Boris-code · commit 85995399df0a · 2026-06-01T09:40:00.000+08:00
diff --git a/feapder/core/runtime_state.py b/feapder/core/runtime_state.py
@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+"""
+Small runtime state helper for background runtime threads.
+"""
+from contextlib import contextmanager
+from threading import RLock
+
+
+class RuntimeState:
+    def __init__(self):
+        self._stop_requested = False
+        self._busy_count = 0
+        self._lock = RLock()
+
+    def request_stop(self):
+        with self._lock:
+            self._stop_requested = True
+
+    @property
+    def is_stop_requested(self):
+        with self._lock:
+            return self._stop_requested
+
+    def mark_busy(self):
+        with self._lock:
+            self._busy_count += 1
+
+    def mark_idle(self):
+        with self._lock:
+            if self._busy_count > 0:
+                self._busy_count -= 1
+
+    @property
+    def busy_count(self):
+        with self._lock:
+            return self._busy_count
+
+    @property
+    def is_idle(self):
+        return self.busy_count == 0
+
+    @contextmanager
+    def busy(self):
+        self.mark_busy()
+        try:
+            yield
+        finally:
+            self.mark_idle()
diff --git a/tests/test_runtime_state.py b/tests/test_runtime_state.py
@@ -0,0 +1,56 @@
+from feapder.core.runtime_state import RuntimeState
+
+
+def test_runtime_state_starts_idle_and_running():
+    state = RuntimeState()
+
+    assert state.is_idle is True
+    assert state.is_stop_requested is False
+    assert state.busy_count == 0
+
+
+def test_runtime_state_tracks_busy_count():
+    state = RuntimeState()
+
+    state.mark_busy()
+    state.mark_busy()
+
+    assert state.is_idle is False
+    assert state.busy_count == 2
+
+    state.mark_idle()
+    state.mark_idle()
+
+    assert state.is_idle is True
+    assert state.busy_count == 0
+
+
+def test_runtime_state_does_not_go_negative():
+    state = RuntimeState()
+
+    state.mark_idle()
+
+    assert state.busy_count == 0
+    assert state.is_idle is True
+
+
+def test_runtime_state_busy_context_resets_on_exception():
+    state = RuntimeState()
+
+    try:
+        with state.busy():
+            assert state.is_idle is False
+            raise RuntimeError("boom")
+    except RuntimeError:
+        pass
+
+    assert state.is_idle is True
+
+
+def test_runtime_state_stop_request_is_sticky():
+    state = RuntimeState()
+
+    state.request_stop()
+    state.request_stop()
+
+    assert state.is_stop_requested is True
