Command Layer (Level-3 System Interface)

class Command(BaseModel):
    """
    Represents a command to be executed by an agent.
    Commands are atomic units of work dispatched by the orchestrator.
    """

    tool_name: str = Field(..., description="Name of the tool to execute")
    parameters: Optional[Dict[str, Any]] = Field(
        default=None, description="Parameters for the tool"
    )
    tool_type: Literal["data_collection", "action"] = Field(
        ..., description="Type of tool: data_collection or action"
    )
    call_id: Optional[str] = Field(
        default=None, description="Unique identifier for this command call"
    )

Command(
    tool_name="click_element",
    parameters={
        "control_id": "Button_InsertChart",
        "process_name": "EXCEL.EXE",
        "app_root_name": "Microsoft Excel"
    },
    tool_type="action"
)

Command(
    tool_name="shell_execute",
    parameters={
        "command": "ls -la /home/user/documents",
        "timeout": 30
    },
    tool_type="action"
)

Command(
    tool_name="read_file",
    parameters={
        "file_path": "/home/user/data.csv",
        "encoding": "utf-8"
    },
    tool_type="data_collection"
)

class Result(BaseModel):
    """
    Represents the result of a command execution.
    Contains status, error information, and the actual result payload.
    """

    status: ResultStatus = Field(..., description="Execution status")
    error: Optional[str] = Field(default=None, description="Error message if failed")
    result: Any = Field(default=None, description="Result payload")
    namespace: Optional[str] = Field(
        default=None, description="Namespace of the executed tool"
    )
    call_id: Optional[str] = Field(
        default=None, description="ID matching the Command.call_id"
    )

class ResultStatus(str, Enum):
    """Represents the status of a command execution result."""
    SUCCESS = "success"      # Command executed successfully
    FAILURE = "failure"      # Command failed with error
    SKIPPED = "skipped"      # Command was skipped
    NONE = "none"            # No result available

Result(
    status=ResultStatus.SUCCESS,
    result={"clicked": True, "control_name": "Insert Chart"},
    call_id="uuid-1234-5678"
)

Result(
    status=ResultStatus.FAILURE,
    result=None,
    error="Control 'Button_123' not found in UI tree",
    call_id="uuid-1234-5678"
)

Result(
    status=ResultStatus.SUCCESS,
    result={
        "stdout": "total 24\ndrwxr-xr-x  5 user user 4096...",
        "stderr": "",
        "exit_code": 0
    },
    call_id="uuid-1234-5678"
)

class BasicCommandDispatcher(ABC):
    """
    Abstract base class for command dispatcher.

    Responsibilities:
    - Send commands to device client
    - Wait for execution results
    - Handle errors and timeouts
    """

    @abstractmethod
    async def execute_commands(
        self, commands: List[Command], timeout: float = 6000
    ) -> Optional[List[Result]]:
        """
        Execute commands and return results.

        :param commands: List of commands to execute
        :param timeout: Execution timeout in seconds
        :return: List of results, or None if timeout
        """
        pass

    def generate_error_results(
        self, commands: List[Command], error: Exception
    ) -> List[Result]:
        """
        Generate error results for failed commands.

        :param commands: Commands that failed
        :param error: The error that occurred
        :return: List of error results
        """
        result_list = []
        for command in commands:
            error_msg = f"Error executing {command}: {error}"
            result = Result(
                status=ResultStatus.FAILURE,
                error=error_msg,
                result=error_msg,
                call_id=command.call_id
            )
            result_list.append(result)
        return result_list

class LocalCommandDispatcher(BasicCommandDispatcher):
    """
    Local command dispatcher - executes commands directly.

    No network communication - calls MCP tools locally.
    """

    def __init__(self, session: BaseSession, mcp_server_manager: MCPServerManager):
        self.session = session
        self.mcp_server_manager = mcp_server_manager

        # Direct local execution
        self.computer_manager = ComputerManager(configs, mcp_server_manager)
        self.command_router = CommandRouter(self.computer_manager)

    async def execute_commands(
        self, commands: List[Command], timeout=6000
    ) -> Optional[List[Result]]:
        """Execute commands locally via CommandRouter"""
        try:
            # Direct invocation (no network)
            action_results = await asyncio.wait_for(
                self.command_router.execute(
                    agent_name=self.session.current_agent_class,
                    root_name=self.session.context.get(ContextNames.APPLICATION_ROOT_NAME),
                    process_name=self.session.context.get(ContextNames.APPLICATION_PROCESS_NAME),
                    commands=commands
                ),
                timeout=timeout
            )
            return action_results
        except Exception as e:
            return self.generate_error_results(commands, e)

class WebSocketCommandDispatcher(BasicCommandDispatcher):
    """
    WebSocket command dispatcher - executes commands remotely via AIP.

    Uses AIP's TaskExecutionProtocol for structured messaging.
    """

    def __init__(self, session: BaseSession, protocol: TaskExecutionProtocol):
        self.session = session
        self.protocol = protocol  # AIP protocol instance
        self.pending: Dict[str, asyncio.Future] = {}
        self.logger = logging.getLogger(__name__)

    async def execute_commands(
        self, commands: List[Command], timeout=6000
    ) -> Optional[List[Result]]:
        """Execute commands remotely via AIP WebSocket"""
        try:
            # Build ServerMessage
            server_msg = self.make_server_response(commands)

            # Send via AIP
            await self.protocol.send_command(server_msg)

            # Wait for results
            results = await asyncio.wait_for(
                self._wait_for_results(server_msg.response_id),
                timeout=timeout
            )

            return results
        except Exception as e:
            return self.generate_error_results(commands, e)

    def make_server_response(self, commands: List[Command]) -> ServerMessage:
        """Create ServerMessage for commands"""
        # Assign call_ids
        for command in commands:
            command.call_id = str(uuid.uuid4())

        return ServerMessage(
            type=ServerMessageType.COMMAND,
            status=TaskStatus.CONTINUE,
            agent_name=self.session.current_agent_class,
            process_name=self.session.context.get(ContextNames.APPLICATION_PROCESS_NAME),
            root_name=self.session.context.get(ContextNames.APPLICATION_ROOT_NAME),
            actions=commands,
            session_id=self.session.id,
            task_name=self.session.task,
            response_id=str(uuid.uuid4()),
            timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat()
        )

# Windows MCP Server registration
mcp_server_manager.register_server(
    server_name="ufo_windows",
    tools=[
        MCPToolInfo(
            name="click_element",
            description="Click UI element by control ID",
            arguments_schema={
                "control_id": {"type": "string", "required": True},
                "process_name": {"type": "string", "required": True}
            }
        ),
        MCPToolInfo(
            name="type_text",
            description="Type text into focused element",
            arguments_schema={
                "text": {"type": "string", "required": True}
            }
        ),
        # ... more tools
    ]
)

# Strategy requests available tools
tools_cmd = Command(
    function="get_mcp_tools",
    arguments={"server_name": "ufo_windows"}
)

results = await dispatcher.execute_commands([tools_cmd])

# LLM receives tool registry
available_tools = results[0].result  # List[MCPToolInfo]

class CommandRouter:
    """Routes and validates commands on device client"""

    async def execute(
        self,
        agent_name: str,
        root_name: str,
        process_name: str,
        commands: List[Command]
    ) -> List[Result]:
        """Execute commands with validation"""
        results = []

        for command in commands:
            # Validate command
            validation_error = self._validate_command(command)
            if validation_error:
                results.append(Result(
                    status=ResultStatus.FAILURE,
                    error=validation_error,
                    call_id=command.call_id
                ))
                continue

            # Execute command
            try:
                result = await self._execute_single_command(command)
                results.append(result)
            except Exception as e:
                results.append(Result(
                    status=ResultStatus.FAILURE,
                    error=str(e),
                    call_id=command.call_id
                ))

        return results

    def _validate_command(self, command: Command) -> Optional[str]:
        """Validate command structure and arguments"""
        # Check function exists
        tool_info = self.mcp_server_manager.get_tool(command.function)
        if not tool_info:
            return f"Unknown command: {command.function}"

        # Check required arguments
        schema = tool_info.arguments_schema
        for arg_name, arg_spec in schema.items():
            if arg_spec.get("required") and arg_name not in command.arguments:
                return f"Missing required argument: {arg_name}"

        # Check argument types
        for arg_name, arg_value in command.arguments.items():
            expected_type = schema.get(arg_name, {}).get("type")
            if expected_type and not self._check_type(arg_value, expected_type):
                return f"Argument '{arg_name}' has wrong type"

        return None  # Valid

{
    "name": "click_element",
    "description": "Click UI element by control ID",
    "arguments_schema": {
        "control_id": {
            "type": "string",
            "required": True,
            "description": "Unique identifier of control to click"
        },
        "process_name": {
            "type": "string",
            "required": True,
            "description": "Process name of application"
        },
        "double_click": {
            "type": "boolean",
            "required": False,
            "default": False,
            "description": "Whether to double-click"
        }
    }
}

# Execute single command
async def execute(self, agent, context):
    command = Command(
        function="screenshot",
        arguments={"region": "fullscreen"}
    )

    results = await context.command_dispatcher.execute_commands([command])

    if results[0].status == ResultStatus.SUCCESS:
        screenshot = results[0].result
        context.update_local({"screenshot": screenshot})
        return ProcessingResult(success=True, data={"screenshot": screenshot})
    else:
        return ProcessingResult(success=False, error=results[0].error)

# Execute multiple commands in parallel
async def execute(self, agent, context):
    commands = [
        Command(function="screenshot", arguments={}),
        Command(function="get_ui_tree", arguments={"process_name": "EXCEL.EXE"}),
        Command(function="get_system_info", arguments={"info_type": "memory"})
    ]

    results = await context.command_dispatcher.execute_commands(commands)

    # Process results
    data = {
        "screenshot": results[0].result if results[0].status == ResultStatus.SUCCESS else None,
        "ui_tree": results[1].result if results[1].status == ResultStatus.SUCCESS else None,
        "memory_info": results[2].result if results[2].status == ResultStatus.SUCCESS else None
    }

    return ProcessingResult(success=True, data=data)

# Execute command based on LLM decision
async def execute(self, agent, context):
    parsed_response = context.require_local("parsed_response")
    action = parsed_response.get("ControlText")

    if action == "click":
        command = Command(
            function="click_element",
            arguments={"control_id": parsed_response.get("ControlID")}
        )
    elif action == "type":
        command = Command(
            function="type_text",
            arguments={"text": parsed_response.get("InputText")}
        )
    else:
        return ProcessingResult(success=False, error=f"Unknown action: {action}")

    results = await context.command_dispatcher.execute_commands([command])
    return ProcessingResult(success=results[0].status == ResultStatus.SUCCESS)

# Retry command on failure
async def execute(self, agent, context):
    command = Command(
        function="click_element",
        arguments={"control_id": "Button_123"}
    )

    max_retries = 3
    for attempt in range(max_retries):
        results = await context.command_dispatcher.execute_commands([command])

        if results[0].status == ResultStatus.SUCCESS:
            return ProcessingResult(success=True, data=results[0].result)

        # Retry with exponential backoff
        await asyncio.sleep(2 ** attempt)

    return ProcessingResult(success=False, error="Max retries exceeded")

# Location: ufo/module/dispatcher.py
class BasicCommandDispatcher(ABC):
    """Abstract base class for command dispatcher handling."""

    @abstractmethod
    async def execute_commands(
        self, commands: List[Command], timeout: float = 6000
    ) -> Optional[List[Result]]:
        """Execute commands and return results."""
        pass

# Location: ufo/module/dispatcher.py
class LocalCommandDispatcher(BasicCommandDispatcher):
    """Command dispatcher for local execution (testing/development)."""
    pass

# Location: ufo/module/dispatcher.py
class WebSocketCommandDispatcher(BasicCommandDispatcher):
    """Command dispatcher using WebSocket/AIP protocol."""
    pass