Implement non-blocking async tool execution and context auto-compaction for 24/7 continuous operation

backend.py

@@ -247,8 +247,8 @@ def repair_arguments(func_name: str, args: dict) -> tuple[dict, list[str]]:
     return repaired_args, notes
 
 
-def execute_tool(name: str, arguments: dict) -> str:
-    """Execute a tool and return its output as a string."""
+async def execute_tool(name: str, arguments: dict) -> str:
+    """Execute a tool and return its output as a string asynchronously."""
     try:
         if name == "read_file":
             path = _safe_path(arguments["path"])
@@ -273,24 +273,35 @@ def execute_tool(name: str, arguments: dict) -> str:
             blocked = ["rm -rf /", "mkfs", "dd if=", ":(){", "fork bomb"]
             if any(b in command.lower() for b in blocked):
                 return "Error: Command blocked for safety reasons"
-            result = subprocess.run(
-                ["bash", "-c", command],
+            
+            # ASYNC SUBPROCESS - This prevents the FastAPI server from freezing!
+            process = await asyncio.create_subprocess_shell(
+                command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
                 cwd=WORKSPACE_DIR,
-                capture_output=True,
-                text=True,
-                timeout=30,
                 env={**os.environ, "HOME": "/tmp", "PATH": os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin")},
             )
-            output = ""
-            if result.stdout:
-                output += result.stdout
-            if result.stderr:
-                output += ("\n" if output else "") + f"[stderr] {result.stderr}"
-            if result.returncode != 0:
-                output += f"\n[exit code: {result.returncode}]"
-            if not output:
-                output = "[command completed with no output]"
-            # Truncate very long outputs
+            
+            try:
+                stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=30)
+                output = ""
+                if stdout:
+                    output += stdout.decode('utf-8', errors='replace')
+                if stderr:
+                    output += ("\n" if output else "") + f"[stderr] {stderr.decode('utf-8', errors='replace')}"
+                if process.returncode != 0:
+                    output += f"\n[exit code: {process.returncode}]"
+                if not output:
+                    output = "[command completed with no output]"
+            except asyncio.TimeoutError:
+                try:
+                    process.kill()
+                except Exception:
+                    pass
+                await process.communicate()
+                return "Error: Command timed out after 30 seconds"
+                
             if len(output) > 20000:
                 output = output[:20000] + f"\n\n[Truncated — output is {len(output)} chars]"
             return output
@@ -320,14 +331,26 @@ def execute_tool(name: str, arguments: dict) -> str:
             pattern = arguments["pattern"]
             search_path = arguments.get("path", ".")
             path = _safe_path(search_path)
-            result = subprocess.run(
-                ["grep", "-rn", "--include=*", pattern, str(path)],
-                capture_output=True,
-                text=True,
-                timeout=10,
+            
+            # Escape single quotes in pattern for safety
+            escaped_pattern = pattern.replace("'", "'\\''")
+            process = await asyncio.create_subprocess_shell(
+                f"grep -rn --include=* '{escaped_pattern}' '{path}'",
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
                 cwd=WORKSPACE_DIR,
             )
-            output = result.stdout if result.stdout else "No matches found"
+            try:
+                stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=10)
+                output = stdout.decode('utf-8', errors='replace') if stdout else "No matches found"
+            except asyncio.TimeoutError:
+                try:
+                    process.kill()
+                except Exception:
+                    pass
+                await process.communicate()
+                return "Error: Grep search timed out"
+                
             if len(output) > 10000:
                 output = output[:10000] + "\n\n[Truncated]"
             return output
@@ -335,10 +358,6 @@ def execute_tool(name: str, arguments: dict) -> str:
         else:
             return f"Error: Unknown tool: {name}"
 
-    except subprocess.TimeoutExpired:
-        return "Error: Command timed out after 30 seconds"
-    except ValueError as e:
-        return f"Error: {str(e)}"
     except Exception as e:
         return f"Error executing {name}: {str(e)}"
 
@@ -572,6 +591,57 @@ IMPORTANT RULES:
 """
 
 
+def compact_history(messages: list) -> list:
+    """
+    Compact conversation history to prevent context window overflow.
+    Replaces massive tool call outputs with concise summaries if history is long.
+    """
+    # Only compact if messages count exceeds 15 (to maintain normal conversation)
+    if len(messages) <= 15:
+        return messages
+        
+    compacted = []
+    # Always keep the system prompt (typically the first message)
+    if messages and messages[0].get("role") == "system":
+        compacted.append(messages[0])
+        start_idx = 1
+    else:
+        start_idx = 0
+        
+    # Keep the last 4 messages exactly as they are to preserve immediate context
+    recent_count = 4
+    mid_messages = messages[start_idx:-recent_count]
+    recent_messages = messages[-recent_count:]
+    
+    for msg in mid_messages:
+        role = msg.get("role")
+        content = msg.get("content") or ""
+        
+        if role == "tool":
+            # Compress massive tool outputs (like bash stdout or file reads)
+            if len(content) > 1000:
+                summary = f"[Tool output compacted: {content[:200]}... (Total {len(content)} chars truncated for context preservation)]"
+                compacted.append({
+                    "role": "tool",
+                    "tool_call_id": msg.get("tool_call_id"),
+                    "content": summary
+                })
+                continue
+        elif role == "assistant" and msg.get("tool_calls"):
+            # Keep tool calls metadata so the model's message-tool call mapping DAG doesn't break
+            pass
+            
+        # Keep general messages, but truncate if they are too long
+        if len(content) > 2000:
+            msg = dict(msg)
+            msg["content"] = content[:2000] + "\n[Content truncated for compaction]"
+        compacted.append(msg)
+        
+    compacted.extend(recent_messages)
+    log_activity(f"[Auto-Compaction] Compressed message history from {len(messages)} down to {len(compacted)}")
+    return compacted
+
+
 @app.post("/v1/chat/completions")
 async def chat_completions(request: Request, authorization: str = Header(None)):
     auth(authorization)
@@ -611,6 +681,9 @@ async def chat_completions(request: Request, authorization: str = Header(None)):
     else:
         final_messages = list(messages)
 
+    # Perform auto-compaction before executing agent loops
+    final_messages = compact_history(final_messages)
+
     # Save the user's message to DB
     user_msg = next((m for m in reversed(messages) if m.get("role") == "user"), None)
     if user_msg:
@@ -646,6 +719,8 @@ async def chat_completions(request: Request, authorization: str = Header(None)):
         async with completions_semaphore:
             try:
                 for round_num in range(MAX_TOOL_ROUNDS + 1):
+                    # Perform auto-compaction before calling NIM API
+                    final_messages = compact_history(final_messages)
                     kwargs = {"model": requested_model, "messages": final_messages, "stream": True}
                     if is_agentic:
                         kwargs["tools"] = TOOLS
@@ -757,7 +832,7 @@ async def chat_completions(request: Request, authorization: str = Header(None)):
                         yield make_chunk(request_id, requested_model, "\n")
 
                     # Execute the tool
-                    result = execute_tool(func_name, repaired_args)
+                    result = await execute_tool(func_name, repaired_args)
 
                     # Append teaching note if repaired
                     if repair_notes: