Modules/Agent_Terminal.py

from __future__ import annotations

import os
import sys
import types
import ast
from io import StringIO
from typing import Annotated
import importlib.metadata

import gradio as gr
from ._docstrings import autodoc
from .File_System import ROOT_DIR, File_System
from .Web_Fetch import Web_Fetch
from .Web_Search import Web_Search
from .Memory_Manager import Memory_Manager
from .Generate_Speech import Generate_Speech, List_Kokoro_Voices, List_Supertonic_Voices
from .Generate_Image import Generate_Image
from .Generate_Video import Generate_Video
from .Deep_Research import Deep_Research
from .Obsidian_Vault import Obsidian_Vault
from .Shell_Command import Shell_Command
from .Code_Interpreter import Code_Interpreter

from app import _log_call_end, _log_call_start, _truncate_for_log

def search_packages(query: str = "") -> str:
    """Search for installed Python packages by name. If query is empty, lists all."""
    packages = []
    query = query.lower()
    for dist in importlib.metadata.distributions():
        name = dist.metadata['Name']
        if query in name.lower():
            packages.append(f"{name} ({dist.version})")
    packages.sort()
    if not packages:
        return f"No packages found matching '{query}'."
    return "\n".join(packages)

def _get_tools_map():
    return {
        "Web_Fetch": Web_Fetch,
        "Web_Search": Web_Search,
        "Memory_Manager": Memory_Manager,
        "Generate_Speech": Generate_Speech,
        "List_Kokoro_Voices": List_Kokoro_Voices,
        "List_Supertonic_Voices": List_Supertonic_Voices,
        "Generate_Image": Generate_Image,
        "Generate_Video": Generate_Video,
        "Deep_Research": Deep_Research,
        "File_System": File_System,
        "Obsidian_Vault": Obsidian_Vault,
        "Shell_Command": Shell_Command,
        "Code_Interpreter": Code_Interpreter,
    }

def list_tools() -> list[str]:
    """List all available tools in the Code Interpreter environment."""
    return list(_get_tools_map().keys())

def search_tools(query: str) -> str:
    """Search for tools by name or description. Returns usage info for matches."""
    query = query.lower()
    matches = []
    tools = _get_tools_map()
    for name, func in tools.items():
        doc = (func.__doc__ or "").lower()
        if query in name.lower() or query in doc:
            matches.append((name, func))
    
    if not matches:
        return f"No tools found matching '{query}'."
    
    output = []
    for name, func in matches:
        output.append(f"--- {name} ---")
        output.append(func.__doc__ or "No documentation available.")
        output.append("")
    return "\n".join(output)

def usage(tool_name: str) -> str:
    """Get detailed usage information for a specific tool."""
    tools = _get_tools_map()
    if tool_name not in tools:
        return f"Tool '{tool_name}' not found. Available tools: {', '.join(tools.keys())}"
    func = tools[tool_name]
    return f"--- {tool_name} ---\n{func.__doc__ or 'No documentation available.'}"

def _initialize_mock_modules():
    """
    Registers a mock 'functions' module in sys.modules so that LLMs
    can do 'from functions import ...' without error.
    """
    mock_module = types.ModuleType("functions")
    
    # Add tools
    for name, tool in _get_tools_map().items():
        setattr(mock_module, name, tool)
        
    # Add helpers
    helpers = {
        "list_tools": list_tools,
        "search_tools": search_tools,
        "usage": usage,
        "search_packages": search_packages,
    }
    for name, func in helpers.items():
        setattr(mock_module, name, func)
        
    sys.modules["functions"] = mock_module

_initialize_mock_modules()

# Single source of truth for the LLM-facing tool description
TOOL_SUMMARY = (
    "Executes Python code as the unified interface for the entire tools ecosystem. "
    "All tool interactions must happen through this code-execution gateway. "
    "Use Agent Terminal repeatedly whenever you need to chain or combine tool operations. "
    "Available tools: `Web_Fetch`, `Web_Search`, `Code_Interpreter`, `Shell_Command`, `File_System`, `Obsidian_Vault`, `Memory_Manager`, `Generate_Speech`, `Generate_Image`, `Generate_Video`, `Deep_Research`."
)



@autodoc(
    summary=TOOL_SUMMARY,
)
def Agent_Terminal(input: Annotated[str, (
    "Python source code to run; stdout is captured and returned. "
    "Execute these commands: "
    "`search_tools('query')` to search for tools by name or capability; "
    "`list_tools()` to list all available tools; "
    "`usage('ToolName')` to inspect a tool’s expected input parameters; "
    "`search_packages('query')` to search for installed Python libraries."
)]) -> str:
    _log_call_start("Agent_Terminal", input=_truncate_for_log(input or "", 300))
    if input is None:
        result = "No code provided."
        _log_call_end("Agent_Terminal", result)
        return result
    old_stdout = sys.stdout
    old_cwd = os.getcwd()
    redirected_output = sys.stdout = StringIO()
    
    # Prepare the execution environment with all tools
    tools_env = {
        "Web_Fetch": Web_Fetch,
        "Web_Search": Web_Search,
        "Memory_Manager": Memory_Manager,
        "Generate_Speech": Generate_Speech,
        "List_Kokoro_Voices": List_Kokoro_Voices,
        "List_Supertonic_Voices": List_Supertonic_Voices,
        "Generate_Image": Generate_Image,
        "Generate_Video": Generate_Video,
        "Deep_Research": Deep_Research,
        "File_System": File_System,
        "Obsidian_Vault": Obsidian_Vault,
        "Shell_Command": Shell_Command,
        "Code_Interpreter": Code_Interpreter,
        "list_tools": list_tools,
        "search_tools": search_tools,
        "usage": usage,
        "search_packages": search_packages,
        "print": print,  # Ensure print is available
        "__builtins__": __builtins__,
    }
    
    try:
        os.chdir(ROOT_DIR)
        
        # Parse code to check if the last statement is an expression
        tree = ast.parse(input)
        if tree.body and isinstance(tree.body[-1], ast.Expr):
            last_node = tree.body.pop()
            
            # Execute preceding statements
            if tree.body:
                exec(compile(tree, filename="<string>", mode="exec"), tools_env)
            
            # Evaluate and print the last expression
            expr = compile(ast.Expression(last_node.value), filename="<string>", mode="eval")
            result_val = eval(expr, tools_env)
            if result_val is not None:
                print(result_val)
        else:
            exec(input, tools_env)
            
        result = redirected_output.getvalue()
    except Exception as exc:  # pylint: disable=broad-except
        result = str(exc)
    finally:
        sys.stdout = old_stdout
        try:
            os.chdir(old_cwd)
        except Exception:
            pass
    _log_call_end("Agent_Terminal", _truncate_for_log(result))
    return result


def build_interface() -> gr.Interface:
    return gr.Interface(
        fn=Agent_Terminal,
        inputs=gr.Code(label="Python Code", language="python"),
        outputs=gr.Textbox(label="Output", lines=5, max_lines=20),
        title="Agent Terminal",
        description="<div style=\"text-align:center\">Interact with all other tools via a Python API. Reduces token usage by 90%.</div>",
        api_description=TOOL_SUMMARY,
        flagging_mode="never",
    )


__all__ = ["Agent_Terminal", "build_interface"]