feat: replace synchronous HTTP calls with async client and add background task tracking

pr/core/api.py

@@ -1,15 +1,14 @@
 import json
 import logging
-import urllib.error
-import urllib.request
 
 from pr.config import DEFAULT_MAX_TOKENS, DEFAULT_TEMPERATURE
 from pr.core.context import auto_slim_messages
+from pr.core.http_client import http_client
 
 logger = logging.getLogger("pr")
 
 
-def call_api(messages, model, api_url, api_key, use_tools, tools_definition, verbose=False):
+async def call_api(messages, model, api_url, api_key, use_tools, tools_definition, verbose=False):
     try:
         messages = auto_slim_messages(messages, verbose=verbose)
 
@@ -42,63 +41,70 @@ def call_api(messages, model, api_url, api_key, use_tools, tools_definition, ver
             data["tool_choice"] = "auto"
             logger.debug(f"Tool calling enabled with {len(tools_definition)} tools")
 
-        request_json = json.dumps(data)
+        request_json = data
         logger.debug(f"Request payload size: {len(request_json)} bytes")
 
-        req = urllib.request.Request(
-            api_url, data=request_json.encode("utf-8"), headers=headers, method="POST"
-        )
-
         logger.debug("Sending HTTP request...")
-        with urllib.request.urlopen(req) as response:
-            response_data = response.read().decode("utf-8")
-            logger.debug(f"Response received: {len(response_data)} bytes")
-            result = json.loads(response_data)
+        response = await http_client.post(api_url, headers=headers, json_data=request_json)
 
-            if "usage" in result:
-                logger.debug(f"Token usage: {result['usage']}")
-            if "choices" in result and result["choices"]:
-                choice = result["choices"][0]
-                if "message" in choice:
-                    msg = choice["message"]
-                    logger.debug(f"Response role: {msg.get('role', 'N/A')}")
-                    if "content" in msg and msg["content"]:
-                        logger.debug(f"Response content length: {len(msg['content'])} chars")
-                    if "tool_calls" in msg:
-                        logger.debug(f"Response contains {len(msg['tool_calls'])} tool call(s)")
+        if response.get("error"):
+            if "status" in response:
+                logger.error(f"API HTTP Error: {response['status']} - {response.get('text', '')}")
+                logger.debug("=== API CALL FAILED ===")
+                return {
+                    "error": f"API Error: {response['status']}",
+                    "message": response.get("text", ""),
+                }
+            else:
+                logger.error(f"API call failed: {response.get('exception', 'Unknown error')}")
+                logger.debug("=== API CALL FAILED ===")
+                return {"error": response.get("exception", "Unknown error")}
 
-            if verbose and "usage" in result:
-                from pr.core.usage_tracker import UsageTracker
+        response_data = response["text"]
+        logger.debug(f"Response received: {len(response_data)} bytes")
+        result = json.loads(response_data)
 
-                usage = result["usage"]
-                input_t = usage.get("prompt_tokens", 0)
-                output_t = usage.get("completion_tokens", 0)
-                cost = UsageTracker._calculate_cost(model, input_t, output_t)
-                print(f"API call cost: €{cost:.4f}")
+        if "usage" in result:
+            logger.debug(f"Token usage: {result['usage']}")
+        if "choices" in result and result["choices"]:
+            choice = result["choices"][0]
+            if "message" in choice:
+                msg = choice["message"]
+                logger.debug(f"Response role: {msg.get('role', 'N/A')}")
+                if "content" in msg and msg["content"]:
+                    logger.debug(f"Response content length: {len(msg['content'])} chars")
+                if "tool_calls" in msg:
+                    logger.debug(f"Response contains {len(msg['tool_calls'])} tool call(s)")
+
+        if verbose and "usage" in result:
+            from pr.core.usage_tracker import UsageTracker
+
+            usage = result["usage"]
+            input_t = usage.get("prompt_tokens", 0)
+            output_t = usage.get("completion_tokens", 0)
+            UsageTracker._calculate_cost(model, input_t, output_t)
 
         logger.debug("=== API CALL END ===")
         return result
 
-    except urllib.error.HTTPError as e:
-        error_body = e.read().decode("utf-8")
-        logger.error(f"API HTTP Error: {e.code} - {error_body}")
-        logger.debug("=== API CALL FAILED ===")
-        return {"error": f"API Error: {e.code}", "message": error_body}
     except Exception as e:
         logger.error(f"API call failed: {e}")
         logger.debug("=== API CALL FAILED ===")
         return {"error": str(e)}
 
 
-def list_models(model_list_url, api_key):
+async def list_models(model_list_url, api_key):
     try:
-        req = urllib.request.Request(model_list_url)
+        headers = {}
         if api_key:
-            req.add_header("Authorization", f"Bearer {api_key}")
+            headers["Authorization"] = f"Bearer {api_key}"
 
-        with urllib.request.urlopen(req) as response:
-            data = json.loads(response.read().decode("utf-8"))
+        response = await http_client.get(model_list_url, headers=headers)
 
+        if response.get("error"):
+            return {"error": response.get("text", "HTTP error")}
+
+        data = json.loads(response["text"])
         return data.get("data", [])
     except Exception as e:
         return {"error": str(e)}

pr/core/assistant.py

@@ -20,7 +20,7 @@ from pr.config import (
 )
 from pr.core.api import call_api
 from pr.core.autonomous_interactions import (
-    get_global_autonomous,
+    start_global_autonomous,
     stop_global_autonomous,
 )
 from pr.core.background_monitor import (
@@ -29,6 +29,7 @@ from pr.core.background_monitor import (
     stop_global_monitor,
 )
 from pr.core.context import init_system_message, truncate_tool_result
+from pr.core.usage_tracker import UsageTracker
 from pr.tools import get_tools_definition
 from pr.tools.agents import (
     collaborate_agents,
@@ -71,7 +72,7 @@ from pr.tools.memory import (
 from pr.tools.patch import apply_patch, create_diff, display_file_diff
 from pr.tools.python_exec import python_exec
 from pr.tools.web import http_fetch, web_search, web_search_news
-from pr.ui import Colors, render_markdown
+from pr.ui import Colors, Spinner, render_markdown
 
 logger = logging.getLogger("pr")
 logger.setLevel(logging.DEBUG)
@@ -109,6 +110,8 @@ class Assistant:
         self.autonomous_mode = False
         self.autonomous_iterations = 0
         self.background_monitoring = False
+        self.usage_tracker = UsageTracker()
+        self.background_tasks = set()
         self.init_database()
         self.messages.append(init_system_message(args))
 
@@ -125,8 +128,7 @@ class Assistant:
         # Initialize background monitoring components
         try:
             start_global_monitor()
-            autonomous = get_global_autonomous()
-            autonomous.start(llm_callback=self._handle_background_updates)
+            start_global_autonomous(llm_callback=self._handle_background_updates)
             self.background_monitoring = True
             if self.debug:
                 logger.debug("Background monitoring initialized")
@@ -236,7 +238,7 @@ class Assistant:
             if self.debug:
                 print(f"{Colors.RED}Error checking background updates: {e}{Colors.RESET}")
 
-    def execute_tool_calls(self, tool_calls):
+    async def execute_tool_calls(self, tool_calls):
         results = []
 
         logger.debug(f"Executing {len(tool_calls)} tool call(s)")
@@ -337,7 +339,7 @@ class Assistant:
 
         return results
 
-    def process_response(self, response):
+    async def process_response(self, response):
         if "error" in response:
             return f"Error: {response['error']}"
 
@@ -348,15 +350,17 @@ class Assistant:
         self.messages.append(message)
 
         if "tool_calls" in message and message["tool_calls"]:
-            if self.verbose:
-                print(f"{Colors.YELLOW}Executing tool calls...{Colors.RESET}")
+            tool_count = len(message["tool_calls"])
+            print(f"{Colors.BLUE}🔧 Executing {tool_count} tool call(s)...{Colors.RESET}")
 
-            tool_results = self.execute_tool_calls(message["tool_calls"])
+            tool_results = await self.execute_tool_calls(message["tool_calls"])
+
+            print(f"{Colors.GREEN}✅ Tool execution completed.{Colors.RESET}")
 
             for result in tool_results:
                 self.messages.append(result)
 
-            follow_up = call_api(
+            follow_up = await call_api(
                 self.messages,
                 self.model,
                 self.api_url,
@@ -365,7 +369,7 @@ class Assistant:
                 get_tools_definition(),
                 verbose=self.verbose,
             )
-            return self.process_response(follow_up)
+            return await self.process_response(follow_up)
 
         content = message.get("content", "")
         return render_markdown(content, self.syntax_highlighting)
@@ -439,12 +443,23 @@ class Assistant:
         readline.set_completer(completer)
         readline.parse_and_bind("tab: complete")
 
-    def run_repl(self):
+    async def run_repl(self):
         self.setup_readline()
         signal.signal(signal.SIGINT, self.signal_handler)
 
-        print(f"{Colors.BOLD}r{Colors.RESET}")
-        print(f"Type 'help' for commands or start chatting")
+        print(
+            f"{Colors.BOLD}{Colors.CYAN}╔══════════════════════════════════════════════╗{Colors.RESET}"
+        )
+        print(
+            f"{Colors.BOLD}{Colors.CYAN}║{Colors.RESET}{Colors.BOLD}              PR Assistant v{__import__('pr').__version__}              {Colors.RESET}{Colors.BOLD}{Colors.CYAN}║{Colors.RESET}"
+        )
+        print(
+            f"{Colors.BOLD}{Colors.CYAN}╚══════════════════════════════════════════════╝{Colors.RESET}"
+        )
+        print(
+            f"{Colors.GRAY}Type 'help' for commands, 'exit' to quit, or start chatting.{Colors.RESET}"
+        )
+        print(f"{Colors.GRAY}AI calls will show costs and progress indicators.{Colors.RESET}\n")
 
         while True:
             try:
@@ -480,7 +495,7 @@ class Assistant:
                 elif cmd_result is True:
                     continue
 
-                process_message(self, user_input)
+                await process_message(self, user_input)
 
             except EOFError:
                 break
@@ -490,7 +505,7 @@ class Assistant:
                 print(f"{Colors.RED}Error: {e}{Colors.RESET}")
                 logging.error(f"REPL error: {e}\n{traceback.format_exc()}")
 
-    def run_single(self):
+    async def run_single(self):
         if self.args.message:
             message = self.args.message
         else:
@@ -498,7 +513,7 @@ class Assistant:
 
         from pr.autonomous.mode import run_autonomous_mode
 
-        run_autonomous_mode(self, message)
+        await run_autonomous_mode(self, message)
 
     def cleanup(self):
         if hasattr(self, "enhanced") and self.enhanced:
@@ -525,22 +540,22 @@ class Assistant:
         if self.db_conn:
             self.db_conn.close()
 
-    def run(self):
+    async def run(self):
         try:
             print(
                 f"DEBUG: interactive={self.args.interactive}, message={self.args.message}, isatty={sys.stdin.isatty()}"
             )
             if self.args.interactive or (not self.args.message and sys.stdin.isatty()):
                 print("DEBUG: calling run_repl")
-                self.run_repl()
+                await self.run_repl()
             else:
                 print("DEBUG: calling run_single")
-                self.run_single()
+                await self.run_single()
         finally:
             self.cleanup()
 
 
-def process_message(assistant, message):
+async def process_message(assistant, message):
     from pr.core.knowledge_context import inject_knowledge_context
 
     inject_knowledge_context(assistant, message)
@@ -550,10 +565,11 @@ def process_message(assistant, message):
     logger.debug(f"Processing user message: {message[:100]}...")
     logger.debug(f"Current message count: {len(assistant.messages)}")
 
-    if assistant.verbose:
-        print(f"{Colors.GRAY}Sending request to API...{Colors.RESET}")
+    # Start spinner for AI call
+    spinner = Spinner("Querying AI...")
+    await spinner.start()
 
-    response = call_api(
+    response = await call_api(
         assistant.messages,
         assistant.model,
         assistant.api_url,
@@ -562,6 +578,19 @@ def process_message(assistant, message):
         get_tools_definition(),
         verbose=assistant.verbose,
     )
-    result = assistant.process_response(response)
+
+    await spinner.stop()
+
+    # Track usage and display cost
+    if "usage" in response:
+        usage = response["usage"]
+        input_tokens = usage.get("prompt_tokens", 0)
+        output_tokens = usage.get("completion_tokens", 0)
+        assistant.usage_tracker.track_request(assistant.model, input_tokens, output_tokens)
+        cost = UsageTracker._calculate_cost(assistant.model, input_tokens, output_tokens)
+        total_cost = assistant.usage_tracker.session_usage["estimated_cost"]
+        print(f"{Colors.YELLOW}💰 Cost: ${cost:.4f} | Total: ${total_cost:.4f}{Colors.RESET}")
+
+    result = await assistant.process_response(response)
 
     print(f"\n{Colors.GREEN}r:{Colors.RESET} {result}\n")

pr/core/enhanced_assistant.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import logging
 import uuid
@@ -128,15 +129,39 @@ class EnhancedAssistant:
     def _api_caller_for_agent(
         self, messages: List[Dict[str, Any]], temperature: float, max_tokens: int
     ) -> Dict[str, Any]:
-        return call_api(
-            messages,
-            self.base.model,
-            self.base.api_url,
-            self.base.api_key,
-            use_tools=False,
-            tools_definition=[],
-            verbose=self.base.verbose,
-        )
+        try:
+            # Try to get the current event loop
+            loop = asyncio.get_running_loop()
+            # If we're in an event loop, use run_coroutine_threadsafe
+            import concurrent.futures
+
+            with concurrent.futures.ThreadPoolExecutor() as executor:
+                future = asyncio.run_coroutine_threadsafe(
+                    call_api(
+                        messages,
+                        self.base.model,
+                        self.base.api_url,
+                        self.base.api_key,
+                        use_tools=False,
+                        tools_definition=[],
+                        verbose=self.base.verbose,
+                    ),
+                    loop,
+                )
+                return future.result()
+        except RuntimeError:
+            # No event loop running, use asyncio.run
+            return asyncio.run(
+                call_api(
+                    messages,
+                    self.base.model,
+                    self.base.api_url,
+                    self.base.api_key,
+                    use_tools=False,
+                    tools_definition=[],
+                    verbose=self.base.verbose,
+                )
+            )
 
     def enhanced_call_api(self, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
         if self.api_cache and CACHE_ENABLED:
@@ -145,15 +170,39 @@ class EnhancedAssistant:
                 logger.debug("API cache hit")
                 return cached_response
 
-        response = call_api(
-            messages,
-            self.base.model,
-            self.base.api_url,
-            self.base.api_key,
-            self.base.use_tools,
-            get_tools_definition(),
-            verbose=self.base.verbose,
-        )
+        try:
+            # Try to get the current event loop
+            loop = asyncio.get_running_loop()
+            # If we're in an event loop, use run_coroutine_threadsafe
+            import concurrent.futures
+
+            with concurrent.futures.ThreadPoolExecutor() as executor:
+                future = asyncio.run_coroutine_threadsafe(
+                    call_api(
+                        messages,
+                        self.base.model,
+                        self.base.api_url,
+                        self.base.api_key,
+                        self.base.use_tools,
+                        get_tools_definition(),
+                        verbose=self.base.verbose,
+                    ),
+                    loop,
+                )
+                response = future.result()
+        except RuntimeError:
+            # No event loop running, use asyncio.run
+            response = asyncio.run(
+                call_api(
+                    messages,
+                    self.base.model,
+                    self.base.api_url,
+                    self.base.api_key,
+                    self.base.use_tools,
+                    get_tools_definition(),
+                    verbose=self.base.verbose,
+                )
+            )
 
         if self.api_cache and CACHE_ENABLED and "error" not in response:
             token_count = response.get("usage", {}).get("total_tokens", 0)

pr/core/http_client.py

@@ -0,0 +1,136 @@
+import asyncio
+import json
+import logging
+import socket
+import time
+import urllib.error
+import urllib.request
+from typing import Dict, Any, Optional
+
+logger = logging.getLogger("pr")
+
+
+class AsyncHTTPClient:
+    def __init__(self):
+        self.session_headers = {}
+
+    async def request(
+        self,
+        method: str,
+        url: str,
+        headers: Optional[Dict[str, str]] = None,
+        data: Optional[bytes] = None,
+        json_data: Optional[Dict[str, Any]] = None,
+        timeout: float = 30.0,
+    ) -> Dict[str, Any]:
+        """Make an async HTTP request using urllib in a thread executor with retry logic."""
+        loop = asyncio.get_event_loop()
+
+        # Prepare headers
+        request_headers = {**self.session_headers}
+        if headers:
+            request_headers.update(headers)
+
+        # Prepare data
+        request_data = data
+        if json_data is not None:
+            request_data = json.dumps(json_data).encode("utf-8")
+            request_headers["Content-Type"] = "application/json"
+
+        # Create request object
+        req = urllib.request.Request(url, data=request_data, headers=request_headers, method=method)
+
+        attempt = 0
+        start_time = time.time()
+
+        while True:
+            attempt += 1
+            try:
+                # Execute in thread pool
+                response = await loop.run_in_executor(
+                    None, lambda: urllib.request.urlopen(req, timeout=timeout)
+                )
+                response_data = await loop.run_in_executor(None, response.read)
+                response_text = response_data.decode("utf-8")
+
+                return {
+                    "status": response.status,
+                    "headers": dict(response.headers),
+                    "text": response_text,
+                    "json": lambda: json.loads(response_text) if response_text else None,
+                }
+
+            except urllib.error.HTTPError as e:
+                error_body = await loop.run_in_executor(None, e.read)
+                error_text = error_body.decode("utf-8")
+                return {
+                    "status": e.code,
+                    "error": True,
+                    "text": error_text,
+                    "json": lambda: json.loads(error_text) if error_text else None,
+                }
+            except socket.timeout:
+                # Handle socket timeouts specifically
+                elapsed = time.time() - start_time
+                elapsed_minutes = int(elapsed // 60)
+                elapsed_seconds = elapsed % 60
+                duration_str = (
+                    f"{elapsed_minutes}m {elapsed_seconds:.1f}s"
+                    if elapsed_minutes > 0
+                    else f"{elapsed_seconds:.1f}s"
+                )
+
+                logger.warning(
+                    f"Request timed out (attempt {attempt}, "
+                    f"duration: {duration_str}). Retrying in {attempt} second(s)..."
+                )
+
+                # Exponential backoff starting at 1 second
+                await asyncio.sleep(attempt)
+            except Exception as e:
+                error_msg = str(e)
+                # For other exceptions, check if they might be timeout-related
+                if "timed out" in error_msg.lower() or "timeout" in error_msg.lower():
+                    elapsed = time.time() - start_time
+                    elapsed_minutes = int(elapsed // 60)
+                    elapsed_seconds = elapsed % 60
+                    duration_str = (
+                        f"{elapsed_minutes}m {elapsed_seconds:.1f}s"
+                        if elapsed_minutes > 0
+                        else f"{elapsed_seconds:.1f}s"
+                    )
+
+                    logger.warning(
+                        f"Request timed out (attempt {attempt}, "
+                        f"duration: {duration_str}). Retrying in {attempt} second(s)..."
+                    )
+
+                    # Exponential backoff starting at 1 second
+                    await asyncio.sleep(attempt)
+                else:
+                    # Non-timeout errors should not be retried
+                    return {"error": True, "exception": error_msg}
+
+    async def get(
+        self, url: str, headers: Optional[Dict[str, str]] = None, timeout: float = 30.0
+    ) -> Dict[str, Any]:
+        return await self.request("GET", url, headers=headers, timeout=timeout)
+
+    async def post(
+        self,
+        url: str,
+        headers: Optional[Dict[str, str]] = None,
+        data: Optional[bytes] = None,
+        json_data: Optional[Dict[str, Any]] = None,
+        timeout: float = 30.0,
+    ) -> Dict[str, Any]:
+        return await self.request(
+            "POST", url, headers=headers, data=data, json_data=json_data, timeout=timeout
+        )
+
+    def set_default_headers(self, headers: Dict[str, str]):
+        self.session_headers.update(headers)
+
+
+# Global client instance
+http_client = AsyncHTTPClient()