Method _handle_vm_tool

src/mcp_cli/chat/tool_processor.py:843–962

Execute a VM tool (page_fault or search_pages) via MemoryManager. VM tools are internal memory operations that bypass the MCP ToolManager and all guard checks (dataflow tracking, $vN references, per-tool limits). Includes: - UI display so tool calls are visible in t

(
        self,
        tool_name: str,
        arguments: dict,
        llm_tool_name: str,
        call_id: str,
    )

        self._add_tool_result_to_history(llm_tool_name, call_id, result_text)

    async def _handle_vm_tool(
        self,
        tool_name: str,
        arguments: dict,
        llm_tool_name: str,
        call_id: str,
    ) -> None:
        """Execute a VM tool (page_fault or search_pages) via MemoryManager.

        VM tools are internal memory operations that bypass the MCP ToolManager
        and all guard checks (dataflow tracking, $vN references, per-tool limits).

        Includes:
        - UI display so tool calls are visible in the chat output
        - Loop prevention: refuses to re-fault the same page_id twice
        - Content truncation for oversized pages
        """
        vm = getattr(getattr(self.context, "session", None), "vm", None)
        if not vm:
            self._add_tool_result_to_history(
                llm_tool_name, call_id, "Error: VM not available."
            )
            return

        logger.info(f"VM tool {tool_name} called with args: {arguments}")

        # Show tool call in UI (so page_fault calls are visible)
        try:
            self.ui_manager.print_tool_call(tool_name, arguments)
            await self.ui_manager.start_tool_execution(tool_name, arguments)
        except Exception as e:
            logger.debug("UI error displaying VM tool call: %s", e)

        content: str | list[dict[str, Any]] = ""
        success = True
        try:
            if tool_name == "page_fault":
                page_id = arguments.get("page_id", "")

                # Loop prevention: don't re-fault the same page
                if page_id in self._faulted_page_ids:
                    content = json.dumps(
                        {
                            "success": True,
                            "already_loaded": True,
                            "page_id": page_id,
                            "message": (
                                "This page was already loaded earlier in the "
                                "conversation. The content is in a previous "
                                "tool result message — use that directly."
                            ),
                        }
                    )
                else:
                    result = await vm.handle_fault(
                        page_id=page_id,
                        target_level=arguments.get("target_level", 2),
                    )