diff --git a/src/sot/_app.py b/src/sot/_app.py index e0e209e..03dbc20 100644 --- a/src/sot/_app.py +++ b/src/sot/_app.py @@ -24,6 +24,7 @@ ProcessesWidget, SotWidget, ) +from .widgets.confirmation_modal import ConfirmationModal # Main SOT Application @@ -51,6 +52,8 @@ def __init__(self, net_interface=None, log_file=None): super().__init__() self.net_interface = net_interface self.log_file = log_file + self.pending_kill = None + self._waiting_for_kill_confirmation = False # Set up logging if specified if log_file: @@ -115,6 +118,20 @@ def on_mount(self) -> None: async def on_load(self, _): self.bind("q", "quit") + def on_key(self, event) -> None: + """Handle key events for kill confirmation.""" + from textual import events + + if self._waiting_for_kill_confirmation: + if event.key == "y": + self._waiting_for_kill_confirmation = False + self._kill_process(self.pending_kill["process_info"]) + else: + # Any other key cancels + self._waiting_for_kill_confirmation = False + self.notify("❌ Kill cancelled", timeout=2) + event.prevent_default() + def on_processes_widget_process_selected( self, message: ProcessesWidget.ProcessSelected ) -> None: @@ -169,6 +186,77 @@ def on_processes_widget_process_selected( timeout=5, ) + def on_processes_widget_kill_request( + self, message: ProcessesWidget.KillRequest + ) -> None: + """Handle kill request by showing a pending confirmation.""" + process_info = message.process_info + process_name = process_info.get("name", "Unknown") + process_id = process_info.get("pid", "N/A") + + # Store pending kill for confirmation + self.pending_kill = { + "process_info": process_info, + "process_name": process_name, + "process_id": process_id, + "timestamp": 0, + } + + # Show confirmation prompt with danger emoji and error severity (red) + self.notify( + f"⚠️ KILL {process_name}? Press 'y' to confirm, any key to cancel", + severity="error", + timeout=10, + ) + + # Listen for confirmation key in the next on_key event + self._waiting_for_kill_confirmation = True + + # Auto-cancel after 10 seconds (matching notification timeout) + def reset_confirmation(): + if self._waiting_for_kill_confirmation: + self._waiting_for_kill_confirmation = False + self.notify("❌ Kill action expired", severity="error", timeout=2) + + self.set_timer(reset_confirmation, 10.0) + + def _kill_process(self, process_info: dict) -> None: + """Execute the kill process action.""" + import psutil + + process_id = process_info.get("pid") + process_name = process_info.get("name", "Unknown") + + if not process_id: + self.notify("❌ Invalid process ID", severity="error", timeout=3) + return + + # Log the action attempt if logging is enabled + if self.log_file: + self.log(f"Attempting to kill process: {process_name} (PID: {process_id})") + + try: + target_process = psutil.Process(process_id) + target_process.kill() + if self.log_file: + self.log( + f"Successfully killed process: {process_name} (PID: {process_id})" + ) + self.notify( + f"💥 Killed {process_name} (PID: {process_id})", + severity="warning", + timeout=4, + ) + + except psutil.ZombieProcess: + self._handle_zombie_process_error(process_name, process_id) + except psutil.NoSuchProcess: + self._handle_no_such_process_error(process_id) + except psutil.AccessDenied: + self._handle_access_denied_error(process_name, process_id) + except Exception as error: + self._handle_general_process_error("kill", process_name, error) + def on_processes_widget_process_action( self, message: ProcessesWidget.ProcessAction ) -> None: diff --git a/src/sot/widgets/confirmation_modal.py b/src/sot/widgets/confirmation_modal.py new file mode 100644 index 0000000..837b644 --- /dev/null +++ b/src/sot/widgets/confirmation_modal.py @@ -0,0 +1,82 @@ +""" +Confirmation Modal Widget + +Simple confirmation dialog using Textual best practices. +""" + +from textual.app import ComposeResult +from textual.containers import Container, Horizontal +from textual.message import Message +from textual.screen import ModalScreen +from textual.widgets import Button, Label + + +class ConfirmationModal(ModalScreen[bool]): + """A modal for confirming actions. Returns True if confirmed, False if cancelled.""" + + class Confirmed(Message): + """Posted when user confirms.""" + + def __init__(self, action_data: dict) -> None: + self.action_data = action_data + super().__init__() + + BINDINGS = [("escape", "cancel")] + + CSS = """ + ConfirmationModal { + align: center middle; + } + + ConfirmationModal > Container { + width: 60; + height: 9; + border: solid $accent; + background: $panel; + } + + ConfirmationModal #title { + margin: 1 2; + width: 1fr; + } + + ConfirmationModal #message { + margin: 0 2; + width: 1fr; + } + + ConfirmationModal #buttons { + margin: 1 2 0 2; + width: 1fr; + } + + ConfirmationModal Button { + margin-right: 1; + } + """ + + def __init__(self, title: str, message: str, action_data: dict = None) -> None: + super().__init__() + self.title_text = title + self.message_text = message + self.action_data = action_data or {} + + def compose(self) -> ComposeResult: + with Container(): + yield Label(self.title_text, id="title") + yield Label(self.message_text, id="message") + with Horizontal(id="buttons"): + yield Button("Kill", variant="error", id="confirm") + yield Button("Cancel", id="cancel") + + def on_button_pressed(self, event: Button.Pressed) -> None: + """Handle button presses.""" + if event.button.id == "confirm": + self.app.post_message(self.Confirmed(self.action_data)) + self.dismiss(True) + else: + self.dismiss(False) + + def action_cancel(self) -> None: + """Handle escape key.""" + self.dismiss(False) diff --git a/src/sot/widgets/processes.py b/src/sot/widgets/processes.py index e65a97f..0e3786d 100644 --- a/src/sot/widgets/processes.py +++ b/src/sot/widgets/processes.py @@ -94,6 +94,13 @@ def __init__(self, action: str, process_info: dict) -> None: self.process_info = process_info super().__init__() + class KillRequest(Message): + """Message sent when kill is requested on a process (requires confirmation).""" + + def __init__(self, process_info: dict) -> None: + self.process_info = process_info + super().__init__() + def __init__(self, **kwargs): super().__init__(title="Processes", **kwargs) self.max_num_procs = 1000 @@ -242,7 +249,7 @@ def handle_action_keys(self, key_pressed: str) -> bool: elif key_pressed == "k": if 0 <= self.selected_process_index < len(self.process_list_data): selected_process = self.process_list_data[self.selected_process_index] - self.post_message(self.ProcessAction("kill", selected_process)) + self.post_message(self.KillRequest(selected_process)) return True elif key_pressed == "t": if 0 <= self.selected_process_index < len(self.process_list_data): @@ -265,6 +272,11 @@ def handle_action_keys(self, key_pressed: str) -> bool: def on_key(self, event: events.Key) -> None: """Handle keyboard navigation and actions with scrolling support.""" + # Check if the app is waiting for kill confirmation + # If so, let it bubble up to the app's on_key handler + if hasattr(self.app, '_waiting_for_kill_confirmation') and self.app._waiting_for_kill_confirmation: + return + if not self.is_interactive_mode or not self.process_list_data: return