diff --git a/src/cursor_flasher/state.py b/src/cursor_flasher/state.py new file mode 100644 index 0000000..b93438e --- /dev/null +++ b/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 diff --git a/tests/test_state.py b/tests/test_state.py new file mode 100644 index 0000000..f973f44 --- /dev/null +++ b/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