Probe Rs

{
  "exe": "probe-rs",
  "gdb_exe": "C:\\Program Files\\Arm\\GNU Toolchain mingw-w64-x86_64-arm-none-eabi\\bin\\arm-none-eabi-gdb.exe",
  "gdb_port": 3333,
  "dap_port": 50000,
  "operation_mode": 1
}

"""probe-rs 基础操作与包装命令。"""

from __future__ import annotations

import argparse
import os
import re
import subprocess
import sys
import time
from pathlib import Path


SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from probe_rs_runtime import (
    build_artifacts,
    default_config_path,
    get_state_entry,
    hidden_subprocess_kwargs,
    is_missing,
    load_json_file,
    load_project_config,
    load_workspace_state,
    make_result,
    make_timing,
    normalize_path,
    now_iso,
    output_json,
    parameter_context,
    save_project_config,
    update_state_entry,
    workspace_root,
)


ALL_ACTIONS = ["list", "info", "flash", "erase", "reset", "read-mem", "write-mem", "attach", "run"]

ERROR_PATTERNS = [
    (r"no probes were found", "no_probe_found", "未检测到调试探针，请检查 USB 连接和驱动"),
    (r"multiple probes were found", "multiple_probes", "检测到多个探针，请通过 --probe 显式指定"),
    (r"chip.*not found", "chip_not_found", "未找到目标芯片描述，请确认 --chip 配置"),
    (r"failed to open probe", "probe_open_failed", "打开调试探针失败，请检查探针占用、驱动和 USB 连接"),
    (r"failed to open the debug probe", "probe_open_failed", "打开调试探针失败，请检查探针占用、驱动和 USB 连接"),
    (r"error while probing target", "probe_open_failed", "打开调试探针失败，请检查探针占用、驱动和 USB 连接"),
    (r"unexpected answer to command", "probe_protocol_error", "探针返回异常响应，请检查固件、驱动和链路稳定性"),
    (r"failed to attach", "attach_failed", "连接目标失败，请检查供电、连线和芯片型号"),
    (r"permission denied", "permission_denied", "访问调试探针被拒绝，请检查驱动和权限"),
    (r"address.*out of bounds", "address_out_of_range", "访问地址超出范围，请确认地址和数据宽度"),
    (r"timed out", "timeout", "操作超时，请检查连接和速度配置"),
]


def infer_binary_format(file_path: str) -> str:
    suffix = Path(file_path).suffix.lower()
    if suffix == ".bin":
        return "bin"
    if suffix in {".hex", ".ihex"}:
        return "hex"
    if suffix == ".uf2":
        return "uf2"
    return "elf"


def parse_output(text: str, action: str) -> dict:
    parsed = {"raw": text}
    for pattern, code, message in ERROR_PATTERNS:
        if re.search(pattern, text, re.IGNORECASE):
            return {"error_code": code, "error_message": message, "raw": text}

    if action == "list":
        probes = []
        for line in text.splitlines():
            item = line.strip()
            if not item or item.lower().startswith("the following debug probes were found"):
                continue
            probes.append(item)
        if probes:
            parsed["probes"] = probes
    elif action == "read-mem":
        words = re.findall(r"\b[0-9a-fA-F]{2,16}\b", text)
        if words:
            parsed["words"] = words
    elif action == "info":
        chip_match = re.search(r"chip[:=]\s*([^\r\n]+)", text, re.IGNORECASE)
        probe_match = re.search(r"probe[:=]\s*([^\r\n]+)", text, re.IGNORECASE)
        if chip_match:
            parsed["chip"] = chip_match.group(1).strip()
        if probe_match:
            parsed["probe"] = probe_match.group(1).strip()
    return parsed


def _summary(action: str, parsed: dict, fallback: str) -> str:
    if action == "list" and parsed.get("probes"):
        return f"已发现 {len(parsed['probes'])} 个调试探针"
    if action == "flash":
        return "烧录成功"
    if action == "erase":
        return "擦除成功"
    if action == "reset":
        return "目标已复位"
    if action == "read-mem" and parsed.get("words"):
        return f"已读取 {len(parsed['words'])} 个内存字"
    if action == "write-mem":
        return "内存写入成功"
    return fallback


def normalize_write_values(value_text: str) -> list[str]:
    values = [item.strip() for item in re.split(r"[\s,]+", value_text) if item.strip()]
    if not values:
        raise ValueError("write-mem 必须提供 --value")

    normalized: list[str] = []
    for value in values:
        lowered = value.lower()
        if lowered.startswith(("0x", "0o", "0b")):
            normalized.append(value)
            continue
        if re.fullmatch(r"[0-9a-fA-F]+", value) and (value.startswith("0") or re.search(r"[a-fA-F]", value)):
            normalized.append(f"0x{value}")
            continue
        normalized.append(value)
    return normalized


def _state_lookup(state: dict) -> dict:
    last_build = get_state_entry(state, "last_build")
    last_flash = get_state_entry(state, "last_flash")
    last_debug = get_state_entry(state, "last_debug")
    artifacts = last_build.get("artifacts", {})
    return {
        "chip": last_debug.get("chip") or last_flash.get("chip"),
        "probe": last_debug.get("probe") or last_flash.get("probe"),
        "protocol": last_debug.get("protocol") or last_flash.get("protocol"),
        "speed": last_debug.get("speed") or last_flash.get("speed"),
        "connect_under_reset": last_debug.get("connect_under_reset") or last_flash.get("connect_under_reset"),
        "elf_file": last_build.get("debug_file") or artifacts.get("debug_file"),
        "flash_file": last_build.get("flash_file") or artifacts.get("flash_file"),
    }


def resolve_probe_params(args, config: dict, project_config: dict, state_lookup: dict, workspace: str) -> tuple[dict, dict]:
    parameter_sources: dict[str, str] = {}

    exe = args.exe if not is_missing(args.exe) else config.get("exe") or "probe-rs"
    parameter_sources["exe"] = "cli" if not is_missing(args.exe) else ("config:exe" if config.get("exe") else "default")

    chip = args.chip
    chip_source = "cli"
    if is_missing(chip):
        chip = project_config.get("chip")
        chip_source = "project_config"
    if is_missing(chip):
        chip = state_lookup.get("chip")
        chip_source = "state"
    parameter_sources["chip"] = chip_source

    protocol = args.protocol
    protocol_source = "cli"
    if is_missing(protocol):
        protocol = project_config.get("protocol")
        protocol_source = "project_config"
    if is_missing(protocol):
        protocol = state_lookup.get("protocol")
        protocol_source = "state"
    if is_missing(protocol):
        protocol = "swd"
        protocol_source = "default"
    parameter_sources["protocol"] = protocol_source

    probe = args.probe
    probe_source = "cli"
    if is_missing(probe):
        probe = project_config.get("probe")
        probe_source = "project_config"
    if is_missing(probe):
        probe = state_lookup.get("probe")
        probe_source = "state"
    parameter_sources["probe"] = probe_source

    speed = args.speed
    speed_source = "cli"
    if is_missing(speed):
        speed = project_config.get("speed")
        speed_source = "project_config"
    if is_missing(speed):
        speed = state_lookup.get("speed")
        speed_source = "state"
    if is_missing(speed):
        speed = "4000"
        speed_source = "default"
    parameter_sources["speed"] = speed_source

    connect_under_reset = args.connect_under_reset
    connect_source = "cli" if args.connect_under_reset else ""
    if not connect_under_reset:
        value = project_config.get("connect_under_reset")
        if value is not None:
            connect_under_reset = bool(value)
            connect_source = "project_config"
    if not connect_under_reset:
        value = state_lookup.get("connect_under_reset")
        if value is not None:
            connect_under_reset = bool(value)
            connect_source = "state"
    if not connect_source:
        connect_source = "default"
    parameter_sources["connect_under_reset"] = connect_source

    file_path = args.file
    file_source = "cli"
    if is_missing(file_path) and args.action == "flash":
        file_path = state_lookup.get("flash_file")
        file_source = "state"
    if is_missing(file_path) and args.action in {"run", "attach"}:
        file_path = state_lookup.get("elf_file")
        file_source = "state"
    if not is_missing(file_path):
        file_path = normalize_path(str(file_path))
    parameter_sources["file"] = file_source

    return (
        {
            "exe": exe,
            "chip": chip,
            "protocol": protocol,
            "probe": probe,
            "speed": str(speed),
            "connect_under_reset": bool(connect_under_reset),
            "file": file_path,
            "workspace": workspace,
        },
        parameter_sources,
    )


def build_probe_args(params: dict, *, require_chip: bool = True) -> list[str]:
    args = ["--non-interactive"]
    if require_chip:
        if is_missing(params["chip"]):
            raise ValueError("缺少必要参数: chip")
        args.extend(["--chip", params["chip"]])
    if params.get("protocol"):
        args.extend(["--protocol", str(params["protocol"]).lower()])
    if params.get("probe"):
        args.extend(["--probe", params["probe"]])
    if params.get("speed"):
        args.extend(["--speed", str(params["speed"])])
    if params.get("connect_under_reset"):
        args.append("--connect-under-reset")
    return args


def build_command(action: str, params: dict, args) -> list[str]:
    exe = params["exe"]
    if action == "list":
        return [exe, "list"]
    if action == "info":
        return [exe, "info", *build_probe_args(params)]
    if action == "reset":
        return [exe, "reset", *build_probe_args(params)]
    if action == "erase":
        return [exe, "erase", *build_probe_args(params)]
    if action == "read-mem":
        return [exe, "read", *build_probe_args(params), args.width, args.address, args.length]
    if action == "write-mem":
        return [exe, "write", *build_probe_args(params), args.width, args.address, *normalize_write_values(args.value)]
    if action == "flash":
        if is_missing(params["file"]):
            raise ValueError("flash 必须提供 --file 固件文件路径")
        if not os.path.isfile(params["file"]):
            raise ValueError(f"固件文件不存在: {params['file']}")
        fmt = infer_binary_format(params["file"])
        cmd = [exe, "download", *build_probe_args(params), "--binary-format", fmt]
        if args.chip_erase:
            cmd.append("--chip-erase")
        if args.verify:
            cmd.append("--verify")
        if fmt == "bin":
            if not args.address:
                raise ValueError(".bin 文件必须提供 --address 烧录地址")
            cmd.extend(["--base-address", args.address])
        cmd.append(params["file"])
        return cmd
    if action in {"attach", "run"}:
        cmd = [exe, action, *build_probe_args(params)]
        if params.get("file"):
            cmd.append(params["file"])
        return cmd
    raise ValueError(f"未知动作: {action}")


def run_command(action: str, cmd: list[str], duration: float = 0) -> dict:
    started = time.time()
    try:
        if action in {"attach", "run"} and duration > 0:
            proc = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                encoding="utf-8",
                errors="replace",
                **hidden_subprocess_kwargs(new_process_group=True),
            )
            time.sleep(duration)
            if proc.poll() is None:
                proc.terminate()
            stdout, stderr = proc.communicate(timeout=5)
        else:
            proc = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=120 if action not in {"attach", "run"} else None,
                **hidden_subprocess_kwargs(),
            )
            stdout, stderr = proc.stdout, proc.stderr
        returncode = proc.returncode
    except FileNotFoundError:
        return {"status": "error", "action": action, "error": {"code": "exe_not_found", "message": f"probe-rs 不存在或不在 PATH 中: {cmd[0]}"}}
    except subprocess.TimeoutExpired:
        return {"status": "error", "action": action, "error": {"code": "timeout", "message": "probe-rs 执行超时(120s)"}}
    except Exception as exc:
        return {"status": "error", "action": action, "error": {"code": "exec_error", "message": str(exc)}}

    elapsed_ms = int((time.time() - started) * 1000)
    combined = "\n".join(part for part in (stdout, stderr) if part)
    parsed = parse_output(combined, action)
    if "error_code" in parsed:
        return {
            "status": "error",
            "action": action,
            "error": {"code": parsed["error_code"], "message": parsed["error_message"]},
            "details": {"elapsed_ms": elapsed_ms, "returncode": returncode},
        }

    status = "ok"
    if returncode != 0 and action not in {"attach", "run"}:
        status = "error"
    return {
        "status": status,
        "action": action,
        "summary": _summary(action, parsed, f"{action} 完成"),
        "details": {"elapsed_ms": elapsed_ms, "returncode": returncode, **{k: v for k, v in parsed.items() if k != "raw"}, "output": combined},
        "error": None if status == "ok" else {"code": "nonzero_exit", "message": combined or f"{action} 失败"},
    }


def state_payload(action: str, params: dict) -> tuple[str, dict] | None:
    payload = {
        "provider": "probe-rs",
        "action": action,
        "chip": params["chip"] or "",
        "probe": params["probe"] or "",
        "protocol": params["protocol"],
        "speed": params["speed"],
        "connect_under_reset": params["connect_under_reset"],
    }
    if action == "flash":
        payload["flash_file"] = params["file"] or ""
        payload["artifacts"] = build_artifacts(flash_file=params["file"])
        return "last_flash", payload
    if action in {"reset", "read-mem", "write-mem", "info"}:
        return "last_debug", payload
    return None


def main() -> None:
    parser = argparse.ArgumentParser(description="probe-rs 基础操作包装")
    parser.add_argument("action", choices=ALL_ACTIONS)
    parser.add_argument("--exe", default=None, help="probe-rs 可执行文件路径或命令名")
    parser.add_argument("--chip", default=None, help="芯片型号")
    parser.add_argument("--protocol", default=None, choices=["swd", "jtag"], help="调试协议")
    parser.add_argument("--probe", default=None, help="探针选择器，格式 VID:PID[:Serial]")
    parser.add_argument("--speed", default=None, help="调试速率 kHz")
    parser.add_argument("--connect-under-reset", action="store_true", help="连接时保持 reset")
    parser.add_argument("--file", default=None, help="固件或 ELF 文件路径")
    parser.add_argument("--address", default="", help="地址（flash .bin / read-mem / write-mem 用）")
    parser.add_argument("--length", default="64", help="读取长度")
    parser.add_argument("--value", default="", help="写入值")
    parser.add_argument("--width", default="b32", choices=["b8", "b16", "b32", "b64"], help="读写位宽")
    parser.add_argument("--duration", type=float, default=0, help="attach/run 时运行秒数，0 表示等待命令自然退出")
    parser.add_argument("--verify", action="store_true", help="烧录后校验")
    parser.add_argument("--chip-erase", action="store_true", help="烧录前整片擦除")
    parser.add_argument("--config", default=None, help="skill config.json 路径")
    parser.add_argument("--workspace", default=None, help="workspace 根目录，默认当前目录")
    parser.add_argument("--json", action="store_true", dest="as_json")
    args = parser.parse_args()

    started_at = now_iso()
    started_ts = time.time()
    workspace = workspace_root(args.workspace)
    config_path = normalize_path(args.config or str(default_config_path(__file__)))
    config = load_json_file(config_path)
    state = load_workspace_state(str(workspace))
    state_lookup = _state_lookup(state)
    project_config = load_project_config(str(workspace))

    params, parameter_sources = resolve_probe_params(args, config, project_config, state_lookup, str(workspace))

    if args.action != "list" and is_missing(params["chip"]):
        result = make_result(
            status="error",
            action=args.action,
            summary="缺少必要参数: chip",
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "missing_chip", "message": "必须提供 --chip，或通过 .embeddedskills/config.json 的 probe-rs 段配置"},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {result['error']['message']}", file=sys.stderr)
        sys.exit(1)

    try:
        cmd = build_command(args.action, params, args)
    except ValueError as exc:
        result = make_result(
            status="error",
            action=args.action,
            summary=str(exc),
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "invalid_args", "message": str(exc)},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {exc}", file=sys.stderr)
        sys.exit(1)

    raw_result = run_command(args.action, cmd, args.duration)
    elapsed_ms = (time.time() - started_ts) * 1000
    if raw_result.get("status") == "ok":
        save_project_config(str(workspace), {
            "chip": params["chip"] or "",
            "protocol": params["protocol"],
            "probe": params["probe"] or "",
            "speed": params["speed"],
            "connect_under_reset": params["connect_under_reset"],
        })
        state_info = {}
        state_entry = state_payload(args.action, params)
        if state_entry:
            state_key, payload = state_entry
            state_info = update_state_entry(state_key, payload, str(workspace))
        result = make_result(
            status="ok",
            action=args.action,
            summary=raw_result.get("summary", f"{args.action} 完成"),
            details={
                "chip": params["chip"] or "",
                "probe": params["probe"] or "",
                "protocol": params["protocol"],
                "speed": params["speed"],
                "command": cmd,
                **(raw_result.get("details") or {}),
            },
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            artifacts=build_artifacts(flash_file=params["file"] if args.action == "flash" else "", debug_file=params["file"] if args.action in {"attach", "run"} else ""),
            state=state_info,
            next_actions=["可继续基于 probe-rs 执行 gdb 或 rtt 观测"] if args.action in {"flash", "info"} else None,
            timing=make_timing(started_at, elapsed_ms),
        )
    else:
        result = make_result(
            status="error",
            action=args.action,
            summary=(raw_result.get("error") or {}).get("message", f"{args.action} 失败"),
            details={
                "chip": params["chip"] or "",
                "probe": params["probe"] or "",
                "protocol": params["protocol"],
                "speed": params["speed"],
                "command": cmd,
                **(raw_result.get("details") or {}),
            },
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error=raw_result.get("error"),
            timing=make_timing(started_at, elapsed_ms),
        )

    if args.as_json:
        output_json(result)
    elif result["status"] == "ok":
        print(f"[probe-rs {args.action}] {result['summary']}")
        output = result.get("details", {}).get("output", "")
        if output:
            print(output)
    else:
        print(f"[probe-rs {args.action}] 失败 — {result['error']['message']}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

"""probe-rs skill 私有 GDB 工具。"""

from __future__ import annotations

import re
import subprocess
from pathlib import Path
from typing import Any

from probe_rs_runtime import hidden_subprocess_kwargs


INTROSPECTION_ACTIONS = {
    "backtrace",
    "locals",
    "frame",
    "print",
    "threads",
    "disassemble",
    "crash-report",
}


def run_gdb_commands(gdb_exe: str, elf_file: str, target_remote: str, commands: list[str], timeout: int = 30) -> dict:
    gdb_init = ["set pagination off", "set confirm off", "set width 0"]
    if elf_file:
        gdb_init.append(f'file "{Path(elf_file).resolve().as_posix()}"')
    gdb_init.append(f"target remote {target_remote}")

    cmd = [gdb_exe, "--batch", "--nx"]
    for item in gdb_init + commands + ["quit"]:
        cmd.extend(["-ex", item])

    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
            encoding="utf-8",
            errors="replace",
            **hidden_subprocess_kwargs(),
        )
        combined_output = "\n".join(part for part in (proc.stdout, proc.stderr) if part)
        return {
            "status": "ok" if proc.returncode == 0 else "error",
            "stdout": combined_output,
            "stderr": proc.stderr,
            "returncode": proc.returncode,
        }
    except subprocess.TimeoutExpired as exc:
        stdout = exc.stdout or ""
        stderr = exc.stderr or ""
        combined_output = "\n".join(part for part in (stdout, stderr) if part)
        return {
            "status": "timeout",
            "stdout": combined_output,
            "stderr": stderr,
            "returncode": None,
            "error": f"GDB 执行超时({timeout}s)",
        }
    except Exception as exc:
        return {"status": "error", "error": str(exc)}


def require_action_expr(action: str, expr: str | None, hint: str) -> str:
    if not expr:
        raise ValueError(f"{action} 必须提供 {hint}")
    return expr


def build_gdb_commands(action: str, expr: str | None = None, *, halt_before: bool = True) -> list[str]:
    commands: list[str] = []
    if halt_before and action in INTROSPECTION_ACTIONS | {"next", "step", "finish", "until"}:
        commands.append("monitor halt")

    if action == "run":
        raise ValueError("run 需要由调用方直接提供 commands")
    if action == "backtrace":
        commands.append("backtrace")
    elif action == "locals":
        commands.append("info locals")
    elif action == "break":
        commands.extend([f"break {require_action_expr(action, expr, '--expr')}", "info breakpoints"])
    elif action == "continue":
        commands.append("continue")
    elif action == "next":
        commands.append("next")
    elif action == "step":
        commands.append("step")
    elif action == "finish":
        commands.append("finish")
    elif action == "until":
        commands.append(f"until {expr}" if expr else "until")
    elif action == "frame":
        commands.append(f"frame {require_action_expr(action, expr, '--expr <帧号>')}")
    elif action == "print":
        commands.append(f"print {require_action_expr(action, expr, '--expr')}")
    elif action == "watch":
        commands.extend([f"watch {require_action_expr(action, expr, '--expr')}", "info breakpoints"])
    elif action == "disassemble":
        commands.append(f"disassemble {expr}" if expr else "disassemble")
    elif action == "threads":
        commands.extend(["info threads", "thread apply all backtrace 1"])
    elif action == "crash-report":
        commands.extend(
            [
                "backtrace full",
                "info registers",
                "frame 0",
                "info locals",
                "info threads",
                "disassemble /m $pc,$pc+32",
            ]
        )
    else:
        raise ValueError(f"未知 GDB 子命令: {action}")

    return commands


def _parse_frames(stdout: str) -> list[dict[str, Any]]:
    frames = []
    for line in stdout.splitlines():
        match = re.match(
            r"#(?P<index>\d+)\s+(?:(?P<address>0x[0-9a-fA-F]+)\s+in\s+)?(?P<function>[^\s(]+)?\s*\((?P<args>[^)]*)\)(?:\s+at\s+(?P<location>.+))?",
            line.strip(),
        )
        if not match:
            continue
        frame = {"frame": int(match.group("index")), "function": match.group("function") or "??"}
        if match.group("address"):
            frame["address"] = match.group("address")
        if match.group("args"):
            frame["args"] = match.group("args").strip()
        if match.group("location"):
            frame["location"] = match.group("location").strip()
        frames.append(frame)
    return frames


def _parse_variables(stdout: str) -> dict[str, str]:
    variables: dict[str, str] = {}
    for line in stdout.splitlines():
        match = re.match(r"^([A-Za-z_][\w.\->\[\]]*)\s*=\s*(.+)$", line.strip())
        if match:
            variables[match.group(1)] = match.group(2).strip()
    return variables


def _parse_registers(stdout: str) -> dict[str, str]:
    registers: dict[str, str] = {}
    for line in stdout.splitlines():
        match = re.match(r"^([A-Za-z_][\w]*)\s+(0x[0-9a-fA-F]+)\b(.*)$", line.strip())
        if match:
            registers[match.group(1)] = match.group(2)
    return registers


def _parse_threads(stdout: str) -> list[dict[str, Any]]:
    threads: list[dict[str, Any]] = []
    for line in stdout.splitlines():
        match = re.match(r"^([* ])\s*(\d+)\s+Thread\s+(.+)$", line.strip())
        if match:
            threads.append({"selected": match.group(1) == "*", "id": int(match.group(2)), "description": match.group(3).strip()})
    return threads


def _parse_disassembly(stdout: str) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    for line in stdout.splitlines():
        match = re.match(r"^(=>)?\s*(0x[0-9a-fA-F]+)(?:\s+<([^>]+)>)?:\s+(.+)$", line.strip())
        if match:
            item = {"address": match.group(2), "instruction": match.group(4).strip()}
            if match.group(1):
                item["selected"] = "true"
            if match.group(3):
                item["symbol"] = match.group(3).strip()
            items.append(item)
    return items


def _extract_source_location(stdout: str, frames: list[dict[str, Any]]) -> str:
    for frame in frames:
        location = frame.get("location", "")
        if location:
            return location
    match = re.search(r'at\s+([A-Za-z]:)?[^:\n]+\:\d+', stdout)
    return match.group(0).replace("at ", "").strip() if match else ""


def _parse_selected_frame(stdout: str) -> dict[str, Any]:
    match = re.search(r"#(?P<index>\d+)\s+.+?(?:at\s+(?P<location>.+))?$", stdout, re.MULTILINE)
    if not match:
        return {}
    selected = {"frame": int(match.group("index"))}
    if match.group("location"):
        selected["location"] = match.group("location").strip()
    return selected


def parse_gdb_output(stdout: str, action: str) -> dict:
    frames = _parse_frames(stdout)
    variables = _parse_variables(stdout)
    registers = _parse_registers(stdout)
    threads = _parse_threads(stdout)
    disassembly = _parse_disassembly(stdout)

    parsed: dict[str, Any] = {"output": stdout}
    if frames:
        parsed["frames"] = frames
    if variables:
        parsed["variables"] = variables
    if registers:
        parsed["registers"] = registers
    if threads:
        parsed["threads"] = threads
    if disassembly:
        parsed["disassembly"] = disassembly

    selected_frame = _parse_selected_frame(stdout)
    if not selected_frame and frames:
        selected_frame = frames[0]
    if selected_frame:
        parsed["selected_frame"] = selected_frame

    source_location = _extract_source_location(stdout, frames)
    if source_location:
        parsed["source_location"] = source_location

    if action == "print":
        match = re.search(r"\$\d+\s*=\s*(.+)", stdout)
        if match:
            parsed["value"] = match.group(1).strip()

    return parsed

"""probe-rs GDB Server 启动与 one-shot 调试。"""

from __future__ import annotations

import argparse
import os
import signal
import socket
import subprocess
import sys
import time
from pathlib import Path
from shutil import which


ROOT_DIR = Path(__file__).resolve().parents[2]
if str(ROOT_DIR) not in sys.path:
    sys.path.insert(0, str(ROOT_DIR))

from probe_rs_gdb_common import build_gdb_commands, parse_gdb_output, run_gdb_commands  # noqa: E402
from probe_rs_runtime import (  # noqa: E402
    build_artifacts,
    default_config_path,
    get_state_entry,
    hidden_subprocess_kwargs,
    is_missing,
    load_json_file,
    load_project_config,
    load_workspace_state,
    make_result,
    make_timing,
    normalize_path,
    now_iso,
    output_json,
    parameter_context,
    save_project_config,
    update_state_entry,
    workspace_root,
)


ALL_COMMANDS = [
    "run",
    "backtrace",
    "locals",
    "break",
    "continue",
    "next",
    "step",
    "finish",
    "until",
    "frame",
    "print",
    "watch",
    "disassemble",
    "threads",
    "crash-report",
]


def find_free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("", 0))
        return sock.getsockname()[1]


def start_gdb_server(exe: str, chip: str, protocol: str, speed: str, probe: str, connect_under_reset: bool, gdb_port: int) -> tuple[subprocess.Popen, int]:
    if not gdb_port:
        gdb_port = find_free_port()

    cmd = [
        exe,
        "gdb",
        "--non-interactive",
        "--chip",
        chip,
        "--protocol",
        protocol,
        "--speed",
        speed,
        "--gdb-connection-string",
        f"localhost:{gdb_port}",
    ]
    if probe:
        cmd.extend(["--probe", probe])
    if connect_under_reset:
        cmd.append("--connect-under-reset")

    proc = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        encoding="utf-8",
        errors="replace",
        **hidden_subprocess_kwargs(new_process_group=True),
    )
    return proc, gdb_port


def wait_gdb_server_ready(proc: subprocess.Popen, port: int, timeout: int = 15) -> tuple[bool, str]:
    started = time.time()
    startup_grace = min(5.0, max(1.0, timeout / 3))
    while time.time() - started < timeout:
        if proc.poll() is not None:
            stdout = proc.stdout.read() if proc.stdout else ""
            stderr = proc.stderr.read() if proc.stderr else ""
            return False, "\n".join(part for part in (stdout, stderr) if part).strip()
        if time.time() - started >= startup_grace:
            return True, f"probe-rs gdb 已启动，假定 localhost:{port} 可用"
        time.sleep(0.2)
    return False, f"GDB Server 在 {timeout}s 内未监听 localhost:{port}"


def cleanup(procs: list[subprocess.Popen]) -> None:
    for proc in procs:
        if proc and proc.poll() is None:
            try:
                if sys.platform == "win32":
                    proc.terminate()
                else:
                    proc.send_signal(signal.SIGTERM)
                proc.wait(timeout=5)
            except (subprocess.TimeoutExpired, OSError):
                proc.kill()


def _state_lookup(state: dict) -> dict:
    last_build = get_state_entry(state, "last_build")
    last_flash = get_state_entry(state, "last_flash")
    last_debug = get_state_entry(state, "last_debug")
    artifacts = last_build.get("artifacts", {})
    return {
        "chip": last_debug.get("chip") or last_flash.get("chip"),
        "probe": last_debug.get("probe") or last_flash.get("probe"),
        "protocol": last_debug.get("protocol") or last_flash.get("protocol"),
        "speed": last_debug.get("speed") or last_flash.get("speed"),
        "connect_under_reset": last_debug.get("connect_under_reset") or last_flash.get("connect_under_reset"),
        "elf_file": last_build.get("debug_file") or last_build.get("elf_file") or artifacts.get("debug_file"),
    }


def resolve_probe_params(args, config: dict, project_config: dict, state_lookup: dict) -> tuple[dict, dict]:
    parameter_sources: dict[str, str] = {}

    exe = args.exe if not is_missing(args.exe) else config.get("exe") or "probe-rs"
    parameter_sources["exe"] = "cli" if not is_missing(args.exe) else ("config:exe" if config.get("exe") else "default")

    gdb_exe = args.gdb_exe if not is_missing(args.gdb_exe) else config.get("gdb_exe")
    gdb_source = "cli" if not is_missing(args.gdb_exe) else ("config:gdb_exe" if config.get("gdb_exe") else "")
    if is_missing(gdb_exe):
        discovered = which("arm-none-eabi-gdb") or which("arm-none-eabi-gdb.exe")
        if discovered:
            gdb_exe = discovered
            gdb_source = "path"
    parameter_sources["gdb_exe"] = gdb_source

    chip = args.chip
    chip_source = "cli"
    if is_missing(chip):
        chip = project_config.get("chip")
        chip_source = "project_config"
    if is_missing(chip):
        chip = state_lookup.get("chip")
        chip_source = "state"
    parameter_sources["chip"] = chip_source

    protocol = args.protocol
    protocol_source = "cli"
    if is_missing(protocol):
        protocol = project_config.get("protocol")
        protocol_source = "project_config"
    if is_missing(protocol):
        protocol = state_lookup.get("protocol")
        protocol_source = "state"
    if is_missing(protocol):
        protocol = "swd"
        protocol_source = "default"
    parameter_sources["protocol"] = protocol_source

    probe = args.probe
    probe_source = "cli"
    if is_missing(probe):
        probe = project_config.get("probe")
        probe_source = "project_config"
    if is_missing(probe):
        probe = state_lookup.get("probe")
        probe_source = "state"
    parameter_sources["probe"] = probe_source

    speed = args.speed
    speed_source = "cli"
    if is_missing(speed):
        speed = project_config.get("speed")
        speed_source = "project_config"
    if is_missing(speed):
        speed = state_lookup.get("speed")
        speed_source = "state"
    if is_missing(speed):
        speed = "4000"
        speed_source = "default"
    parameter_sources["speed"] = speed_source

    connect_under_reset = args.connect_under_reset
    connect_source = "cli" if args.connect_under_reset else ""
    if not connect_under_reset:
        value = project_config.get("connect_under_reset")
        if value is not None:
            connect_under_reset = bool(value)
            connect_source = "project_config"
    if not connect_under_reset:
        value = state_lookup.get("connect_under_reset")
        if value is not None:
            connect_under_reset = bool(value)
            connect_source = "state"
    if not connect_source:
        connect_source = "default"
    parameter_sources["connect_under_reset"] = connect_source

    elf_file = args.elf
    elf_source = "cli"
    if is_missing(elf_file):
        elf_file = state_lookup.get("elf_file")
        elf_source = "state"
    if not is_missing(elf_file):
        elf_file = normalize_path(str(elf_file))
    parameter_sources["elf"] = elf_source

    gdb_port = args.gdb_port if args.gdb_port else config.get("gdb_port", 3333)
    parameter_sources["gdb_port"] = "cli" if args.gdb_port else ("config:gdb_port" if config.get("gdb_port") else "default")

    return (
        {
            "exe": exe,
            "gdb_exe": gdb_exe,
            "chip": chip,
            "protocol": protocol,
            "probe": probe,
            "speed": str(speed),
            "connect_under_reset": bool(connect_under_reset),
            "elf_file": elf_file,
            "gdb_port": int(gdb_port or 0),
        },
        parameter_sources,
    )


def _summary(command: str, parsed: dict) -> str:
    if command == "continue" and parsed.get("timed_out"):
        return "continue 已执行，目标在超时窗口内未停下"
    if command == "backtrace" and parsed.get("frames"):
        return f"backtrace 完成，frames={len(parsed['frames'])}"
    if command == "locals" and parsed.get("variables"):
        return f"locals 完成，variables={len(parsed['variables'])}"
    if command == "threads" and parsed.get("threads"):
        return f"threads 完成，threads={len(parsed['threads'])}"
    if command == "print" and parsed.get("value"):
        return f"print 完成，value={parsed['value']}"
    return f"gdb {command} 完成"


def _metrics(parsed: dict) -> dict:
    metrics: dict[str, int] = {}
    if parsed.get("frames"):
        metrics["frames"] = len(parsed["frames"])
    if parsed.get("variables"):
        metrics["variables"] = len(parsed["variables"])
    if parsed.get("registers"):
        metrics["registers"] = len(parsed["registers"])
    if parsed.get("threads"):
        metrics["threads"] = len(parsed["threads"])
    if parsed.get("disassembly"):
        metrics["instructions"] = len(parsed["disassembly"])
    return metrics


def stepping_fallback_commands(action: str) -> list[str] | None:
    if action == "next":
        return ["monitor halt", "nexti"]
    if action == "step":
        return ["monitor halt", "stepi"]
    return None


def main() -> None:
    parser = argparse.ArgumentParser(description="probe-rs GDB Server 调试")
    sub = parser.add_subparsers(dest="command")
    for name in ALL_COMMANDS:
        sub_parser = sub.add_parser(name, help=f"GDB {name}")
        sub_parser.add_argument("--exe", default=None, help="probe-rs 可执行文件")
        sub_parser.add_argument("--gdb-exe", default=None, help="arm-none-eabi-gdb 路径")
        sub_parser.add_argument("--chip", default=None, help="芯片型号")
        sub_parser.add_argument("--elf", default=None, help="ELF 文件路径")
        sub_parser.add_argument("--protocol", default=None, choices=["swd", "jtag"], help="调试协议")
        sub_parser.add_argument("--probe", default=None, help="探针选择器")
        sub_parser.add_argument("--speed", default=None, help="调试速率 kHz")
        sub_parser.add_argument("--connect-under-reset", action="store_true", help="连接时保持 reset")
        sub_parser.add_argument("--gdb-port", type=int, default=0, help="GDB 端口，0=自动")
        sub_parser.add_argument("--config", default=None, help="skill config.json 路径")
        sub_parser.add_argument("--workspace", default=None, help="workspace 根目录，默认当前目录")
        sub_parser.add_argument("--json", action="store_true", dest="as_json")
        if name == "run":
            sub_parser.add_argument("--commands", nargs="+", required=True, help="GDB 命令序列")
        elif name in {"break", "frame", "print", "watch"}:
            sub_parser.add_argument("--expr", required=True, help="表达式或参数")
        elif name in {"until", "disassemble"}:
            sub_parser.add_argument("--expr", default=None, help="表达式或参数")

    args = parser.parse_args()
    if not args.command:
        parser.print_help()
        sys.exit(1)

    started_at = now_iso()
    started_ts = time.time()
    workspace = workspace_root(args.workspace)
    config_path = normalize_path(args.config or str(default_config_path(__file__)))
    config = load_json_file(config_path)
    state = load_workspace_state(str(workspace))
    state_lookup = _state_lookup(state)
    project_config = load_project_config(str(workspace))

    params, parameter_sources = resolve_probe_params(args, config, project_config, state_lookup)
    if is_missing(params["chip"]):
        result = make_result(
            status="error",
            action=args.command,
            summary="缺少必要参数: chip",
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "missing_chip", "message": "必须提供 --chip，或通过 .embeddedskills/config.json 的 probe-rs 段配置"},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {result['error']['message']}", file=sys.stderr)
        sys.exit(1)

    if is_missing(params["gdb_exe"]) or not os.path.isfile(str(params["gdb_exe"])):
        message = f"arm-none-eabi-gdb 不存在: {params['gdb_exe']}"
        result = make_result(
            status="error",
            action=args.command,
            summary=message,
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "gdb_not_found", "message": message},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {message}", file=sys.stderr)
        sys.exit(1)

    try:
        gdb_commands = list(args.commands) if args.command == "run" else build_gdb_commands(args.command, getattr(args, "expr", None))
    except ValueError as exc:
        result = make_result(
            status="error",
            action=args.command,
            summary=str(exc),
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "invalid_args", "message": str(exc)},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {exc}", file=sys.stderr)
        sys.exit(1)

    procs: list[subprocess.Popen] = []
    try:
        server_proc, gdb_port = start_gdb_server(
            exe=params["exe"],
            chip=params["chip"],
            protocol=params["protocol"],
            speed=params["speed"],
            probe=params["probe"] or "",
            connect_under_reset=params["connect_under_reset"],
            gdb_port=params["gdb_port"],
        )
        procs.append(server_proc)

        ready, server_output = wait_gdb_server_ready(server_proc, gdb_port)
        if not ready:
            result = make_result(
                status="error",
                action=args.command,
                summary="GDB Server 启动失败",
                details={"chip": params["chip"], "server_output": server_output},
                context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                error={"code": "gdbserver_failed", "message": server_output or "GDB Server 启动失败"},
                timing=make_timing(started_at, (time.time() - started_ts) * 1000),
            )
            if args.as_json:
                output_json(result)
            else:
                print(f"[probe-rs gdb-{args.command}] 失败 — {result['error']['message']}", file=sys.stderr)
            sys.exit(1)

        gdb_result = run_gdb_commands(str(params["gdb_exe"]), params["elf_file"] or "", f"localhost:{gdb_port}", gdb_commands)
        if gdb_result["status"] == "timeout":
            fallback_commands = stepping_fallback_commands(args.command)
            if fallback_commands:
                gdb_result = run_gdb_commands(str(params["gdb_exe"]), params["elf_file"] or "", f"localhost:{gdb_port}", fallback_commands)
                if gdb_result["status"] == "ok":
                    gdb_commands = fallback_commands
        elapsed_ms = (time.time() - started_ts) * 1000

        if gdb_result["status"] == "timeout" and args.command == "continue":
            parsed = parse_gdb_output(gdb_result.get("stdout", ""), args.command)
            parsed["timed_out"] = True
        elif gdb_result["status"] in {"error", "timeout"}:
            result = make_result(
                status="error",
                action=args.command,
                summary="GDB 执行失败",
                details={"chip": params["chip"], "gdb_port": gdb_port, "server_output": server_output},
                context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                artifacts=build_artifacts(debug_file=params["elf_file"]),
                error={"code": "gdb_error", "message": gdb_result.get("error", gdb_result.get("stderr", "GDB 执行失败"))},
                timing=make_timing(started_at, elapsed_ms),
            )
            if args.as_json:
                output_json(result)
            else:
                print(f"[probe-rs gdb-{args.command}] 失败 — {result['error']['message']}", file=sys.stderr)
            sys.exit(1)
        else:
            parsed = parse_gdb_output(gdb_result["stdout"], args.command)

        artifacts = build_artifacts(debug_file=params["elf_file"])
        state_info = update_state_entry(
            "last_debug",
            {
                "provider": "probe-rs",
                "action": args.command,
                "chip": params["chip"],
                "probe": params["probe"] or "",
                "protocol": params["protocol"],
                "speed": params["speed"],
                "connect_under_reset": params["connect_under_reset"],
                "debug_file": params["elf_file"] or "",
                "artifacts": artifacts,
            },
            str(workspace),
        )
        save_project_config(str(workspace), {
            "chip": params["chip"],
            "protocol": params["protocol"],
            "probe": params["probe"] or "",
            "speed": params["speed"],
            "connect_under_reset": params["connect_under_reset"],
        })
        result = make_result(
            status="ok",
            action=args.command,
            summary=_summary(args.command, parsed),
            details={
                "chip": params["chip"],
                "probe": params["probe"] or "",
                "protocol": params["protocol"],
                "gdb_port": gdb_port,
                "commands": gdb_commands,
                "output": parsed.get("output", ""),
                "server_output": server_output,
                "returncode": gdb_result.get("returncode", 0),
                **{key: value for key, value in parsed.items() if key != "output"},
            },
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            artifacts=artifacts,
            metrics=_metrics(parsed),
            state=state_info,
            next_actions=["可继续基于 last_debug 复用 chip/debug_file"],
            timing=make_timing(started_at, elapsed_ms),
        )

        if args.as_json:
            output_json(result)
        elif result["status"] == "ok":
            print(f"[probe-rs gdb-{args.command}] {result['summary']}")
            output = result.get("details", {}).get("output", "")
            if output:
                print(output)
        else:
            print(f"[probe-rs gdb-{args.command}] 失败 — {result['error']['message']}", file=sys.stderr)
            sys.exit(1)
    finally:
        cleanup(procs)


if __name__ == "__main__":
    main()

"""probe-rs RTT 日志读取。"""

from __future__ import annotations

import argparse
import queue
import re
import signal
import subprocess
import sys
import threading
import time
from pathlib import Path


ROOT_DIR = Path(__file__).resolve().parents[2]
if str(ROOT_DIR) not in sys.path:
    sys.path.insert(0, str(ROOT_DIR))

from probe_rs_runtime import (  # noqa: E402
    default_config_path,
    emit_stream_record,
    get_state_entry,
    hidden_subprocess_kwargs,
    is_missing,
    load_json_file,
    load_project_config,
    load_workspace_state,
    make_result,
    make_timing,
    normalize_path,
    now_iso,
    output_json,
    parameter_context,
    save_project_config,
    update_state_entry,
    workspace_root,
)


ERROR_PATTERNS = [
    (r"no probes were found", "no_probe_found", "未检测到调试探针，请检查 USB 连接和驱动"),
    (r"multiple probes were found", "multiple_probes", "检测到多个探针，请通过 --probe 显式指定"),
    (r"chip.*not found", "chip_not_found", "未找到目标芯片描述，请确认 --chip 配置"),
    (r"failed to open probe", "probe_open_failed", "打开调试探针失败，请检查探针占用、驱动和 USB 连接"),
    (r"failed to open the debug probe", "probe_open_failed", "打开调试探针失败，请检查探针占用、驱动和 USB 连接"),
    (r"error while probing target", "probe_open_failed", "打开调试探针失败，请检查探针占用、驱动和 USB 连接"),
    (r"unexpected answer to command", "probe_protocol_error", "探针返回异常响应，请检查固件、驱动和链路稳定性"),
    (r"permission denied", "permission_denied", "访问调试探针被拒绝，请检查驱动和权限"),
    (r"timed out", "timeout", "操作超时，请检查连接和速度配置"),
]


def cleanup(proc: subprocess.Popen | None) -> None:
    if proc and proc.poll() is None:
        try:
            if sys.platform == "win32":
                proc.terminate()
            else:
                proc.send_signal(signal.SIGTERM)
            proc.wait(timeout=5)
        except (subprocess.TimeoutExpired, OSError):
            proc.kill()


def start_stream_reader(stream) -> queue.Queue:
    line_queue: queue.Queue = queue.Queue()

    def _reader() -> None:
        try:
            for line in iter(stream.readline, ""):
                line_queue.put(line)
        finally:
            line_queue.put(None)

    threading.Thread(target=_reader, daemon=True).start()
    return line_queue


def _state_lookup(state: dict) -> dict:
    last_build = get_state_entry(state, "last_build")
    last_debug = get_state_entry(state, "last_debug")
    last_flash = get_state_entry(state, "last_flash")
    artifacts = last_build.get("artifacts", {})
    last_debug_artifacts = (last_debug.get("artifacts") or {}) if isinstance(last_debug, dict) else {}
    return {
        "chip": last_debug.get("chip") or last_flash.get("chip"),
        "probe": last_debug.get("probe") or last_flash.get("probe"),
        "protocol": last_debug.get("protocol") or last_flash.get("protocol"),
        "speed": last_debug.get("speed") or last_flash.get("speed"),
        "connect_under_reset": last_debug.get("connect_under_reset") or last_flash.get("connect_under_reset"),
        "elf_file": last_debug.get("debug_file") or last_debug_artifacts.get("debug_file") or last_build.get("debug_file") or artifacts.get("debug_file"),
    }


def resolve_probe_params(args, config: dict, project_config: dict, state_lookup: dict) -> tuple[dict, dict]:
    parameter_sources: dict[str, str] = {}

    exe = args.exe if not is_missing(args.exe) else config.get("exe") or "probe-rs"
    parameter_sources["exe"] = "cli" if not is_missing(args.exe) else ("config:exe" if config.get("exe") else "default")

    chip = args.chip
    chip_source = "cli"
    if is_missing(chip):
        chip = project_config.get("chip")
        chip_source = "project_config"
    if is_missing(chip):
        chip = state_lookup.get("chip")
        chip_source = "state"
    parameter_sources["chip"] = chip_source

    protocol = args.protocol
    protocol_source = "cli"
    if is_missing(protocol):
        protocol = project_config.get("protocol")
        protocol_source = "project_config"
    if is_missing(protocol):
        protocol = state_lookup.get("protocol")
        protocol_source = "state"
    if is_missing(protocol):
        protocol = "swd"
        protocol_source = "default"
    parameter_sources["protocol"] = protocol_source

    probe = args.probe
    probe_source = "cli"
    if is_missing(probe):
        probe = project_config.get("probe")
        probe_source = "project_config"
    if is_missing(probe):
        probe = state_lookup.get("probe")
        probe_source = "state"
    parameter_sources["probe"] = probe_source

    speed = args.speed
    speed_source = "cli"
    if is_missing(speed):
        speed = project_config.get("speed")
        speed_source = "project_config"
    if is_missing(speed):
        speed = state_lookup.get("speed")
        speed_source = "state"
    if is_missing(speed):
        speed = "4000"
        speed_source = "default"
    parameter_sources["speed"] = speed_source

    connect_under_reset = args.connect_under_reset
    connect_source = "cli" if args.connect_under_reset else ""
    if not connect_under_reset:
        value = project_config.get("connect_under_reset")
        if value is not None:
            connect_under_reset = bool(value)
            connect_source = "project_config"
    if not connect_under_reset:
        value = state_lookup.get("connect_under_reset")
        if value is not None:
            connect_under_reset = bool(value)
            connect_source = "state"
    if not connect_source:
        connect_source = "default"
    parameter_sources["connect_under_reset"] = connect_source

    elf_file = args.elf
    elf_source = "cli"
    if is_missing(elf_file):
        elf_file = state_lookup.get("elf_file")
        elf_source = "state"
    if not is_missing(elf_file):
        elf_file = normalize_path(str(elf_file))
        if not Path(elf_file).is_file():
            elf_file = ""
            elf_source = "state:missing"
    parameter_sources["elf"] = elf_source

    return (
        {
            "exe": exe,
            "chip": chip,
            "protocol": protocol,
            "probe": probe,
            "speed": str(speed),
            "connect_under_reset": bool(connect_under_reset),
            "elf_file": elf_file,
        },
        parameter_sources,
    )


def build_attach_command(params: dict) -> list[str]:
    if is_missing(params["chip"]):
        raise ValueError("缺少必要参数: chip")
    if is_missing(params["elf_file"]):
        raise ValueError("缺少必要参数: elf，必须提供 --elf，或保证 last_debug/last_build 中存在有效 ELF 文件")
    cmd = [
        params["exe"],
        "attach",
        "--non-interactive",
        "--chip",
        params["chip"],
        "--protocol",
        params["protocol"],
        "--speed",
        params["speed"],
    ]
    if params["probe"]:
        cmd.extend(["--probe", params["probe"]])
    if params["connect_under_reset"]:
        cmd.append("--connect-under-reset")
    cmd.append(params["elf_file"])
    return cmd


def detect_runtime_error(text: str) -> tuple[str, str] | None:
    for pattern, code, message in ERROR_PATTERNS:
        if re.search(pattern, text, re.IGNORECASE):
            return code, message
    return None


def main() -> None:
    parser = argparse.ArgumentParser(description="probe-rs RTT 日志读取")
    parser.add_argument("--exe", default=None, help="probe-rs 可执行文件")
    parser.add_argument("--chip", default=None, help="芯片型号")
    parser.add_argument("--elf", default=None, help="ELF 文件路径")
    parser.add_argument("--protocol", default=None, choices=["swd", "jtag"], help="调试协议")
    parser.add_argument("--probe", default=None, help="探针选择器")
    parser.add_argument("--speed", default=None, help="调试速率 kHz")
    parser.add_argument("--connect-under-reset", action="store_true", help="连接时保持 reset")
    parser.add_argument("--duration", type=float, default=0, help="读取时长(秒)，0=持续运行")
    parser.add_argument("--config", default=None, help="skill config.json 路径")
    parser.add_argument("--workspace", default=None, help="workspace 根目录，默认当前目录")
    parser.add_argument("--json", action="store_true", dest="as_json")
    args = parser.parse_args()

    started_at = now_iso()
    started_ts = time.time()
    workspace = workspace_root(args.workspace)
    config_path = normalize_path(args.config or str(default_config_path(__file__)))
    config = load_json_file(config_path)
    state = load_workspace_state(str(workspace))
    state_lookup = _state_lookup(state)
    project_config = load_project_config(str(workspace))

    params, parameter_sources = resolve_probe_params(args, config, project_config, state_lookup)
    try:
        cmd = build_attach_command(params)
    except ValueError as exc:
        result = make_result(
            status="error",
            action="rtt",
            summary=str(exc),
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "invalid_args", "message": str(exc)},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {exc}", file=sys.stderr)
        sys.exit(1)

    proc: subprocess.Popen | None = None
    try:
        proc = subprocess.Popen(
            cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            encoding="utf-8",
            errors="replace",
            **hidden_subprocess_kwargs(new_process_group=True),
        )
    except FileNotFoundError:
        result = make_result(
            status="error",
            action="rtt",
            summary=f"probe-rs 不存在或不在 PATH 中: {params['exe']}",
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "exe_not_found", "message": f"probe-rs 不存在或不在 PATH 中: {params['exe']}"},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {result['error']['message']}", file=sys.stderr)
        sys.exit(1)

    stdout_queue = start_stream_reader(proc.stdout)
    stderr_queue = start_stream_reader(proc.stderr)

    startup_deadline = time.time() + 1.5
    buffered_stdout: list[str] = []
    buffered_stderr: list[str] = []
    while time.time() < startup_deadline:
        while True:
            try:
                item = stdout_queue.get_nowait()
            except queue.Empty:
                break
            if item is None:
                break
            buffered_stdout.append(item)
        while True:
            try:
                item = stderr_queue.get_nowait()
            except queue.Empty:
                break
            if item is None:
                break
            buffered_stderr.append(item)
        combined_startup = "\n".join(buffered_stderr + buffered_stdout)
        startup_error = detect_runtime_error(combined_startup)
        if startup_error:
            cleanup(proc)
            code, message = startup_error
            result = make_result(
                status="error",
                action="rtt",
                summary=message,
                details={
                    "chip": params["chip"],
                    "command": cmd,
                    "returncode": proc.poll() if proc.poll() is not None else 1,
                    "output": combined_startup,
                },
                context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
                error={"code": code, "message": message},
                timing=make_timing(started_at, (time.time() - started_ts) * 1000),
            )
            if args.as_json:
                output_json(result)
            else:
                print(f"错误: {message}", file=sys.stderr)
            sys.exit(1)
        if proc.poll() is not None:
            break
        time.sleep(0.05)

    if proc.poll() is not None:
        combined_startup = "\n".join(buffered_stderr + buffered_stdout)
        message = combined_startup or "probe-rs RTT 启动失败"
        result = make_result(
            status="error",
            action="rtt",
            summary="probe-rs RTT 启动失败",
            details={
                "chip": params["chip"],
                "command": cmd,
                "returncode": proc.returncode,
                "output": combined_startup,
            },
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            error={"code": "startup_failed", "message": message},
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        if args.as_json:
            output_json(result)
        else:
            print(f"错误: {message}", file=sys.stderr)
        sys.exit(1)

    state_info = update_state_entry(
        "last_observe",
        {
            "provider": "probe-rs",
            "action": "rtt",
            "chip": params["chip"],
            "probe": params["probe"] or "",
            "protocol": params["protocol"],
            "speed": params["speed"],
        },
        str(workspace),
    )
    save_project_config(str(workspace), {
        "chip": params["chip"],
        "protocol": params["protocol"],
        "probe": params["probe"] or "",
        "speed": params["speed"],
        "connect_under_reset": params["connect_under_reset"],
    })

    if args.as_json:
        header = make_result(
            status="ok",
            action="rtt",
            summary="probe-rs RTT 已启动",
            details={"command": cmd, "chip": params["chip"], "probe": params["probe"] or ""},
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            state=state_info,
            timing=make_timing(started_at, (time.time() - started_ts) * 1000),
        )
        output_json(header)
    else:
        print(f"[probe-rs rtt] 已启动，chip={params['chip']}")

    deadline = time.time() + args.duration if args.duration > 0 else None
    stdout_done = False
    stderr_done = False
    lines = len(buffered_stdout)

    for item in buffered_stdout:
        emit_stream_record(source="probe-rs", channel_type="rtt", text=item, as_json=args.as_json)
    for item in buffered_stderr:
        emit_stream_record(source="probe-rs", channel_type="rtt", stream_type="stderr", text=item, as_json=args.as_json)

    try:
        while True:
            if deadline and time.time() >= deadline:
                break
            if proc.poll() is not None and stdout_done and stderr_done:
                break

            handled = False
            try:
                item = stdout_queue.get(timeout=0.1)
                handled = True
                if item is None:
                    stdout_done = True
                else:
                    lines += 1
                    emit_stream_record(source="probe-rs", channel_type="rtt", text=item, as_json=args.as_json)
            except queue.Empty:
                pass

            try:
                item = stderr_queue.get_nowait()
                handled = True
                if item is None:
                    stderr_done = True
                else:
                    emit_stream_record(source="probe-rs", channel_type="rtt", stream_type="stderr", text=item, as_json=args.as_json)
            except queue.Empty:
                pass

            if not handled:
                time.sleep(0.05)
    finally:
        cleanup(proc)

    elapsed_ms = (time.time() - started_ts) * 1000
    if args.as_json:
        footer = make_result(
            status="ok",
            action="rtt",
            summary="probe-rs RTT 已结束",
            details={"chip": params["chip"], "lines": lines},
            context=parameter_context(provider="probe-rs", workspace=str(workspace), parameter_sources=parameter_sources, config_path=config_path),
            metrics={"lines": lines},
            timing=make_timing(started_at, elapsed_ms),
        )
        output_json(footer)
    else:
        print(f"[probe-rs rtt] 已结束，lines={lines}")


if __name__ == "__main__":
    main()