#!/usr/bin/env python3
"""mcpdoctor.py -- why is my MCP server unhealthy? One command, one answer.

WHY THIS EXISTS
---------------
In the four weeks to 2026-09-12, 130 GitHub issues were opened whose *title*
contains both "MCP server" and "connection closed", across completely unrelated
projects -- a chip-vendor SDK, an editor, a browser agent, a Qwen desktop
client, Claude Code itself. Fifteen were read in full. Not one reporter had
worked out which of three repeatable causes they had hit; every one is "it just
says connection closed", followed by somebody else doing the archaeology in the
comments. The evidence pack is `candidates/2026-09-15-scout-5.md`.

The three causes, and what each looks like from the outside:

  1. THE LAUNCHER ISN'T THERE. `uvx` or `npx` is not installed, or a `${VAR}`
     in the config was never expanded, so the command cannot start. The client
     reports "Connection closed" and hides the real error.
  2. A DEPENDENCY MOVED UNDER IT. `mcp>=1.0.0` resolves to a breaking 2.x on
     the next install; the server starts, throws on import, and exits. The
     traceback goes to stderr, which the client throws away.
  3. THE TRANSPORT DOESN'T MATCH. Remotely: the server speaks the legacy
     HTTP+SSE transport and the client POSTs at it (or the reverse). Locally:
     the server prints a log line to stdout before its `initialize` reply, and
     that line corrupts the JSON-RPC stream.

A fourth was added on 2026-09-29 from a second, later sweep of the same search:

  4. THE LAUNCHER IS A POSIX SHELL AND THE MACHINE IS WINDOWS. A config that
     hardcodes `"command": "bash"` or `"sh"` works everywhere its author tried
     it and nowhere on native Windows, where the name resolves off the Windows
     PATH -- to WSL's shell, or to nothing, but never to Git Bash. Two issues
     opened a day apart, `AxiomCodeAI/axiomcodegraph#1233` and
     `aboul3ata/lazyweb-skill#156`, both reported as "connection closed". The
     evidence pack is `candidates/2026-09-28-scout.md`.

`tools/mcphealth.py` already decides exactly this, once a day, for ~19,000
registry endpoints -- but only for servers on the registry, and only as a label
on a page. This is the same logic pointed at one server, on demand, on your own
machine, answering in sentences.

The remote path is not a re-implementation: it calls `mcphealth.probe_endpoint`
directly, so the two-transport dispatcher has one implementation and cannot
drift. The stdio path is new -- `mcphealth.py` never had one, because a local
server has no address to probe.

The per-endpoint history comes from our own daily series. Inside this repo that
is `state/mcp-health/latest.json`; anywhere else -- a pip install, a copied
file -- it is the published feed, fetched once a day into `~/.cache/mcp-doctor/`
and never mentioned if the fetch fails.

USAGE
-----
    python3 tools/mcpdoctor.py https://mcp.example.com/mcp
    python3 tools/mcpdoctor.py -- uvx some-mcp-server --flag
    python3 tools/mcpdoctor.py --config .mcp.json              # list servers
    python3 tools/mcpdoctor.py --config .mcp.json --server foo
    python3 tools/mcpdoctor.py <url> --json                    # machine form

`--config` reads the client config you name and nothing else; a server is only
started when you name it with `--server` (or pass a command after `--`). The
process is spawned with an argument list, never through a shell.

Written 3.8-safe, stdlib only, no model tokens, like the rest of `tools/`.
"""
from __future__ import annotations

import argparse
import collections
import gzip
import json
import os
import pathlib
import platform
import re
import shutil
import subprocess
import sys
import threading
import time
import urllib.parse
import urllib.request

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "tools"))

import mcphealth                                        # noqa: E402

LATEST = ROOT / "state" / "mcp-health" / "latest.json"
SITE = "https://referencesource.org/mcp-health/"
FEED = SITE + "feed.json"

# The history is the only thing this tool has that a generic MCP inspector
# does not, and in a copy installed from a package there is no `state/` dir to
# read it out of. So when the state file is absent we fetch the published feed
# instead -- the same numbers, from the surface anyone can check -- keep it for
# a day, and say nothing at all if that fails. A diagnostic tool that printed a
# network error of its own while you are diagnosing a network problem would be
# actively unhelpful.
FEED_MAX_AGE = 24 * 3600    # seconds before the cached feed is refetched
FEED_RETRY_AFTER = 3600     # seconds to leave it alone after a failed fetch
FEED_TIMEOUT = 12.0
# Deliberately not mcphealth's UA: "feed fetches by distinct UA" is one of E6's
# counters, so a packaged copy asking for the feed should be legible as itself
# in the access log rather than as the daily prober. Deliberately not a bare
# "mcp-doctor" either -- that name is already an unrelated PyPI package
# (github.com/Jiansen/mcp-doctor, a tool-description linter, checked
# 2026-09-15), and a log line has to say whose tool it was.
FEED_UA = "referencesource-mcp-doctor/0.1 (+" + SITE + ")"

STDIO_TIMEOUT = 20.0        # seconds to wait for the initialize reply
STDIO_TOOLS_TIMEOUT = 20.0
STDERR_KEEP = 40            # lines of the server's own stderr to keep

# A stderr line matching one of these is quoted back as the likely cause. The
# rule is the same one the rest of this project runs on: never assert a cause
# without showing the line it was read off. If nothing matches, we say the
# server exited and print the tail verbatim rather than inventing a reason.
DEP_BREAK = [
    (r"ModuleNotFoundError: No module named ['\"]?([\w\.\-]+)",
     "dependency_break",
     "a Python import failed -- the package is not installed in the "
     "environment the launcher built"),
    (r"ImportError: cannot import name ['\"]?([\w\.]+)",
     "dependency_break",
     "a Python import failed on a NAME, which is what a dependency bumping a "
     "major version looks like: the module is there, the symbol moved"),
    (r"AttributeError: module ['\"]?([\w\.]+)['\"]? has no attribute",
     "dependency_break",
     "a dependency is present but the wrong version -- the attribute this "
     "code calls does not exist in the installed release"),
    (r"TypeError: .*unexpected keyword argument",
     "dependency_break",
     "a dependency changed a function signature under this code, which is a "
     "breaking version bump"),
    (r"pydantic.*ValidationError",
     "dependency_break",
     "a validation error at import time, which usually means a pinned "
     "library moved"),
    (r"ImportError: ", "dependency_break", "a Python import failed"),
    # Resolution failures are a DIFFERENT problem with different advice: the
    # environment was never built, so there is no version to pin back to.
    (r"No matching distribution found for ([\w\.\-\[\]<>=,]+)",
     "unresolvable",
     "the launcher could not find that package or version at all"),
    (r"(?:No solution found|Failed to build|failed to resolve|"
     r"could not find a version)",
     "unresolvable",
     "the launcher could not build the environment -- the package or the "
     "version it needs does not resolve"),
    (r"npm ERR! code ([A-Z_]+)", "unresolvable",
     "npm failed before the server ran"),
    (r"npm ERR! 404", "unresolvable", "npm could not find that package"),
]

LAUNCHERS = ("uvx", "uv", "npx", "npm", "node", "python", "python3", "bunx",
             "deno", "pipx", "docker")

# A fourth cause, added 2026-09-29 off two issues opened a day apart and fetched
# in full: `AxiomCodeAI/axiomcodegraph#1233` (2026-09-24) and
# `aboul3ata/lazyweb-skill#156` (2026-09-23). Both hardcode a bare POSIX shell
# name as the launcher `command`, on native Windows, and fail in two different
# shapes:
#
#   * `sh` is not on the Windows PATH at all -- Git for Windows ships `sh.exe`
#     but does not put its `bin\` on PATH -- so the spawn fails outright. That
#     shape is ALREADY caught by the `launcher_missing` branch below, because
#     `shutil.which("sh")` returns None. What it did not do is say why, on
#     Windows, specifically; `windows_shell` on the result adds that.
#   * `bash` DOES resolve, to `%LOCALAPPDATA%\Microsoft\WindowsApps\bash.exe`,
#     the placeholder that forwards to WSL's `/bin/bash`. So the process starts
#     and then dies handed a Windows-format path WSL cannot see, exit 127. That
#     shape was not caught at all: `which` succeeds, the spawn succeeds, and it
#     fell into `crashed_on_start` whose DEP_BREAK list has no pattern for
#     "No such file or directory" or exit 127 -- the user got "the server exited
#     with code 127 before answering" and no cause.
POSIX_SHELLS = ("bash", "sh")
SHELL_NOT_FOUND = re.compile(r"No such file or directory", re.I)


# --------------------------------------------------------------------------
# what our own daily probe already knows


def _cache_dir():
    base = (os.environ.get("MCP_DOCTOR_CACHE")
            or os.environ.get("XDG_CACHE_HOME")
            or os.path.join(os.path.expanduser("~"), ".cache"))
    if os.environ.get("MCP_DOCTOR_CACHE"):
        return pathlib.Path(base)
    return pathlib.Path(base) / "mcp-doctor"


def _age(path):
    try:
        return time.time() - path.stat().st_mtime
    except Exception:                                    # noqa: BLE001
        return None


def _usable_feed(text):
    """Parse a feed body, or None. A 200 is not evidence of the feed.

    A captive portal, a WAF challenge and an error page all come back 200 with
    a body, and caching one of those for a day would mean a day of silence with
    no way to tell it from "you are not a server we probe". So the shape is
    checked before anything is written: the three fields this tool reads have
    to be there and be the right type.
    """
    try:
        d = json.loads(text)
    except Exception:                                    # noqa: BLE001
        return None
    if not isinstance(d, dict):
        return None
    if not isinstance(d.get("servers"), list):
        return None
    if not isinstance(d.get("counts"), dict) or not d["counts"]:
        return None
    if not d.get("latest_date"):
        return None
    return d


def _fetch_feed():
    req = urllib.request.Request(FEED, headers={
        "User-Agent": FEED_UA,
        "Accept": "application/json",
        "Accept-Encoding": "gzip",
    })
    with urllib.request.urlopen(req, timeout=FEED_TIMEOUT) as r:
        raw = r.read()
        if (r.headers.get("Content-Encoding") or "").lower() == "gzip":
            raw = gzip.decompress(raw)
    return raw.decode("utf-8", "replace")


def _feed_copy():
    """The published feed, cached for a day. Silent on every failure."""
    cache = _cache_dir() / "feed.json"
    failed = _cache_dir() / "feed.failed"
    fresh = _age(cache)
    if fresh is not None and fresh < FEED_MAX_AGE:
        d = _usable_feed(_read(cache))
        if d:
            return d

    recent_failure = _age(failed)
    if recent_failure is not None and recent_failure < FEED_RETRY_AFTER:
        # Offline, or the site is down. Do not spend the timeout again on
        # every invocation -- somebody diagnosing a broken setup runs this
        # repeatedly. A stale copy still answers "when did this start", and
        # every line printed off it names the date it was measured.
        return _usable_feed(_read(cache))

    try:
        body = _fetch_feed()
    except Exception:                                    # noqa: BLE001
        _touch(failed)
        return _usable_feed(_read(cache))
    d = _usable_feed(body)
    if not d:
        _touch(failed)
        return _usable_feed(_read(cache))
    _write_cache(cache, body)
    try:
        failed.unlink()
    except Exception:                                    # noqa: BLE001
        pass
    return d


def _read(path):
    try:
        return path.read_text(encoding="utf-8")
    except Exception:                                    # noqa: BLE001
        return ""


def _touch(path):
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                        encoding="utf-8")
    except Exception:                                    # noqa: BLE001
        pass


def _write_cache(path, body):
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        tmp = path.with_name(path.name + ".tmp")
        tmp.write_text(body, encoding="utf-8")
        os.replace(str(tmp), str(path))
    except Exception:                                    # noqa: BLE001
        pass


def _latest(allow_fetch=True):
    """Our own daily series: the state file in the repo, else the feed.

    `allow_fetch` is False when nothing being checked is a remote address --
    a local stdio server is not in the probed population, so neither the
    history nor the "for scale" line applies to it and there is nothing to
    fetch the feed for.
    """
    try:
        with LATEST.open() as fh:
            return json.load(fh)
    except Exception:                                    # noqa: BLE001
        pass
    return _feed_copy() if allow_fetch else None


def ecosystem_line(cls, data):
    """How common this verdict is, from our own run. Never invented."""
    if not data:
        return None
    counts = data.get("counts") or {}
    n = counts.get(cls)
    total = sum(counts.values()) or 0
    if not n or not total:
        return None
    return ("{0:,} of the {1:,} registry endpoints we probed on {2} landed in "
            "this same bucket ({3}%) -- {4}".format(
                n, total, data.get("latest_date") or "the last run",
                round(100.0 * n / total, 1), SITE))


def history_for(url, data):
    """Our per-endpoint history, if this exact address is one we probe.

    This is the part no generic inspector can do: "it is down right now" and
    "it has been down since Tuesday" are different problems with different
    fixes, and only a series can tell them apart.
    """
    if not data:
        return None
    want = (url or "").rstrip("/")
    for s in data.get("servers") or []:
        if (s.get("url") or "").rstrip("/") == want:
            return s
    return None


def _strip_english(strip):
    """`HHGGGGGG` -> the days in words, oldest first."""
    if not strip:
        return None
    out = []
    for ch in strip:
        for k, v in mcphealth.CLASS_CHAR.items():
            if v == ch:
                out.append(mcphealth.CLASS_LABEL.get(k, k))
                break
        else:
            out.append("no check")
    return out


def _strip_story(strip):
    """The history as a sentence: the runs, not eight repetitions of one word.

    `HHGGGGGG` reads "2 days answered, tools listed, then 6 days needs
    credentials", which is the shape of the question somebody actually has --
    when did this change -- rather than a list they have to scan.
    """
    days = _strip_english(strip)
    if not days:
        return None
    runs = []
    for d in days:
        if runs and runs[-1][0] == d:
            runs[-1][1] += 1
        else:
            runs.append([d, 1])
    bits = ["{0} day{1} {2}".format(n, "" if n == 1 else "s", d)
            for d, n in runs]
    if len(bits) == 1:
        return bits[0]
    return ", then ".join(bits)


# --------------------------------------------------------------------------
# the remote path -- mcphealth's own dispatcher, called, not copied


def diagnose_url(url, transport=None):
    ep = {"url": url, "transport": (transport or "").lower(),
          "name": url, "title": "", "description": "", "version": ""}
    res = mcphealth.probe_endpoint(ep)
    res["mode"] = "remote"
    res["target"] = url
    # `probe_endpoint` returns `via`: which transport actually answered. When
    # that disagrees with what the config declared, that IS the diagnosis, and
    # it is cause (3) in the header of this file.
    declared = (transport or "").lower()
    via = res.get("via")
    res["cause"] = None
    if res.get("class") in ("healthy", "no_tools") and declared and via:
        spoke_sse = (via == "sse")
        said_sse = (declared == "sse")
        if spoke_sse != said_sse:
            res["cause"] = "transport_mismatch"
    if res.get("class") == "not_mcp":
        res["cause"] = "not_mcp"
    elif res.get("class") == "gated":
        res["cause"] = "needs_credentials"
    elif res.get("class") == "unreachable":
        res["cause"] = "unreachable"
    elif res.get("class") == "no_tools":
        res["cause"] = "no_tools"
    return res


# --------------------------------------------------------------------------
# the local path -- stdio, which mcphealth.py has never had


def _unexpanded(bits):
    """`${CLAUDE_PROJECT_DIR}` still sitting in a command or an env value."""
    out = []
    for b in bits:
        for m in re.finditer(r"\$\{?([A-Za-z_][A-Za-z0-9_]*)\}?", str(b)):
            if not os.environ.get(m.group(1)):
                out.append(m.group(0))
    return sorted(set(out))


def _reader(stream, sink):
    try:
        for raw in iter(stream.readline, b""):
            sink.append(raw.decode("utf-8", "replace").rstrip("\r\n"))
    except Exception:                                    # noqa: BLE001
        pass


def _await_reply(lines, cursor, noise, want_id, proc, deadline):
    """Pull the next JSON-RPC object with `want_id` off the stdout queue.

    Every stdout line that is NOT a JSON-RPC object is kept in `noise`. That is
    the whole point: on stdio, stdout is the wire, so a log line printed there
    is not untidiness, it is a protocol violation that will close the
    connection -- and the client will report it as "connection closed".

    `cursor` is a one-element list so the position survives between calls. It
    started as a local, which made the second call re-read every line the first
    had already consumed and file each one as noise a second time -- the count
    doubled and a stale reply could have been handed back as a fresh one.
    """
    while time.time() < deadline:
        seen = cursor[0]
        if len(lines) > seen:
            line = lines[seen]
            cursor[0] = seen + 1
            if not line.strip():
                continue
            try:
                msg = json.loads(line)
            except ValueError:
                noise.append(line)
                continue
            if not isinstance(msg, dict) or "jsonrpc" not in msg:
                noise.append(line)
                continue
            if msg.get("id") == want_id:
                return msg
            continue                    # a notification or another id: fine
        if proc.poll() is not None and len(lines) <= cursor[0]:
            return None
        time.sleep(0.05)
    return None


def diagnose_stdio(argv, env=None, cwd=None, timeout=STDIO_TIMEOUT):
    """Start a local MCP server the way a client would, and say what happened."""
    res = {"mode": "stdio", "target": " ".join(argv), "class": "unreachable",
           "cause": None, "ms": None, "tools": None, "protocol": None,
           "server_name": None, "server_version": None, "via": "stdio",
           "detail": "", "stderr": [], "stdout_noise": [], "exit_code": None}

    full_env = dict(os.environ)
    full_env.update(env or {})

    # (1a) an unexpanded variable -- `Atiqul-Islam/genesis#25` in the evidence.
    bits = list(argv) + list((env or {}).values())
    missing_vars = _unexpanded(bits)
    if missing_vars:
        res["cause"] = "unexpanded_variable"
        res["detail"] = ("the command still contains " + ", ".join(missing_vars)
                         + ", and nothing in the environment expands it")
        return res

    # (1b) the launcher itself.
    exe = shutil.which(argv[0], path=full_env.get("PATH"))
    if not exe:
        res["cause"] = "launcher_missing"
        res["detail"] = ("`" + argv[0] + "` is not on PATH, so the command "
                         "cannot start at all")
        # `lazyweb-skill#156`. The generic advice ("install it, or use an
        # absolute path") is right but tells a Windows user to go looking for
        # something that is already on their disk.
        if _bare_posix_shell(argv[0]) and _system() == "Windows":
            res["windows_shell"] = True
            res["detail"] += ("; Git for Windows ships `" + argv[0]
                              + ".exe`, but its `bin\\` directory is not on "
                                "PATH in a default install")
        return res
    res["launcher"] = exe

    t0 = time.time()
    try:
        proc = subprocess.Popen(
            argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
            stderr=subprocess.PIPE, env=full_env,
            cwd=str(cwd) if cwd else None, bufsize=0)
    except Exception as e:                               # noqa: BLE001
        res["cause"] = "launcher_failed"
        res["detail"] = "{0}: {1}".format(type(e).__name__, e)[:200]
        return res

    out_lines = []
    err_lines = collections.deque(maxlen=STDERR_KEEP)
    threading.Thread(target=_reader, args=(proc.stdout, out_lines),
                     daemon=True).start()
    threading.Thread(target=_reader, args=(proc.stderr, err_lines),
                     daemon=True).start()

    noise = []
    try:
        init = {"jsonrpc": "2.0", "id": 1, "method": "initialize",
                "params": {"protocolVersion": mcphealth.PROTOCOL_VERSION,
                           "capabilities": {},
                           "clientInfo": {"name": "mcpdoctor",
                                          "version": "0.1"}}}
        try:
            proc.stdin.write((json.dumps(init) + "\n").encode("utf-8"))
            proc.stdin.flush()
        except Exception as e:                           # noqa: BLE001
            res["detail"] = "could not write to the server's stdin: " + str(e)[:120]

        cursor = [0]
        msg = _await_reply(out_lines, cursor, noise, 1, proc,
                           time.time() + timeout)
        res["ms"] = int((time.time() - t0) * 1000)
        res["stderr"] = list(err_lines)
        res["stdout_noise"] = noise[:10]

        # (2) it started and died. The traceback is on stderr, which the client
        #     discarded; that is why nobody in those 130 issues could say why.
        if msg is None and proc.poll() is not None:
            res["exit_code"] = proc.returncode
            res["cause"] = "crashed_on_start"
            res["detail"] = ("the server exited with code {0} before "
                             "answering".format(proc.returncode))
            # Asked before DEP_BREAK: this one is decided on the argument list,
            # the platform and the exit code as well as the stderr line, so it
            # is the more specific claim of the two and must not be shadowed by
            # a generic string match.
            shell_hit = _posix_shell_on_windows(
                argv, proc.returncode, err_lines)
            if shell_hit:
                res["cause"] = "posix_shell_on_windows"
                res["windows_shell"] = True
                res["evidence"] = shell_hit
                alias = "windowsapps" in (exe or "").lower()
                res["detail"] = (
                    "`{0}` resolved to {1}, and that shell could not find the "
                    "script it was handed: exit 127".format(argv[0], exe)
                    + (" -- that path is the WindowsApps placeholder, which "
                       "forwards to WSL's /bin/bash" if alias else ""))
                return res
            hit = _dep_signature(err_lines)
            if hit:
                res["evidence"], res["cause"], res["detail"] = hit
            return res

        # (3, local form) it is alive but printed something that is not
        #     JSON-RPC on stdout, which corrupts the stream.
        if msg is None and noise:
            res["class"] = "not_mcp"
            res["cause"] = "stdout_polluted"
            res["evidence"] = noise[0]
            res["detail"] = ("the server wrote a non-JSON line to stdout. On "
                             "stdio, stdout IS the wire: logs belong on stderr")
            return res

        if msg is None:
            res["cause"] = "no_response"
            res["detail"] = ("no reply to `initialize` in {0:.0f}s; the "
                             "process is still running".format(timeout))
            return res

        if "error" in msg and "result" not in msg:
            err = msg.get("error") or {}
            text = str(err.get("message") or "")[:200]
            # `[-_ ]?`, not `[- ]?`: the commonest way a server words this is
            # the environment variable it wanted, and those are written
            # `SOMETHING_API_KEY`. Matching only the hyphen and the space files
            # a server that merely wants a key under `not_mcp`, which is the
            # one verdict that makes a public negative claim about somebody
            # else's software.
            if re.search(
                    r"unauthor|forbidden|api[-_ ]?key|token|auth|credential",
                    text, re.I):
                res["class"] = "gated"
                res["cause"] = "needs_credentials"
            else:
                res["class"] = "not_mcp"
                res["cause"] = "initialize_refused"
            res["detail"] = "initialize returned an error: " + text
            return res

        result = msg.get("result")
        if not isinstance(result, dict) or "protocolVersion" not in result:
            res["class"] = "not_mcp"
            res["cause"] = "not_mcp"
            res["detail"] = "the initialize reply carried no protocolVersion"
            return res

        res["protocol"] = str(result.get("protocolVersion"))[:32]
        info = result.get("serverInfo") or {}
        res["server_name"] = str(info.get("name") or "")[:80]
        res["server_version"] = str(info.get("version") or "")[:32]
        res["class"] = "no_tools"

        try:
            proc.stdin.write((json.dumps(
                {"jsonrpc": "2.0",
                 "method": "notifications/initialized"}) + "\n").encode("utf-8"))
            proc.stdin.flush()
            proc.stdin.write((json.dumps(
                {"jsonrpc": "2.0", "id": 2, "method": "tools/list",
                 "params": {}}) + "\n").encode("utf-8"))
            proc.stdin.flush()
        except Exception as e:                           # noqa: BLE001
            res["detail"] = "initialize ok; could not send tools/list: " \
                            + str(e)[:120]
            return res

        msg3 = _await_reply(out_lines, cursor, noise, 2, proc,
                            time.time() + STDIO_TOOLS_TIMEOUT)
        res["stdout_noise"] = noise[:10]
        res["stderr"] = list(err_lines)
        tools = None
        if isinstance(msg3, dict) and isinstance(msg3.get("result"), dict):
            tools = msg3["result"].get("tools")
        if isinstance(tools, list):
            res["tools"] = len(tools)
            bad = 0
            for t in tools:
                if not isinstance(t, dict) or not t.get("name"):
                    bad += 1
                    continue
                sch = t.get("inputSchema")
                if not isinstance(sch, dict) or sch.get("type") != "object":
                    bad += 1
            res["bad_schema"] = bad
            res["class"] = "healthy" if tools else "no_tools"
            if not tools:
                res["detail"] = "initialize ok; tools/list returned an empty array"
            else:
                res["detail"] = "initialize ok; tools/list returned {0} tool{1}" \
                    .format(len(tools), "" if len(tools) == 1 else "s")
        elif msg3 is None:
            res["detail"] = "initialize ok; no reply to tools/list"
        else:
            err = (msg3 or {}).get("error") or {}
            res["detail"] = ("initialize ok; tools/list error: "
                             + str(err.get("message") or "unparseable")[:160])
        return res
    finally:
        # Noise on stdout is a finding even when the handshake got through.
        # This reader skips a line it cannot parse and carries on; a strict
        # client does not, it closes the connection -- which is the whole of
        # `sd2k/mcp-tokens#11` and is reported to the user as "connection
        # closed". So a server that works for us and printed on stdout is
        # still broken for them, and saying "healthy" would be the more
        # comfortable answer rather than the true one. Mutating `res` here is
        # deliberate: it is the object already being returned.
        res["stdout_noise"] = noise[:10]
        if noise and res.get("cause") is None:
            res["cause"] = "stdout_polluted"
            res["evidence"] = noise[0]
            res["detail"] = (
                (res.get("detail") or "").rstrip(". ")
                + ("; " if res.get("detail") else "")
                + "but it also wrote {0} non-JSON line{1} to stdout, which a "
                  "strict client treats as a broken stream".format(
                      len(noise), "" if len(noise) == 1 else "s")).lstrip("; ")
        try:
            proc.stdin.close()
        except Exception:                                # noqa: BLE001
            pass
        try:
            proc.terminate()
            proc.wait(timeout=3)
        except Exception:                                # noqa: BLE001
            try:
                proc.kill()
            except Exception:                            # noqa: BLE001
                pass


def _system():
    """Indirection so the Windows-only branch can be exercised from a Mac.

    There is no other way to test it here: the failure it names only happens on
    native Windows, and an untested alarm branch is how this project has been
    caught before. `ops/mcp-doctor-selftest.py` patches this.
    """
    return platform.system()


def _bare_posix_shell(arg0):
    """True for `bash`/`sh` written as a bare name, so it is resolved off PATH.

    An absolute path to Git's `sh.exe` must NOT match: that is the documented
    workaround in `lazyweb-skill#156`, not the fault. Checked separator-wise
    rather than with `os.path.basename`, which does not split a Windows path
    when this runs on a posix machine -- as it does in the self-test.
    """
    s = str(arg0 or "")
    if not s or "/" in s or "\\" in s or ":" in s:
        return False
    name = s.lower()
    if name.endswith(".exe"):
        name = name[:-4]
    return name in POSIX_SHELLS


def _posix_shell_on_windows(argv, exit_code, err_lines):
    """The quoted stderr line for `axiomcodegraph#1233`'s shape, or None.

    Four conditions, all needed, none of them a guess: we are on Windows, the
    config names a bare POSIX shell, the process exited 127 (the shell's own
    "I could not run that"), and the shell said so on stderr. On Linux and macOS
    the same three signals mean something else and true -- the script really is
    missing -- so the platform test is load-bearing, not decoration.
    """
    if _system() != "Windows":
        return None
    if not argv or not _bare_posix_shell(argv[0]):
        return None
    if exit_code != 127:
        return None
    for line in reversed(list(err_lines)):
        if SHELL_NOT_FOUND.search(line):
            return line.strip()[:300]
    return None


def _dep_signature(err_lines):
    """(quoted line, cause, plain-English reading) or None. Never guesses."""
    for line in reversed(list(err_lines)):
        for pat, cause, english in DEP_BREAK:
            if re.search(pat, line):
                return line.strip()[:300], cause, english
    return None


# --------------------------------------------------------------------------
# reading a client config the user names


def load_config(path):
    """`.mcp.json`, Claude Desktop, Cursor, VS Code -- all the same two shapes."""
    with open(path) as fh:
        raw = json.load(fh)
    block = raw.get("mcpServers")
    if not isinstance(block, dict):
        block = raw.get("servers")
    if not isinstance(block, dict):
        block = raw if all(isinstance(v, dict) for v in raw.values()) else {}
    out = {}
    for name, cfg in (block or {}).items():
        if not isinstance(cfg, dict):
            continue
        entry = {"name": name}
        if cfg.get("url"):
            entry["url"] = cfg["url"]
            entry["transport"] = (cfg.get("type") or cfg.get("transport") or "")
        elif cfg.get("command"):
            entry["argv"] = [cfg["command"]] + list(cfg.get("args") or [])
            entry["env"] = cfg.get("env") or {}
            entry["cwd"] = cfg.get("cwd")
        else:
            continue
        out[name] = entry
    return out


# --------------------------------------------------------------------------
# saying it in English


NEXT_STEPS = {
    "launcher_missing": [
        "Install the launcher, or give the config an absolute path to it.",
        "Client configs do not inherit your shell's PATH. A command that "
        "works in your terminal can still be missing here -- `which {cmd}` "
        "in the terminal, then put that full path in the config.",
    ],
    "unexpanded_variable": [
        "Nothing expanded the variable. Either set it in the environment the "
        "client runs in, or write the literal value in the config.",
        "Config files are not shells: most clients do no substitution at all, "
        "and the ones that do only know their own variables.",
    ],
    "dependency_break": [
        "Pin the dependency. A range like `mcp>=1.0.0` picks up the next "
        "major release on the next clean install, and `uvx`/`npx` do a clean "
        "install every time -- which is why this breaks with no change of "
        "yours.",
        "Re-run with the version pinned (`uvx pkg==<last good>`) to confirm "
        "before you change the pin for good.",
    ],
    "unresolvable": [
        "The environment was never built, so there is no working version to "
        "go back to: check the package name, and that you are pointing at the "
        "registry that has it.",
        "A private or scoped package needs the registry flag or the auth the "
        "launcher is not getting here.",
    ],
    "crashed_on_start": [
        "The server's own stderr is printed above. Your client threw it away; "
        "that is the whole reason this looked like 'connection closed'.",
        "Run the same command in a terminal to see it in full.",
    ],
    "posix_shell_on_windows": [
        "Do not name a bare `bash` or `sh` as the command on Windows. The "
        "client spawns it from the Windows PATH, not from Git Bash, so it "
        "reaches WSL's shell or nothing -- never the shell your script "
        "expects.",
        "The absolute path to Git's shell (`C:\\Program Files\\Git\\bin\\"
        "sh.exe`) works and is worth trying now to confirm the diagnosis, but "
        "it is lost on the next install or plugin update.",
        "The durable fix is a launcher that exists under the same name on "
        "every platform, with the shell logic moved into it -- `node` and a "
        "small script, typically. Note that `python3` has the identical trap: "
        "on Windows it is a Microsoft Store placeholder that exits 9009, while "
        "`python` works there and is missing on machines that only have "
        "`python3`. No single interpreter name is portable.",
    ],
    "stdout_polluted": [
        "Move that output to stderr. On stdio, stdout carries the JSON-RPC "
        "messages and nothing else -- one stray print closes the connection.",
        "In Python that is `print(..., file=sys.stderr)` or a logging handler "
        "on stderr; in Node, `console.error` rather than `console.log`.",
    ],
    "transport_mismatch": [
        "Your config declares the wrong transport for this server. It "
        "answered on the other one.",
        "MCP has two remote transports: streamable HTTP (one POST per "
        "message) and the legacy HTTP+SSE (a GET stream that names a separate "
        "POST address). A client aimed at the wrong one gets nothing back, "
        "however healthy the server is.",
    ],
    "not_mcp": [
        "Something is answering at this address, but it is not speaking MCP.",
        "The usual causes: the path is wrong (`/mcp` vs `/sse` vs `/`), a "
        "proxy or login page is in front of it, or the server is on the other "
        "transport -- we tried both, so if you have a third address, try it.",
    ],
    "needs_credentials": [
        "This server works; it wants credentials. Not a fault.",
        "Add the header or token its documentation asks for.",
    ],
    "no_response": [
        "The process is alive and said nothing. It is probably blocked before "
        "its own read loop -- a synchronous login, a prompt, a missing file.",
        "Its stderr, if it wrote any, is above.",
    ],
    "initialize_refused": [
        "The server answered the handshake with an error rather than a "
        "protocol version. Its message is above.",
    ],
    "launcher_failed": [
        "The launcher was found but could not be executed -- check the "
        "execute bit and the architecture.",
    ],
    "unreachable": [
        "Nothing usable came back: DNS, TLS, a timeout or a 5xx. The exact "
        "failure is above.",
        "A 5xx is the server telling you it cannot serve right now, which is "
        "a different problem from the address being wrong.",
    ],
    "no_tools": [
        "It completed the handshake and then did not give a tool list, so a "
        "client connects and finds nothing to call.",
        "Commonly the tool list itself is behind the credentials the "
        "handshake did not ask for.",
    ],
}

# Appended to whatever the cause's own advice said, for both Windows shell
# shapes -- the missing `sh` and the `bash` that resolved to WSL's. It is a
# second, distinct bug, in the client rather than in the config, and not ours
# to fix; it is here because a user who has been sent round an auth loop needs
# telling that the loop is a dead end, or they will keep walking it.
# `aboul3ata/lazyweb-skill#156` documents it alongside its own evidence that a
# direct POST with the stored token succeeded -- HTTP 200, all 41 tools listed
# -- so auth was demonstrably never the problem there.
WINDOWS_SHELL_AUTH = [
    "If your client has started asking you to re-authenticate this server, "
    "that is a separate bug and not worth following. A stdio server whose "
    "process dies can be misread as an auth failure and cached as one (Claude "
    "Code writes it into `mcp-needs-auth-cache.json`); no amount of "
    "re-authenticating fixes a launcher that never ran.",
]


# A local process that never started has not "given no usable response (DNS,
# TLS, timeout or 5xx)" -- that is the dataset's wording for a network
# endpoint, and it is the wrong sentence to show somebody whose problem is that
# `uvx` is not installed. Causes with no entry here fall back to the dataset's
# own label, which is what the remote verdicts should keep saying.
HEADLINE = {
    "launcher_missing": "the command cannot start, because the launcher is "
                        "not there",
    "unexpanded_variable": "the command cannot start, because a ${...} in it "
                           "was never expanded",
    "launcher_failed": "the launcher is there but could not be executed",
    "dependency_break": "it starts and dies, because a dependency broke "
                        "under it",
    "unresolvable": "it never starts, because the launcher cannot install it",
    "crashed_on_start": "it starts and exits before answering",
    "posix_shell_on_windows": "the config starts it with a bare `bash`/`sh`, "
                              "which on Windows is not the shell it needs",
    "stdout_polluted": "it writes lines to stdout that are not JSON-RPC",
    "no_response": "it is running and never answered",
    "initialize_refused": "it refused the handshake",
}


def render_text(res, data, out=sys.stdout):
    def w(s=""):
        out.write(s + "\n")

    cls = res.get("class")
    cause = res.get("cause")
    label = HEADLINE.get(cause) or mcphealth.CLASS_LABEL.get(
        cls, cls or "no verdict")

    head = res.get("target", "")
    if res.get("name") and res["name"] != head:
        head = res["name"] + "  --  " + head
    w()
    w("  " + head)
    w("  " + "-" * max(4, min(72, len(head))))
    w()

    if cause == "stdout_polluted" and cls in ("healthy", "no_tools"):
        # The uncomfortable middle: it answered US, and it will still fail in
        # a client that does not skip a line it cannot parse. Saying "working"
        # here would be the friendlier answer and the wrong one.
        w("  IT ANSWERED US, AND IT WILL STILL FAIL IN A STRICT CLIENT: "
          + label + ".")
    elif cls == "healthy":
        w("  WORKING. " + ("It completed the MCP handshake and listed {0} tool{1}."
                           .format(res["tools"], "" if res["tools"] == 1 else "s")))
    elif cls == "gated":
        w("  WORKING, BUT IT WANTS CREDENTIALS.")
    else:
        w("  NOT WORKING: " + label + ".")
    w()

    if res.get("detail"):
        w("  What happened: " + res["detail"] + ".")
    if res.get("evidence"):
        w("  Read off its own output:")
        w("      " + res["evidence"])
    if res.get("protocol"):
        w("  It speaks MCP {0}{1}{2}.".format(
            res["protocol"],
            ", server " + res["server_name"] if res.get("server_name") else "",
            " " + res["server_version"] if res.get("server_version") else ""))
    if res.get("via") and res["via"] != "stdio":
        w("  Transport that answered: " + (
            "legacy HTTP+SSE" if res["via"] == "sse" else "streamable HTTP")
            + ("; HTTP {0}".format(res["http"]) if res.get("http") else ""))
    if res.get("ms") is not None:
        w("  Time to the verdict: {0} ms.".format(res["ms"]))
    w()

    if res.get("stdout_noise"):
        w("  Lines it wrote to stdout that are not JSON-RPC "
          "(these break the protocol):")
        for line in res["stdout_noise"][:5]:
            w("      " + line[:160])
        w()
    if res.get("stderr") and cls != "healthy":
        w("  Its stderr (the part your client discarded), last {0} lines:"
          .format(min(12, len(res["stderr"]))))
        for line in list(res["stderr"])[-12:]:
            w("      " + line[:200])
        w()

    steps = list(NEXT_STEPS.get(cause or "") or [])
    if res.get("windows_shell"):
        steps += WINDOWS_SHELL_AUTH
    if steps:
        w("  What to do")
        for s in steps:
            w("    * " + s.replace("{cmd}", res.get("target", "").split(" ")[0]))
        w()

    hist = res.get("history")
    if hist:
        days = _strip_story(hist.get("strip"))
        # Name the run these numbers come from. A packaged copy can be reading
        # a cached feed from several days ago if the fetch has been failing,
        # and "up 14.3% over 7 days" with no date attached would read as today.
        asof = (data or {}).get("latest_date")
        w("  We have been probing this exact address daily{0}.".format(
            ", last run " + str(asof) if asof else ""))
        w("    {0} check{1}; up {2}% over 7 days, {3}% over 30."
          .format(hist.get("checks"), "" if hist.get("checks") == 1 else "s",
                  hist.get("up7"), hist.get("up30")))
        if hist.get("last_healthy"):
            w("    Last seen answering with a tool list: "
              + str(hist["last_healthy"]) + ".")
        else:
            w("    It has never answered us with a tool list.")
        if days:
            w("    Oldest first: " + days + ".")
        w("    " + SITE)
        w()

    # Only for remote servers. Those 19,000-odd are registry endpoints with an
    # address; a local stdio server is not in that population and never could
    # be, so printing its percentage beside a local verdict would invite a
    # comparison the denominator does not support.
    if res.get("mode") == "remote":
        eco = ecosystem_line(cls, data)
        if eco:
            w("  For scale: " + eco)
            w()


def split_argv(args):
    """`(our flags, the server's command)`, split at the first bare `--`.

    `nargs=argparse.REMAINDER` looks like the right tool and is not: it is
    greedy from the first positional onward, so `mcpdoctor.py <url> --json`
    quietly filed `--json` as part of a command to run and printed the human
    form anyway. Everything after `--` belongs to the server being started,
    including anything that looks like one of our flags.
    """
    if "--" in args:
        i = args.index("--")
        return args[:i], args[i + 1:]
    return list(args), []


def main():
    ap = argparse.ArgumentParser(
        description="Why is my MCP server unhealthy? One command, one answer.",
        epilog="Pass a URL, or a command after `--`, or --config/--server.")
    ap.add_argument("target", nargs="?",
                    help="the server's URL (remote servers)")
    ap.add_argument("--transport", default=None,
                    help="what your config declares: sse | streamable-http")
    ap.add_argument("--config", type=pathlib.Path, default=None,
                    help="an .mcp.json / claude_desktop_config.json to read")
    ap.add_argument("--server", default=None,
                    help="which server in --config to check")
    ap.add_argument("--all", action="store_true",
                    help="check every server in --config")
    ap.add_argument("--timeout", type=float, default=STDIO_TIMEOUT)
    ap.add_argument("--json", action="store_true", help="machine-readable")

    args, cmd = split_argv(sys.argv[1:])
    a = ap.parse_args(args)

    jobs = []

    if a.config:
        cfg = load_config(a.config)
        if not cfg:
            print("no servers found in " + str(a.config))
            return 2
        if a.all:
            jobs = list(cfg.values())
        elif a.server:
            if a.server not in cfg:
                print("no server called '{0}' in {1}. It has: {2}".format(
                    a.server, a.config, ", ".join(sorted(cfg))))
                return 2
            jobs = [cfg[a.server]]
        else:
            print("\n{0} defines {1} server(s):\n".format(a.config, len(cfg)))
            for name, e in sorted(cfg.items()):
                what = e.get("url") or " ".join(e.get("argv") or [])
                print("  {0:<28} {1}".format(name, what[:90]))
            print("\nCheck one with --server <name>, or all of them with "
                  "--all.\n")
            return 0
    elif cmd:
        jobs = [{"name": cmd[0], "argv": cmd}]
    elif a.target and re.match(r"^https?://", a.target):
        jobs = [{"name": a.target, "url": a.target, "transport": a.transport}]
    elif a.target:
        jobs = [{"name": a.target, "argv": [a.target]}]
    else:
        ap.print_help()
        return 2

    # Only remote targets have anything to look up, so a person checking a
    # local server never waits on a fetch they cannot use.
    data = _latest(allow_fetch=any(j.get("url") for j in jobs))

    results = []
    for job in jobs:
        if job.get("url"):
            res = diagnose_url(job["url"], job.get("transport") or a.transport)
            res["history"] = history_for(job["url"], data)
        else:
            res = diagnose_stdio(job["argv"], env=job.get("env"),
                                 cwd=job.get("cwd"), timeout=a.timeout)
        res["name"] = job.get("name")
        results.append(res)

    if a.json:
        json.dump(results if len(results) > 1 else results[0],
                  sys.stdout, indent=1, default=str)
        sys.stdout.write("\n")
    else:
        for res in results:
            render_text(res, data)

    return 0 if all(r.get("class") in ("healthy", "gated")
                    for r in results) else 1


if __name__ == "__main__":
    sys.exit(main())