cottongin/cursor-flasher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix per-workspace pulse dismiss for multi-window setups

Browse Source 
Interacting with one Cursor window no longer dismisses overlays from
other windows. The overlay manager now tracks panels by tag (workspace),
the daemon maintains per-workspace pending/active/cooldown state, and
dismiss logic identifies the focused window via AXFocusedWindow to
target only that workspace's pulse.

Made-with: Cursor
This commit is contained in:
cottongin
2026-03-10 09:33:42 -04:00
parent a3203e2970
commit 5fc378e558
4 changed files with 465 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files

280
tests/test_daemon.py
View File

@@ -9,16 +9,21 @@ from cursor_flasher.daemon import FlasherDaemon
PATCH_APPEARANCE = "cursor_flasher.daemon._get_system_appearance"
PATCH_FOCUSED = "cursor_flasher.daemon.get_focused_cursor_window"
class TestFlasherDaemon:
def _make_daemon(self, **config_overrides) -> FlasherDaemon:
config = Config(**config_overrides)
with patch("cursor_flasher.daemon.OverlayManager"):
with patch("cursor_flasher.daemon.OverlayManager") as MockOverlay:
daemon = FlasherDaemon(config)
daemon.overlay.is_pulsing = False
daemon.overlay.active_tags = set()
daemon.overlay.has_tag = lambda tag: tag in daemon.overlay.active_tags
return daemon
# --- preToolUse / pending ---
def test_preToolUse_queues_pending(self):
daemon = self._make_daemon()
@@ -26,9 +31,9 @@ class TestFlasherDaemon:
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert daemon._pending is not None
assert daemon._pending.tool == "Shell"
daemon.overlay.pulse.assert_not_called()
assert "/path" in daemon._pending_approvals
assert daemon._pending_approvals["/path"].tool == "Shell"
daemon.overlay.add_pulse.assert_not_called()
def test_pending_promotes_after_delay(self):
daemon = self._make_daemon(approval_delay=0.0, flash_mode="screen")
@@ -46,7 +51,11 @@ class TestFlasherDaemon:
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.pulse.assert_called_once_with([screen], daemon.config.dark.running)
daemon.overlay.add_pulse.assert_called_once_with(
"/path", [screen], daemon.config.dark.running
)
assert "/path" in daemon._active_pulses
assert "/path" not in daemon._pending_approvals
def test_postToolUse_cancels_pending(self):
"""Auto-approved tools: postToolUse arrives before delay, cancels the pulse."""
@@ -55,13 +64,13 @@ class TestFlasherDaemon:
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert daemon._pending is not None
assert "/path" in daemon._pending_approvals
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUse", "tool": "Shell"}).encode()
)
assert daemon._pending is None
daemon.overlay.pulse.assert_not_called()
assert "/path" not in daemon._pending_approvals
daemon.overlay.add_pulse.assert_not_called()
def test_preToolUse_skips_non_approval_tool(self):
daemon = self._make_daemon()
@@ -70,7 +79,7 @@ class TestFlasherDaemon:
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Read"}).encode()
)
assert daemon._pending is None
assert not daemon._pending_approvals
def test_preToolUse_respects_custom_tool_list(self):
daemon = self._make_daemon(approval_delay=0.0, flash_mode="screen", approval_tools=["Shell", "MCP"])
@@ -87,7 +96,9 @@ class TestFlasherDaemon:
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.pulse.assert_called_once()
daemon.overlay.add_pulse.assert_called_once()
# --- stop / flash ---
def test_stop_flashes_briefly(self):
daemon = self._make_daemon(flash_mode="screen")
@@ -102,7 +113,9 @@ class TestFlasherDaemon:
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.flash.assert_called_once_with([screen], daemon.config.dark.completed)
daemon.overlay.add_flash.assert_called_once_with(
"/path", [screen], daemon.config.dark.completed
)
def test_stop_flashes_window_frame_when_window_mode(self):
daemon = self._make_daemon(flash_mode="window")
@@ -115,8 +128,8 @@ class TestFlasherDaemon:
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.flash.assert_called_once_with(
[((0, 0), (800, 600))], daemon.config.light.completed
daemon.overlay.add_flash.assert_called_once_with(
"/path", [((0, 0), (800, 600))], daemon.config.light.completed
)
def test_allscreens_mode_uses_all_screens(self):
@@ -135,18 +148,24 @@ class TestFlasherDaemon:
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.pulse.assert_called_once_with(screens, daemon.config.dark.running)
def test_stop_dismisses_active_pulse(self):
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
daemon.overlay.add_pulse.assert_called_once_with(
"/path", screens, daemon.config.dark.running
)
daemon.overlay.dismiss.assert_called_once()
daemon.overlay.flash.assert_not_called()
# --- stop interactions with active pulse ---
def test_stop_dismisses_active_pulse_for_workspace(self):
daemon = self._make_daemon()
daemon._active_pulses["/path"] = MagicMock()
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=None), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.dismiss_tag.assert_called_with("/path")
assert "/path" not in daemon._active_pulses
def test_stop_clears_pending(self):
daemon = self._make_daemon(approval_delay=10.0)
@@ -154,32 +173,37 @@ class TestFlasherDaemon:
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert daemon._pending is not None
assert "/path" in daemon._pending_approvals
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
assert daemon._pending is None
assert "/path" not in daemon._pending_approvals
# --- postToolUse / dismiss ---
def test_postToolUse_dismisses_active_pulse(self):
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._active_pulses["/path"] = MagicMock()
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUse", "tool": "Shell"}).encode()
)
daemon.overlay.dismiss.assert_called_once()
daemon.overlay.dismiss_tag.assert_called_with("/path")
assert "/path" not in daemon._active_pulses
def test_postToolUseFailure_dismisses_pulse(self):
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._active_pulses["/path"] = MagicMock()
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUseFailure", "tool": "Shell"}).encode()
)
daemon.overlay.dismiss.assert_called_once()
daemon.overlay.dismiss_tag.assert_called_with("/path")
# --- cooldown ---
def test_cooldown_prevents_rapid_triggers(self):
daemon = self._make_daemon(cooldown=5.0)
@@ -187,19 +211,30 @@ class TestFlasherDaemon:
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
daemon._last_flash = time.monotonic()
daemon._pending = None
daemon._last_flash["/path"] = time.monotonic()
daemon._pending_approvals.clear()
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert daemon._pending is None
assert "/path" not in daemon._pending_approvals
def test_cooldown_is_per_workspace(self):
daemon = self._make_daemon(cooldown=5.0)
daemon._last_flash["/pathA"] = time.monotonic()
daemon._handle_message(
json.dumps({"workspace": "/pathB", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert "/pathB" in daemon._pending_approvals
# --- misc ---
def test_invalid_json_ignored(self):
daemon = self._make_daemon()
daemon._handle_message(b"not json")
daemon.overlay.pulse.assert_not_called()
daemon.overlay.flash.assert_not_called()
daemon.overlay.add_pulse.assert_not_called()
daemon.overlay.add_flash.assert_not_called()
def test_no_window_found(self):
daemon = self._make_daemon(approval_delay=0.0)
@@ -211,108 +246,127 @@ class TestFlasherDaemon:
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=None):
daemon._check_pending()
daemon.overlay.pulse.assert_not_called()
daemon.overlay.add_pulse.assert_not_called()
def test_focus_transition_dismisses_pulse(self):
"""Pulse dismisses when user switches TO Cursor from another app."""
# --- focus dismiss ---
def test_focus_transition_dismisses_focused_workspace(self):
"""Pulse dismisses when user switches TO Cursor, only for the focused window."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._active_pulses["/pathA"] = _ActivePulse("/pathA", "project-a", time.monotonic())
daemon._active_pulses["/pathB"] = _ActivePulse("/pathB", "project-b", time.monotonic())
daemon._cursor_was_frontmost = False
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True):
focused = {"title": "project-a", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch(PATCH_FOCUSED, return_value=focused):
daemon._check_focus()
daemon.overlay.dismiss.assert_called_once()
daemon.overlay.dismiss_tag.assert_called_once_with("/pathA")
assert "/pathA" not in daemon._active_pulses
assert "/pathB" in daemon._active_pulses
def test_focus_no_dismiss_when_cursor_already_frontmost(self):
"""No dismiss if Cursor was already frontmost (no transition)."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic())
daemon._cursor_was_frontmost = True
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True):
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch(PATCH_FOCUSED, return_value={"title": "proj", "frame": ((0, 0), (800, 600))}):
daemon._check_focus()
daemon.overlay.dismiss.assert_not_called()
daemon.overlay.dismiss_tag.assert_not_called()
def test_focus_tracks_state_changes(self):
"""_cursor_was_frontmost updates each tick."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic())
daemon._cursor_was_frontmost = True
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False):
daemon._check_focus()
assert daemon._cursor_was_frontmost is False
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True):
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch(PATCH_FOCUSED, return_value={"title": "proj", "frame": ((0, 0), (800, 600))}):
daemon._check_focus()
daemon.overlay.dismiss.assert_called_once()
daemon.overlay.dismiss_tag.assert_called_once_with("/path")
def test_focus_no_dismiss_when_not_pulsing(self):
daemon = self._make_daemon()
daemon.overlay.is_pulsing = False
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True):
daemon._check_focus()
daemon.overlay.dismiss.assert_not_called()
daemon.overlay.dismiss_tag.assert_not_called()
def test_input_dismiss_when_pulsing_and_cursor_focused(self):
"""Recent input + Cursor frontmost + past grace period = dismiss."""
# --- input dismiss ---
def test_input_dismiss_targets_focused_workspace(self):
"""Recent input + Cursor frontmost + past grace period = dismiss focused workspace only."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._pulse_started_at = time.monotonic() - 2.0
daemon._active_pulses["/pathA"] = _ActivePulse("/pathA", "project-a", time.monotonic() - 2.0)
daemon._active_pulses["/pathB"] = _ActivePulse("/pathB", "project-b", time.monotonic() - 2.0)
focused = {"title": "project-a", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.2):
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.2), \
patch(PATCH_FOCUSED, return_value=focused):
daemon._check_input_dismiss()
daemon.overlay.dismiss.assert_called_once()
daemon.overlay.dismiss_tag.assert_called_once_with("/pathA")
assert "/pathA" not in daemon._active_pulses
assert "/pathB" in daemon._active_pulses
def test_input_dismiss_skipped_during_grace_period(self):
"""No dismiss if pulse just started (within grace period)."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._pulse_started_at = time.monotonic() - 0.1
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 0.1)
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.05):
daemon._check_input_dismiss()
daemon.overlay.dismiss.assert_not_called()
daemon.overlay.dismiss_tag.assert_not_called()
def test_input_dismiss_skipped_when_not_pulsing(self):
daemon = self._make_daemon()
daemon.overlay.is_pulsing = False
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.1):
daemon._check_input_dismiss()
daemon.overlay.dismiss.assert_not_called()
daemon.overlay.dismiss_tag.assert_not_called()
def test_input_dismiss_skipped_when_cursor_not_frontmost(self):
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._pulse_started_at = time.monotonic() - 2.0
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 2.0)
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False):
daemon._check_input_dismiss()
daemon.overlay.dismiss.assert_not_called()
daemon.overlay.dismiss_tag.assert_not_called()
def test_input_dismiss_ignores_old_input(self):
"""Input from before the pulse started should not trigger dismiss."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon.overlay.is_pulsing = True
daemon._pulse_started_at = time.monotonic() - 2.0
daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 2.0)
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=5.0):
daemon._check_input_dismiss()
daemon.overlay.dismiss.assert_not_called()
daemon.overlay.dismiss_tag.assert_not_called()
# --- sound ---
def test_running_style_sound_plays_on_approval(self):
"""Running style with sound configured plays on approval pulse."""
@@ -385,12 +439,13 @@ class TestFlasherDaemon:
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
daemon.overlay.pulse.assert_called_once_with(
[((0, 0), (800, 600))], running
daemon.overlay.add_pulse.assert_called_once_with(
"/path", [((0, 0), (800, 600))], running
)
daemon.overlay.pulse.reset_mock()
daemon._last_flash = 0
daemon.overlay.add_pulse.reset_mock()
daemon._last_flash.clear()
daemon._active_pulses.clear()
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.play_alert"), \
@@ -399,8 +454,8 @@ class TestFlasherDaemon:
json.dumps({"workspace": "/path", "event": "stop"}).encode()
)
daemon.overlay.flash.assert_called_once_with(
[((0, 0), (800, 600))], completed
daemon.overlay.add_flash.assert_called_once_with(
"/path", [((0, 0), (800, 600))], completed
)
def test_theme_auto_uses_light_when_system_light(self):
@@ -425,6 +480,85 @@ class TestFlasherDaemon:
patch(PATCH_APPEARANCE, return_value="light"):
daemon._check_pending()
daemon.overlay.pulse.assert_called_once_with(
[((0, 0), (800, 600))], light_running
daemon.overlay.add_pulse.assert_called_once_with(
"/path", [((0, 0), (800, 600))], light_running
)
# --- multi-workspace scenarios ---
def test_two_workspaces_pulse_simultaneously(self):
"""Two workspaces can pulse at the same time."""
daemon = self._make_daemon(approval_delay=0.0, flash_mode="window")
win_a = {"title": "project-a", "frame": ((0, 0), (800, 600))}
win_b = {"title": "project-b", "frame": ((900, 0), (800, 600))}
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "preToolUse", "tool": "Shell"}).encode()
)
daemon._handle_message(
json.dumps({"workspace": "/b", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", side_effect=[win_a, win_b]), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending()
assert daemon.overlay.add_pulse.call_count == 2
assert "/a" in daemon._active_pulses
assert "/b" in daemon._active_pulses
def test_dismiss_one_workspace_keeps_other_pulsing(self):
"""postToolUse for workspace A only dismisses A, B keeps pulsing."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon()
daemon._active_pulses["/a"] = _ActivePulse("/a", "project-a", time.monotonic())
daemon._active_pulses["/b"] = _ActivePulse("/b", "project-b", time.monotonic())
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "postToolUse", "tool": "Shell"}).encode()
)
daemon.overlay.dismiss_tag.assert_called_once_with("/a")
assert "/a" not in daemon._active_pulses
assert "/b" in daemon._active_pulses
def test_stop_only_affects_its_workspace(self):
"""Stop for workspace A dismisses A's pulse and flashes A, B keeps pulsing."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon(flash_mode="window")
daemon._active_pulses["/a"] = _ActivePulse("/a", "project-a", time.monotonic())
daemon._active_pulses["/b"] = _ActivePulse("/b", "project-b", time.monotonic())
window = {"title": "project-a", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "stop"}).encode()
)
daemon.overlay.dismiss_tag.assert_called_with("/a")
daemon.overlay.add_flash.assert_called_once()
assert "/a" not in daemon._active_pulses
assert "/b" in daemon._active_pulses
def test_postToolUse_only_cancels_matching_workspace_pending(self):
"""postToolUse for workspace A doesn't cancel workspace B's pending."""
daemon = self._make_daemon(approval_delay=10.0)
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "preToolUse", "tool": "Shell"}).encode()
)
daemon._handle_message(
json.dumps({"workspace": "/b", "event": "preToolUse", "tool": "Shell"}).encode()
)
assert "/a" in daemon._pending_approvals
assert "/b" in daemon._pending_approvals
daemon._handle_message(
json.dumps({"workspace": "/a", "event": "postToolUse", "tool": "Shell"}).encode()
)
assert "/a" not in daemon._pending_approvals
assert "/b" in daemon._pending_approvals