Merge branch 'feat/mcp_server_monitor' into 'master'

feat(tools): mcp-server added monitor-device tool

Closes IDFGH-17421 and IDFGH-17857

See merge request espressif/esp-idf!51734

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
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

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 智能体
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

tools/idf_py_actions/mcp_ext.py

@@ -2,16 +2,25 @@
 # 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 esp_pylib.logger import log
+from esp_pylib.serial_ports import get_port_names
 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 +48,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.
@@ -60,7 +100,7 @@ def _is_valid_project_dir(directory: str) -> bool:
     normalised_patterns = [''.join(p.split()) for p in CMAKE_PROJECT_LINE]
 
     try:
-        with open(str(cmakelists_path), encoding='utf-8') as f:
+        with open(cmakelists_path, encoding='utf-8') as f:
             for line in f:
                 line_normalised = ''.join(line.split())
                 if any(line_normalised.startswith(pattern) for pattern in normalised_patterns):
@@ -71,6 +111,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 ``<project>/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,11 +304,95 @@ 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"""
 
     def start_mcp_server(action_name: str, ctx: Context, args: PropertyDict, **kwargs: Any) -> None:
         """Start MCP server for ESP-IDF project integration"""
+        # MCP stdio transport consumes stdout; keep informational output on stderr.
+        log.set_info_stream(sys.stderr)
 
         if not MCP_AVAILABLE:
             raise FatalError(
@@ -129,7 +407,7 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
         # the MCP client knows when (not) to pass project_dir.
         startup_default_dir = resolve_default_project_dir(project_path)
         if startup_default_dir is not None:
-            print(f'INFO: Starting ESP-IDF MCP Server. Default project: {startup_default_dir}', file=sys.stderr)
+            log.note(f'Starting ESP-IDF MCP Server. Default project: {startup_default_dir}')
             bound_hint = (
                 f"This MCP server was launched with '{startup_default_dir}' as the default ESP-IDF "
                 'project. Leave project_dir as None to operate on this project. Only set project_dir '
@@ -137,11 +415,10 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
                 'user explicitly asks to operate on a different ESP-IDF project.'
             )
         else:
-            print(
-                'INFO: Starting ESP-IDF MCP Server. No project directory configured at startup. '
+            log.note(
+                'Starting ESP-IDF MCP Server. No project directory configured at startup. '
                 'Pass project_dir in each tool call, or set IDF_MCP_WORKSPACE_FOLDER, '
-                'or restart with: idf.py -C <project_dir> mcp-server',
-                file=sys.stderr,
+                'or restart with: idf.py -C <project_dir> mcp-server'
             )
             bound_hint = (
                 'This MCP server was launched without a project context. You MUST pass project_dir '
@@ -178,16 +455,16 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
                     effective_dir,
                     'build',
                 ]
-                print(f'INFO: Building project with command: {" ".join(cmd)} in path: {effective_dir}', file=sys.stderr)
+                log.note(f'Building project with command: {" ".join(cmd)} in path: {effective_dir}')
                 result = subprocess.run(cmd, capture_output=True, text=True, stdin=subprocess.DEVNULL)
                 if result.returncode == 0:
-                    print('INFO: Build successful', file=sys.stderr)
+                    log.note('Build successful')
                     return 'Successfully built project'
                 else:
-                    print(f'ERROR: Build failed: {result.stderr}', file=sys.stderr)
+                    log.err(f'Build failed: {result.stderr}')
                     return f'Build failed: {result.stderr}'
             except Exception as e:
-                print(f'ERROR: Build failed: {str(e)}', file=sys.stderr)
+                log.err(f'Build failed: {e}')
                 return f'Build failed: {str(e)}'
 
         @mcp.tool(
@@ -219,16 +496,16 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
                     'set-target',
                     target,
                 ]
-                print(f'INFO: Setting target with command: {" ".join(cmd)} in path: {effective_dir}', file=sys.stderr)
+                log.note(f'Setting target with command: {" ".join(cmd)} in path: {effective_dir}')
                 result = subprocess.run(cmd, capture_output=True, text=True, stdin=subprocess.DEVNULL)
                 if result.returncode == 0:
-                    print(f'INFO: Target set to: {target}', file=sys.stderr)
+                    log.note(f'Target set to: {target}')
                     return f'Target set to: {target}'
                 else:
-                    print(f'ERROR: Failed to set target: {result.stderr}', file=sys.stderr)
+                    log.err(f'Failed to set target: {result.stderr}')
                     return f'Failed to set target: {result.stderr}'
             except Exception as e:
-                print(f'ERROR: Failed to set target: {str(e)}', file=sys.stderr)
+                log.err(f'Failed to set target: {e}')
                 return f'Error setting target: {str(e)}'
 
         @mcp.tool(
@@ -261,19 +538,139 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
                     '-C',
                     effective_dir,
                 ] + flash_args
-                print(f'INFO: Flashing project with command: {" ".join(cmd)} in path: {effective_dir}', file=sys.stderr)
+                log.note(f'Flashing project with command: {" ".join(cmd)} in path: {effective_dir}')
                 result = subprocess.run(cmd, capture_output=True, text=True, stdin=subprocess.DEVNULL)
 
                 if result.returncode == 0:
-                    print('INFO: Flash successful', file=sys.stderr)
+                    log.note('Flash successful')
                     return f'Successfully flashed project{" to port " + port if port else ""}'
                 else:
-                    print(f'ERROR: Flash failed: {result.stderr}', file=sys.stderr)
+                    log.err(f'Flash failed: {result.stderr}')
                     return f'Flash failed: {result.stderr}'
             except Exception as e:
-                print(f'ERROR: Flash failed: {str(e)}', file=sys.stderr)
+                log.err(f'Flash failed: {e}')
                 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 <seconds> <regex>". '
+                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 <seconds> <regex>" (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')
+
+            log.note(f'Running monitor: {" ".join(cmd)} (hard timeout {hard_timeout:g}s)')
+            log.note(f'Monitor script:\n{script}')
+
+            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:
+                log.note(f'Monitor killed after {hard_timeout:g}s')
+                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:
+                log.err(f'Monitor failed to run: {e}')
+                return f'Failed to run the monitor: {e}'
+
+            log.note(f'Monitor exited with code {result.returncode}')
+            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`). '
@@ -302,18 +699,18 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
                     'create-project',
                     name,
                 ]
-                print(f'INFO: Creating project "{name}" in {parent_dir}', file=sys.stderr)
+                log.note(f'Creating project "{name}" in {parent_dir}')
                 result = subprocess.run(cmd, capture_output=True, text=True, stdin=subprocess.DEVNULL)
                 if result.returncode == 0:
                     project_path_new = os.path.join(parent_dir, name)
-                    print(f'INFO: Project "{name}" created at {project_path_new}', file=sys.stderr)
+                    log.note(f'Project "{name}" created at {project_path_new}')
                     return f'Project "{name}" created at {project_path_new}'
                 else:
                     output = result.stderr or result.stdout
-                    print(f'ERROR: Failed to create project: {output}', file=sys.stderr)
+                    log.err(f'Failed to create project: {output}')
                     return f'Failed to create project "{name}": {output}'
             except Exception as e:
-                print(f'ERROR: Failed to create project: {str(e)}', file=sys.stderr)
+                log.err(f'Failed to create project: {e}')
                 return f'Failed to create project "{name}": {str(e)}'
 
         @mcp.tool(description=f'Remove build artifacts from the ESP-IDF project (runs `idf.py clean`). {bound_hint}')
@@ -338,23 +735,23 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
                     effective_dir,
                     'clean',
                 ]
-                print(f'INFO: Cleaning project with command: {" ".join(cmd)} in path: {effective_dir}', file=sys.stderr)
+                log.note(f'Cleaning project with command: {" ".join(cmd)} in path: {effective_dir}')
                 result = subprocess.run(cmd, capture_output=True, text=True, stdin=subprocess.DEVNULL)
                 if result.returncode == 0:
-                    print('INFO: Project cleaned successfully', file=sys.stderr)
+                    log.note('Project cleaned successfully')
                     return 'Project cleaned successfully'
                 else:
-                    print(f'ERROR: Clean failed: {result.stderr}', file=sys.stderr)
+                    log.err(f'Clean failed: {result.stderr}')
                     return f'Clean failed: {result.stderr}'
             except Exception as e:
-                print(f'ERROR: Error cleaning: {str(e)}', file=sys.stderr)
+                log.err(f'Error cleaning: {e}')
                 return f'Error cleaning: {str(e)}'
 
         # === RESOURCES (Data Access) ===
         @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 +762,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 +793,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']
@@ -428,10 +813,8 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
         def get_connected_devices() -> str:
             """Get list of connected devices"""
             try:
-                import serial.tools.list_ports
-
-                devices_on_ports = [p.device.strip() for p in serial.tools.list_ports.comports()]
-                print(f'Devices: {devices_on_ports}', file=sys.stderr)
+                devices_on_ports = [name.strip() for name in get_port_names()]
+                log.note(f'Devices: {devices_on_ports}')
                 devices = {'available_ports': devices_on_ports if devices_on_ports else []}
                 return json.dumps(devices, indent=2)
             except Exception as e:
@@ -439,14 +822,14 @@ def action_extensions(base_actions: dict, project_path: str) -> dict:
 
         # Start the MCP server. Diagnostics should go to stderr — stdout is reserved
         # for the JSON-RPC transport and any non-JSON bytes can confuse MCP clients.
-        print('MCP Server running on stdio...', file=sys.stderr)
+        log.note('MCP Server running on stdio...')
 
         try:
             mcp.run()
         except KeyboardInterrupt:
-            print('\nMCP Server stopped.', file=sys.stderr)
+            log.note('MCP Server stopped.')
         except Exception as e:
-            print(f'MCP Server error: {e}', file=sys.stderr)
+            log.err(f'MCP Server error: {e}')
 
     # Return the action extension
     return {

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
@@ -93,6 +95,24 @@ def mcp_ext(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[types.Modu
     rich_click = types.ModuleType('rich_click')
     rich_click.Context = object  # type: ignore[attr-defined]
 
+    # Stub esp_pylib modules used by mcp_ext
+    esp_pylib_pkg = types.ModuleType('esp_pylib')
+    logger_mod = types.ModuleType('esp_pylib.logger')
+    serial_ports_mod = types.ModuleType('esp_pylib.serial_ports')
+
+    class _MockLog:
+        def note(self, *args: Any, **kwargs: Any) -> None:
+            pass
+
+        def err(self, *args: Any, **kwargs: Any) -> None:
+            pass
+
+        def set_info_stream(self, *args: Any, **kwargs: Any) -> None:
+            pass
+
+    logger_mod.log = _MockLog()  # type: ignore[attr-defined]
+    serial_ports_mod.get_port_names = lambda **kwargs: []  # type: ignore[attr-defined]
+
     # Stub idf_py_actions hierarchy
     idf_py_actions_pkg = types.ModuleType('idf_py_actions')
     errors_mod = types.ModuleType('idf_py_actions.errors')
@@ -106,6 +126,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,13 +136,26 @@ 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,
+        'esp_pylib': esp_pylib_pkg,
+        'esp_pylib.logger': logger_mod,
+        'esp_pylib.serial_ports': serial_ports_mod,
         'idf_py_actions': idf_py_actions_pkg,
         'idf_py_actions.errors': errors_mod,
         '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 +210,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 +519,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 +936,7 @@ class TestServerStartsOutsideProject:
         assert 'set_target' in tools
         assert 'flash_project' in tools
         assert 'clean_project' in tools
+        assert 'monitor_device' in tools
 
 
 # ---------------------------------------------------------------------------