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

Compare commits

base: cottongin:eefb908268fab3e2535bfd85aa19e324597ae9db
Branches Tags
cottongin:master
...
compare: cottongin:5fc378e558edcb7e9afa5bb6aca057d04d94c98f
Branches Tags
cottongin:master

3 Commits
eefb908268 ... 5fc378e558

Author SHA1 Message Date
cottongin
5fc378e558 Fix per-workspace pulse dismiss for multi-window setups 
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
 2026-03-10 09:33:42 -04:00
cottongin
a3203e2970 Add dark/light theme support with real-time OS appearance detection 
Config now uses top-level dark/light sections (each with running/completed
styles) and a theme option ("dark", "light", "auto"). The daemon resolves
the active theme at flash time via NSApp.effectiveAppearance().

Made-with: Cursor
 2026-03-10 09:06:09 -04:00
cottongin
49c03e4b71 Add design doc for dark/light theme support 
Defines the config format, data model, and daemon integration
approach for theme-aware styling.

Made-with: Cursor
 2026-03-10 08:34:43 -04:00
8 changed files with 865 additions and 313 deletions
Expand all files Collapse all files

56
README.md
View File

@@ -49,26 +49,44 @@ uv run cursor-flasher stop
Optional config file at `~/.cursor-flasher/config.yaml`: Optional config file at `~/.cursor-flasher/config.yaml`:
```yaml ```yaml
running: # approval pulse (continuous until you interact) theme: "auto" # "dark", "light", or "auto" (follows macOS appearance)
color: "#FF9500" # border color (hex)
width: 4 # border thickness in pixels
opacity: 0.85 # max border opacity
pulse_speed: 1.5 # pulse cycle speed in seconds
sound: "Glass" # macOS system sound ("" to disable)
volume: 0.5 # 0.0 to 1.0
# sounds: Basso, Blow, Bottle, Frog, Funk, Glass, Hero,
# Morse, Ping, Pop, Purr, Sosumi, Submarine, Tink
completed: # agent stop flash (brief fade-in/out) dark: # styles used when OS is in dark mode
color: "#00FF00" # different color for completion running: # approval pulse (continuous until you interact)
width: 4 color: "#FF9500" # border color (hex)
opacity: 0.85 width: 4 # border thickness in pixels
duration: 1.5 # flash duration in seconds opacity: 0.85 # max border opacity
sound: "" # no sound by default (Cursor plays its own) pulse_speed: 1.5 # pulse cycle speed in seconds
volume: 0.0 sound: "Glass" # macOS system sound ("" to disable)
volume: 0.5 # 0.0 to 1.0
# sounds: Basso, Blow, Bottle, Frog, Funk, Glass, Hero,
# Morse, Ping, Pop, Purr, Sosumi, Submarine, Tink
completed: # agent stop flash (brief fade-in/out)
color: "#00FF00" # different color for completion
width: 4
opacity: 0.85
duration: 1.5 # flash duration in seconds
sound: "" # no sound by default (Cursor plays its own)
volume: 0.0
light: # styles used when OS is in light mode
running:
color: "#3B82F6"
width: 4
opacity: 0.9
pulse_speed: 1.5
sound: "Glass"
volume: 0.5
completed:
color: "#22C55E"
width: 4
opacity: 0.9
duration: 1.5
sound: ""
volume: 0.0
flash: flash:
mode: "screen" # "window", "screen", or "allscreens" mode: "screen" # "window", "screen", or "allscreens"
# Tools that trigger the pulse + sound (approval mode). # Tools that trigger the pulse + sound (approval mode).
# Others are silently ignored (e.g., Read, Grep, Glob, Task). # Others are silently ignored (e.g., Read, Grep, Glob, Task).
@@ -78,11 +96,11 @@ approval_tools:
- Delete - Delete
general: general:
approval_delay: 2.5 # seconds to wait before pulsing (filters auto-approvals) approval_delay: 2.5 # seconds to wait before pulsing (filters auto-approvals)
cooldown: 2.0 # minimum seconds between flashes cooldown: 2.0 # minimum seconds between flashes
``` ```
Each mode (`running` and `completed`) has its own color, border style, and sound settings. Set `sound: ""` to disable sound for a particular mode. Styles are organized under `dark` and `light` theme sections, each containing `running` (approval pulse) and `completed` (stop flash) modes with their own color, border, and sound settings. The `theme` option controls which styles are active: set `"auto"` to follow macOS appearance in real-time, or force `"dark"` / `"light"`. Set `sound: ""` to disable sound for a particular mode.
## Uninstall ## Uninstall

84
docs/plans/2026-03-10-dark-light-theme-design.md Normal file
View File

@@ -0,0 +1,84 @@
# Dark/Light Theme Support — Design
## Summary
Add dark/light mode theme support to cursor-flasher. Users can define separate border styles for each OS appearance mode via new `dark` and `light` config sections. A `theme` option controls which styles are active: `"dark"`, `"light"`, or `"auto"` (follows macOS appearance in real-time).
## Config Format
The old top-level `running`/`completed` format is replaced (breaking change). Modes are now nested under theme sections:
```yaml
theme: auto # "dark" | "light" | "auto"
dark:
running:
color: "#FF9500"
width: 4
opacity: 0.85
pulse_speed: 1.5
sound: "Glass"
volume: 0.5
completed:
color: "#00FF00"
width: 4
opacity: 0.85
duration: 1.5
sound: ""
volume: 0.0
light:
running:
color: "#3B82F6"
width: 4
opacity: 0.9
pulse_speed: 1.5
sound: "Glass"
volume: 0.5
completed:
color: "#22C55E"
width: 4
opacity: 0.9
duration: 1.5
sound: ""
volume: 0.0
flash:
mode: "screen"
approval_tools:
- Shell
- Write
- Delete
general:
approval_delay: 2.5
cooldown: 2.0
```
## Data Model
- `StyleConfig` — unchanged (color, width, opacity, duration, pulse_speed, sound, volume).
- New `ThemeStyles` dataclass — groups `running: StyleConfig` and `completed: StyleConfig` for one theme.
- `Config` — replaces `running`/`completed` with `dark: ThemeStyles` and `light: ThemeStyles`. Adds `theme: str` field. Exposes `active_styles(system_appearance: str) -> ThemeStyles` method that resolves the correct theme based on the `theme` setting and the passed-in system appearance string.
## Appearance Detection
The daemon detects macOS appearance via `NSApplication.sharedApplication().effectiveAppearance().name()`. If the name contains "Dark", the appearance is `"dark"`; otherwise `"light"`. This check happens at flash/pulse trigger time (not polled), so it picks up OS appearance changes between flashes with zero overhead.
## Daemon Integration
Two call sites change: `_check_pending()` and `_handle_stop()`. Each resolves the active theme styles at trigger time:
```python
styles = self.config.active_styles(_get_system_appearance())
self.overlay.pulse(frames, styles.running)
play_alert(styles.running)
```
## Decisions
- **Modes under themes** (not themes under modes) — `dark.running` rather than `running.dark`.
- **Old format not supported** — top-level `running`/`completed` keys are ignored.
- **Real-time detection** — appearance checked at each flash trigger, not just at startup.
- **Config stays pure** — no Cocoa imports in config.py; appearance detection lives in daemon.py.

54
src/cursor_flasher/config.py
View File

@@ -15,6 +15,8 @@ STYLE_FIELDS = {
"volume": float, "volume": float,
} }
VALID_THEMES = {"dark", "light", "auto"}
@dataclass @dataclass
class StyleConfig: class StyleConfig:
@@ -36,10 +38,19 @@ def _default_completed() -> StyleConfig:
@dataclass @dataclass
class Config: class ThemeStyles:
"""Running and completed styles for a single theme (dark or light)."""
running: StyleConfig = field(default_factory=_default_running) running: StyleConfig = field(default_factory=_default_running)
completed: StyleConfig = field(default_factory=_default_completed) completed: StyleConfig = field(default_factory=_default_completed)
@dataclass
class Config:
dark: ThemeStyles = field(default_factory=ThemeStyles)
light: ThemeStyles = field(default_factory=ThemeStyles)
theme: str = "auto"
flash_mode: str = "screen" flash_mode: str = "screen"
approval_tools: list[str] = field( approval_tools: list[str] = field(
@@ -49,6 +60,16 @@ class Config:
approval_delay: float = 2.5 approval_delay: float = 2.5
cooldown: float = 2.0 cooldown: float = 2.0
def active_styles(self, system_appearance: str) -> ThemeStyles:
"""Return the ThemeStyles matching the current theme setting.
Args:
system_appearance: "dark" or "light" as detected from the OS.
"""
if self.theme == "auto":
return self.dark if system_appearance == "dark" else self.light
return self.dark if self.theme == "dark" else self.light
GENERAL_FIELD_MAP: dict[str, str] = { GENERAL_FIELD_MAP: dict[str, str] = {
"approval_delay": "approval_delay", "approval_delay": "approval_delay",
@@ -77,6 +98,21 @@ def _parse_style(raw_section: dict, defaults: StyleConfig) -> StyleConfig:
) )
def _parse_theme_styles(raw_section: dict) -> ThemeStyles:
"""Build a ThemeStyles from a YAML theme section (dark or light)."""
kwargs: dict[str, StyleConfig] = {}
running_raw = raw_section.get("running")
if isinstance(running_raw, dict):
kwargs["running"] = _parse_style(running_raw, _default_running())
completed_raw = raw_section.get("completed")
if isinstance(completed_raw, dict):
kwargs["completed"] = _parse_style(completed_raw, _default_completed())
return ThemeStyles(**kwargs)
def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config: def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
"""Load config from YAML, falling back to defaults for missing values.""" """Load config from YAML, falling back to defaults for missing values."""
if not path.exists(): if not path.exists():
@@ -90,13 +126,17 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
config_kwargs: dict[str, Any] = {} config_kwargs: dict[str, Any] = {}
running_raw = raw.get("running") theme = raw.get("theme")
if isinstance(running_raw, dict): if isinstance(theme, str) and theme in VALID_THEMES:
config_kwargs["running"] = _parse_style(running_raw, _default_running()) config_kwargs["theme"] = theme
completed_raw = raw.get("completed") dark_raw = raw.get("dark")
if isinstance(completed_raw, dict): if isinstance(dark_raw, dict):
config_kwargs["completed"] = _parse_style(completed_raw, _default_completed()) config_kwargs["dark"] = _parse_theme_styles(dark_raw)
light_raw = raw.get("light")
if isinstance(light_raw, dict):
config_kwargs["light"] = _parse_theme_styles(light_raw)
flash_raw = raw.get("flash") flash_raw = raw.get("flash")
if isinstance(flash_raw, dict) and "mode" in flash_raw: if isinstance(flash_raw, dict) and "mode" in flash_raw:

185
src/cursor_flasher/daemon.py
View File

@@ -20,6 +20,7 @@ from cursor_flasher.overlay import OverlayManager
from cursor_flasher.sound import play_alert from cursor_flasher.sound import play_alert
from cursor_flasher.windows import ( from cursor_flasher.windows import (
find_window_by_workspace, find_window_by_workspace,
get_focused_cursor_window,
screen_frame_for_window, screen_frame_for_window,
all_screen_frames, all_screen_frames,
is_cursor_frontmost, is_cursor_frontmost,
@@ -33,6 +34,13 @@ SOCKET_PATH = os.path.join(SOCKET_DIR, "flasher.sock")
INPUT_DISMISS_GRACE = 0.5 INPUT_DISMISS_GRACE = 0.5
def _get_system_appearance() -> str:
"""Return "dark" or "light" based on the current macOS appearance."""
app = NSApplication.sharedApplication()
name = app.effectiveAppearance().name()
return "dark" if "Dark" in name else "light"
class _PendingApproval: class _PendingApproval:
"""An approval trigger waiting for the delay to expire before pulsing.""" """An approval trigger waiting for the delay to expire before pulsing."""
__slots__ = ("workspace", "tool", "timestamp") __slots__ = ("workspace", "tool", "timestamp")
@@ -43,15 +51,25 @@ class _PendingApproval:
self.timestamp = timestamp self.timestamp = timestamp
class _ActivePulse:
"""A workspace that is currently pulsing."""
__slots__ = ("workspace", "window_title", "started_at")
def __init__(self, workspace: str, window_title: str, started_at: float):
self.workspace = workspace
self.window_title = window_title
self.started_at = started_at
class FlasherDaemon: class FlasherDaemon:
def __init__(self, config: Config): def __init__(self, config: Config):
self.config = config self.config = config
self.overlay = OverlayManager() self.overlay = OverlayManager()
self._running = False self._running = False
self._server: socket.socket | None = None self._server: socket.socket | None = None
self._last_flash: float = 0 self._last_flash: dict[str, float] = {}
self._pending: _PendingApproval | None = None self._pending_approvals: dict[str, _PendingApproval] = {}
self._pulse_started_at: float = 0 self._active_pulses: dict[str, _ActivePulse] = {}
self._cursor_was_frontmost: bool = False self._cursor_was_frontmost: bool = False
def run(self) -> None: def run(self) -> None:
@@ -108,47 +126,52 @@ class FlasherDaemon:
pass pass
def _check_pending(self) -> None: def _check_pending(self) -> None:
"""Promote a pending approval to an active pulse after the delay expires.""" """Promote pending approvals whose delay has expired."""
if self._pending is None: promoted: list[str] = []
return for workspace, pending in self._pending_approvals.items():
elapsed = time.monotonic() - pending.timestamp
if elapsed < self.config.approval_delay:
continue
elapsed = time.monotonic() - self._pending.timestamp window = find_window_by_workspace(pending.workspace)
if elapsed < self.config.approval_delay: if window is None:
return logger.warning(
"No Cursor window found for pending approval: %s", workspace
)
promoted.append(workspace)
continue
pending = self._pending frames = self._resolve_frames(window["frame"])
self._pending = None styles = self.config.active_styles(_get_system_appearance())
logger.info(
"Pulsing for approval (after %.1fs delay): tool=%s window=%s",
elapsed, pending.tool, window["title"],
)
self.overlay.add_pulse(workspace, frames, styles.running)
self._active_pulses[workspace] = _ActivePulse(
workspace, window["title"], time.monotonic()
)
self._cursor_was_frontmost = is_cursor_frontmost()
play_alert(styles.running)
self._last_flash[workspace] = time.monotonic()
promoted.append(workspace)
window = find_window_by_workspace(pending.workspace) for workspace in promoted:
if window is None: self._pending_approvals.pop(workspace, None)
logger.warning("No Cursor window found for pending approval")
return
frames = self._resolve_frames(window["frame"])
logger.info(
"Pulsing for approval (after %.1fs delay): tool=%s window=%s",
elapsed, pending.tool, window["title"],
)
self.overlay.pulse(frames, self.config.running)
self._pulse_started_at = time.monotonic()
self._cursor_was_frontmost = is_cursor_frontmost()
play_alert(self.config.running)
self._last_flash = time.monotonic()
def _check_input_dismiss(self) -> None: def _check_input_dismiss(self) -> None:
"""Dismiss pulse when user clicks or types while Cursor is frontmost. """Dismiss the focused window's pulse when the user clicks or types.
Polls CGEventSourceSecondsSinceLastEventType which reads system-wide Identifies which Cursor window has focus and only dismisses that
HID counters — works reliably from forked daemon processes unlike workspace's pulse, leaving other workspaces pulsing.
CGEventTap callbacks which silently fail without a window server
connection.
""" """
if not self.overlay.is_pulsing: if not self._active_pulses:
return return
if not is_cursor_frontmost(): if not is_cursor_frontmost():
return return
pulse_age = time.monotonic() - self._pulse_started_at oldest_start = min(p.started_at for p in self._active_pulses.values())
pulse_age = time.monotonic() - oldest_start
if pulse_age < INPUT_DISMISS_GRACE: if pulse_age < INPUT_DISMISS_GRACE:
return return
@@ -163,32 +186,50 @@ class FlasherDaemon:
) )
last_input = min(last_click, last_rclick, last_key) last_input = min(last_click, last_rclick, last_key)
if last_input >= (pulse_age - INPUT_DISMISS_GRACE):
return
if last_input < (pulse_age - INPUT_DISMISS_GRACE): workspace = self._find_focused_workspace()
logger.info( if workspace is None:
"User input in Cursor — dismissing pulse " return
"(input %.1fs ago, pulse %.1fs old)",
last_input, pulse_age, logger.info(
) "User input in Cursor — dismissing pulse for %s "
self._dismiss_pulse() "(input %.1fs ago, pulse %.1fs old)",
workspace, last_input, pulse_age,
)
self._dismiss_workspace(workspace)
def _check_focus(self) -> None: def _check_focus(self) -> None:
"""Dismiss pulse when user switches TO Cursor via Cmd+Tab or similar. """Dismiss the focused window's pulse when user switches TO Cursor."""
if not self._active_pulses:
Detects the transition from another app to Cursor. self._cursor_was_frontmost = is_cursor_frontmost()
"""
if not self.overlay.is_pulsing:
return return
frontmost = is_cursor_frontmost() frontmost = is_cursor_frontmost()
if frontmost and not self._cursor_was_frontmost: if frontmost and not self._cursor_was_frontmost:
logger.info("Cursor became frontmost — dismissing pulse") workspace = self._find_focused_workspace()
self._dismiss_pulse() if workspace is not None:
logger.info(
"Cursor became frontmost — dismissing pulse for %s", workspace
)
self._dismiss_workspace(workspace)
self._cursor_was_frontmost = frontmost self._cursor_was_frontmost = frontmost
def _dismiss_pulse(self) -> None: def _find_focused_workspace(self) -> str | None:
"""Centralized pulse dismissal.""" """Match the currently focused Cursor window to an active pulse."""
self.overlay.dismiss() focused = get_focused_cursor_window()
if focused is None:
return None
for workspace, pulse in self._active_pulses.items():
if pulse.window_title == focused["title"]:
return workspace
return None
def _dismiss_workspace(self, workspace: str) -> None:
"""Dismiss a single workspace's pulse."""
self._active_pulses.pop(workspace, None)
self.overlay.dismiss_tag(workspace)
def _handle_message(self, raw: bytes) -> None: def _handle_message(self, raw: bytes) -> None:
try: try:
@@ -203,12 +244,12 @@ class FlasherDaemon:
logger.info("Received: event=%s tool=%s pulsing=%s pending=%s", logger.info("Received: event=%s tool=%s pulsing=%s pending=%s",
event, tool, self.overlay.is_pulsing, event, tool, self.overlay.is_pulsing,
self._pending is not None) bool(self._pending_approvals))
if event == "preToolUse": if event == "preToolUse":
self._handle_approval(workspace, tool) self._handle_approval(workspace, tool)
elif event in ("postToolUse", "postToolUseFailure"): elif event in ("postToolUse", "postToolUseFailure"):
self._handle_dismiss(event, tool) self._handle_dismiss(workspace, event, tool)
elif event == "stop": elif event == "stop":
self._handle_stop(workspace) self._handle_stop(workspace)
else: else:
@@ -219,44 +260,48 @@ class FlasherDaemon:
return return
now = time.monotonic() now = time.monotonic()
if (now - self._last_flash) < self.config.cooldown: last = self._last_flash.get(workspace, 0)
if (now - last) < self.config.cooldown:
return return
logger.debug("Queuing approval: tool=%s workspace=%s", tool, workspace) logger.debug("Queuing approval: tool=%s workspace=%s", tool, workspace)
self._pending = _PendingApproval(workspace, tool, now) self._pending_approvals[workspace] = _PendingApproval(workspace, tool, now)
def _handle_dismiss(self, event: str, tool: str) -> None: def _handle_dismiss(self, workspace: str, event: str, tool: str) -> None:
if self._pending is not None: if workspace in self._pending_approvals:
logger.debug( logger.debug(
"Cancelled pending approval (auto-approved): %s tool=%s", "Cancelled pending approval (auto-approved): %s tool=%s workspace=%s",
event, tool, event, tool, workspace,
) )
self._pending = None self._pending_approvals.pop(workspace, None)
if self.overlay.is_pulsing: if workspace in self._active_pulses:
logger.info("Dismissing pulse: %s tool=%s", event, tool) logger.info(
self._dismiss_pulse() "Dismissing pulse: %s tool=%s workspace=%s", event, tool, workspace
)
self._dismiss_workspace(workspace)
def _handle_stop(self, workspace: str) -> None: def _handle_stop(self, workspace: str) -> None:
self._pending = None self._pending_approvals.pop(workspace, None)
if self.overlay.is_pulsing: if workspace in self._active_pulses:
self._dismiss_pulse() self._dismiss_workspace(workspace)
return
now = time.monotonic() now = time.monotonic()
if (now - self._last_flash) < self.config.cooldown: last = self._last_flash.get(workspace, 0)
if (now - last) < self.config.cooldown:
return return
window = find_window_by_workspace(workspace) window = find_window_by_workspace(workspace)
if window is None: if window is None:
return return
styles = self.config.active_styles(_get_system_appearance())
frames = self._resolve_frames(window["frame"]) frames = self._resolve_frames(window["frame"])
logger.info("Flash for stop: window=%s", window["title"]) logger.info("Flash for stop: window=%s", window["title"])
self.overlay.flash(frames, self.config.completed) self.overlay.add_flash(workspace, frames, styles.completed)
play_alert(self.config.completed) play_alert(styles.completed)
self._last_flash = now self._last_flash[workspace] = now
def _resolve_frames(self, window_frame: tuple) -> list[tuple]: def _resolve_frames(self, window_frame: tuple) -> list[tuple]:
"""Return frame(s) based on flash_mode config.""" """Return frame(s) based on flash_mode config."""

210
src/cursor_flasher/overlay.py
View File

@@ -62,79 +62,99 @@ class _Mode(enum.Enum):
PULSE = "pulse" PULSE = "pulse"
class OverlayManager: class _TagState:
"""Manages overlay borders on one or more frames simultaneously. """Per-tag state: mode, style, panels, and (for FLASH) its own elapsed."""
__slots__ = ("mode", "style", "panels", "elapsed")
Two modes: def __init__(
- flash(): brief fade-in/hold/fade-out, auto-dismisses self,
- pulse(): continuous sine-wave pulsing until dismiss() is called mode: _Mode,
style: StyleConfig,
panels: list[tuple[NSWindow, FlashBorderView]],
):
self.mode = mode
self.style = style
self.panels = panels
self.elapsed = 0.0
class OverlayManager:
"""Manages overlay borders grouped by tag (typically workspace path).
Supports multiple simultaneous pulse/flash groups. All PULSE tags share
a single elapsed counter so they animate in sync. Each FLASH tag has its
own elapsed counter for independent fade-in/hold/fade-out.
""" """
def __init__(self): def __init__(self):
self._panels: list[tuple[NSWindow, FlashBorderView]] = [] self._tags: dict[str, _TagState] = {}
self._timer: NSTimer | None = None self._timer: NSTimer | None = None
self._elapsed: float = 0.0 self._pulse_elapsed: float = 0.0
self._mode: _Mode = _Mode.IDLE
self._style: StyleConfig = StyleConfig()
@property @property
def is_pulsing(self) -> bool: def is_pulsing(self) -> bool:
return self._mode == _Mode.PULSE return any(ts.mode == _Mode.PULSE for ts in self._tags.values())
def flash(self, frames: list[tuple], style: StyleConfig) -> None: @property
"""Brief flash: fade in, hold, fade out, auto-dismiss.""" def active_tags(self) -> set[str]:
self._show(frames, _Mode.FLASH, style) return set(self._tags.keys())
def pulse(self, frames: list[tuple], style: StyleConfig) -> None: def has_tag(self, tag: str) -> bool:
"""Continuous pulse: sine-wave opacity until dismiss() is called.""" return tag in self._tags
self._show(frames, _Mode.PULSE, style)
def dismiss(self) -> None: def add_pulse(self, tag: str, frames: list[tuple], style: StyleConfig) -> None:
"""Stop any animation and hide all overlays.""" """Start pulsing panels for this tag. Reuses tag if it already exists."""
self._stop_timer() self._remove_tag(tag)
self._mode = _Mode.IDLE panels = [self._create_overlay(f) for f in frames]
for window, view in self._panels: self._tags[tag] = _TagState(_Mode.PULSE, style, panels)
view.setAlpha_(0.0) for window, view in panels:
window.setAlphaValue_(0.0)
window.orderOut_(None)
logger.debug("Overlay dismissed (%d panels hidden)", len(self._panels))
def hide(self) -> None:
self.dismiss()
def _show(self, frames: list[tuple], mode: _Mode, style: StyleConfig) -> None:
self._stop_timer()
self._elapsed = 0.0
self._mode = mode
self._style = style
self._ensure_panels(len(frames))
for i, frame in enumerate(frames):
window, view = self._panels[i]
view._style = style view._style = style
window.setFrame_display_(frame, True)
content_rect = ((0, 0), (frame[1][0], frame[1][1]))
view.setFrame_(content_rect)
view.setAlpha_(style.opacity) view.setAlpha_(style.opacity)
window.setAlphaValue_(1.0) window.setAlphaValue_(1.0)
window.orderFrontRegardless() window.orderFrontRegardless()
self._ensure_timer()
for j in range(len(frames), len(self._panels)): def add_flash(self, tag: str, frames: list[tuple], style: StyleConfig) -> None:
self._panels[j][0].orderOut_(None) """Start a brief flash for this tag. Auto-removes when done."""
self._remove_tag(tag)
panels = [self._create_overlay(f) for f in frames]
self._tags[tag] = _TagState(_Mode.FLASH, style, panels)
for window, view in panels:
view._style = style
view.setAlpha_(0.0)
window.setAlphaValue_(1.0)
window.orderFrontRegardless()
self._ensure_timer()
interval = 1.0 / 30.0 def dismiss_tag(self, tag: str) -> None:
self._timer = NSTimer.scheduledTimerWithTimeInterval_target_selector_userInfo_repeats_( """Hide and remove panels for a specific tag."""
interval, self, "_tick:", None, True self._remove_tag(tag)
) if not self._tags:
self._stop_timer()
self._pulse_elapsed = 0.0
def _ensure_panels(self, count: int) -> None: def dismiss_all(self) -> None:
"""Grow the panel pool if needed.""" """Hide everything and stop the timer."""
while len(self._panels) < count: for tag in list(self._tags):
dummy = ((0, 0), (1, 1)) self._remove_tag(tag)
self._panels.append(self._create_overlay(dummy)) self._stop_timer()
self._pulse_elapsed = 0.0
logger.debug("All overlays dismissed")
def _create_overlay(self, frame) -> tuple: def hide(self) -> None:
self.dismiss_all()
def _remove_tag(self, tag: str) -> None:
state = self._tags.pop(tag, None)
if state is None:
return
for window, view in state.panels:
view.setAlpha_(0.0)
window.setAlphaValue_(0.0)
window.orderOut_(None)
logger.debug("Tag '%s' dismissed (%d panels hidden)", tag, len(state.panels))
def _create_overlay(self, frame) -> tuple[NSWindow, FlashBorderView]:
window = NSWindow.alloc().initWithContentRect_styleMask_backing_defer_( window = NSWindow.alloc().initWithContentRect_styleMask_backing_defer_(
frame, NSBorderlessWindowMask, 2, False frame, NSBorderlessWindowMask, 2, False
) )
@@ -149,6 +169,14 @@ class OverlayManager:
window.setContentView_(view) window.setContentView_(view)
return window, view return window, view
def _ensure_timer(self) -> None:
if self._timer is not None:
return
interval = 1.0 / 30.0
self._timer = NSTimer.scheduledTimerWithTimeInterval_target_selector_userInfo_repeats_(
interval, self, "_tick:", None, True
)
def _stop_timer(self) -> None: def _stop_timer(self) -> None:
if self._timer is not None: if self._timer is not None:
self._timer.invalidate() self._timer.invalidate()
@@ -156,47 +184,59 @@ class OverlayManager:
@objc.python_method @objc.python_method
def _tick_impl(self): def _tick_impl(self):
if not self._tags:
self._stop_timer()
return
dt = 1.0 / 30.0 dt = 1.0 / 30.0
self._elapsed += dt self._pulse_elapsed += dt
match self._mode: tags_to_remove: list[str] = []
case _Mode.FLASH: for tag, state in self._tags.items():
self._tick_flash() match state.mode:
case _Mode.PULSE: case _Mode.PULSE:
self._tick_pulse() self._tick_pulse_tag(state)
case _Mode.IDLE: case _Mode.FLASH:
self._stop_timer() state.elapsed += dt
if not self._tick_flash_tag(tag, state):
tags_to_remove.append(tag)
def _tick_flash(self) -> None: for tag in tags_to_remove:
duration = self._style.duration self._remove_tag(tag)
if not self._tags:
self._stop_timer()
self._pulse_elapsed = 0.0
def _tick_pulse_tag(self, state: _TagState) -> None:
speed = state.style.pulse_speed
phase = (2.0 * math.pi * self._pulse_elapsed) / speed
opacity_min = 0.3
t = 0.5 + 0.5 * math.sin(phase)
alpha = opacity_min + (state.style.opacity - opacity_min) * t
for _, view in state.panels:
view.setAlpha_(alpha)
def _tick_flash_tag(self, tag: str, state: _TagState) -> bool:
"""Tick a flash tag. Returns False when the flash is finished."""
duration = state.style.duration
fade_in = 0.15 fade_in = 0.15
fade_out = 0.4 fade_out = 0.4
hold_end = duration - fade_out hold_end = duration - fade_out
elapsed = state.elapsed
if self._elapsed < fade_in: if elapsed < fade_in:
alpha = self._style.opacity * (self._elapsed / fade_in) alpha = state.style.opacity * (elapsed / fade_in)
elif self._elapsed < hold_end: elif elapsed < hold_end:
alpha = self._style.opacity alpha = state.style.opacity
elif self._elapsed < duration: elif elapsed < duration:
progress = (self._elapsed - hold_end) / fade_out progress = (elapsed - hold_end) / fade_out
alpha = self._style.opacity * (1.0 - progress) alpha = state.style.opacity * (1.0 - progress)
else: else:
self.dismiss() return False
return
self._set_all_alpha(alpha) for _, view in state.panels:
def _tick_pulse(self) -> None:
speed = self._style.pulse_speed
phase = (2.0 * math.pi * self._elapsed) / speed
opacity_min = 0.3
t = 0.5 + 0.5 * math.sin(phase)
alpha = opacity_min + (self._style.opacity - opacity_min) * t
self._set_all_alpha(alpha)
def _set_all_alpha(self, alpha: float) -> None:
for _, view in self._panels:
view.setAlpha_(alpha) view.setAlpha_(alpha)
return True
def _tick_(self, timer) -> None: def _tick_(self, timer) -> None:
self._tick_impl() self._tick_impl()

27
src/cursor_flasher/windows.py
View File

@@ -104,6 +104,33 @@ def find_window_by_workspace(workspace_path: str) -> dict | None:
return windows[0] if len(windows) == 1 else None return windows[0] if len(windows) == 1 else None
def get_focused_cursor_window() -> dict | None:
"""Return the currently focused (key) Cursor window.
Uses the AXFocusedWindow attribute to identify which specific Cursor
window has keyboard focus. Returns {"title": str, "frame": tuple} or
None if Cursor isn't running or has no focused window.
"""
pid = find_cursor_pid()
if pid is None:
return None
app = AXUIElementCreateApplication(pid)
err, focused = AXUIElementCopyAttributeValue(app, "AXFocusedWindow", None)
if err or focused is None:
return None
err, title = AXUIElementCopyAttributeValue(focused, "AXTitle", None)
title = str(title) if not err and title else ""
screen_height = NSScreen.mainScreen().frame().size.height
frame = _read_frame(focused, screen_height)
if frame is None:
return None
return {"title": title, "frame": frame}
def screen_frame_for_window(window_frame: tuple) -> tuple: def screen_frame_for_window(window_frame: tuple) -> tuple:
"""Return the NSScreen frame of the monitor containing the window's center. """Return the NSScreen frame of the monitor containing the window's center.

212
tests/test_config.py
View File

@@ -1,23 +1,42 @@
from cursor_flasher.config import Config, StyleConfig, load_config from cursor_flasher.config import (
Config,
StyleConfig,
ThemeStyles,
load_config,
)
class TestDefaultConfig: class TestDefaultConfig:
def test_running_defaults(self): def test_dark_running_defaults(self):
c = Config() c = Config()
assert c.running.color == "#FF9500" assert c.dark.running.color == "#FF9500"
assert c.running.width == 4 assert c.dark.running.width == 4
assert c.running.duration == 1.5 assert c.dark.running.duration == 1.5
assert c.running.opacity == 0.85 assert c.dark.running.opacity == 0.85
assert c.running.pulse_speed == 1.5 assert c.dark.running.pulse_speed == 1.5
assert c.running.sound == "Glass" assert c.dark.running.sound == "Glass"
assert c.running.volume == 0.5 assert c.dark.running.volume == 0.5
def test_completed_defaults(self): def test_dark_completed_defaults(self):
c = Config() c = Config()
assert c.completed.color == "#FF9500" assert c.dark.completed.color == "#FF9500"
assert c.completed.width == 4 assert c.dark.completed.width == 4
assert c.completed.sound == "" assert c.dark.completed.sound == ""
assert c.completed.volume == 0.0 assert c.dark.completed.volume == 0.0
def test_light_running_defaults(self):
c = Config()
assert c.light.running.color == "#FF9500"
assert c.light.running.sound == "Glass"
def test_light_completed_defaults(self):
c = Config()
assert c.light.completed.sound == ""
assert c.light.completed.volume == 0.0
def test_theme_defaults_to_auto(self):
c = Config()
assert c.theme == "auto"
def test_has_approval_tools(self): def test_has_approval_tools(self):
c = Config() c = Config()
@@ -32,6 +51,41 @@ class TestDefaultConfig:
assert c.flash_mode == "screen" assert c.flash_mode == "screen"
class TestActiveStyles:
def test_auto_returns_dark_when_system_dark(self):
c = Config(theme="auto")
dark_styles = ThemeStyles(running=StyleConfig(color="#111111"))
light_styles = ThemeStyles(running=StyleConfig(color="#EEEEEE"))
c.dark = dark_styles
c.light = light_styles
assert c.active_styles("dark").running.color == "#111111"
def test_auto_returns_light_when_system_light(self):
c = Config(theme="auto")
dark_styles = ThemeStyles(running=StyleConfig(color="#111111"))
light_styles = ThemeStyles(running=StyleConfig(color="#EEEEEE"))
c.dark = dark_styles
c.light = light_styles
assert c.active_styles("light").running.color == "#EEEEEE"
def test_explicit_dark_ignores_system(self):
c = Config(theme="dark")
c.dark = ThemeStyles(running=StyleConfig(color="#111111"))
c.light = ThemeStyles(running=StyleConfig(color="#EEEEEE"))
assert c.active_styles("light").running.color == "#111111"
def test_explicit_light_ignores_system(self):
c = Config(theme="light")
c.dark = ThemeStyles(running=StyleConfig(color="#111111"))
c.light = ThemeStyles(running=StyleConfig(color="#EEEEEE"))
assert c.active_styles("dark").running.color == "#EEEEEE"
def test_active_styles_includes_completed(self):
c = Config(theme="dark")
c.dark = ThemeStyles(completed=StyleConfig(color="#AA0000"))
assert c.active_styles("light").completed.color == "#AA0000"
class TestLoadConfig: class TestLoadConfig:
def test_missing_file_returns_defaults(self, tmp_path): def test_missing_file_returns_defaults(self, tmp_path):
c = load_config(tmp_path / "nope.yaml") c = load_config(tmp_path / "nope.yaml")
@@ -43,26 +97,80 @@ class TestLoadConfig:
c = load_config(p) c = load_config(p)
assert c == Config() assert c == Config()
def test_loads_running_overrides(self, tmp_path): def test_loads_theme(self, tmp_path):
p = tmp_path / "config.yaml" p = tmp_path / "config.yaml"
p.write_text( p.write_text("theme: dark\n")
"running:\n color: '#00FF00'\n duration: 2.0\n sound: Ping\n"
)
c = load_config(p) c = load_config(p)
assert c.running.color == "#00FF00" assert c.theme == "dark"
assert c.running.duration == 2.0
assert c.running.sound == "Ping"
assert c.running.width == 4
def test_loads_completed_overrides(self, tmp_path): def test_invalid_theme_uses_default(self, tmp_path):
p = tmp_path / "config.yaml"
p.write_text("theme: neon\n")
c = load_config(p)
assert c.theme == "auto"
def test_loads_dark_running_overrides(self, tmp_path):
p = tmp_path / "config.yaml" p = tmp_path / "config.yaml"
p.write_text( p.write_text(
"completed:\n color: '#0000FF'\n sound: Hero\n volume: 0.8\n" "dark:\n"
" running:\n"
" color: '#00FF00'\n"
" duration: 2.0\n"
" sound: Ping\n"
) )
c = load_config(p) c = load_config(p)
assert c.completed.color == "#0000FF" assert c.dark.running.color == "#00FF00"
assert c.completed.sound == "Hero" assert c.dark.running.duration == 2.0
assert c.completed.volume == 0.8 assert c.dark.running.sound == "Ping"
assert c.dark.running.width == 4
def test_loads_dark_completed_overrides(self, tmp_path):
p = tmp_path / "config.yaml"
p.write_text(
"dark:\n"
" completed:\n"
" color: '#0000FF'\n"
" sound: Hero\n"
" volume: 0.8\n"
)
c = load_config(p)
assert c.dark.completed.color == "#0000FF"
assert c.dark.completed.sound == "Hero"
assert c.dark.completed.volume == 0.8
def test_loads_light_running_overrides(self, tmp_path):
p = tmp_path / "config.yaml"
p.write_text(
"light:\n"
" running:\n"
" color: '#3B82F6'\n"
" opacity: 0.9\n"
)
c = load_config(p)
assert c.light.running.color == "#3B82F6"
assert c.light.running.opacity == 0.9
assert c.light.running.width == 4
def test_missing_dark_section_uses_defaults(self, tmp_path):
p = tmp_path / "config.yaml"
p.write_text(
"light:\n"
" running:\n"
" color: '#FFFFFF'\n"
)
c = load_config(p)
assert c.dark == ThemeStyles()
def test_missing_running_within_theme_uses_defaults(self, tmp_path):
p = tmp_path / "config.yaml"
p.write_text(
"dark:\n"
" completed:\n"
" color: '#FF0000'\n"
)
c = load_config(p)
assert c.dark.running == StyleConfig()
assert c.dark.completed.color == "#FF0000"
def test_loads_flash_mode(self, tmp_path): def test_loads_flash_mode(self, tmp_path):
p = tmp_path / "config.yaml" p = tmp_path / "config.yaml"
@@ -86,16 +194,25 @@ class TestLoadConfig:
def test_full_config(self, tmp_path): def test_full_config(self, tmp_path):
p = tmp_path / "config.yaml" p = tmp_path / "config.yaml"
p.write_text( p.write_text(
"running:\n" "theme: light\n"
" color: '#FF0000'\n" "dark:\n"
" width: 6\n" " running:\n"
" opacity: 0.9\n" " color: '#FF0000'\n"
" pulse_speed: 2.0\n" " width: 6\n"
" sound: Glass\n" " opacity: 0.9\n"
" volume: 0.8\n" " pulse_speed: 2.0\n"
"completed:\n" " sound: Glass\n"
" color: '#00FF00'\n" " volume: 0.8\n"
" sound: ''\n" " completed:\n"
" color: '#00FF00'\n"
" sound: ''\n"
"light:\n"
" running:\n"
" color: '#3B82F6'\n"
" width: 3\n"
" completed:\n"
" color: '#22C55E'\n"
" duration: 2.0\n"
"flash:\n" "flash:\n"
" mode: window\n" " mode: window\n"
"general:\n" "general:\n"
@@ -105,14 +222,19 @@ class TestLoadConfig:
" - Shell\n" " - Shell\n"
) )
c = load_config(p) c = load_config(p)
assert c.running.color == "#FF0000" assert c.theme == "light"
assert c.running.width == 6 assert c.dark.running.color == "#FF0000"
assert c.running.opacity == 0.9 assert c.dark.running.width == 6
assert c.running.pulse_speed == 2.0 assert c.dark.running.opacity == 0.9
assert c.running.sound == "Glass" assert c.dark.running.pulse_speed == 2.0
assert c.running.volume == 0.8 assert c.dark.running.sound == "Glass"
assert c.completed.color == "#00FF00" assert c.dark.running.volume == 0.8
assert c.completed.sound == "" assert c.dark.completed.color == "#00FF00"
assert c.dark.completed.sound == ""
assert c.light.running.color == "#3B82F6"
assert c.light.running.width == 3
assert c.light.completed.color == "#22C55E"
assert c.light.completed.duration == 2.0
assert c.flash_mode == "window" assert c.flash_mode == "window"
assert c.approval_delay == 1.0 assert c.approval_delay == 1.0
assert c.cooldown == 3.0 assert c.cooldown == 3.0

350
tests/test_daemon.py
View File

@@ -4,18 +4,26 @@ import time
from unittest.mock import patch, MagicMock from unittest.mock import patch, MagicMock
from cursor_flasher.config import Config, StyleConfig from cursor_flasher.config import Config, StyleConfig, ThemeStyles
from cursor_flasher.daemon import FlasherDaemon 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: class TestFlasherDaemon:
def _make_daemon(self, **config_overrides) -> FlasherDaemon: def _make_daemon(self, **config_overrides) -> FlasherDaemon:
config = Config(**config_overrides) config = Config(**config_overrides)
with patch("cursor_flasher.daemon.OverlayManager"): with patch("cursor_flasher.daemon.OverlayManager") as MockOverlay:
daemon = FlasherDaemon(config) daemon = FlasherDaemon(config)
daemon.overlay.is_pulsing = False daemon.overlay.is_pulsing = False
daemon.overlay.active_tags = set()
daemon.overlay.has_tag = lambda tag: tag in daemon.overlay.active_tags
return daemon return daemon
# --- preToolUse / pending ---
def test_preToolUse_queues_pending(self): def test_preToolUse_queues_pending(self):
daemon = self._make_daemon() daemon = self._make_daemon()
@@ -23,9 +31,9 @@ class TestFlasherDaemon:
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode() json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
) )
assert daemon._pending is not None assert "/path" in daemon._pending_approvals
assert daemon._pending.tool == "Shell" assert daemon._pending_approvals["/path"].tool == "Shell"
daemon.overlay.pulse.assert_not_called() daemon.overlay.add_pulse.assert_not_called()
def test_pending_promotes_after_delay(self): def test_pending_promotes_after_delay(self):
daemon = self._make_daemon(approval_delay=0.0, flash_mode="screen") daemon = self._make_daemon(approval_delay=0.0, flash_mode="screen")
@@ -39,10 +47,15 @@ class TestFlasherDaemon:
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.screen_frame_for_window", return_value=screen), \ patch("cursor_flasher.daemon.screen_frame_for_window", return_value=screen), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \ patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"): patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending() daemon._check_pending()
daemon.overlay.pulse.assert_called_once_with([screen], daemon.config.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): def test_postToolUse_cancels_pending(self):
"""Auto-approved tools: postToolUse arrives before delay, cancels the pulse.""" """Auto-approved tools: postToolUse arrives before delay, cancels the pulse."""
@@ -51,13 +64,13 @@ class TestFlasherDaemon:
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode() json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
) )
assert daemon._pending is not None assert "/path" in daemon._pending_approvals
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUse", "tool": "Shell"}).encode() json.dumps({"workspace": "/path", "event": "postToolUse", "tool": "Shell"}).encode()
) )
assert daemon._pending is None assert "/path" not in daemon._pending_approvals
daemon.overlay.pulse.assert_not_called() daemon.overlay.add_pulse.assert_not_called()
def test_preToolUse_skips_non_approval_tool(self): def test_preToolUse_skips_non_approval_tool(self):
daemon = self._make_daemon() daemon = self._make_daemon()
@@ -66,7 +79,7 @@ class TestFlasherDaemon:
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Read"}).encode() 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): def test_preToolUse_respects_custom_tool_list(self):
daemon = self._make_daemon(approval_delay=0.0, flash_mode="screen", approval_tools=["Shell", "MCP"]) daemon = self._make_daemon(approval_delay=0.0, flash_mode="screen", approval_tools=["Shell", "MCP"])
@@ -79,10 +92,13 @@ class TestFlasherDaemon:
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.screen_frame_for_window", return_value=((0, 0), (1920, 1080))), \ patch("cursor_flasher.daemon.screen_frame_for_window", return_value=((0, 0), (1920, 1080))), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \ patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"): patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending() daemon._check_pending()
daemon.overlay.pulse.assert_called_once() daemon.overlay.add_pulse.assert_called_once()
# --- stop / flash ---
def test_stop_flashes_briefly(self): def test_stop_flashes_briefly(self):
daemon = self._make_daemon(flash_mode="screen") daemon = self._make_daemon(flash_mode="screen")
@@ -91,25 +107,29 @@ class TestFlasherDaemon:
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.screen_frame_for_window", return_value=screen), \ patch("cursor_flasher.daemon.screen_frame_for_window", return_value=screen), \
patch("cursor_flasher.daemon.play_alert"): patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode() json.dumps({"workspace": "/path", "event": "stop"}).encode()
) )
daemon.overlay.flash.assert_called_once_with([screen], daemon.config.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): def test_stop_flashes_window_frame_when_window_mode(self):
daemon = self._make_daemon(flash_mode="window") daemon = self._make_daemon(flash_mode="window")
window = {"title": "my-project", "frame": ((0, 0), (800, 600))} window = {"title": "my-project", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.play_alert"): patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="light"):
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode() json.dumps({"workspace": "/path", "event": "stop"}).encode()
) )
daemon.overlay.flash.assert_called_once_with( daemon.overlay.add_flash.assert_called_once_with(
[((0, 0), (800, 600))], daemon.config.completed "/path", [((0, 0), (800, 600))], daemon.config.light.completed
) )
def test_allscreens_mode_uses_all_screens(self): def test_allscreens_mode_uses_all_screens(self):
@@ -124,21 +144,28 @@ class TestFlasherDaemon:
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.all_screen_frames", return_value=screens), \ patch("cursor_flasher.daemon.all_screen_frames", return_value=screens), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \ patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"): patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending() daemon._check_pending()
daemon.overlay.pulse.assert_called_once_with(screens, daemon.config.running) daemon.overlay.add_pulse.assert_called_once_with(
"/path", 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.dismiss.assert_called_once() # --- stop interactions with active pulse ---
daemon.overlay.flash.assert_not_called()
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): def test_stop_clears_pending(self):
daemon = self._make_daemon(approval_delay=10.0) daemon = self._make_daemon(approval_delay=10.0)
@@ -146,32 +173,37 @@ class TestFlasherDaemon:
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode() json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
) )
assert daemon._pending is not None assert "/path" in daemon._pending_approvals
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode() 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): def test_postToolUse_dismisses_active_pulse(self):
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = True daemon._active_pulses["/path"] = MagicMock()
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUse", "tool": "Shell"}).encode() 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): def test_postToolUseFailure_dismisses_pulse(self):
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = True daemon._active_pulses["/path"] = MagicMock()
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "postToolUseFailure", "tool": "Shell"}).encode() 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): def test_cooldown_prevents_rapid_triggers(self):
daemon = self._make_daemon(cooldown=5.0) daemon = self._make_daemon(cooldown=5.0)
@@ -179,19 +211,30 @@ class TestFlasherDaemon:
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode() json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
) )
daemon._last_flash = time.monotonic() daemon._last_flash["/path"] = time.monotonic()
daemon._pending = None daemon._pending_approvals.clear()
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode() 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): def test_invalid_json_ignored(self):
daemon = self._make_daemon() daemon = self._make_daemon()
daemon._handle_message(b"not json") daemon._handle_message(b"not json")
daemon.overlay.pulse.assert_not_called() daemon.overlay.add_pulse.assert_not_called()
daemon.overlay.flash.assert_not_called() daemon.overlay.add_flash.assert_not_called()
def test_no_window_found(self): def test_no_window_found(self):
daemon = self._make_daemon(approval_delay=0.0) daemon = self._make_daemon(approval_delay=0.0)
@@ -203,113 +246,133 @@ class TestFlasherDaemon:
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=None): with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=None):
daemon._check_pending() daemon._check_pending()
daemon.overlay.pulse.assert_not_called() daemon.overlay.add_pulse.assert_not_called()
def test_focus_transition_dismisses_pulse(self): # --- focus dismiss ---
"""Pulse dismisses when user switches TO Cursor from another app."""
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 = 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 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._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): def test_focus_no_dismiss_when_cursor_already_frontmost(self):
"""No dismiss if Cursor was already frontmost (no transition).""" """No dismiss if Cursor was already frontmost (no transition)."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = True daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic())
daemon._cursor_was_frontmost = True 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._check_focus()
daemon.overlay.dismiss.assert_not_called() daemon.overlay.dismiss_tag.assert_not_called()
def test_focus_tracks_state_changes(self): def test_focus_tracks_state_changes(self):
"""_cursor_was_frontmost updates each tick.""" """_cursor_was_frontmost updates each tick."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = True daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic())
daemon._cursor_was_frontmost = True daemon._cursor_was_frontmost = True
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False): with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False):
daemon._check_focus() daemon._check_focus()
assert daemon._cursor_was_frontmost is False 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._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): def test_focus_no_dismiss_when_not_pulsing(self):
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = False
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True): with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True):
daemon._check_focus() 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): # --- input dismiss ---
"""Recent input + Cursor frontmost + past grace period = 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 = self._make_daemon()
daemon.overlay.is_pulsing = True daemon._active_pulses["/pathA"] = _ActivePulse("/pathA", "project-a", time.monotonic() - 2.0)
daemon._pulse_started_at = 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), \ 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._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): def test_input_dismiss_skipped_during_grace_period(self):
"""No dismiss if pulse just started (within grace period).""" """No dismiss if pulse just started (within grace period)."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = True daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 0.1)
daemon._pulse_started_at = time.monotonic() - 0.1
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \ with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.05): patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.05):
daemon._check_input_dismiss() 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): def test_input_dismiss_skipped_when_not_pulsing(self):
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = False
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \ with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.1): patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=0.1):
daemon._check_input_dismiss() 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): def test_input_dismiss_skipped_when_cursor_not_frontmost(self):
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = True daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 2.0)
daemon._pulse_started_at = time.monotonic() - 2.0
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False): with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False):
daemon._check_input_dismiss() 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): def test_input_dismiss_ignores_old_input(self):
"""Input from before the pulse started should not trigger dismiss.""" """Input from before the pulse started should not trigger dismiss."""
from cursor_flasher.daemon import _ActivePulse
daemon = self._make_daemon() daemon = self._make_daemon()
daemon.overlay.is_pulsing = True daemon._active_pulses["/path"] = _ActivePulse("/path", "proj", time.monotonic() - 2.0)
daemon._pulse_started_at = time.monotonic() - 2.0
with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \ with patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=True), \
patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=5.0): patch("cursor_flasher.daemon.CGEventSourceSecondsSinceLastEventType", return_value=5.0):
daemon._check_input_dismiss() 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): def test_running_style_sound_plays_on_approval(self):
"""Running style with sound configured plays on approval pulse.""" """Running style with sound configured plays on approval pulse."""
running = StyleConfig(sound="Glass", volume=0.5) running = StyleConfig(sound="Glass", volume=0.5)
daemon = self._make_daemon(approval_delay=0.0, flash_mode="window", running=running) dark = ThemeStyles(running=running)
daemon = self._make_daemon(approval_delay=0.0, flash_mode="window", dark=dark)
window = {"title": "proj", "frame": ((0, 0), (800, 600))} window = {"title": "proj", "frame": ((0, 0), (800, 600))}
daemon._handle_message( daemon._handle_message(
@@ -318,7 +381,8 @@ class TestFlasherDaemon:
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \ patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert") as mock_alert: patch("cursor_flasher.daemon.play_alert") as mock_alert, \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending() daemon._check_pending()
mock_alert.assert_called_once_with(running) mock_alert.assert_called_once_with(running)
@@ -326,11 +390,13 @@ class TestFlasherDaemon:
def test_completed_style_sound_plays_on_stop(self): def test_completed_style_sound_plays_on_stop(self):
"""Completed style with sound configured plays on stop flash.""" """Completed style with sound configured plays on stop flash."""
completed = StyleConfig(sound="Ping", volume=0.7) completed = StyleConfig(sound="Ping", volume=0.7)
daemon = self._make_daemon(flash_mode="window", completed=completed) dark = ThemeStyles(completed=completed)
daemon = self._make_daemon(flash_mode="window", dark=dark)
window = {"title": "proj", "frame": ((0, 0), (800, 600))} window = {"title": "proj", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.play_alert") as mock_alert: patch("cursor_flasher.daemon.play_alert") as mock_alert, \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode() json.dumps({"workspace": "/path", "event": "stop"}).encode()
) )
@@ -340,11 +406,13 @@ class TestFlasherDaemon:
def test_no_sound_when_style_sound_empty(self): def test_no_sound_when_style_sound_empty(self):
"""No sound plays when the style has sound="" (the completed default).""" """No sound plays when the style has sound="" (the completed default)."""
completed = StyleConfig(sound="", volume=0.0) completed = StyleConfig(sound="", volume=0.0)
daemon = self._make_daemon(flash_mode="window", completed=completed) dark = ThemeStyles(completed=completed)
daemon = self._make_daemon(flash_mode="window", dark=dark)
window = {"title": "proj", "frame": ((0, 0), (800, 600))} window = {"title": "proj", "frame": ((0, 0), (800, 600))}
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.play_alert") as mock_alert: patch("cursor_flasher.daemon.play_alert") as mock_alert, \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode() json.dumps({"workspace": "/path", "event": "stop"}).encode()
) )
@@ -355,9 +423,9 @@ class TestFlasherDaemon:
"""Different colors for running vs completed are passed through.""" """Different colors for running vs completed are passed through."""
running = StyleConfig(color="#FF0000") running = StyleConfig(color="#FF0000")
completed = StyleConfig(color="#00FF00") completed = StyleConfig(color="#00FF00")
dark = ThemeStyles(running=running, completed=completed)
daemon = self._make_daemon( daemon = self._make_daemon(
approval_delay=0.0, flash_mode="window", approval_delay=0.0, flash_mode="window", dark=dark,
running=running, completed=completed,
) )
window = {"title": "proj", "frame": ((0, 0), (800, 600))} window = {"title": "proj", "frame": ((0, 0), (800, 600))}
@@ -367,22 +435,130 @@ class TestFlasherDaemon:
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \ patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"): patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._check_pending() daemon._check_pending()
daemon.overlay.pulse.assert_called_once_with( daemon.overlay.add_pulse.assert_called_once_with(
[((0, 0), (800, 600))], running "/path", [((0, 0), (800, 600))], running
) )
daemon.overlay.pulse.reset_mock() daemon.overlay.add_pulse.reset_mock()
daemon._last_flash = 0 daemon._last_flash.clear()
daemon._active_pulses.clear()
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \ with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.play_alert"): patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="dark"):
daemon._handle_message( daemon._handle_message(
json.dumps({"workspace": "/path", "event": "stop"}).encode() json.dumps({"workspace": "/path", "event": "stop"}).encode()
) )
daemon.overlay.flash.assert_called_once_with( daemon.overlay.add_flash.assert_called_once_with(
[((0, 0), (800, 600))], completed "/path", [((0, 0), (800, 600))], completed
) )
def test_theme_auto_uses_light_when_system_light(self):
"""Auto theme resolves to light styles when system is in light mode."""
dark_running = StyleConfig(color="#111111")
light_running = StyleConfig(color="#EEEEEE")
daemon = self._make_daemon(
approval_delay=0.0, flash_mode="window",
dark=ThemeStyles(running=dark_running),
light=ThemeStyles(running=light_running),
theme="auto",
)
window = {"title": "proj", "frame": ((0, 0), (800, 600))}
daemon._handle_message(
json.dumps({"workspace": "/path", "event": "preToolUse", "tool": "Shell"}).encode()
)
with patch("cursor_flasher.daemon.find_window_by_workspace", return_value=window), \
patch("cursor_flasher.daemon.is_cursor_frontmost", return_value=False), \
patch("cursor_flasher.daemon.play_alert"), \
patch(PATCH_APPEARANCE, return_value="light"):
daemon._check_pending()
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