tools/__init__.py

"""JARVIS Tool System — extensible tool registry."""

import json
import logging
import platform
from typing import Callable

_log = logging.getLogger("jarvis.tools")

TOOL_REGISTRY: dict[str, dict] = {}

# Tools that require macOS (osascript/subprocess) and should be delegated
# to a connected device when the server runs on Linux (HF Space / Render).
_MACOS_ONLY_TOOLS: set[str] = set()

IS_MACOS = platform.system() == "Darwin"


def tool(name: str, description: str, parameters: dict = None, macos_only: bool = False):
    """Decorator to register a tool.

    Args:
        macos_only: If True, this tool requires macOS (osascript, AppleScript, etc.)
                    and will be delegated to a connected device when running on Linux.
    """
    def decorator(func: Callable):
        TOOL_REGISTRY[name] = {
            "name": name,
            "description": description,
            "parameters": parameters or {},
            "function": func,
        }
        if macos_only:
            _MACOS_ONLY_TOOLS.add(name)
        return func
    return decorator


async def _delegate_to_device(name: str, args: dict) -> str:
    """Delegate a macOS-only tool call to a connected device via WebSocket push."""
    try:
        from user_device_registry import list_devices, send_command_to_device
    except ImportError:
        return f"Error: {name} requires macOS but this server runs on Linux. No device delegation available."

    # Find an online device to delegate to
    devices = await list_devices("default")
    online_devices = [d for d in devices if d.get("status") == "online"]
    if not online_devices:
        return (
            f"Error: '{name}' requires macOS but this server runs in the cloud. "
            f"No connected Mac device found to execute this command. "
            f"Make sure the JARVIS listener is running on your Mac."
        )

    target = online_devices[0]
    device_id = target.get("device_id", "")
    alias = target.get("alias", "device")

    # Try WebSocket push first (instant)
    try:
        # Import at call time to avoid circular imports
        import importlib
        server_mod = importlib.import_module("server")
        push_fn = getattr(server_mod, "push_command_to_device", None)
        if push_fn:
            pushed = await push_fn(device_id, {
                "action": "execute_tool",
                "tool": name,
                "args": args,
            })
            if pushed:
                return f"Command '{name}' sent to {alias} for execution."
    except Exception:
        pass

    # Fallback: queue command via REST
    result = await send_command_to_device(
        target_alias=alias,
        command=json.dumps({"tool": name, "args": args}),
        user_id="default",
    )
    if "error" in result:
        return f"Error delegating '{name}' to {alias}: {result['error']}"
    return f"Command '{name}' queued for execution on {alias}."


async def execute_tool(name: str, args: dict) -> str:
    if name not in TOOL_REGISTRY:
        return f"Error: Unknown tool '{name}'"

    # If running on Linux (HF Space) and this tool needs macOS, delegate it
    if not IS_MACOS and name in _MACOS_ONLY_TOOLS:
        return await _delegate_to_device(name, args)

    # Validate required parameters and basic types before calling
    schema = TOOL_REGISTRY[name].get("parameters", {})
    properties = schema.get("properties", {})
    required = schema.get("required", [])

    for req_param in required:
        if req_param not in args:
            return f"Error: {name} requires parameter '{req_param}'"

    # Basic type coercion for common mismatches from LLM output
    for param_name, param_value in list(args.items()):
        if param_name in properties:
            expected_type = properties[param_name].get("type")
            if expected_type == "integer" and isinstance(param_value, str):
                try:
                    args[param_name] = int(param_value)
                except ValueError:
                    return f"Error: {name} parameter '{param_name}' must be an integer"
            elif expected_type == "boolean" and isinstance(param_value, str):
                args[param_name] = param_value.lower() in ("true", "1", "yes")

    try:
        result = TOOL_REGISTRY[name]["function"](**args)
        if hasattr(result, "__await__"):
            result = await result
        return str(result)
    except TypeError as e:
        return f"Error executing {name}: invalid arguments"
    except Exception as e:
        _log.error(f"Tool {name} failed: {type(e).__name__}: {e}")
        return f"Error executing {name}: an unexpected error occurred"


def register_macos_tools():
    """Register all macOS-only tool names.

    Called after all tool modules are imported so TOOL_REGISTRY is populated.
    These tools use osascript, AppleScript, or macOS-specific CLI commands.
    On Linux (HF Space / Render), they get automatically delegated to a
    connected Mac device via WebSocket push.
    """
    macos_tools = {
        # system_control.py — all 22 tools
        "set_volume", "get_volume", "set_brightness", "toggle_dark_mode",
        "wifi_control", "bluetooth_control", "do_not_disturb", "screenshot",
        "media_control", "clipboard", "send_notification", "list_running_apps",
        "quit_app", "set_timer", "lock_screen", "sleep_display", "empty_trash",
        "announce", "open_url", "search_files", "create_reminder", "get_events",
        # app_automation.py — all 36 tools
        "spotify_play", "spotify_play_uri", "spotify_search", "spotify_queue",
        "spotify_status", "spotify_control",
        "notes_create", "notes_append", "notes_list", "notes_search",
        "notes_read", "notes_delete",
        "reminders_add", "reminders_list", "reminders_complete",
        "calendar_create_event", "calendar_today",
        "send_imessage", "read_messages",
        "mail_compose", "mail_unread",
        "browser_open", "browser_tabs", "browser_read_page",
        "spotlight_search", "finder_open", "finder_move", "finder_copy", "trash_file",
        "contacts_search", "maps_directions",
        "app_keystroke", "app_menu_click", "app_window_manage", "textedit_create",
        "photos_recent", "photos_search", "photos_albums", "photos_open",
        "focus_status", "focus_set",
        # builtin.py — macOS-specific subset
        "open_app", "open_terminal", "run_in_terminal",
        # vscode_tools.py — all 7 tools
        "vscode_open", "vscode_open_terminal", "vscode_run_command",
        "copilot_chat", "copilot_inline", "vscode_list_extensions", "vscode_diff",
        # device_control.py — macOS-specific subset
        "bluetooth_scan", "bluetooth_pair", "bluetooth_connect",
        "bluetooth_disconnect", "bluetooth_info", "network_scan",
        "bonjour_discover", "homekit_control", "homekit_list_shortcuts",
        "find_my_devices", "siri_command", "airdrop_file",
    }
    _MACOS_ONLY_TOOLS.update(macos_tools)


def get_tool_definitions() -> list[dict]:
    """Get tool definitions formatted for LLM function calling."""
    tools = []
    for name, info in TOOL_REGISTRY.items():
        tools.append({
            "name": info["name"],
            "description": info["description"],
            "parameters": info["parameters"],
        })
    return tools


def get_tools_prompt() -> str:
    """Generate a tools description for the system prompt."""
    lines = ["You have access to the following tools:\n"]
    for name, info in TOOL_REGISTRY.items():
        params = info["parameters"]
        param_str = ", ".join(
            f"{p}: {params['properties'][p].get('description', '')}"
            for p in params.get("properties", {})
        )
        lines.append(f"- **{name}**({param_str}): {info['description']}")
    lines.append(
        "\n## TOOL CALL FORMAT (MANDATORY)\n"
        "To use a tool, you MUST respond with a JSON block inside ```tool fences.\n"
        "The format is EXACTLY:\n"
        '```tool\n{"tool": "tool_name", "args": {"param": "value"}}\n```\n\n'
        "## TOOL CALL EXAMPLES\n"
        "User: \"Open Safari\"\n"
        "Response: Opening Safari for you, sir.\n"
        '```tool\n{"tool": "open_app", "args": {"app_name": "Safari"}}\n```\n\n'
        "User: \"Open VS Code\"\n"
        "Response: Launching Visual Studio Code now.\n"
        '```tool\n{"tool": "open_app", "args": {"app_name": "Visual Studio Code"}}\n```\n\n'
        "User: \"What's the weather in London?\"\n"
        "Response: Let me check London's weather.\n"
        '```tool\n{"tool": "get_weather", "args": {"location": "London"}}\n```\n\n'
        "User: \"Turn on dark mode\"\n"
        "Response: Switching to dark mode.\n"
        '```tool\n{"tool": "toggle_dark_mode", "args": {"enable": true}}\n```\n\n'
        "User: \"Set volume to 50\"\n"
        "Response: Setting volume to 50%.\n"
        '```tool\n{"tool": "set_volume", "args": {"level": 50}}\n```\n\n'
        "## CRITICAL RULES\n"
        "- ALWAYS emit a ```tool block when the user wants you to DO something (open app, play music, set volume, search, etc.)\n"
        "- NEVER just describe what you would do — actually call the tool.\n"
        "- You can include speech text BEFORE the tool block.\n"
        "- You can chain multiple ```tool blocks in one response.\n"
        "- After tool results come back, synthesize them into a natural response.\n"
        "- If a command maps to ANY registered tool, USE IT. Do not respond with just text.\n"
    )
    return "\n".join(lines)