tools: correct port map + esp8266 USB monitor smoke + interactive USB gate

hardware/firmware/tools/dev/ports_map.json

@@ -1,7 +1,7 @@
 {
   "location": {
-    "20-6.1.1": "esp8266",
-    "20-6.1.2": "esp32"
+    "20-6.1.1": "esp32",
+    "20-6.1.2": "esp8266_usb"
   },
   "vidpid": {
     "2e8a:0005": "rp2040",

hardware/firmware/tools/dev/serial_smoke.py

@@ -18,21 +18,34 @@ except ImportError:
     sys.exit(2)
 
 PING_COMMAND = "PING"
-PING_OK_PATTERN = re.compile(r"\b(PONG|OK|ACK|UNKNOWN)\b|BOOT|STATUS|HELLO|STAT", re.IGNORECASE)
+PING_OK_PATTERN = re.compile(r"\b(PONG|ACK|OK|UNKNOWN|HELLO|STAT|STATUS)\b", re.IGNORECASE)
+READY_PATTERN = re.compile(r"\[SCREEN\]\s*Ready\.", re.IGNORECASE)
+FATAL_PATTERN = re.compile(
+    r"(User exception|Exception|panic|abort|assert|rst cause|stack smashing|Guru Meditation|Fatal)",
+    re.IGNORECASE,
+)
+REBOOT_PATTERN = re.compile(r"(ets Jan|rst:|boot mode|load:0x|entry 0x)", re.IGNORECASE)
 
 ROOT = Path(__file__).resolve().parents[2]
 PORTS_MAP_PATH = ROOT / "tools" / "dev" / "ports_map.json"
 DEFAULT_PORTS_MAP = {
     "location": {
         "20-6.1.1": "esp32",
-        "20-6.1.2": "esp8266",
+        "20-6.1.2": "esp8266_usb",
     },
     "vidpid": {
         "2e8a:0005": "rp2040",
         "2e8a:000a": "rp2040",
     },
 }
-ROLE_PRIORITY = ["esp32", "esp8266", "rp2040"]
+ROLE_PRIORITY = ["esp32", "esp8266_usb", "rp2040"]
+
+
+def normalize_role(role: str) -> str:
+    value = (role or "").strip().lower()
+    if value in ("esp8266", "esp8266_usb", "ui", "oled"):
+        return "esp8266_usb"
+    return value
 
 
 def exit_no_hw(wait_port, allow, reason=None):
@@ -69,16 +82,16 @@ def normalize_ports_map(raw_map):
         if isinstance(raw_location, dict):
             for key, value in raw_location.items():
                 if isinstance(value, str):
-                    location[str(key).lower()] = value
+                    location[str(key).lower()] = normalize_role(value)
                 elif isinstance(value, dict) and isinstance(value.get("role"), str):
-                    location[str(key).lower()] = value["role"]
+                    location[str(key).lower()] = normalize_role(value["role"])
         raw_vidpid = raw_map.get("vidpid", {})
         if isinstance(raw_vidpid, dict):
             for key, value in raw_vidpid.items():
                 if isinstance(value, str):
-                    vidpid[str(key).lower()] = value
+                    vidpid[str(key).lower()] = normalize_role(value)
                 elif isinstance(value, dict) and isinstance(value.get("role"), str):
-                    vidpid[str(key).lower()] = value["role"]
+                    vidpid[str(key).lower()] = normalize_role(value["role"])
         return {"location": location, "vidpid": vidpid}
 
     # Backward compatibility with the previous shape:
@@ -89,9 +102,9 @@ def normalize_ports_map(raw_map):
             continue
         for key, value in platform_map.items():
             if isinstance(value, dict) and isinstance(value.get("role"), str):
-                location[str(key).lower()] = value["role"]
+                location[str(key).lower()] = normalize_role(value["role"])
             elif isinstance(value, str):
-                location[str(key).lower()] = value
+                location[str(key).lower()] = normalize_role(value)
     return {"location": location, "vidpid": vidpid}
 
 
@@ -207,16 +220,12 @@ def find_port_by_name(name, wait_port):
     return None
 
 
-def read_until_match(ser: Serial, pattern: re.Pattern, timeout_s: float) -> bool:
+def read_lines(ser: Serial, timeout_s: float):
     deadline = time.time() + timeout_s
     while time.time() < deadline:
         line = ser.readline().decode("utf-8", errors="ignore").strip()
-        if not line:
-            continue
-        print(f"[rx] {line}")
-        if pattern.search(line):
-            return True
-    return False
+        if line:
+            yield line
 
 
 def run_smoke(device, baud, timeout):
@@ -225,13 +234,91 @@ def run_smoke(device, baud, timeout):
         with Serial(device, baud, timeout=0.2) as ser:
             time.sleep(0.15)
             ser.reset_input_buffer()
-            print(f"[tx] {PING_COMMAND}")
-            ser.write((PING_COMMAND + "\n").encode("ascii"))
-            if read_until_match(ser, PING_OK_PATTERN, timeout):
-                print(f"[ok] {PING_COMMAND}")
-                return True
-            print(f"[fail] no expected response for {PING_COMMAND}", file=sys.stderr)
-            return False
+            handshake_hits = 0
+            post_handshake = False
+            handshake_deadline = time.time() + max(timeout * 2, 1.2)
+
+            while time.time() < handshake_deadline and handshake_hits < 2:
+                print(f"[tx] {PING_COMMAND}")
+                ser.write((PING_COMMAND + "\n").encode("ascii"))
+                for line in read_lines(ser, max(0.7, timeout)):
+                    print(f"[rx] {line}")
+                    if FATAL_PATTERN.search(line):
+                        print(f"[fail] fatal marker detected: {line}", file=sys.stderr)
+                        return False
+                    if REBOOT_PATTERN.search(line):
+                        # Boot noise can happen before a stable handshake.
+                        continue
+                    if PING_OK_PATTERN.search(line):
+                        handshake_hits += 1
+                        print(f"[ok] handshake {handshake_hits}/2")
+                        if handshake_hits >= 2:
+                            post_handshake = True
+                            break
+                if handshake_hits < 2:
+                    time.sleep(0.15)
+
+            if not post_handshake:
+                print(f"[fail] handshake incomplete ({handshake_hits}/2)", file=sys.stderr)
+                return False
+
+            stable_until = time.time() + 3.0
+            while time.time() < stable_until:
+                for line in read_lines(ser, 0.35):
+                    print(f"[rx] {line}")
+                    if FATAL_PATTERN.search(line):
+                        print(f"[fail] fatal marker after handshake: {line}", file=sys.stderr)
+                        return False
+                    if REBOOT_PATTERN.search(line):
+                        print(f"[fail] reboot marker after handshake: {line}", file=sys.stderr)
+                        return False
+            print(f"[ok] {PING_COMMAND} stable")
+            return True
+    except Exception as exc:
+        print(f"[error] serial failure on {device}: {exc}", file=sys.stderr)
+        return False
+
+
+def run_monitor_smoke(device, baud):
+    print(f"[smoke] monitor-only on {device} (baud={baud})")
+    try:
+        with Serial(device, baud, timeout=0.2) as ser:
+            time.sleep(0.15)
+            ser.reset_input_buffer()
+            ready_deadline = time.time() + 4.0
+            saw_ready = False
+            while time.time() < ready_deadline:
+                line = ser.readline().decode("utf-8", errors="ignore").strip()
+                if not line:
+                    continue
+                print(f"[rx] {line}")
+                if FATAL_PATTERN.search(line):
+                    print(f"[fail] fatal marker detected: {line}", file=sys.stderr)
+                    return False
+                if REBOOT_PATTERN.search(line):
+                    print(f"[fail] reboot marker detected: {line}", file=sys.stderr)
+                    return False
+                if READY_PATTERN.search(line):
+                    saw_ready = True
+                    break
+            if not saw_ready:
+                print("[fail] ready marker missing", file=sys.stderr)
+                return False
+
+            stable_until = time.time() + 3.0
+            while time.time() < stable_until:
+                line = ser.readline().decode("utf-8", errors="ignore").strip()
+                if not line:
+                    continue
+                print(f"[rx] {line}")
+                if FATAL_PATTERN.search(line):
+                    print(f"[fail] fatal marker after ready: {line}", file=sys.stderr)
+                    return False
+                if REBOOT_PATTERN.search(line):
+                    print(f"[fail] reboot marker after ready: {line}", file=sys.stderr)
+                    return False
+            print("[ok] monitor stable")
+            return True
     except Exception as exc:
         print(f"[error] serial failure on {device}: {exc}", file=sys.stderr)
         return False
@@ -241,9 +328,9 @@ def main() -> int:
     parser = argparse.ArgumentParser(description="Run quick serial smoke with auto port detection.")
     parser.add_argument("--port", help="Explicit serial port to use (optional)")
     parser.add_argument("--wait-port", type=int, default=30, help="Seconds to wait for a port (default 30)")
-    parser.add_argument("--role", choices=["auto", "all", "esp32", "esp8266", "rp2040"], default="auto")
+    parser.add_argument("--role", choices=["auto", "all", "esp32", "esp8266", "esp8266_usb", "rp2040"], default="auto")
     parser.add_argument("--all", action="store_true", help="Run smoke on all detected roles (when not auto)")
-    parser.add_argument("--baud", type=int, default=19200, help="Serial baud (default 19200)")
+    parser.add_argument("--baud", type=int, default=0, help="Serial baud override (role default when omitted)")
     parser.add_argument("--timeout", type=float, default=1.0, help="Per-command timeout (seconds)")
     parser.add_argument(
         "--allow-no-hardware",
@@ -252,6 +339,7 @@ def main() -> int:
     )
     parser.add_argument("--prefer-cu", action="store_true", default=platform.system() == "Darwin", help="Prefer /dev/cu.* on macOS")
     args = parser.parse_args()
+    args.role = normalize_role(args.role)
     allow_no_hardware = args.allow_no_hardware or os.environ.get("ZACUS_ALLOW_NO_HW") == "1"
 
     ports_map = load_ports_map()
@@ -266,6 +354,13 @@ def main() -> int:
         if not detection and args.role == "auto":
             print("[error] failed to classify the explicit port", file=sys.stderr)
             return 1
+        if args.role in ("esp32", "esp8266_usb", "rp2040") and args.role not in detection:
+            location = parse_location(getattr(port, "hwid", "")) or "unknown"
+            detection[args.role] = {
+                "device": port.device,
+                "hwid": getattr(port, "hwid", ""),
+                "location": location,
+            }
     else:
         baseline_ports = list(list_ports.comports())
         if baseline_ports:
@@ -307,6 +402,15 @@ def main() -> int:
             targets.append({"role": role, **info})
         elif args.role != "auto":
             print(f"[warn] requested role {role} not detected", file=sys.stderr)
+            if args.port:
+                targets.append(
+                    {
+                        "role": role,
+                        "device": args.port,
+                        "hwid": "",
+                        "location": "explicit",
+                    }
+                )
 
     if not targets:
         return 1
@@ -316,8 +420,12 @@ def main() -> int:
         role = entry["role"]
         device = entry["device"]
         location = entry["location"]
+        baud = args.baud or (115200 if role in ("esp32", "esp8266_usb") else 19200)
         print(f"[detect] role={role} device={device} location={location}")
-        ok = run_smoke(device, args.baud, args.timeout)
+        if role == "esp8266_usb":
+            ok = run_monitor_smoke(device, baud)
+        else:
+            ok = run_smoke(device, baud, args.timeout)
         overall_ok = overall_ok and ok
 
     return 0 if overall_ok else 1

tools/test/hw_now.sh

@@ -197,13 +197,17 @@ print("RESOLVE_PORT_ESP32=" + shlex.quote(str(v.get("ports", {}).get("esp32", ""
 print("RESOLVE_PORT_ESP8266=" + shlex.quote(str(v.get("ports", {}).get("esp8266", ""))))
 print("RESOLVE_REASON_ESP32=" + shlex.quote(str(v.get("reasons", {}).get("esp32", ""))))
 print("RESOLVE_REASON_ESP8266=" + shlex.quote(str(v.get("reasons", {}).get("esp8266", ""))))
+print("RESOLVE_LOCATION_ESP32=" + shlex.quote(str(v.get("details", {}).get("esp32", {}).get("location", ""))))
+print("RESOLVE_LOCATION_ESP8266=" + shlex.quote(str(v.get("details", {}).get("esp8266", {}).get("location", ""))))
+print("RESOLVE_ROLE_ESP32=" + shlex.quote(str(v.get("details", {}).get("esp32", {}).get("role", ""))))
+print("RESOLVE_ROLE_ESP8266=" + shlex.quote(str(v.get("details", {}).get("esp8266", {}).get("role", ""))))
 print("RESOLVE_NOTES=" + shlex.quote(" | ".join(v.get("notes", []))))
 ')"
 
   PORT_ESP32="$RESOLVE_PORT_ESP32"
   PORT_ESP8266="$RESOLVE_PORT_ESP8266"
-  action_log "[port] ESP32 = ${PORT_ESP32:-n/a} (${RESOLVE_REASON_ESP32:-unresolved})"
-  action_log "[port] ESP8266 = ${PORT_ESP8266:-n/a} (${RESOLVE_REASON_ESP8266:-unresolved})"
+  action_log "[port] ESP32 = ${PORT_ESP32:-n/a} location=${RESOLVE_LOCATION_ESP32:-unknown} role=${RESOLVE_ROLE_ESP32:-esp32} (${RESOLVE_REASON_ESP32:-unresolved})"
+  action_log "[port] ESP8266 = ${PORT_ESP8266:-n/a} location=${RESOLVE_LOCATION_ESP8266:-unknown} role=${RESOLVE_ROLE_ESP8266:-esp8266_usb} (${RESOLVE_REASON_ESP8266:-unresolved})"
   if [[ -n "$RESOLVE_NOTES" ]]; then
     action_log "[port] notes: $RESOLVE_NOTES"
   fi
@@ -211,19 +215,17 @@ print("RESOLVE_NOTES=" + shlex.quote(" | ".join(v.get("notes", []))))
 }
 
 countdown_usb_prompt() {
-  if [[ "$DRY_RUN" == "1" ]]; then
-    action_log "[dry-run] skip countdown"
+  if [[ "$DRY_RUN" == "1" || "${ZACUS_NO_CONFIRM:-0}" == "1" ]]; then
+    action_log "[info] USB confirm prompt skipped"
     return
   fi
-  local left="$COUNTDOWN_SEC"
-  while (( left > 0 )); do
-    if (( left % 5 == 0 || left == COUNTDOWN_SEC )); then
-      action_log "⚠️ BRANCHE L'USB MAINTENANT ⚠️ ($left s)"
-      printf '\a\a\a' | tee -a "$run_log" >/dev/null
-    fi
-    sleep 1
-    left=$((left - 1))
+  local i
+  for i in 1 2 3; do
+    action_log "⚠️ BRANCHE L’USB MAINTENANT ⚠️"
+    printf '\a' >>"$run_log"
   done
+  read -r -p "USB branchés ? Appuie sur Entrée pour continuer..." < /dev/tty
+  action_log "[info] USB confirmation received"
 }
 
 run_smoke_with_policy() {
@@ -236,8 +238,10 @@ run_smoke_with_policy() {
     bauds=("$SMOKE_BAUD_OVERRIDE")
   elif [[ "$role" == "esp32" ]]; then
     bauds=("115200" "19200")
+  elif [[ "$role" == "esp8266_usb" ]]; then
+    bauds=("115200")
   else
-    bauds=("115200" "19200")
+    bauds=("115200")
   fi
 
   local b
@@ -318,11 +322,35 @@ cd "$FW"
 action_log "[info] FW=$FW"
 action_log "[info] artifacts: $artifact_dir"
 
-if ! resolve_ports; then
-  append_step "resolve_ports" "FAIL" "2" "$artifact_dir/ports_resolve.json" "port resolution failed"
-  exit 2
+if [[ "$AUTO_PORTS" == "1" ]]; then
+  while true; do
+    countdown_usb_prompt
+    if ! resolve_ports 1; then
+      append_step "resolve_ports" "FAIL" "2" "$artifact_dir/ports_resolve.json" "port resolution failed"
+      if [[ "${ZACUS_REQUIRE_HW:-1}" == "1" && "${ZACUS_NO_CONFIRM:-0}" != "1" && "$DRY_RUN" != "1" ]]; then
+        action_log "[warn] hardware required: reconnect USB and retry"
+        continue
+      fi
+      exit 2
+    fi
+    if [[ "${ZACUS_REQUIRE_HW:-1}" == "1" && ( -z "$PORT_ESP32" || -z "$PORT_ESP8266" ) ]]; then
+      append_step "resolve_ports" "FAIL" "2" "$artifact_dir/ports_resolve.json" "missing required roles"
+      if [[ "${ZACUS_NO_CONFIRM:-0}" != "1" && "$DRY_RUN" != "1" ]]; then
+        action_log "[warn] missing role ports, reconnect then confirm again"
+        continue
+      fi
+      exit 2
+    fi
+    append_step "resolve_ports" "PASS" "0" "$artifact_dir/ports_resolve.json" "ok"
+    break
+  done
+else
+  if ! resolve_ports; then
+    append_step "resolve_ports" "FAIL" "2" "$artifact_dir/ports_resolve.json" "port resolution failed"
+    exit 2
+  fi
+  append_step "resolve_ports" "PASS" "0" "$artifact_dir/ports_resolve.json" "ok"
 fi
-append_step "resolve_ports" "PASS" "0" "$artifact_dir/ports_resolve.json" "ok"
 
 if [[ "$SKIP_UPLOAD" == "0" ]]; then
   if [[ "$SKIP_BUILD" == "0" ]]; then
@@ -341,18 +369,8 @@ else
   append_step "upload_esp8266" "SKIP" "0" "$artifact_dir/esp8266_upload.log" "--skip-upload"
 fi
 
-countdown_usb_prompt
-if [[ "$AUTO_PORTS" == "1" && "$DRY_RUN" == "0" ]]; then
-  prev_esp32="$PORT_ESP32"
-  prev_esp8266="$PORT_ESP8266"
-  if resolve_ports 1; then
-    if [[ "$PORT_ESP32" != "$prev_esp32" || "$PORT_ESP8266" != "$prev_esp8266" ]]; then
-      action_log "[port] refreshed after countdown: ESP32=$PORT_ESP32 ESP8266=$PORT_ESP8266"
-    fi
-  fi
-fi
 run_smoke_with_policy "esp32" "$PORT_ESP32" "$artifact_dir/smoke_esp32.log" || true
-run_smoke_with_policy "esp8266" "$PORT_ESP8266" "$artifact_dir/smoke_esp8266.log" || true
+run_smoke_with_policy "esp8266_usb" "$PORT_ESP8266" "$artifact_dir/smoke_esp8266.log" || true
 check_ui_link_status || true
 
 run_step "gate_s1" "$artifact_dir/gate_s1.log" bash "$GATE_SCRIPT" --sprint s1 --port-esp32 "$PORT_ESP32" --port-esp8266 "$PORT_ESP8266" --no-auto-ports --require-hw || true
@@ -370,6 +388,10 @@ port_esp32 = sys.argv[4]
 port_esp8266 = sys.argv[5]
 env_esp32 = sys.argv[6]
 env_esp8266 = sys.argv[7]
+ports_resolve = artifact_dir / "ports_resolve.json"
+resolve_data = {}
+if ports_resolve.exists():
+    resolve_data = json.loads(ports_resolve.read_text(encoding="utf-8"))
 
 steps = []
 for raw in steps_tsv.read_text(encoding="utf-8").splitlines():
@@ -378,7 +400,7 @@ for raw in steps_tsv.read_text(encoding="utf-8").splitlines():
     name, status, exit_code, log_file, details = raw.split("\t", 4)
     steps.append({"name": name, "status": status, "exit_code": int(exit_code), "log_file": log_file, "details": details})
 
-critical = {"resolve_ports", "upload_esp32", "upload_esp8266", "smoke_esp32", "smoke_esp8266", "gate_s1", "gate_s2"}
+critical = {"resolve_ports", "upload_esp32", "upload_esp8266", "smoke_esp32", "smoke_esp8266_usb", "gate_s1", "gate_s2"}
 overall = "PASS"
 if any(s["name"] in critical and s["status"] == "FAIL" for s in steps):
     overall = "FAIL"
@@ -388,7 +410,20 @@ elif all(s["status"] == "SKIP" for s in steps):
 ui_status = next((s for s in steps if s["name"] == "ui_link"), None)
 summary = {
     "timestamp": timestamp,
-    "ports": {"esp32": port_esp32, "esp8266": port_esp8266, "detection_reason": "see ports_resolve.json"},
+    "ports": {
+        "esp32": {
+            "port": port_esp32,
+            "location": resolve_data.get("details", {}).get("esp32", {}).get("location", ""),
+            "role": "esp32",
+            "reason": resolve_data.get("details", {}).get("esp32", {}).get("reason", ""),
+        },
+        "esp8266_usb": {
+            "port": port_esp8266,
+            "location": resolve_data.get("details", {}).get("esp8266", {}).get("location", ""),
+            "role": "esp8266_usb",
+            "reason": resolve_data.get("details", {}).get("esp8266", {}).get("reason", ""),
+        },
+    },
     "envs": {"esp32": env_esp32, "esp8266": env_esp8266},
     "steps": steps,
     "result": overall,
@@ -404,8 +439,8 @@ rows = [
     "# RC live summary",
     "",
     f"- Result: **{overall}**",
-    f"- ESP32 port: `{port_esp32}`",
-    f"- ESP8266 port: `{port_esp8266}`",
+    f"- ESP32 port: `{port_esp32}` (location `{resolve_data.get('details', {}).get('esp32', {}).get('location', '')}`)",
+    f"- ESP8266 USB port: `{port_esp8266}` (location `{resolve_data.get('details', {}).get('esp8266', {}).get('location', '')}`)",
     f"- UI link: `{ui_status['status'] if ui_status else 'n/a'}`",
     "",
     "| Step | Status | Exit | Log | Details |",

tools/test/resolve_ports.py

@@ -0,0 +1,446 @@
+#!/usr/bin/env python3
+"""Resolve ESP32/ESP8266 serial ports deterministically for local hardware runs."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+try:
+    from serial.tools import list_ports
+except ImportError:
+    print(json.dumps({"status": "fail", "notes": ["pyserial missing: pip install pyserial"]}))
+    raise SystemExit(2)
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+FW_ROOT = REPO_ROOT / "hardware" / "firmware"
+PORTS_MAP_PATH = FW_ROOT / "tools" / "dev" / "ports_map.json"
+SERIAL_SMOKE = FW_ROOT / "tools" / "dev" / "serial_smoke.py"
+
+USB_HINTS = ("slab", "usbserial", "wch", "ch340", "cp210", "usbmodem")
+ROLE_HINTS = {
+    "esp32": ("esp32",),
+    "esp8266_usb": ("esp8266", "nodemcu", "ch340", "cp210"),
+}
+REQUEST_TO_CANONICAL = {"esp32": "esp32", "esp8266": "esp8266_usb"}
+CANONICAL_TO_REQUEST = {"esp32": "esp32", "esp8266_usb": "esp8266"}
+
+
+def is_bluetooth_port(device: str) -> bool:
+    lower = device.lower()
+    return "bluetooth" in lower or lower.endswith(".blth")
+
+
+def normalize_role(value: Optional[str]) -> Optional[str]:
+    if not value:
+        return None
+    lower = value.strip().lower()
+    if lower in ("ui", "oled", "esp8266", "esp8266_usb", "port-ui"):
+        return "esp8266_usb"
+    if lower == "esp32":
+        return "esp32"
+    return None
+
+
+def parse_location(port) -> str:
+    location = getattr(port, "location", None)
+    if location:
+        return str(location).lower()
+    hwid = str(getattr(port, "hwid", "") or "")
+    m = re.search(r"LOCATION=([\w\-.]+)", hwid)
+    if m:
+        return m.group(1).lower()
+    return ""
+
+
+def load_ports_map() -> Dict[str, Dict[str, str]]:
+    default = {"location": {}, "vidpid": {}}
+    if not PORTS_MAP_PATH.exists():
+        return default
+    try:
+        raw = json.loads(PORTS_MAP_PATH.read_text(encoding="utf-8"))
+    except Exception:
+        return default
+    location: Dict[str, str] = {}
+    for key, value in (raw.get("location") or {}).items():
+        role = normalize_role(str(value))
+        if role:
+            location[str(key).lower()] = role
+    vidpid: Dict[str, str] = {}
+    for key, value in (raw.get("vidpid") or {}).items():
+        role = normalize_role(str(value))
+        if role:
+            vidpid[str(key).lower()] = role
+    return {"location": location, "vidpid": vidpid}
+
+
+def is_usb_like(port) -> bool:
+    text = " ".join(
+        [
+            str(getattr(port, "device", "") or ""),
+            str(getattr(port, "description", "") or ""),
+            str(getattr(port, "manufacturer", "") or ""),
+            str(getattr(port, "product", "") or ""),
+            str(getattr(port, "hwid", "") or ""),
+        ]
+    ).lower()
+    if "bluetooth" in text or ".blth" in text:
+        return False
+    return any(token in text for token in USB_HINTS)
+
+
+def is_candidate_port(port) -> bool:
+    device = str(getattr(port, "device", "") or "")
+    if not device:
+        return False
+    if is_bluetooth_port(device):
+        return False
+    if device.startswith("/dev/cu."):
+        return True
+    return is_usb_like(port)
+
+
+def score_port(port, prefer_cu: bool) -> int:
+    device = str(getattr(port, "device", "") or "")
+    score = 50
+    if prefer_cu and device.startswith("/dev/cu."):
+        score -= 20
+    elif device.startswith("/dev/cu."):
+        score -= 10
+    if "slab" in device.lower() or "usbserial" in device.lower() or "wch" in device.lower():
+        score -= 6
+    if is_usb_like(port):
+        score -= 8
+    return score
+
+
+def role_from_map(port, ports_map: Dict[str, Dict[str, str]]) -> Optional[str]:
+    location = parse_location(port)
+    if location and location in ports_map["location"]:
+        return ports_map["location"][location]
+    vid = getattr(port, "vid", None)
+    pid = getattr(port, "pid", None)
+    if vid is not None and pid is not None:
+        key = f"{vid:04x}:{pid:04x}".lower()
+        if key in ports_map["vidpid"]:
+            return ports_map["vidpid"][key]
+    return None
+
+
+def role_from_hint(port) -> Optional[str]:
+    text = " ".join(
+        [
+            str(getattr(port, "description", "") or ""),
+            str(getattr(port, "manufacturer", "") or ""),
+            str(getattr(port, "product", "") or ""),
+            str(getattr(port, "hwid", "") or ""),
+        ]
+    ).lower()
+    for role, hints in ROLE_HINTS.items():
+        if any(h in text for h in hints):
+            return role
+    return None
+
+
+def probe_port(device: str, role: str, baud: int) -> bool:
+    cmd = [
+        sys.executable,
+        str(SERIAL_SMOKE),
+        "--role",
+        role,
+        "--port",
+        device,
+        "--baud",
+        str(baud),
+        "--timeout",
+        "0.6",
+        "--wait-port",
+        "1",
+    ]
+    try:
+        proc = subprocess.run(
+            cmd,
+            cwd=str(FW_ROOT),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            timeout=3.0,
+            check=False,
+        )
+    except Exception:
+        return False
+    return proc.returncode == 0
+
+
+def choose_interactive(candidates: List[dict], role: str) -> Optional[str]:
+    print(f"Select port for {role}:", file=sys.stderr)
+    for idx, item in enumerate(candidates, start=1):
+        print(f"  {idx}) {item['device']} [{item['reason']}]", file=sys.stderr)
+    try:
+        raw = input(f"{role} port number [1-{len(candidates)}]: ").strip()
+    except EOFError:
+        return None
+    if not raw.isdigit():
+        return None
+    pos = int(raw)
+    if pos < 1 or pos > len(candidates):
+        return None
+    return candidates[pos - 1]["device"]
+
+
+def gather_ports(wait_port: int) -> List:
+    deadline = time.monotonic() + max(1, wait_port)
+    while True:
+        ports = list(list_ports.comports())
+        if ports:
+            return ports
+        if time.monotonic() >= deadline:
+            return []
+        time.sleep(0.4)
+
+
+def classify(ports: List, prefer_cu: bool, ports_map: Dict[str, Dict[str, str]], allow_probe: bool) -> Tuple[Dict[str, str], Dict[str, str], Dict[str, str], Dict[str, dict], List[str]]:
+    filtered = [p for p in ports if is_candidate_port(p)]
+    usb_ports = [p for p in filtered if is_usb_like(p)]
+    ports = usb_ports if usb_ports else filtered
+    candidates: List[dict] = []
+    for port in ports:
+        device = str(getattr(port, "device", "") or "")
+        if not device:
+            continue
+        mapped_role = role_from_map(port, ports_map)
+        hinted_role = role_from_hint(port)
+        role = mapped_role or hinted_role
+        location = parse_location(port)
+        reason = ""
+        if mapped_role:
+            reason = f"location-map:{location or 'unknown'}"
+        elif hinted_role:
+            reason = "usb-hint"
+        else:
+            reason = "fallback"
+        candidates.append(
+            {
+                "device": device,
+                "role": role,
+                "location": location,
+                "reason": reason,
+                "score": score_port(port, prefer_cu),
+            }
+        )
+
+    candidates.sort(key=lambda x: (x["score"], x["device"]))
+    by_role: Dict[str, List[dict]] = {"esp32": [], "esp8266": []}
+    fallback: List[dict] = []
+    for cand in candidates:
+        canonical_role = normalize_role(cand.get("role"))
+        requested_role = CANONICAL_TO_REQUEST.get(canonical_role or "", "")
+        if requested_role in by_role:
+            by_role[requested_role].append(cand)
+        else:
+            fallback.append(cand)
+
+    selected: Dict[str, str] = {}
+    reasons: Dict[str, str] = {}
+    probe_baud: Dict[str, str] = {}
+    details: Dict[str, dict] = {}
+    notes: List[str] = []
+
+    for role in ("esp32", "esp8266"):
+        if by_role[role]:
+            selected[role] = by_role[role][0]["device"]
+            reasons[role] = by_role[role][0]["reason"]
+            details[role] = {
+                "port": by_role[role][0]["device"],
+                "location": by_role[role][0]["location"] or "unknown",
+                "role": REQUEST_TO_CANONICAL[role],
+                "reason": by_role[role][0]["reason"],
+            }
+
+    if allow_probe:
+        unresolved = [r for r in ("esp32", "esp8266") if r not in selected]
+        if unresolved and len(candidates) <= 4:
+            for cand in candidates:
+                for role in unresolved:
+                    if role in selected:
+                        continue
+                    for baud in (19200, 115200):
+                        if probe_port(cand["device"], REQUEST_TO_CANONICAL[role], baud):
+                            selected[role] = cand["device"]
+                            reasons[role] = f"probe:{baud}"
+                            probe_baud[role] = str(baud)
+                            details[role] = {
+                                "port": cand["device"],
+                                "location": cand["location"] or "unknown",
+                                "role": REQUEST_TO_CANONICAL[role],
+                                "reason": f"probe:{baud}",
+                            }
+                            break
+                    if role in selected:
+                        break
+
+    used = set(selected.values())
+    for role in ("esp32", "esp8266"):
+        if role in selected:
+            continue
+        for cand in candidates:
+            if cand["device"] in used:
+                continue
+            selected[role] = cand["device"]
+            reasons[role] = f"deterministic:{cand['reason']}"
+            details[role] = {
+                "port": cand["device"],
+                "location": cand["location"] or "unknown",
+                "role": REQUEST_TO_CANONICAL[role],
+                "reason": f"deterministic:{cand['reason']}",
+            }
+            used.add(cand["device"])
+            break
+
+    if len(candidates) > 2:
+        notes.append(f"multiple candidates: {len(candidates)}")
+
+    return selected, reasons, probe_baud, details, notes
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Resolve serial ports for ESP32 + ESP8266")
+    parser.add_argument("--port-esp32", default="")
+    parser.add_argument("--port-esp8266", default="")
+    parser.add_argument("--wait-port", type=int, default=20)
+    parser.add_argument("--need-esp32", action="store_true")
+    parser.add_argument("--need-esp8266", action="store_true")
+    parser.add_argument("--allow-no-hardware", action="store_true")
+    parser.add_argument("--auto-ports", dest="auto_ports", action="store_true", default=True)
+    parser.add_argument("--no-auto-ports", dest="auto_ports", action="store_false")
+    parser.add_argument("--prefer-cu", action="store_true")
+    parser.add_argument("--interactive", action="store_true")
+    args = parser.parse_args()
+
+    prefer_cu = args.prefer_cu or sys.platform == "darwin"
+    ports_map = load_ports_map()
+
+    result = {
+        "status": "pass",
+        "ports": {"esp32": args.port_esp32, "esp8266": args.port_esp8266},
+        "reasons": {
+            "esp32": "explicit" if args.port_esp32 else "",
+            "esp8266": "explicit" if args.port_esp8266 else "",
+        },
+        "probe_baud": {"esp32": "", "esp8266": ""},
+        "details": {
+            "esp32": {
+                "port": args.port_esp32,
+                "location": "explicit" if args.port_esp32 else "",
+                "role": "esp32" if args.port_esp32 else "",
+                "reason": "explicit" if args.port_esp32 else "",
+            },
+            "esp8266": {
+                "port": args.port_esp8266,
+                "location": "explicit" if args.port_esp8266 else "",
+                "role": "esp8266_usb" if args.port_esp8266 else "",
+                "reason": "explicit" if args.port_esp8266 else "",
+            },
+        },
+        "notes": [],
+    }
+
+    missing_roles = []
+    if args.need_esp32 and not result["ports"]["esp32"]:
+        missing_roles.append("esp32")
+    if args.need_esp8266 and not result["ports"]["esp8266"]:
+        missing_roles.append("esp8266")
+
+    candidates = []
+    if missing_roles and args.auto_ports:
+        ports = gather_ports(args.wait_port)
+        if not ports:
+            if args.allow_no_hardware:
+                result["status"] = "skip"
+                result["notes"].append("no serial ports detected")
+                print(json.dumps(result))
+                return 0
+            result["status"] = "fail"
+            result["notes"].append("no serial ports detected")
+            print(json.dumps(result))
+            return 1
+
+        selected, reasons, probe_baud, details, notes = classify(
+            ports=ports,
+            prefer_cu=prefer_cu,
+            ports_map=ports_map,
+            allow_probe=True,
+        )
+        result["notes"].extend(notes)
+
+        for role in missing_roles:
+            if role in selected:
+                result["ports"][role] = selected[role]
+                result["reasons"][role] = reasons.get(role, "auto")
+                if role in probe_baud:
+                    result["probe_baud"][role] = probe_baud[role]
+                result["details"][role] = details.get(role, result["details"][role])
+
+        # optional interactive disambiguation when both ports still unresolved or duplicated
+        if args.interactive and sys.stdin.isatty():
+            if result["ports"].get("esp32") and result["ports"].get("esp8266") and result["ports"]["esp32"] == result["ports"]["esp8266"]:
+                candidates = []
+                for p in list_ports.comports():
+                    candidates.append(
+                        {
+                            "device": p.device,
+                            "reason": f"location={parse_location(p) or 'unknown'}",
+                        }
+                    )
+                candidates.sort(key=lambda x: x["device"])
+                if candidates:
+                    pick1 = choose_interactive(candidates, "esp32")
+                    if pick1:
+                        result["ports"]["esp32"] = pick1
+                        result["reasons"]["esp32"] = "interactive"
+                    remaining = [c for c in candidates if c["device"] != result["ports"]["esp32"]]
+                    if remaining:
+                        pick2 = choose_interactive(remaining, "esp8266")
+                        if pick2:
+                            result["ports"]["esp8266"] = pick2
+                            result["reasons"]["esp8266"] = "interactive"
+
+    unresolved = []
+    if args.need_esp32 and not result["ports"].get("esp32"):
+        unresolved.append("esp32")
+    if args.need_esp8266 and not result["ports"].get("esp8266"):
+        unresolved.append("esp8266")
+
+    if unresolved:
+        if args.allow_no_hardware:
+            result["status"] = "skip"
+            result["notes"].append(f"unresolved roles: {','.join(unresolved)}")
+        else:
+            result["status"] = "fail"
+            result["notes"].append(f"unresolved roles: {','.join(unresolved)}")
+            result["notes"].append("check USB data cable, CP210x/CH340 driver, and press BOOT for ESP32 upload")
+            print(json.dumps(result))
+            return 1
+
+    if result["status"] == "pass":
+        if result["ports"].get("esp32") and result["ports"].get("esp8266") and result["ports"]["esp32"] == result["ports"]["esp8266"]:
+            result["status"] = "fail"
+            result["notes"].append("esp32 and esp8266 resolved to the same device")
+            if not args.allow_no_hardware:
+                print(json.dumps(result))
+                return 1
+
+    print(json.dumps(result))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())