diff --git a/docs/en/api-guides/tools/idf-py.rst b/docs/en/api-guides/tools/idf-py.rst index f5688c99886..648c565d1e3 100644 --- a/docs/en/api-guides/tools/idf-py.rst +++ b/docs/en/api-guides/tools/idf-py.rst @@ -337,8 +337,9 @@ The MCP server provides the following tools: - ``flash project``: Flash the built project to a connected device. Specify it by port name - ``clean project``: Clean build artifacts - ``create project``: Create a new ESP-IDF project from the sample template. Can be used before any project exists +- ``monitor device``: Run a scripted serial monitor session against a flashed device and wait for expected output. See :ref:`mcp-monitor-device` -All tools accept an optional ``project_dir`` argument. When omitted, the tool operates on the directory configured at startup (``-C`` flag or ``IDF_MCP_WORKSPACE_FOLDER``). You can instruct the AI model to use a specific project directory explicitly, for example when working with multiple projects or when no default project was configured at startup. +The project tools accept an optional ``project_dir`` argument. When omitted, the tool operates on the directory configured at startup (``-C`` flag or ``IDF_MCP_WORKSPACE_FOLDER``). You can instruct the AI model to use a specific project directory explicitly, for example when working with multiple projects or when no default project was configured at startup. The ``create project`` and ``monitor device`` tools are the exception: the former takes the parent ``path`` for the new project, and the latter talks to a device rather than to a project directory. The MCP server also provides these resources: @@ -346,6 +347,15 @@ The MCP server also provides these resources: - ``project://status``: Get current project build status and artifacts - ``project://devices``: Get list of connected devices +.. _mcp-monitor-device: + +Monitoring a Device +^^^^^^^^^^^^^^^^^^^ + +The ``monitor device`` tool lets an AI assistant observe what a flashed device prints. Ask in ordinary language, for example "flash it and check that the device prints ``Minimum free heap size`` within 30 seconds". The assistant waits for that output and then tells you whether it appeared. + +The full serial log is kept in a file. Ask the assistant to search that log if you want more detail than the short status it reports. + Adding ESP-IDF MCP Server to IDEs and AI agents ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ diff --git a/docs/zh_CN/api-guides/tools/idf-py.rst b/docs/zh_CN/api-guides/tools/idf-py.rst index c612bb6b0ab..78226001736 100644 --- a/docs/zh_CN/api-guides/tools/idf-py.rst +++ b/docs/zh_CN/api-guides/tools/idf-py.rst @@ -337,8 +337,9 @@ MCP 服务器提供以下工具： - ``flash project``：将已构建的项目烧录到已连接的设备，通过端口名称进行指定 - ``clean project``：清理构建产物 - ``create project``：基于示例模板创建新的 ESP-IDF 项目，可在尚无项目时使用 +- ``monitor device``：在已烧录的设备上运行一次脚本化的串口监视会话，并等待预期的输出。参见 :ref:`mcp-monitor-device` -所有工具都接受可选的 ``project_dir`` 参数。当省略该参数时，工具将默认使用启动时配置的目录（该目录可通过 ``-C`` 参数或 ``IDF_MCP_WORKSPACE_FOLDER`` 环境变量指定）。你可以要求 AI 模型明确指定某个项目目录，例如当同时处理多个项目，或启动时未配置默认项目的情况下。 +项目工具接受可选的 ``project_dir`` 参数。当省略该参数时，工具将默认使用启动时配置的目录（该目录可通过 ``-C`` 参数或 ``IDF_MCP_WORKSPACE_FOLDER`` 环境变量指定）。在同时处理多个项目，或启动时未配置默认项目的情况下，你可以明确指示 AI 模型使用某个特定的项目目录。``create project`` 和 ``monitor device`` 工具是例外：前者使用新项目的上级 ``path``；后者与设备交互，而不是针对项目目录。 同时提供以下资源： @@ -346,6 +347,15 @@ MCP 服务器提供以下工具： - ``project://status``：获取当前项目的构建状态和构建产物 - ``project://devices``：获取已连接的设备列表 +.. _mcp-monitor-device: + +监视设备 +^^^^^^^^ + +AI 助手可以通过 ``monitor device`` 工具观察已烧录设备的打印输出，只需用自然语言提问即可。例如“烧录并检查设备是否在 30 秒内打印 ``Minimum free heap size``”。AI 助手会等待该输出，然后告知你结果。 + +完整的串口日志会保存在一个文件中。如果你需要了解详细信息，可以让 AI 助手搜索日志文件。 + 将 ESP-IDF MCP 服务器添加至 IDE 和 AI 智能体 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ diff --git a/tools/idf_py_actions/mcp_ext.py b/tools/idf_py_actions/mcp_ext.py index 16588f2e16f..3879b70695b 100644 --- a/tools/idf_py_actions/mcp_ext.py +++ b/tools/idf_py_actions/mcp_ext.py @@ -2,16 +2,23 @@ # SPDX-License-Identifier: Apache-2.0 import json +import math import os import subprocess import sys +import tempfile +import time from pathlib import Path +from typing import Annotated from typing import Any +from esp_idf_monitor.base.constants import EXIT_EXPECT_TIMEOUT +from esp_idf_monitor.base.constants import EXIT_SCRIPT_ERROR from rich_click import Context from idf_py_actions.errors import FatalError from idf_py_actions.tools import PropertyDict +from idf_py_actions.tools import get_sdkconfig_value from idf_py_actions.tools import get_target from idf_py_actions.tools import idf_version @@ -39,6 +46,37 @@ except ImportError: MCP_AVAILABLE = False +try: + # Field() feeds per-argument descriptions into a tool's inputSchema (see the + # Annotated[...] argument hints on monitor_device below). + # Annotated evaluates Field(...) at import time, and this whole module is imported + # unconditionally by idf.py on every run, regardless of the command invoked - so without + # this fallback, idf.py would break entirely on installs that lack the optional + # "mcp" feature. pydantic ships as a dependency of mcp, so this import only + # fails when mcp itself is not installed either (see MCP_AVAILABLE above). + from pydantic import Field +except ImportError: + + def Field(**kwargs: Any) -> Any: # type: ignore[misc] # noqa: N802 + return None + + +# Root folder (inside the OS temp directory) that ESP-IDF MCP tool logs are +# written under, one subfolder per tool. +MCP_LOG_ROOT_DIR_NAME = 'esp_idf_mcp_log' + +# Bounded wait injected into 'expect' lines that arrive without --timeout. +DEFAULT_MONITOR_TIMEOUT_SEC = 20.0 +# Whole-script cap: sum of expect and sleep durations. +MONITOR_MAX_SCRIPT_SEC = 600.0 +# Commands the MCP monitor script may contain. Everything else (flash, log, ...) +# is rejected so this tool cannot trigger a rebuild or toggle monitor UI state. +MONITOR_SCRIPT_COMMANDS = frozenset({'expect', 'send', 'sleep', 'reset', 'bootloader', 'exit'}) +# Size limit for text embedded directly in a tool result, in characters. Serial +# output belongs in the log file, not in the agent's context. +MONITOR_TAIL_CHARS = 1500 + + def _is_valid_project_dir(directory: str) -> bool: """ Determine if the given directory is a valid ESP-IDF project directory. @@ -71,6 +109,160 @@ def _is_valid_project_dir(directory: str) -> bool: return False +def _derive_build_dir(effective_dir: str, launch_dir: str, args: PropertyDict) -> str: + """Use idf.py's build_dir when it belongs to this project; otherwise ``/build``.""" + if launch_dir == effective_dir: + build_dir = args.get('build_dir', '') + if build_dir: + return str(build_dir) + return os.path.join(effective_dir, 'build') + + +def _load_project_description( + launch_dir: str, args: PropertyDict +) -> tuple[str | None, str | None, dict[str, Any] | None]: + """Resolve project dir, build dir, and ``project_description.json`` without failing. + + Returns ``(project_dir, build_dir, description)``. Any piece that cannot be + resolved is ``None``. Missing project, build directory, or description is not + an error. + """ + project_dir = resolve_default_project_dir(launch_dir) + if project_dir is None: + return None, None, None + build_dir = _derive_build_dir(project_dir, launch_dir, args) + if not os.path.exists(build_dir): + return project_dir, None, None + desc_path = os.path.join(build_dir, 'project_description.json') + try: + with open(desc_path, encoding='utf-8') as f: + description = json.load(f) + except (OSError, ValueError): + return project_dir, build_dir, None + if not isinstance(description, dict): + return project_dir, build_dir, None + return project_dir, build_dir, description + + +def _monitor_normalize_expect_line(line: str, default_expect_duration: float) -> tuple[str, float | None]: + """Inject '--timeout default_expect_duration' into a bare 'expect' line. + + Returns (line, expect_duration): expect_duration is the bound this line + ends up with, or None if the line already carries a --timeout the monitor + will reject - left untouched so the monitor reports the script error + itself. Timeout syntax is parsed the same way CommandReader parses it. + """ + command, _, argument = line.partition(' ') + argument = argument.strip() + if argument.startswith('--timeout'): + parts = argument.split(None, 2) + try: + expect_duration = float(parts[1]) + except (IndexError, ValueError): + return line, None + if not math.isfinite(expect_duration) or expect_duration <= 0: + return line, None + return line, expect_duration + return ( + ' '.join(filter(None, [command, '--timeout', f'{default_expect_duration:g}', argument])), + default_expect_duration, + ) + + +def _monitor_parse_sleep_duration(line: str) -> float | None: + """Return the duration of a 'sleep' line, or None if the monitor would + reject it (such a line is skipped, so it costs no time).""" + argument = line.partition(' ')[2].strip() + try: + sleep_duration = float(argument) + except ValueError: + return None + if not math.isfinite(sleep_duration) or sleep_duration <= 0: + return None + return sleep_duration + + +def _save_monitor_output(output: str) -> str: + """Save the monitor's output to a log file and describe where it went, or + inline a tail on failure - keeps the full serial dump out of the agent's context. + + *output* is the monitor's merged stdout and stderr: the chip prints serial + data on stdout while the monitor's own '--- ' lines (decoded panic + backtraces among them) go to stderr, so both streams are captured through + one pipe to keep a decoded backtrace next to the raw one that triggered it. + """ + log_dir = Path(tempfile.gettempdir()) / MCP_LOG_ROOT_DIR_NAME / 'action_monitor' + log_path = log_dir / f'monitor_{time.strftime("%Y%m%d_%H%M%S")}.log' + try: + log_dir.mkdir(parents=True, exist_ok=True) + log_path.write_text(output, encoding='utf-8', errors='replace') + # Path on its own line so an agent cannot paraphrase the character + # count and drop the location. + return f'Log file: {log_path}\nMonitor output: {len(output)} characters. Read or grep that file.' + except OSError as e: + return f'Log file could not be written ({e}), so here is its tail instead:\n{tail(output, MONITOR_TAIL_CHARS)}' + + +def _monitor_status(returncode: int) -> str: + """Explain a monitor exit code, so the agent does not have to scrape stderr.""" + if returncode == 0: + return 'Monitor script completed successfully (exit code 0).' + if returncode == EXIT_EXPECT_TIMEOUT: + return ( + f'An expect pattern was not seen before its --timeout elapsed (exit code {returncode}). ' + 'The monitor aborted the rest of the script.' + ) + if returncode == EXIT_SCRIPT_ERROR: + return ( + f'The monitor rejected the script (exit code {returncode}): invalid expect syntax, ' + 'timeout value or regular expression.' + ) + return f'The monitor exited with code {returncode}.' + + +def _monitor_project_args(baud: str | int | None, launch_dir: str, args: PropertyDict) -> list[str]: + """Baud, optional decode metadata, and ELF files derived from the project build.""" + extra: list[str] = [] + _, build_dir, description = _load_project_description(launch_dir, args) + + # baud: explicit argument, else monitor_baud from the description + if baud is not None: + extra += ['-b', str(baud)] + elif description is not None and description.get('monitor_baud') is not None: + extra += ['-b', str(description['monitor_baud'])] + + if description is None or not build_dir: + return extra + + # ELF files (app ELF first, same order as idf.py monitor) + elf_file = os.path.join(build_dir, description.get('app_elf', '') or '') + elf_list = [str(elf) for elf in Path(build_dir).rglob('*.elf')] + if elf_file and elf_file in elf_list: + elf_list.insert(0, elf_list.pop(elf_list.index(elf_file))) + # without ELF files, metadata has no additional value to monitor + if not elf_list: + return extra + + # toolchain / target / decode flags (only useful with ELFs) + toolchain_prefix = description.get('monitor_toolprefix') + if toolchain_prefix: + extra += ['--toolchain-prefix', str(toolchain_prefix)] + config_file = str(description.get('config_file') or '') + coredump_decode = get_sdkconfig_value(config_file, 'CONFIG_ESP_COREDUMP_DECODE') + if coredump_decode is not None: + extra += ['--decode-coredumps', coredump_decode] + target = description.get('target') + if target: + extra += ['--target', str(target)] + revision = description.get('min_rev') + if revision: + extra += ['--revision', str(revision)] + if get_sdkconfig_value(config_file, 'CONFIG_IDF_TARGET_ARCH_RISCV'): + extra += ['--decode-panic', 'backtrace'] + extra += elf_list + return extra + + def resolve_default_project_dir(launch_dir: str) -> str | None: """ Returns the first valid ESP-IDF project directory from the server's launch @@ -110,6 +302,88 @@ def resolve_tool_project_dir(explicit_dir: str | None, launch_dir: str) -> tuple ) +def assemble_monitor_script_from_agent_commands( + commands: str, + timeout_sec: float = DEFAULT_MONITOR_TIMEOUT_SEC, +) -> tuple[str, float]: + """Frame the agent's command body into a runnable monitor script. + + Bounds every 'expect' with --timeout (default *timeout_sec*) and appends + 'exit' if missing, so the monitor always terminates on its own. A leading + 'reset' is not added: the monitor already resets the chip when it opens the + port. Agent-supplied 'reset' lines are left in place. + + Returns (script, effective_timeout). Raises ValueError if *commands* has no + command at all, if a line is not one of MONITOR_SCRIPT_COMMANDS, or if the + script would run longer than MONITOR_MAX_SCRIPT_SEC. + + effective_timeout is the sum of every bounded expect duration plus every + sleep duration. If the script has neither (for example only 'send'), + *timeout_sec* is used instead so the monitor process still has a kill bound. + """ + lines: list[str] = [] + effective_timeout = 0.0 + has_command = False + allowed = ', '.join(sorted(MONITOR_SCRIPT_COMMANDS)) + + for raw_line in commands.splitlines(): + line = raw_line.strip() + if not line: + continue + if line.startswith('#'): + lines.append(line) + continue + command = line.partition(' ')[0].lower() + if command not in MONITOR_SCRIPT_COMMANDS: + raise ValueError(f'Unsupported monitor command {command!r}. Allowed commands: {allowed}.') + has_command = True + if command == 'expect': + line, expect_duration = _monitor_normalize_expect_line(line, timeout_sec) + if expect_duration is not None: + effective_timeout += expect_duration + elif command == 'sleep': + sleep_duration = _monitor_parse_sleep_duration(line) + if sleep_duration is not None: + effective_timeout += sleep_duration + lines.append(line) + + if not has_command: + raise ValueError( + 'No monitor commands given. Pass a commands body with at least one command, ' + 'for example: expect --timeout 20 Hello world!' + ) + + if not lines or lines[-1].lower() != 'exit': + lines.append('exit') + + if effective_timeout <= 0: + effective_timeout = timeout_sec + if effective_timeout > MONITOR_MAX_SCRIPT_SEC: + raise ValueError(f'This script would run for {effective_timeout:g}s. The limit is {MONITOR_MAX_SCRIPT_SEC:g}s.') + return '\n'.join(lines) + '\n', effective_timeout + + +def decode_stream(stream: Any) -> str: + """Decode a subprocess stream that may be bytes even in text mode. + + TimeoutExpired carries the output collected so far as bytes on POSIX, while + a completed run in text mode yields str. + """ + if stream is None: + return '' + if isinstance(stream, bytes): + return stream.decode('utf-8', errors='replace') + return str(stream) + + +def tail(text: str, limit: int) -> str: + """Return at most the last *limit* characters of *text*, marking a cut.""" + text = text.strip() + if len(text) <= limit: + return text + return f'[...truncated...]\n{text[-limit:]}' + + def action_extensions(base_actions: dict, project_path: str) -> dict: """ESP-IDF MCP Server Extension""" @@ -274,6 +548,126 @@ def action_extensions(base_actions: dict, project_path: str) -> dict: print(f'ERROR: Flash failed: {str(e)}', file=sys.stderr) return f'Error flashing: {str(e)}' + @mcp.tool( + description=( + 'Run a serial monitor session against a flashed device, driven by a script ' + '(runs `python -m esp_idf_monitor` in non-interactive command mode). ' + 'Use this to observe what a device prints and to wait for specific output. ' + 'Write the commands body from the project context or from the exact instructions ' + 'of the user; allowed commands are expect, send, sleep, reset, bootloader and exit. ' + 'The monitor already resets the chip when it opens the port, so do not start the ' + 'script with reset unless you need an extra reset later. Set no_reset to skip that ' + 'connection reset. ALWAYS end the script with an explicit exit as its last line. ' + 'Every expect should be bounded: "expect --timeout ". ' + f'The whole script may take at most {MONITOR_MAX_SCRIPT_SEC:g} seconds. ' + 'Exit code 0 means the script finished, 110 means an expect pattern never appeared, ' + '2 means the script itself was invalid. ' + 'Returns a short status plus a Log file: line with the absolute path of the monitor ' + 'log, which holds the serial output together with the monitor messages. ' + 'Always quote that path when answering the user; read or ' + 'grep the file instead of expecting the whole log inline. ' + 'The port is optional and is autodetected when omitted; see project://devices ' + 'when several devices are connected.' + ) + ) + def monitor_device( + commands: Annotated[ + str, + Field( + description=( + 'Monitor script body, one command per line (expect, send, sleep, reset, bootloader, exit). ' + 'Bound every wait as "expect --timeout " (default 20 seconds). ' + f'The whole script may take at most {MONITOR_MAX_SCRIPT_SEC:g} seconds. ' + 'A reset at the start is unnecessary: the monitor resets on connect unless no_reset. ' + 'Leave any later reset in the script. End with exit; the server appends one if missing. ' + 'Lines starting with # are comments.' + ) + ), + ], + timeout_sec: Annotated[ + float, + Field( + description=( + 'Wait, in seconds, injected into expect lines that omit --timeout. ' + f'The whole script may take at most {MONITOR_MAX_SCRIPT_SEC:g} seconds, ' + 'the sum of expect and sleep durations. The monitor process is killed after ' + 'twice that time. If the script has no expect or sleep, twice timeout_sec is used.' + ) + ), + ] = DEFAULT_MONITOR_TIMEOUT_SEC, + port: str | None = None, + baud: str | int | None = None, + no_reset: bool = False, + ) -> str: + """Run a scripted esp-idf-monitor session and return a status plus a log path. + + Args: + port: Optional serial port such as /dev/ttyUSB0 or COM3. Leave as + None to let the monitor autodetect it. + baud: Optional monitor baud rate. When omitted, ``monitor_baud`` + from ``project_description.json`` is used if that file exists. + no_reset: Pass --no-reset so the monitor does not reset the chip + when it opens the port. Agent-supplied reset commands in the + script are left unchanged. + """ + if not math.isfinite(timeout_sec) or timeout_sec <= 0: + return f'timeout_sec must be a finite number greater than 0, got {timeout_sec}.' + + try: + script, effective_timeout = assemble_monitor_script_from_agent_commands( + commands, timeout_sec=timeout_sec + ) + except ValueError as e: + return str(e) + hard_timeout = 2 * effective_timeout + + cmd = [sys.executable, '-m', 'esp_idf_monitor'] + if port: + cmd.extend(['-p', port]) + cmd.extend(_monitor_project_args(baud, project_path, args)) + if no_reset: + cmd.append('--no-reset') + + print(f'INFO: Running monitor: {" ".join(cmd)} (hard timeout {hard_timeout:g}s)', file=sys.stderr) + print(f'INFO: Monitor script:\n{script}', file=sys.stderr) + + try: + result = subprocess.run( + cmd, + input=script, + stdout=subprocess.PIPE, + # One pipe for both streams keeps the monitor's decoded backtraces + # in order with the serial lines they belong to. + stderr=subprocess.STDOUT, + text=True, + timeout=hard_timeout, + ) + except subprocess.TimeoutExpired as e: + print(f'ERROR: Monitor killed after {hard_timeout:g}s', file=sys.stderr) + output = decode_stream(e.output) + parts = [ + f'The monitor did not exit on its own and was killed after {hard_timeout:g} seconds ' + '(twice the time the script should have taken). ' + 'The serial output captured until then was kept.', + _save_monitor_output(output), + ] + output_tail = tail(output, MONITOR_TAIL_CHARS) + if output_tail: + parts.append(f'Last output before the kill:\n{output_tail}') + return '\n'.join(parts) + except Exception as e: + print(f'ERROR: Monitor failed to run: {e}', file=sys.stderr) + return f'Failed to run the monitor: {e}' + + print(f'INFO: Monitor exited with code {result.returncode}', file=sys.stderr) + output = result.stdout or '' + parts = [_monitor_status(result.returncode), _save_monitor_output(output)] + if result.returncode != 0: + output_tail = tail(output, MONITOR_TAIL_CHARS) + if output_tail: + parts.append(f'Last output before the exit:\n{output_tail}') + return '\n'.join(parts) + @mcp.tool( description=( 'Create a new ESP-IDF project from the sample template (runs `idf.py create-project`). ' @@ -354,7 +748,7 @@ def action_extensions(base_actions: dict, project_path: str) -> dict: @mcp.resource('project://config') def get_project_config() -> str: """Get current project configuration""" - effective_dir = resolve_default_project_dir(project_path) + effective_dir, build_dir, description = _load_project_description(project_path, args) config: dict[str, Any] = {} if effective_dir is None: @@ -365,24 +759,15 @@ def action_extensions(base_actions: dict, project_path: str) -> dict: return json.dumps(config, indent=2) config['project_path'] = effective_dir - # Use the build_dir from idf.py args when the project matches; otherwise derive it. - build_dir = ( - args.get('build_dir', '') if project_path == effective_dir else os.path.join(effective_dir, 'build') - ) - - if not os.path.exists(build_dir): + if build_dir is None: config['build_dir_exists'] = False return json.dumps(config, indent=2) config['build_dir'] = build_dir - proj_desc_fn = os.path.join(build_dir, 'project_description.json') - config['project_description'] = 'Project description does not exist' - - try: - with open(proj_desc_fn, encoding='utf-8') as f: - config['project_description'] = json.load(f) - except (OSError, ValueError): - pass + if description is not None: + config['project_description'] = description + else: + config['project_description'] = 'Project description does not exist' return json.dumps(config, indent=2) @@ -405,10 +790,7 @@ def action_extensions(base_actions: dict, project_path: str) -> dict: status['target'] = get_target(effective_dir) status['idf_version'] = idf_version() - # Use the build_dir from idf.py args when the project matches; otherwise derive it. - build_dir = ( - args.get('build_dir', '') if project_path == effective_dir else os.path.join(effective_dir, 'build') - ) + build_dir = _derive_build_dir(effective_dir, project_path, args) if os.path.exists(build_dir): status['build_dir'] = build_dir artifacts = ['bootloader', 'partition_table', 'app-flash', 'flash_args'] diff --git a/tools/test_idf_py/test_mcp_ext.py b/tools/test_idf_py/test_mcp_ext.py index 638e17978b7..54287880704 100644 --- a/tools/test_idf_py/test_mcp_ext.py +++ b/tools/test_idf_py/test_mcp_ext.py @@ -11,7 +11,9 @@ import importlib import importlib.util import json import os +import subprocess import sys +import tempfile import types from collections.abc import Callable from pathlib import Path @@ -106,6 +108,7 @@ def mcp_ext(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[types.Modu tools_mod.PropertyDict = dict # type: ignore[attr-defined] tools_mod.get_target = mock.Mock(return_value='esp32') # type: ignore[attr-defined] tools_mod.idf_version = mock.Mock(return_value='5.4.0') # type: ignore[attr-defined] + tools_mod.get_sdkconfig_value = mock.Mock(return_value=None) # type: ignore[attr-defined] idf_py_actions_pkg.errors = errors_mod # type: ignore[attr-defined] idf_py_actions_pkg.tools = tools_mod # type: ignore[attr-defined] @@ -115,6 +118,13 @@ def mcp_ext(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[types.Modu mcp_server_pkg = types.ModuleType('mcp.server') mcp_server_pkg.MCPServer = lambda name: mock_mcp_instance # type: ignore[attr-defined] + # Stub monitor exit codes (same values as esp_idf_monitor.base.constants). + esp_idf_monitor_pkg = types.ModuleType('esp_idf_monitor') + esp_idf_monitor_base = types.ModuleType('esp_idf_monitor.base') + esp_idf_monitor_constants = types.ModuleType('esp_idf_monitor.base.constants') + esp_idf_monitor_constants.EXIT_EXPECT_TIMEOUT = 110 # type: ignore[attr-defined] + esp_idf_monitor_constants.EXIT_SCRIPT_ERROR = 2 # type: ignore[attr-defined] + stubs = { 'rich_click': rich_click, 'idf_py_actions': idf_py_actions_pkg, @@ -122,6 +132,9 @@ def mcp_ext(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[types.Modu 'idf_py_actions.tools': tools_mod, 'mcp': mcp_pkg, 'mcp.server': mcp_server_pkg, + 'esp_idf_monitor': esp_idf_monitor_pkg, + 'esp_idf_monitor.base': esp_idf_monitor_base, + 'esp_idf_monitor.base.constants': esp_idf_monitor_constants, } for name, stub_mod in stubs.items(): monkeypatch.setitem(sys.modules, name, stub_mod) @@ -176,13 +189,17 @@ class TestIsValidProjectDir: mod, _ = mcp_ext assert mod._is_valid_project_dir(str(tmp_path / 'nonexistent')) is False - def test_directory_without_cmakelists(self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockMCPServer]) -> None: + def test_directory_without_cmakelists( + self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockMCPServer] + ) -> None: mod, _ = mcp_ext d = tmp_path / 'no_cmake' d.mkdir() assert mod._is_valid_project_dir(str(d)) is False - def test_cmakelists_without_idf_line(self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockMCPServer]) -> None: + def test_cmakelists_without_idf_line( + self, tmp_path: Path, mcp_ext: tuple[types.ModuleType, _MockMCPServer] + ) -> None: mod, _ = mcp_ext proj = _make_invalid_project(tmp_path / 'plain') assert mod._is_valid_project_dir(str(proj)) is False @@ -481,6 +498,387 @@ class TestCleanProject: assert cmd[cmd.index('-C') + 1] == str(proj) +# --------------------------------------------------------------------------- +# Tests: monitor script building +# --------------------------------------------------------------------------- + + +class TestBuildMonitorScript: + """Covers only what the MCP server itself is responsible for: framing the + agent's command body (exit/comments) and computing the effective + timeout. Validity of esp-idf-monitor's own 'expect --timeout' syntax is + esp-idf-monitor's job, not duplicated here.""" + + def test_commands_without_an_actual_command_are_rejected( + self, mcp_ext: tuple[types.ModuleType, _MockMCPServer] + ) -> None: + mod, _ = mcp_ext + with pytest.raises(ValueError, match='No monitor commands given'): + mod.assemble_monitor_script_from_agent_commands(' \n# just a note\n') + + def test_bare_expect_gets_default_timeout_and_framing( + self, mcp_ext: tuple[types.ModuleType, _MockMCPServer] + ) -> None: + mod, _ = mcp_ext + script, effective = mod.assemble_monitor_script_from_agent_commands('expect Hello world!') + assert script == 'expect --timeout 20 Hello world!\nexit\n' + assert effective == 20.0 + + @pytest.mark.parametrize( + 'commands, timeout_sec, expected_effective, expected_snippet', + [ + ('expect --timeout 45 ALL TESTS PASSED', 20.0, 45.0, 'expect --timeout 45 ALL TESTS PASSED'), + ( + 'expect --timeout 5 first\nexpect --timeout 90 second\nexpect third', + 30.0, + 125.0, + 'expect --timeout 30 third', + ), + ('sleep 120\nexpect ready\nsleep 30', 20.0, 170.0, 'expect --timeout 20 ready'), + ('expect --timeout abc pattern', 20.0, 20.0, 'expect --timeout abc pattern'), + ('send hello', 20.0, 20.0, 'send hello'), + ], + ids=[ + 'keeps_explicit_timeout', + 'expect_timeouts_are_summed_and_default_is_injected', + 'sleeps_are_added_to_the_sum', + 'malformed_timeout_left_to_the_monitor', + 'send_only_uses_timeout_sec_as_kill_bound', + ], + ) + def test_timeout_handling( + self, + mcp_ext: tuple[types.ModuleType, _MockMCPServer], + commands: str, + timeout_sec: float, + expected_effective: float, + expected_snippet: str, + ) -> None: + mod, _ = mcp_ext + script, effective = mod.assemble_monitor_script_from_agent_commands(commands, timeout_sec=timeout_sec) + assert expected_snippet in script + assert effective == expected_effective + + def test_script_at_max_duration_is_accepted(self, mcp_ext: tuple[types.ModuleType, _MockMCPServer]) -> None: + mod, _ = mcp_ext + _, effective = mod.assemble_monitor_script_from_agent_commands( + 'sleep 580\nexpect --timeout 20 ready', timeout_sec=20.0 + ) + assert effective == 600.0 + + @pytest.mark.parametrize( + 'commands, timeout_sec', + [ + ('sleep 99999\nexpect ready', 20.0), + ('expect --timeout 601 ready', 20.0), + ('expect --timeout 400 a\nexpect --timeout 250 b', 20.0), + ], + ids=['long_sleep', 'long_expect', 'sum_of_expects'], + ) + def test_script_over_max_duration_is_rejected( + self, mcp_ext: tuple[types.ModuleType, _MockMCPServer], commands: str, timeout_sec: float + ) -> None: + mod, _ = mcp_ext + with pytest.raises(ValueError, match='The limit is'): + mod.assemble_monitor_script_from_agent_commands(commands, timeout_sec=timeout_sec) + + @pytest.mark.parametrize( + 'commands, expected_script', + [ + ('expect uptime', 'expect --timeout 20 uptime\nexit\n'), + ('reset\nexpect done\nexit\n', 'reset\nexpect --timeout 20 done\nexit\n'), + ( + 'expect first\nreset\nexpect second', + 'expect --timeout 20 first\nreset\nexpect --timeout 20 second\nexit\n', + ), + ], + ids=['exit_appended_and_reset_not_prepended', 'agent_reset_and_exit_kept', 'mid_script_reset_kept'], + ) + def test_script_framing( + self, mcp_ext: tuple[types.ModuleType, _MockMCPServer], commands: str, expected_script: str + ) -> None: + mod, _ = mcp_ext + script, _ = mod.assemble_monitor_script_from_agent_commands(commands) + assert script == expected_script + + @pytest.mark.parametrize('command', ['flash', 'app-flash', 'log', 'output']) + def test_disallowed_commands_are_rejected( + self, mcp_ext: tuple[types.ModuleType, _MockMCPServer], command: str + ) -> None: + mod, _ = mcp_ext + with pytest.raises(ValueError, match=rf'Unsupported monitor command {command!r}'): + mod.assemble_monitor_script_from_agent_commands(f'{command}\nexpect ready') + + +# --------------------------------------------------------------------------- +# Tests: monitor_device tool +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def monitor_tools( + mcp_ext: tuple[types.ModuleType, _MockMCPServer], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> dict[str, Callable[..., Any]]: + """Registered tools with monitor logs redirected into tmp_path.""" + _, mock_mcp = mcp_ext + monkeypatch.setattr(tempfile, 'tempdir', str(tmp_path)) + tools, _ = _start_server(mcp_ext, mock_mcp, str(tmp_path)) + return tools + + +def _log_path_from(result: str) -> Path: + """Extract the log file path the tool reported.""" + for line in result.splitlines(): + if line.startswith('Log file: '): + return Path(line.split(': ', 1)[1]) + raise AssertionError(f'no serial log path in result:\n{result}') + + +class TestMonitorDevice: + """Covers what the MCP tool itself is responsible for: turning a script and + exit code into a status message, wiring subprocess/log-file plumbing, and + forwarding its own arguments. Behaviour of esp-idf-monitor's non-interactive + command mode (e.g. what makes an 'expect --timeout' line valid) is exercised + by esp-idf-monitor's own tests, not duplicated here.""" + + @pytest.mark.parametrize( + 'commands, timeout_sec, expected_snippet', + [ + (' ', 20.0, 'No monitor commands given'), + ('expect ready', 0, 'timeout_sec must be a finite number greater than 0'), + ('flash\nexpect ready', 20.0, "Unsupported monitor command 'flash'"), + ], + ids=['empty_commands', 'invalid_timeout_sec', 'disallowed_command'], + ) + def test_guard_clauses_return_error_without_running( + self, monitor_tools: dict[str, Callable[..., Any]], commands: str, timeout_sec: float, expected_snippet: str + ) -> None: + with mock.patch('subprocess.run') as mock_run: + result = monitor_tools['monitor_device'](commands=commands, timeout_sec=timeout_sec) + + assert expected_snippet in result + mock_run.assert_not_called() + + @pytest.mark.parametrize( + 'returncode, output, expected_status', + [ + (0, 'Hello world!\n', 'completed successfully (exit code 0)'), + (110, "boot\nExpect pattern 'nope' timed out after 20.0s", 'not seen before its --timeout elapsed'), + (2, 'Invalid expect timeout value: (must be a finite number > 0)', 'rejected the script'), + (1, 'could not open port', 'exited with code 1'), + ], + ids=['success', 'expect_timeout', 'script_error', 'unexpected_code'], + ) + def test_status_message_and_log_reflect_exit_code( + self, + monitor_tools: dict[str, Callable[..., Any]], + returncode: int, + output: str, + expected_status: str, + ) -> None: + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=returncode, stdout=output, stderr=None) + result = monitor_tools['monitor_device'](commands='expect ready') + + assert expected_status in result + if returncode != 0: + # Why the monitor stopped is its last output, so the tail carries it. + assert output in result + log_file = _log_path_from(result) + assert log_file.is_absolute() + assert log_file.read_text(encoding='utf-8') == output + + @pytest.mark.parametrize( + 'call_kwargs, expected_in_cmd, expected_not_in_cmd', + [ + ({'port': '/dev/ttyUSB0'}, ['-p', '/dev/ttyUSB0'], []), + ({'no_reset': True}, ['--no-reset'], []), + ], + ids=['port_forwarded', 'no_reset_passed_to_the_monitor'], + ) + def test_cli_arguments_are_forwarded( + self, + monitor_tools: dict[str, Callable[..., Any]], + call_kwargs: dict[str, Any], + expected_in_cmd: list[str], + expected_not_in_cmd: list[str], + ) -> None: + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='') + monitor_tools['monitor_device'](commands='expect ready', **call_kwargs) + + cmd = mock_run.call_args[0][0] + assert cmd[:3] == [sys.executable, '-m', 'esp_idf_monitor'] + for item in expected_in_cmd: + assert item in cmd + for item in expected_not_in_cmd: + assert item not in cmd + + def test_hard_timeout_is_twice_the_effective_timeout(self, monitor_tools: dict[str, Callable[..., Any]]) -> None: + """The effective-timeout math itself (sum of waits plus sleeps) is covered + by TestBuildMonitorScript; this only checks the tool wires 2x it in.""" + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='') + monitor_tools['monitor_device'](commands='sleep 60\nexpect --timeout 20 ready') + + assert mock_run.call_args[1]['timeout'] == 2 * 80.0 + + def test_process_timeout_keeps_partial_output(self, monitor_tools: dict[str, Callable[..., Any]]) -> None: + with mock.patch('subprocess.run') as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired( + cmd=['python', '-m', 'esp_idf_monitor'], + timeout=60.0, + output=b'partial serial output\nExpect still waiting\n', + ) + result = monitor_tools['monitor_device'](commands='expect ready') + + assert 'was killed after 40 seconds' in result + assert 'Expect still waiting' in result + logged = _log_path_from(result).read_text(encoding='utf-8') + assert logged == 'partial serial output\nExpect still waiting\n' + + def test_launch_failure_is_reported(self, monitor_tools: dict[str, Callable[..., Any]]) -> None: + with mock.patch('subprocess.run') as mock_run: + mock_run.side_effect = OSError('no such interpreter') + result = monitor_tools['monitor_device'](commands='expect ready') + + assert 'Failed to run the monitor' in result + assert 'no such interpreter' in result + + def test_log_write_failure_falls_back_to_inline_tail( + self, + monitor_tools: dict[str, Callable[..., Any]], + monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.setattr(Path, 'mkdir', mock.Mock(side_effect=OSError('disk full'))) + + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout='important tail\n', stderr='') + result = monitor_tools['monitor_device'](commands='expect ready') + + assert 'Log file could not be written (disk full)' in result + assert 'important tail' in result + + def test_long_output_is_not_inlined(self, monitor_tools: dict[str, Callable[..., Any]]) -> None: + """The whole point of the log file: a huge serial dump must not come back + through the tool result.""" + big_output = 'x' * 200000 + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout=big_output, stderr='') + result = monitor_tools['monitor_device'](commands='expect ready') + + assert len(result) < 500 + assert _log_path_from(result).read_text(encoding='utf-8') == big_output + + def test_both_streams_are_captured_through_one_pipe(self, monitor_tools: dict[str, Callable[..., Any]]) -> None: + """esp-idf-monitor prints decoded panic backtraces on stderr, so a log + built from stdout alone would lose exactly what the ELF files enable.""" + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr=None) + monitor_tools['monitor_device'](commands='expect ready') + + assert mock_run.call_args[1]['stderr'] == subprocess.STDOUT + + def test_no_baud_elf_or_metadata_when_description_absent( + self, monitor_tools: dict[str, Callable[..., Any]] + ) -> None: + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='') + monitor_tools['monitor_device'](commands='expect ready') + + cmd = mock_run.call_args[0][0] + assert '-b' not in cmd + assert '--toolchain-prefix' not in cmd + assert '--target' not in cmd + assert '--revision' not in cmd + assert '--decode-coredumps' not in cmd + assert '--decode-panic' not in cmd + assert not any(str(arg).endswith('.elf') for arg in cmd) + + def test_forwards_baud_elf_order_and_metadata_when_description_has_elfs( + self, + mcp_ext: tuple[types.ModuleType, _MockMCPServer], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + mod, mock_mcp = mcp_ext + monkeypatch.setattr(tempfile, 'tempdir', str(tmp_path)) + monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '') + proj = _make_valid_project(tmp_path / 'proj') + build_dir = proj / 'build' + boot_dir = build_dir / 'bootloader' + boot_dir.mkdir(parents=True) + app_elf = build_dir / 'hello_world.elf' + boot_elf = boot_dir / 'bootloader.elf' + app_elf.write_bytes(b'') + boot_elf.write_bytes(b'') + (build_dir / 'project_description.json').write_text( + json.dumps( + { + 'app_elf': 'hello_world.elf', + 'monitor_baud': '115200', + 'monitor_toolprefix': 'xtensa-esp32-elf-', + 'target': 'esp32', + 'min_rev': '3', + 'config_file': str(proj / 'sdkconfig'), + } + ), + encoding='utf-8', + ) + monkeypatch.setattr( + mod, + 'get_sdkconfig_value', + lambda _path, key: { + 'CONFIG_ESP_COREDUMP_DECODE': 'info', + 'CONFIG_IDF_TARGET_ARCH_RISCV': 'y', + }.get(key), + ) + tools, _ = _start_server(mcp_ext, mock_mcp, str(proj)) + + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='') + tools['monitor_device'](commands='expect ready') + + cmd = mock_run.call_args[0][0] + assert cmd[cmd.index('-b') + 1] == '115200' + elf_args = [arg for arg in cmd if str(arg).endswith('.elf')] + assert elf_args[0] == str(app_elf) + assert str(boot_elf) in elf_args + assert cmd[cmd.index('--toolchain-prefix') + 1] == 'xtensa-esp32-elf-' + assert cmd[cmd.index('--target') + 1] == 'esp32' + assert cmd[cmd.index('--revision') + 1] == '3' + assert cmd[cmd.index('--decode-coredumps') + 1] == 'info' + assert cmd[cmd.index('--decode-panic') + 1] == 'backtrace' + + def test_explicit_baud_overrides_description_monitor_baud( + self, + mcp_ext: tuple[types.ModuleType, _MockMCPServer], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + _, mock_mcp = mcp_ext + monkeypatch.setattr(tempfile, 'tempdir', str(tmp_path)) + monkeypatch.setenv('IDF_MCP_WORKSPACE_FOLDER', '') + proj = _make_valid_project(tmp_path / 'proj') + build_dir = proj / 'build' + build_dir.mkdir() + (build_dir / 'project_description.json').write_text( + json.dumps({'monitor_baud': '115200', 'app_elf': 'hello_world.elf'}), + encoding='utf-8', + ) + tools, _ = _start_server(mcp_ext, mock_mcp, str(proj)) + + with mock.patch('subprocess.run') as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout='', stderr='') + tools['monitor_device'](commands='expect ready', baud='9600') + + cmd = mock_run.call_args[0][0] + assert cmd[cmd.index('-b') + 1] == '9600' + assert '115200' not in cmd + + # --------------------------------------------------------------------------- # Test: server starts without error when project_path is not valid # --------------------------------------------------------------------------- @@ -517,6 +915,7 @@ class TestServerStartsOutsideProject: assert 'set_target' in tools assert 'flash_project' in tools assert 'clean_project' in tools + assert 'monitor_device' in tools # ---------------------------------------------------------------------------