gns3-server/gns3server/agent/gns3_copilot/tools_v2/window_controller.py

# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This file is part of FlowNet-Lab.
#
# FlowNet-Lab is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by the
# Free Software Foundation, either version 3 of the License, or (at your
# option) any later version.
#
# FlowNet-Lab is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License
# for more details.
#
# You should have received a copy of the GNU Affero General Public License along
# with FlowNet-Lab. If not, see <https://www.gnu.org/licenses/>.

"""
Window Controller Tool - Frontend Proxy Tool

This tool acts as a proxy to frontend window operations.
It sends commands to the frontend via WebSocket and waits for execution results.

Architecture:
    Agent → WindowControllerTool._arun() → WebSocket → Frontend
         ← (waits for future)             ← execution result
"""

import json
import uuid
from typing import Optional, Type, Any

from langchain.tools import BaseTool
from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from pydantic import BaseModel, Field

import logging
from backend.core.session_manager import get_session_manager

logger = logging.getLogger(__name__)


# Window type configurations with minimal required parameters
# Size and position should be dynamically generated by the LLM based on context
WINDOW_CONFIGS = {
    "ai_chat": {
        "window_type": "ai_chat",
        "title": "AI Chat",
        "min_size": {"width": 400, "height": 500},
        "description": "AI conversation interface for chatting with the assistant",
    },
    "network_topology": {
        "window_type": "network_topology",
        "title": "Network Topology",
        "min_size": {"width": 600, "height": 400},
        "description": "GNS3 network topology visualization view",
    },
    "terminal": {
        "window_type": "terminal",
        "title": "Terminal",
        "min_size": {"width": 500, "height": 300},
        "description": "Device terminal emulator for command-line access",
    },
    "calibre_books": {
        "window_type": "calibre_books",
        "title": "Calibre Books",
        "min_size": {"width": 500, "height": 400},
        "description": "Calibre ebook library browser",
    },
    "notes": {
        "window_type": "notes",
        "title": "Notes",
        "min_size": {"width": 300, "height": 300},
        "description": "Note-taking and management interface",
    },
    "pdf_reader": {
        "window_type": "pdf_reader",
        "title": "PDF Reader",
        "min_size": {"width": 500, "height": 400},
        "description": "PDF document viewer and reader",
    },
    "gns3_projects": {
        "window_type": "gns3_projects",
        "title": "GNS3 Projects",
        "min_size": {"width": 500, "height": 400},
        "description": "GNS3 project selection and management",
    },
    "video_recorder": {
        "window_type": "video_recorder",
        "title": "Video Recorder",
        "min_size": {"width": 400, "height": 300},
        "description": "Screen recording tool",
    },
    "settings": {
        "window_type": "settings",
        "title": "Settings",
        "min_size": {"width": 400, "height": 300},
        "description": "System settings and configuration",
    },
}

VALID_ACTIONS = {
    "open": ["window_type"],
    "close": ["window_title"],
    "focus": ["window_title"],
    "minimize": ["window_title"],
    "maximize": ["window_title"],
    "restore": ["window_title"],
    "center": ["window_title"],
    "move": ["window_title", "x", "y"],
    "resize": ["window_title", "width", "height"],
}


class WindowControllerInput(BaseModel):
    """Input schema for WindowControllerTool."""

    action: str = Field(
        ...,
        description=f"Window action to perform. Valid actions: {list(VALID_ACTIONS.keys())}"
    )
    window_type: Optional[str] = Field(
        None,
        description=f"Type of window to open (required for 'open' action). Valid types: {list(WINDOW_CONFIGS.keys())}"
    )
    window_title: Optional[str] = Field(
        None,
        description="Title of the target window (required for close, focus, minimize, maximize, center actions)"
    )
    x: Optional[float] = Field(None, description="X coordinate for move action")
    y: Optional[float] = Field(None, description="Y coordinate for move action")
    width: Optional[float] = Field(None, description="Width for resize action or when opening a new window")
    height: Optional[float] = Field(None, description="Height for resize action or when opening a new window")
    size: Optional[dict] = Field(None, description="Complete size object: {\"width\": 800, \"height\": 600} (for open action)")
    position: Optional[dict] = Field(None, description="Complete position object: {\"x\": 200, \"y\": 150} (for open action)")


class WindowControllerTool(BaseTool):
    """
    Proxy tool for frontend window control operations.

    This tool sends window control commands to the frontend via WebSocket
    and waits for the frontend to execute and return results.

    The actual window operations (open, close, focus, etc.) are executed
    in the browser, not on the backend.
    """

    name: str = "window_controller"
    description: str = """
Control windows in the frontend web interface.

Supported actions:
- open: Open a new window (requires: window_type)
  You SHOULD intelligently generate size and position based on window type and context.
  Example: {"action": "open", "window_type": "terminal", "size": {"width": 900, "height": 600}, "position": {"x": 250, "y": 150}}

- close: Close a window (requires: window_title)
  Example: {"action": "close", "window_title": "Notes"}

- focus: Bring a window to front - increases z-index to make it visible above other windows (requires: window_title)
  Use this when a window is visible but behind other windows.
  Does NOT change minimized/maximized state.
  Example: {"action": "focus", "window_title": "Notes"}

- minimize: Minimize a window - hide it from view (requires: window_title)
  Window becomes hidden and can be restored later.
  Example: {"action": "minimize", "window_title": "Notes"}

- maximize: Maximize a window - expand to fullscreen (requires: window_title)
  Window fills the entire screen. Can be restored to previous size.
  Example: {"action": "maximize", "window_title": "Notes"}

- restore: Restore a minimized or maximized window to its normal state (requires: window_title)
  Use this when user says "restore", "recover", "show", "unhide", "unminimize", "unmaximize" a window.
  This is the CORRECT action when window is minimized or maximized and user wants it back to normal.
  Example: {"action": "restore", "window_title": "Notes"}

- center: Center a window on screen (requires: window_title)
  Window occupies 90%% of viewport in center.
  Example: {"action": "center", "window_title": "Notes"}

- move: Move a window (requires: window_title, x, y)
  Example: {"action": "move", "window_title": "Notes", "x": 100, "y": 100}

- resize: Resize a window (requires: window_title, width, height)
  Example: {"action": "resize", "window_title": "Notes", "width": 800, "height": 600}

Available window types and recommended sizes:
- "ai_chat": AI Chat interface (600x800 default, good for right side of screen)
- "network_topology": Network topology visualization (1000x700 default, needs large space)
- "terminal": Command execution terminal (900x600 default, good for bottom or left)
- "calibre_books": E-book library browser (1200x800 default, needs wide space)
- "notes": Note-taking interface (700x600 default, flexible)
- "pdf_reader": PDF document viewer (1000x700 default, needs large space)
- "gns3_projects": GNS3 project selector (600x500 default, medium size)
- "video_recorder": Screen recording tool (800x600 default, medium size)
- "settings": Application configuration (800x600 default, medium size)

Guidelines for intelligent window placement:
- For productivity tools (notes, pdf_reader, calibre_books): Place on right side (x: 800-1200, y: 100-200)
- For system tools (terminal, settings): Place on left side (x: 100-300, y: 100-200)
- For visualization (network_topology): Place in center or large area (x: 200-400, y: 100-200)
- Avoid overlapping windows by checking their purpose
- Consider typical screen resolution (1920x1080) when positioning

When opening windows, ALWAYS include both size and position parameters for better UX.
"""
    args_schema: Type[BaseModel] = WindowControllerInput

    # Session ID is injected when creating the tool instance
    session_id: str = "default"

    def _run(
        self,
        action: str,
        window_type: Optional[str] = None,
        window_title: Optional[str] = None,
        x: Optional[float] = None,
        y: Optional[float] = None,
        width: Optional[float] = None,
        height: Optional[float] = None,
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        """
        Synchronous version - not supported.

        Use the async version (_arun) instead.
        """
        raise NotImplementedError(
            "WindowControllerTool requires async execution. "
            "Use the async version (_arun) or ensure your agent supports async tools."
        )

    async def _arun(
        self,
        action: str,
        window_type: Optional[str] = None,
        window_title: Optional[str] = None,
        x: Optional[float] = None,
        y: Optional[float] = None,
        width: Optional[float] = None,
        height: Optional[float] = None,
        size: Optional[dict] = None,
        position: Optional[dict] = None,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """
        Async implementation: Send command to frontend and wait for result.

        Args:
            action: Window action to perform
            window_type: Type of window (for 'open' action)
            window_title: Title of target window (for other actions)
            x, y: Position coordinates (for 'move' action)
            width, height: Size dimensions (for 'resize' action or 'open' action)
            size: Complete size dict for 'open' action
            position: Complete position dict for 'open' action
            run_manager: LangChain callback manager

        Returns:
            JSON string with execution result from frontend
        """
        import asyncio

        session_manager = get_session_manager()
        session = session_manager.get_session(self.session_id)

        if not session:
            logger.error(f"Session not found: {self.session_id}")
            return json.dumps({
                "status": "error",
                "error": f"Session not found: {self.session_id}"
            })

        # Validate action
        if action not in VALID_ACTIONS:
            return json.dumps({
                "status": "error",
                "error": f"Invalid action '{action}'. Valid actions: {list(VALID_ACTIONS.keys())}"
            })

        # Build input data
        input_data = {
            "action": action,
        }

        # Add optional parameters based on action
        if action == "open":
            if not window_type:
                return json.dumps({
                    "status": "error",
                    "error": "Missing required parameter: window_type"
                })
            if window_type not in WINDOW_CONFIGS:
                return json.dumps({
                    "status": "error",
                    "error": f"Invalid window_type '{window_type}'. Available: {list(WINDOW_CONFIGS.keys())}"
                })
            input_data["window_type"] = window_type

            # Get base configuration
            base_config = WINDOW_CONFIGS[window_type]
            window_config = base_config.copy()

            # Merge LLM-provided size (supports both formats: dict or width/height)
            if size:
                window_config["size"] = size
            elif width is not None and height is not None:
                window_config["size"] = {"width": width, "height": height}
            else:
                # Use default size based on window type
                window_config["size"] = _get_default_size(window_type)

            # Merge LLM-provided position (supports both formats: dict or x/y)
            if position:
                window_config["position"] = position
            elif x is not None and y is not None:
                window_config["position"] = {"x": x, "y": y}
            else:
                # Use default position based on window type
                window_config["position"] = _get_default_position(window_type)

            # Add min_size from base config
            if "min_size" in base_config:
                window_config["min_size"] = base_config["min_size"]

            # Add window_config to input_data
            input_data["window_config"] = window_config

            logger.info(f"Opening window: type={window_type}, config={window_config}")

        elif action in ["close", "focus", "minimize", "maximize", "restore", "center"]:
            if not window_title:
                return json.dumps({
                    "status": "error",
                    "error": f"Missing required parameter: window_title (for action '{action}')"
                })

            # Check if window exists in backend state before sending command
            from backend.core.window_state_manager import get_window_state_manager
            state_manager = get_window_state_manager()

            logger.info(
                f"[WindowController] Looking for window '{window_title}' "
                f"in session {self.session_id}..."
            )

            window = state_manager.get_window_by_title(self.session_id, window_title)

            if window:
                logger.info(
                    f"[WindowController] ✅ Found window '{window_title}' "
                    f"(id={window.id}, type={window.window_type})"
                )
            else:
                # Log all available windows for debugging
                state = state_manager.get_state(self.session_id)
                if state:
                    available_titles = [w.title for w in state.windows]
                    logger.warning(
                        f"[WindowController] ❌ Window '{window_title}' not found. "
                        f"Available windows: {available_titles}"
                    )
                else:
                    logger.warning(
                        f"[WindowController] ❌ Window '{window_title}' not found. "
                        f"No state exists for session {self.session_id}"
                    )

            if not window:
                # Window not found - return success (idempotent operation)
                logger.info(
                    f"Window '{window_title}' not found in state, "
                    f"treating as already {'closed' if action == 'close' else 'not existent'}"
                )
                return json.dumps({
                    "status": "success",
                    "action": action,
                    "window_title": window_title,
                    "already_closed": True if action == "close" else False,
                    "not_found": True
                })

            input_data["window_title"] = window_title

        elif action == "move":
            if not window_title:
                return json.dumps({
                    "status": "error",
                    "error": "Missing required parameter: window_title"
                })
            if x is None or y is None:
                return json.dumps({
                    "status": "error",
                    "error": "Missing required parameters: x and y (for move action)"
                })
            input_data["window_title"] = window_title
            input_data["x"] = x
            input_data["y"] = y

        elif action == "resize":
            if not window_title:
                return json.dumps({
                    "status": "error",
                    "error": "Missing required parameter: window_title"
                })
            # Support both formats: separate width/height or size dict
            if size:
                input_data["window_title"] = window_title
                input_data["width"] = size.get("width")
                input_data["height"] = size.get("height")
            elif width is not None and height is not None:
                input_data["window_title"] = window_title
                input_data["width"] = width
                input_data["height"] = height
            else:
                return json.dumps({
                    "status": "error",
                    "error": "Missing required parameters: width and height (or size dict) for resize action"
                })

        # Generate task_id
        task_id = str(uuid.uuid4())

        # Create Future for waiting on frontend result
        future = session.create_future(task_id)

        # Send tool_call to frontend
        try:
            await session.websocket.send_json({
                "type": "tool_call",
                "task_id": task_id,
                "tool": "window_controller",
                "args": input_data,  # Include all parameters including window_config
                "session_id": self.session_id
            })

            logger.info(f"WindowController: Sent tool_call to frontend, task_id={task_id}, action={action}")

        except Exception as e:
            session.pending_futures.pop(task_id, None)
            logger.error(f"WindowController: Failed to send tool_call: {e}")
            return json.dumps({
                "status": "error",
                "error": f"Failed to send command to frontend: {str(e)}"
            })

        # Wait for frontend to execute and return result (timeout 30 seconds)
        try:
            result = await asyncio.wait_for(future, timeout=30.0)

            logger.info(f"WindowController: Received result from frontend, task_id={task_id}")

            # Return result to Agent (should already be a dict or JSON string)
            if isinstance(result, dict):
                return json.dumps(result, ensure_ascii=False)
            return result

        except asyncio.TimeoutError:
            # Timeout: remove pending future
            session.pending_futures.pop(task_id, None)

            logger.warning(f"WindowController: Timeout waiting for frontend, task_id={task_id}, action={action}")

            return json.dumps({
                "status": "error",
                "error": "Timeout waiting for frontend to execute window operation",
                "action": action,
                "timeout_seconds": 30
            })

        except Exception as e:
            # Error: remove pending future
            session.pending_futures.pop(task_id, None)

            logger.error(f"WindowController: Error waiting for result: {e}")

            return json.dumps({
                "status": "error",
                "error": f"Error waiting for frontend response: {str(e)}"
            })


def _get_default_size(window_type: str) -> dict:
    """
    Get default size for a window type based on its purpose.
    """
    default_sizes = {
        "ai_chat": {"width": 600, "height": 800},
        "network_topology": {"width": 1000, "height": 700},
        "terminal": {"width": 900, "height": 600},
        "calibre_books": {"width": 1200, "height": 800},
        "notes": {"width": 700, "height": 600},
        "pdf_reader": {"width": 1000, "height": 700},
        "gns3_projects": {"width": 600, "height": 500},
        "video_recorder": {"width": 800, "height": 600},
        "settings": {"width": 800, "height": 600},
    }
    return default_sizes.get(window_type, {"width": 800, "height": 600})


def _get_default_position(window_type: str) -> dict:
    """
    Get default position for a window type based on its category.
    Strategy: Place windows in different areas of screen to minimize overlap.
    """
    # Productivity tools: right side
    if window_type in ["notes", "pdf_reader", "calibre_books"]:
        return {"x": 900, "y": 150}

    # System tools: left side
    if window_type in ["terminal", "settings", "gns3_projects"]:
        return {"x": 250, "y": 150}

    # Visualization: center
    if window_type == "network_topology":
        return {"x": 200, "y": 100}

    # Chat: right edge
    if window_type == "ai_chat":
        return {"x": 1100, "y": 100}

    # Tools: middle-right
    if window_type == "video_recorder":
        return {"x": 700, "y": 200}

    # Default fallback
    return {"x": 200, "y": 150}


def create_window_controller_tool(session_id: str) -> WindowControllerTool:
    """
    Factory function to create WindowControllerTool with specific session_id.

    Args:
        session_id: Session identifier for WebSocket connection

    Returns:
        WindowControllerTool instance configured for this session
    """
    return WindowControllerTool(session_id=session_id)


# ============================================================================
# TextInputTool - Smart text input to frontend windows
# ============================================================================


class TextInputInput(BaseModel):
    """Input schema for TextInputTool."""
    text: str = Field(..., description="The text content to input")
    window_title: Optional[str] = Field(
        None,
        description="Target window title (strongly recommended for reliable input). Examples: 'AI Chat', 'Notes', 'Terminal'"
    )
    selector: Optional[str] = Field(
        None,
        description="CSS selector for specific input element (advanced use only)"
    )


class TextInputTool(BaseTool):
    """
    A tool for smart text input to frontend windows.

    This tool sends text input commands to the frontend via WebSocket
    and waits for the frontend to execute and return results.

    The actual text input operations are executed in the browser.
    """

    name: str = "text_input"
    description: str = """
    Inputs text into frontend windows in FlowNet-Lab.

    IMPORTANT: You MUST specify window_title parameter for reliable text input.
    The focused window does NOT necessarily have a focused input field.

    How to use:
    1. Use window_controller to focus/open the target window first
    2. Use text_input with window_title parameter to send text

    Examples:
    - {"text": "Hello, how are you?", "window_title": "AI Chat"}
    - {"text": "Meeting notes", "window_title": "Notes"}
    - {"text": "user@example.com", "selector": "#email"}  # Advanced: CSS selector

    Returns a dictionary with operation status and target information.
    """
    args_schema: Type[BaseModel] = TextInputInput

    # Session ID is injected when creating the tool instance
    session_id: str = "default"

    def _run(
        self,
        text: str,
        window_title: Optional[str] = None,
        selector: Optional[str] = None,
        run_manager: CallbackManagerForToolRun | None = None,
    ) -> str:
        """
        Synchronous version - not supported.

        Use the async version (_arun) instead.
        """
        raise NotImplementedError(
            "TextInputTool requires async execution. "
            "Use the async version (_arun) or ensure your agent supports async tools."
        )

    async def _arun(
        self,
        text: str,
        window_title: Optional[str] = None,
        selector: Optional[str] = None,
        tool_input: Optional[str] = None,  # Legacy format support
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """
        Async implementation: Send command to frontend and wait for result.

        Args:
            text: The text content to input (from args_schema)
            window_title: Target window title (from args_schema)
            selector: CSS selector for specific element (from args_schema)
            tool_input: Legacy JSON string format (backward compatibility)
            run_manager: LangChain callback manager

        Returns:
            JSON string with execution result from frontend
        """
        import asyncio

        session_manager = get_session_manager()
        session = session_manager.get_session(self.session_id)

        if not session:
            logger.error(f"Session not found: {self.session_id}")
            return json.dumps({
                "status": "error",
                "error": f"Session not found: {self.session_id}"
            })

        try:
            # Handle legacy tool_input format
            if tool_input:
                input_data = json.loads(tool_input)
                text = input_data.get("text", text)
                window_title = input_data.get("window_title", window_title)
                selector = input_data.get("selector", selector)

            # Validate text
            if not text:
                return json.dumps({"error": "Missing text field."})

            # Build command data
            command_data = {
                "text": text,
            }
            if window_title:
                command_data["window_title"] = window_title
            if selector:
                command_data["selector"] = selector

            # Generate task_id
            task_id = str(uuid.uuid4())

            # Create Future for waiting on frontend result
            future = session.create_future(task_id)

            # Send tool_call to frontend
            await session.websocket.send_json({
                "type": "tool_call",
                "task_id": task_id,
                "tool": "text_input",
                "args": command_data,
                "session_id": self.session_id
            })

            logger.info(f"TextInput: Sent tool_call to frontend, task_id={task_id}")

            # Wait for frontend to execute and return result (timeout 30 seconds)
            result = await asyncio.wait_for(future, timeout=30.0)

            logger.info(f"TextInput: Received result from frontend, task_id={task_id}")

            # Return result to Agent
            if isinstance(result, dict):
                return json.dumps(result, ensure_ascii=False)
            return result

        except asyncio.TimeoutError:
            session.pending_futures.pop(task_id, None)
            logger.warning(f"TextInput: Timeout waiting for frontend, task_id={task_id}")
            return json.dumps({
                "status": "error",
                "error": "Timeout waiting for frontend to execute text input",
                "timeout_seconds": 30
            })

        except json.JSONDecodeError as e:
            logger.error(f"Invalid JSON input: {e}")
            return json.dumps({"error": f"Invalid JSON input: {e}"})

        except Exception as e:
            session.pending_futures.pop(task_id, None)
            logger.error(f"TextInput: Error waiting for result: {e}")
            return json.dumps({"error": f"Error waiting for frontend response: {str(e)}"})


def create_text_input_tool(session_id: str) -> TextInputTool:
    """
    Factory function to create TextInputTool with specific session_id.

    Args:
        session_id: Session identifier for WebSocket connection

    Returns:
        TextInputTool instance configured for this session
    """
    return TextInputTool(session_id=session_id)