feat: add state machine for agent activity detection

Made-with: Cursor

src/cursor_flasher/state.py

@@ -0,0 +1,53 @@
+import enum
+import time
+
+
+class FlasherState(enum.Enum):
+    IDLE = "idle"
+    AGENT_WORKING = "agent_working"
+    WAITING_FOR_USER = "waiting_for_user"
+
+
+class StateMachine:
+    def __init__(self, cooldown: float = 3.0):
+        self.state = FlasherState.IDLE
+        self.cooldown = cooldown
+        self._last_dismiss_time: float = 0
+
+    def update(self, *, agent_working: bool, approval_needed: bool) -> bool:
+        """Update state based on detected signals. Returns True if state changed."""
+        old = self.state
+
+        if approval_needed and self.state != FlasherState.WAITING_FOR_USER:
+            if not self._in_cooldown():
+                self.state = FlasherState.WAITING_FOR_USER
+            return self.state != old
+
+        match self.state:
+            case FlasherState.IDLE:
+                if agent_working:
+                    self.state = FlasherState.AGENT_WORKING
+            case FlasherState.AGENT_WORKING:
+                if not agent_working:
+                    if not self._in_cooldown():
+                        self.state = FlasherState.WAITING_FOR_USER
+                    else:
+                        self.state = FlasherState.IDLE
+            case FlasherState.WAITING_FOR_USER:
+                if agent_working:
+                    self.state = FlasherState.AGENT_WORKING
+
+        return self.state != old
+
+    def dismiss(self) -> bool:
+        """User interaction detected — dismiss the flash."""
+        if self.state == FlasherState.WAITING_FOR_USER:
+            self.state = FlasherState.IDLE
+            self._last_dismiss_time = time.monotonic()
+            return True
+        return False
+
+    def _in_cooldown(self) -> bool:
+        if self._last_dismiss_time == 0:
+            return False
+        return (time.monotonic() - self._last_dismiss_time) < self.cooldown

tests/test_state.py

@@ -0,0 +1,75 @@
+import pytest
+from cursor_flasher.state import FlasherState, StateMachine
+
+
+class TestStateMachine:
+    def test_initial_state_is_idle(self):
+        sm = StateMachine()
+        assert sm.state == FlasherState.IDLE
+
+    def test_idle_to_agent_working(self):
+        sm = StateMachine()
+        changed = sm.update(agent_working=True, approval_needed=False)
+        assert sm.state == FlasherState.AGENT_WORKING
+        assert changed is True
+
+    def test_agent_working_to_waiting(self):
+        sm = StateMachine()
+        sm.update(agent_working=True, approval_needed=False)
+        changed = sm.update(agent_working=False, approval_needed=False)
+        assert sm.state == FlasherState.WAITING_FOR_USER
+        assert changed is True
+
+    def test_approval_needed_triggers_waiting(self):
+        sm = StateMachine()
+        sm.update(agent_working=True, approval_needed=False)
+        changed = sm.update(agent_working=False, approval_needed=True)
+        assert sm.state == FlasherState.WAITING_FOR_USER
+        assert changed is True
+
+    def test_idle_does_not_jump_to_waiting(self):
+        sm = StateMachine()
+        changed = sm.update(agent_working=False, approval_needed=False)
+        assert sm.state == FlasherState.IDLE
+        assert changed is False
+
+    def test_waiting_to_user_interacting(self):
+        sm = StateMachine()
+        sm.update(agent_working=True, approval_needed=False)
+        sm.update(agent_working=False, approval_needed=False)
+        assert sm.state == FlasherState.WAITING_FOR_USER
+        changed = sm.dismiss()
+        assert sm.state == FlasherState.IDLE
+        assert changed is True
+
+    def test_waiting_to_agent_working(self):
+        sm = StateMachine()
+        sm.update(agent_working=True, approval_needed=False)
+        sm.update(agent_working=False, approval_needed=False)
+        changed = sm.update(agent_working=True, approval_needed=False)
+        assert sm.state == FlasherState.AGENT_WORKING
+        assert changed is True
+
+    def test_no_change_returns_false(self):
+        sm = StateMachine()
+        sm.update(agent_working=True, approval_needed=False)
+        changed = sm.update(agent_working=True, approval_needed=False)
+        assert changed is False
+
+    def test_cooldown_prevents_immediate_retrigger(self):
+        sm = StateMachine(cooldown=5.0)
+        sm.update(agent_working=True, approval_needed=False)
+        sm.update(agent_working=False, approval_needed=False)
+        assert sm.state == FlasherState.WAITING_FOR_USER
+        sm.dismiss()
+        sm.update(agent_working=True, approval_needed=False)
+        changed = sm.update(agent_working=False, approval_needed=False)
+        assert sm.cooldown == 5.0
+
+    def test_direct_approval_from_idle(self):
+        """If we detect approval buttons without seeing agent_working first,
+        still transition to WAITING_FOR_USER."""
+        sm = StateMachine()
+        changed = sm.update(agent_working=False, approval_needed=True)
+        assert sm.state == FlasherState.WAITING_FOR_USER
+        assert changed is True