Allow pausing agent runs via ESC

Author: latent-variable

mini_agent/agent.py

@@ -73,10 +73,34 @@ def __init__(
         # Initialize logger
         self.logger = AgentLogger()
 
+        # Stop control flag, allows external interrupt requests
+        self._stop_requested = False
+        self._stop_notified = False
+        self.current_step = 0
+        self._run_active = False
+
     def add_user_message(self, content: str):
         """Add a user message to history."""
         self.messages.append(Message(role="user", content=content))
 
+    def request_stop(self):
+        """Signal the agent loop to stop at the next safe checkpoint."""
+        self._stop_requested = True
+        self._stop_notified = False
+
+    def _reset_stop_request(self):
+        self._stop_requested = False
+        self._stop_notified = False
+
+    def _check_stop_requested(self) -> bool:
+        """Return True if a stop was requested and emit a single notification."""
+        if not self._stop_requested:
+            return False
+        if not self._stop_notified:
+            print(f"\n{Colors.BRIGHT_YELLOW}⏸️  Agent paused by user (press Enter to continue interacting).{Colors.RESET}\n")
+            self._stop_notified = True
+        return True
+
     def _estimate_tokens(self) -> int:
         """Accurately calculate token count for message history using tiktoken
 
@@ -258,13 +282,20 @@ async def _create_summary(self, messages: list[Message], round_num: int) -> str:
 
     async def run(self) -> str:
         """Execute agent loop until task is complete or max steps reached."""
-        # Start new run, initialize log file
-        self.logger.start_new_run()
-        print(f"{Colors.DIM}📝 Log file: {self.logger.get_log_file_path()}{Colors.RESET}")
+        if not self._run_active:
+            # Start new run, initialize log file
+            self.logger.start_new_run()
+            print(f"{Colors.DIM}📝 Log file: {self.logger.get_log_file_path()}{Colors.RESET}")
+            self.current_step = 0
+            self._run_active = True
 
-        step = 0
+        step = self.current_step
+        self._reset_stop_request()
 
         while step < self.max_steps:
+            if self._check_stop_requested():
+                return "Agent run interrupted by user."
+
             # Check and summarize message history to prevent context overflow
             await self._summarize_messages()
 
@@ -296,8 +327,13 @@ async def run(self) -> str:
                 else:
                     error_msg = f"LLM call failed: {str(e)}"
                     print(f"\n{Colors.BRIGHT_RED}❌ Error:{Colors.RESET} {error_msg}")
+                self._run_active = False
+                self.current_step = 0
                 return error_msg
 
+            if self._check_stop_requested():
+                return "Agent run interrupted by user."
+
             # Log LLM response
             self.logger.log_response(
                 content=response.content,
@@ -327,10 +363,15 @@ async def run(self) -> str:
 
             # Check if task is complete (no tool calls)
             if not response.tool_calls:
+                self._run_active = False
+                self.current_step = 0
                 return response.content
 
             # Execute tool calls
             for tool_call in response.tool_calls:
+                if self._check_stop_requested():
+                    return "Agent run interrupted by user."
+
                 tool_call_id = tool_call.id
                 function_name = tool_call.function.name
                 arguments = tool_call.function.arguments
@@ -403,10 +444,13 @@ async def run(self) -> str:
                 self.messages.append(tool_msg)
 
             step += 1
+            self.current_step = step
 
         # Max steps reached
         error_msg = f"Task couldn't be completed after {self.max_steps} steps."
         print(f"\n{Colors.BRIGHT_YELLOW}⚠️  {error_msg}{Colors.RESET}")
+        self._run_active = False
+        self.current_step = 0
         return error_msg
 
     def get_history(self) -> list[Message]:

mini_agent/cli.py

@@ -11,10 +11,18 @@
 
 import argparse
 import asyncio
+import sys
 from datetime import datetime
 from pathlib import Path
 from typing import List
 
+try:
+    import termios
+    import tty
+except ImportError:  # pragma: no cover - Windows fallback
+    termios = None  # type: ignore[assignment]
+    tty = None  # type: ignore[assignment]
+
 from prompt_toolkit import PromptSession
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.completion import WordCompleter
@@ -70,6 +78,75 @@ class Colors:
     BG_BLUE = "\033[44m"
 
 
+class EscapeKeyListener:
+    """Listen for ESC key presses during agent execution to request a stop."""
+
+    def __init__(self, agent: Agent):
+        self.agent = agent
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._fd: int | None = None
+        self._old_settings = None
+        self._cbreak_enabled = False
+        self._reader_registered = False
+
+    async def __aenter__(self):
+        self._loop = asyncio.get_running_loop()
+        await self._loop.run_in_executor(None, self._enable_cbreak_mode)
+        if (
+            self._cbreak_enabled
+            and self._loop
+            and self._fd is not None
+            and hasattr(self._loop, "add_reader")
+        ):
+            self._loop.add_reader(self._fd, self._handle_keypress)
+            self._reader_registered = True
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if self._reader_registered and self._loop and self._fd is not None:
+            self._loop.remove_reader(self._fd)
+            self._reader_registered = False
+        if self._loop:
+            await self._loop.run_in_executor(None, self._restore_terminal)
+
+    def _enable_cbreak_mode(self):
+        if termios is None or tty is None:
+            return
+        if not sys.stdin.isatty():
+            return
+        try:
+            self._fd = sys.stdin.fileno()
+            self._old_settings = termios.tcgetattr(self._fd)
+            tty.setcbreak(self._fd)
+            self._cbreak_enabled = True
+        except Exception:
+            self._cbreak_enabled = False
+
+    def _restore_terminal(self):
+        if (
+            self._cbreak_enabled
+            and self._fd is not None
+            and self._old_settings is not None
+            and termios is not None
+        ):
+            termios.tcsetattr(self._fd, termios.TCSADRAIN, self._old_settings)
+        self._cbreak_enabled = False
+
+    def _handle_keypress(self):
+        if not self._cbreak_enabled:
+            return
+        try:
+            ch = sys.stdin.read(1)
+        except Exception:
+            return
+        if ch == "\x1b":  # ESC key
+            print(f"\n{Colors.BRIGHT_YELLOW}⏹️  Escape detected, requesting agent pause...{Colors.RESET}")
+            self.agent.request_stop()
+            if self._loop and self._reader_registered and self._fd is not None:
+                self._loop.remove_reader(self._fd)
+                self._reader_registered = False
+
+
 def print_banner():
     """Print welcome banner with proper alignment"""
     BOX_WIDTH = 58
@@ -105,13 +182,15 @@ def print_help():
   {Colors.BRIGHT_CYAN}Tab{Colors.RESET}        - Auto-complete commands
   {Colors.BRIGHT_CYAN}↑/↓{Colors.RESET}        - Browse command history
   {Colors.BRIGHT_CYAN}→{Colors.RESET}          - Accept auto-suggestion
+  {Colors.BRIGHT_CYAN}Esc{Colors.RESET}        - Pause the current agent run
 
 {Colors.BOLD}{Colors.BRIGHT_YELLOW}Usage:{Colors.RESET}
   - Enter your task directly, Agent will help you complete it
   - Agent remembers all conversation content in this session
   - Use {Colors.BRIGHT_GREEN}/clear{Colors.RESET} to start a new session
   - Press {Colors.BRIGHT_CYAN}Enter{Colors.RESET} to submit your message
   - Use {Colors.BRIGHT_CYAN}Ctrl+J{Colors.RESET} to insert line breaks within your message
+  - Press {Colors.BRIGHT_CYAN}Esc{Colors.RESET} anytime during execution to stop the agent
 """
     print(help_text)
 
@@ -551,7 +630,8 @@ def _(event):
             # Run Agent
             print(f"\n{Colors.BRIGHT_BLUE}Agent{Colors.RESET} {Colors.DIM}›{Colors.RESET} {Colors.DIM}Thinking...{Colors.RESET}\n")
             agent.add_user_message(user_input)
-            _ = await agent.run()
+            async with EscapeKeyListener(agent):
+                _ = await agent.run()
 
             # Visual separation - keep it simple like the reference code
             print(f"\n{Colors.DIM}{'─' * 60}{Colors.RESET}\n")