"""Notification manager — decides when to show action cards to the user. Only pushes a new notification when the proposed action meaningfully changes. In production (Swift), this becomes a native macOS notification / floating card. Here it's a terminal prompt for testing. Swift portability notes: - NotificationManager becomes a class with @Published properties - show_card() triggers a SwiftUI overlay or UNUserNotificationCenter - user_response() is wired to button taps instead of stdin """ from __future__ import annotations import asyncio import logging from dataclasses import dataclass, field log = logging.getLogger(__name__) @dataclass class ActionCard: """A proposed action shown to the user.""" friction_type: str description: str actions: list[dict] source: str target: str vlm_payload: dict # full VLM result for the executor class NotificationManager: """Deduplicates notifications and manages the pending action card. Only shows a new card when the friction type or proposed action labels change from the previous notification. """ def __init__(self): self._last_fingerprint: str = "" self._pending_card: ActionCard | None = None self._response_event: asyncio.Event = asyncio.Event() self._user_accepted: bool = False def _fingerprint(self, vlm_payload: dict) -> str: """Generate a dedup key from friction type + action labels.""" friction = vlm_payload.get("friction", {}) ftype = friction.get("type", "none") labels = tuple( a.get("label", "") for a in friction.get("proposed_actions", []) ) return f"{ftype}:{labels}" # Action types that the executor can actually perform ACTIONABLE_TYPES = {"auto_extract", "brain_dump", "auto_fill", "summarize"} def should_notify(self, vlm_payload: dict) -> bool: """Check if this VLM result warrants a friction card (executor action). Only fires when proposed_actions contain something the executor can act on. Generic suggestions (action_type: "other") are nudges, not executor actions. """ friction = vlm_payload.get("friction", {}) # No friction or no proposed actions → no notification if friction.get("type") == "none": return False actions = friction.get("proposed_actions", []) if not actions: return False # Only notify if at least one action is executor-actionable has_actionable = any( a.get("action_type") in self.ACTIONABLE_TYPES for a in actions ) if not has_actionable: return False # Same as last notification → skip fp = self._fingerprint(vlm_payload) if fp == self._last_fingerprint: return False return True def create_card(self, vlm_payload: dict) -> ActionCard: """Create an action card from VLM output and mark it as pending.""" friction = vlm_payload.get("friction", {}) card = ActionCard( friction_type=friction.get("type", ""), description=friction.get("description", ""), actions=friction.get("proposed_actions", []), source=friction.get("source_context", ""), target=friction.get("target_context", ""), vlm_payload=vlm_payload, ) self._pending_card = card self._last_fingerprint = self._fingerprint(vlm_payload) self._response_event.clear() self._user_accepted = False return card def show_card_terminal(self, card: ActionCard) -> None: """Print the action card to terminal. (Swift: show native UI.)""" print() print("┌" + "─" * 58 + "┐") print(f"│ {'🔧 Friction detected':^58} │") print("│" + " " * 58 + "│") desc_lines = _wrap(card.description, 56) for line in desc_lines: print(f"│ {line:<56} │") print("│" + " " * 58 + "│") for i, action in enumerate(card.actions): label = action.get("label", "?") print(f"│ [{i + 1}] {label:<53} │") print(f"│ [0] Not now{' ' * 44} │") print("└" + "─" * 58 + "┘") print() async def wait_for_response(self, timeout: float = 30.0) -> tuple[bool, int]: """Wait for user response. Returns (accepted, action_index). Swift portability: this becomes a button tap callback. """ try: await asyncio.wait_for(self._response_event.wait(), timeout=timeout) return self._user_accepted, self._selected_index except asyncio.TimeoutError: log.debug("Notification timed out, dismissing") self._pending_card = None return False, -1 def submit_response(self, choice: int) -> None: """Submit user's choice. Called from input reader task. Swift portability: called from button tap handler. """ if choice > 0 and self._pending_card: self._user_accepted = True self._selected_index = choice - 1 else: self._user_accepted = False self._selected_index = -1 self._response_event.set() @property def pending(self) -> ActionCard | None: return self._pending_card def dismiss(self) -> None: """Dismiss current card without action.""" self._pending_card = None self._last_fingerprint = "" def _wrap(text: str, width: int) -> list[str]: """Simple word wrap.""" words = text.split() lines: list[str] = [] current = "" for word in words: if len(current) + len(word) + 1 <= width: current = f"{current} {word}" if current else word else: if current: lines.append(current) current = word if current: lines.append(current) return lines or [""]