MCP OAuth stale client registration auto-recovery

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #52936 fix(mcp): auto-recover from invalid_client on stale OAuth client registration (salvage #37096)

  merged to main

  files: hermes_cli/mcp_config.py, hermes_cli/subcommands/mcp.py, tests/hermes_cli/test_mcp_config.py, tests/tools/test_mcp_oauth.py, tests/tools/test_mcp_oauth_manager.py, tools/mcp_oauth.py, tools/mcp_oauth_manager.py

  Show the diff

  • hermes_cli/mcp_config.py

    @@ -665,44 +665,28 @@ def cmd_mcp_test(args):
     
     # ─── hermes mcp login ────────────────────────────────────────────────────────
     
    -def cmd_mcp_login(args):
    -    """Force re-authentication for an OAuth-based MCP server.
    +def _reauth_oauth_server(name: str, server_config: dict) -> bool:
    +    """Force a fresh OAuth flow for one server. Returns True on success.
     
    -    Deletes cached tokens (both on disk and in the running process's
    -    MCPOAuthManager cache) and triggers a fresh OAuth flow via the
    -    existing probe path.
    -
    -    Use this when:
    -      - Tokens are stuck in a bad state (server revoked, refresh token
    -        consumed by an external process, etc.)
    -      - You want to re-authenticate to change scopes or account
    -      - A tool call returned ``needs_reauth: true``
    +    Wipes cached OAuth state (disk + in-process MCPOAuthManager cache),
    +    re-probes to trigger the browser flow, and verifies a token actually
    +    landed before reporting success. Shared by ``hermes mcp login`` and
    +    ``hermes mcp reauth`` so both behave identically for a single server.
         """
    -    name = args.name
    -    servers = _get_mcp_servers()
    -
    -    if name not in servers:
    -        _error(f"Server '{name}' not found in config.")
    -        if servers:
    -            _info(f"Available servers: {', '.join(servers)}")
    -        return
    -
    -    server_config = servers[name]
         url = server_config.get("url")
         if not url:
             _error(f"Server '{name}' has no URL — not an OAuth-capable server")
    -        return
    +        return False
         if server_config.get("auth") != "oauth":
             _error(f"Server '{name}' is not configured for OAuth (auth={server_config.get('auth')})")
             _info("Use `hermes mcp remove` + `hermes mcp add` to reconfigure auth.")
    -        return
    +        return False
     
         # Wipe both disk and in-memory cache so the next probe forces a fresh
         # OAuth flow.
         try:
             from tools.mcp_oauth_manager import get_manager
    -        mgr = get_manager()
    -        mgr.remove(name)
    +        get_manager().remove(name)
         except Exception as exc:
             _warning(f"Could not clear existing OAuth state: {exc}")
     
    @@ -741,13 +725,90 @@ def cmd_mcp_login(args):
                 print(color(f"          client_secret: \"[redacted:secret-assignment]"", Colors.DIM))
                 print()
                 _info("Then re-run `hermes mcp login " + name + "`.")
    -            return
    +            return False
             if tools:
                 _success(f"Authenticated — {len(tools)} tool(s) available")
             else:
                 _success("Authenticated (server reported no tools)")
    +        return True
         except Exception as exc:
             _error(f"Authentication failed: {exc}")
    +        return False
    +
    +
    +def cmd_mcp_login(args):
    +    """Force re-authentication for an OAuth-based MCP server.
    +
    +    Deletes cached tokens (both on disk and in the running process's
    +    MCPOAuthManager cache) and triggers a fresh OAuth flow via the
    +    existing probe path.
    +
    +    Use this when:
    +      - Tokens are stuck in a bad state (server revoked, refresh token
    +        consumed by an external process, etc.)
    +      - You want to re-authenticate to change scopes or account
    +      - A tool call returned ``needs_reauth: true``
    +    """
    +    name = args.name
    +    servers = _get_mcp_servers()
    +
    +    if name not in servers:
    +        _error(f"Server '{name}' not found in config.")
    +        if servers:
    +            _info(f"Available servers: {', '.join(servers)}")
    +        return
    +
    +    _reauth_oauth_server(name, servers[name])
    +
    +
    +def cmd_mcp_reauth(args):
    +    """Re-authenticate one OAuth MCP server, or all of them sequentially.
    +
    +    ``hermes mcp reauth <name>`` re-auths a single server (same as ``login``).
    +    ``hermes mcp reauth --all`` discovers every ``auth: oauth`` server in
    +    config and re-auths them ONE AT A TIME.
    +
    +    Serial-by-design: a human can only complete one browser OAuth flow at a
    +    time, so re-authing all servers concurrently would open N tabs at once
    +    and N-1 would time out. This is the self-service fix for the recurring
    +    stale-client ritual in GH#36767 (and avoids the startup popup storm when
    +    several servers go stale at once).
    +    """
    +    servers = _get_mcp_servers()
    +    do_all = getattr(args, "all", False)
    +    name = getattr(args, "name", None)
    +
    +    if do_all:
    +        oauth_servers = [
    +            (n, c) for n, c in servers.items()
    +            if c.get("auth") == "oauth" and c.get("url")
    +        ]
    +        if not oauth_servers:
    +            _info("No OAuth-based MCP servers found in config.")
    +            return
    +        print()
    +        _info(f"Re-authenticating {len(oauth_servers)} OAuth server(s) one at a time...")
    +        succeeded = 0
    +        for n, c in oauth_servers:
    +            print()
    +            print(color(f"  ── {n} ──", Colors.CYAN + Colors.BOLD))
    +            if _reauth_oauth_server(n, c):
    +                succeeded += 1
    +        print()
    +        _success(f"Re-authenticated {succeeded}/{len(oauth_servers)} server(s)")
    +        return
    +
    +    if not name:
    +        _error("Specify a server name, or use --all to re-auth every OAuth server.")
    +        _info("Usage: hermes mcp reauth <name>   |   hermes mcp reauth --all")
    +        return
    +    if name not in servers:
    +        _error(f"Server '{name}' not found in config.")
    +        if servers:
    +            _info(f"Available servers: {', '.join(servers)}")
    +        return
    +
    +    _reauth_oauth_server(name, servers[name])
     
     
     # ─── hermes mcp configure ────────────────────────────────────────────────────
    @@ -888,6 +949,7 @@ def mcp_command(args):
             "configure": cmd_mcp_configure,
             "config": cmd_mcp_configure,
             "login": cmd_mcp_login,
    +        "reauth": cmd_mcp_reauth,
         }
     
         handler = handlers.get(action)
    @@ -911,4 +973,5 @@ def mcp_command(args):
             _info("hermes mcp test <name>                        Test connection")
             _info("hermes mcp configure <name>                   Toggle tools")
             _info("hermes mcp login <name>                       Re-authenticate OAuth")
    +        _info("hermes mcp reauth <name> | --all              Re-auth one or all OAuth servers")
             print()

  • hermes_cli/subcommands/mcp.py

    @@ -86,6 +86,19 @@ def build_mcp_parser(subparsers, *, cmd_mcp: Callable) -> None:
         )
         mcp_login_p.add_argument("name", help="Server name to re-authenticate")
     
    +    mcp_reauth_p = mcp_sub.add_parser(
    +        "reauth",
    +        help="Re-authenticate one OAuth MCP server, or all of them (--all)",
    +    )
    +    mcp_reauth_p.add_argument(
    +        "name", nargs="?", help="Server name to re-authenticate (omit with --all)"
    +    )
    +    mcp_reauth_p.add_argument(
    +        "--all",
    +        action="store_true",
    +        help="Re-authenticate every OAuth server in config, one at a time",
    +    )
    +
         # ── Catalog (Nous-approved MCPs shipped with the repo) ─────────────────
         mcp_sub.add_parser(
             "picker",

  • tests/hermes_cli/test_mcp_config.py

    @@ -754,3 +754,98 @@ def mock_probe(name, cfg):
     
             assert "Authenticated — 3 tool(s) available" in out
             assert "no OAuth token" not in out
    +
    +
    +# ---------------------------------------------------------------------------
    +# Tests: cmd_mcp_reauth (GH#36767)
    +# ---------------------------------------------------------------------------
    +
    +class TestMcpReauth:
    +    def test_reauth_all_visits_only_oauth_servers_in_order(
    +        self, tmp_path, capsys, monkeypatch
    +    ):
    +        """--all re-auths every oauth server (skipping non-oauth), serially."""
    +        _seed_config(tmp_path, {
    +            "gh": {"url": "https://gh.example.com/mcp", "auth": "oauth"},
    +            "jira": {"url": "https://jira.example.com/mcp", "auth": "oauth"},
    +            "localstdio": {"command": "foo"},  # no url / no oauth → skipped
    +            "apikey": {"url": "https://k.example.com/mcp", "headers": {"x": "y"}},
    +        })
    +        visited = []
    +        monkeypatch.setattr(
    +            "hermes_cli.mcp_config._reauth_oauth_server",
    +            lambda name, cfg: visited.append(name) or True,
    +        )
    +        from hermes_cli.mcp_config import cmd_mcp_reauth
    +
    +        cmd_mcp_reauth(_make_args(name=None, all=True))
    +        out = capsys.readouterr().out
    +
    +        assert visited == ["gh", "jira"]
    +        assert "Re-authenticated 2/2 server(s)" in out
    +
    +    def test_reauth_all_reports_partial_failures(self, tmp_path, capsys, monkeypatch):
    +        """A server that fails to re-auth is counted but doesn't abort the rest."""
    +        _seed_config(tmp_path, {
    +            "a": {"url": "https://a.example.com/mcp", "auth": "oauth"},
    +            "b": {"url": "https://b.example.com/mcp", "auth": "oauth"},
    +        })
    +        monkeypatch.setattr(
    +            "hermes_cli.mcp_config._reauth_oauth_server",
    +            lambda name, cfg: name == "a",  # only 'a' succeeds
    +        )
    +        from hermes_cli.mcp_config import cmd_mcp_reauth
    +
    +        cmd_mcp_reauth(_make_args(name=None, all=True))
    +        out = capsys.readouterr().out
    +
    +        assert "Re-authenticated 1/2 server(s)" in out
    +
    +    def test_reauth_all_no_oauth_servers(self, tmp_path, capsys, monkeypatch):
    +        _seed_config(tmp_path, {"localstdio": {"command": "foo"}})
    +        called = []
    +        monkeypatch.setattr(
    +            "hermes_cli.mcp_config._reauth_oauth_server",
    +            lambda name, cfg: called.append(name) or True,
    +        )
    +        from hermes_cli.mcp_config import cmd_mcp_reauth
    +
    +        cmd_mcp_reauth(_make_args(name=None, all=True))
    +        out = capsys.readouterr().out
    +
    +        assert "No OAuth-based MCP servers found" in out
    +        assert called == []
    +
    +    def test_reauth_single_server(self, tmp_path, capsys, monkeypatch):
    +        _seed_config(tmp_path, {
    +            "gh": {"url": "https://gh.example.com/mcp", "auth": "oauth"},
    +        })
    +        visited = []
    +        monkeypatch.setattr(
    +            "hermes_cli.mcp_config._reauth_oauth_server",
    +            lambda name, cfg: visited.append(name) or True,
    +        )
    +        from hermes_cli.mcp_config import cmd_mcp_reauth
    +
    +        cmd_mcp_reauth(_make_args(name="gh", all=False))
    +        assert visited == ["gh"]
    +
    +    def test_reauth_requires_name_or_all(self, tmp_path, capsys):
    +        _seed_config(tmp_path, {
    +            "gh": {"url": "https://gh.example.com/mcp", "auth": "oauth"},
    +        })
    +        from hermes_cli.mcp_config import cmd_mcp_reauth
    +
    +        cmd_mcp_reauth(_make_args(name=None, all=False))
    +        out = capsys.readouterr().out
    +        assert "Specify a server name" in out
    +
    +    def test_reauth_unknown_server(self, tmp_path, capsys):
    +        _seed_config(tmp_path, {
    +            "gh": {"url": "https://gh.example.com/mcp", "auth": "oauth"},
    +        })
    +        from hermes_cli.mcp_config import cmd_mcp_reauth
    +
    +        cmd_mcp_reauth(_make_args(name="ghost", all=False))
    +        out = capsys.readouterr().out
    +        assert "not found" in out

  • tests/tools/test_mcp_oauth.py

    @@ -827,3 +827,34 @@ async def instant_sleep(_):
                     asyncio.run(_wait_for_callback())
             err = capsys.readouterr().err
             assert "skip" in err.lower()
    +
    +
    +# ---------------------------------------------------------------------------
    +# poison_client_registration (GH#36767)
    +# ---------------------------------------------------------------------------
    +
    +class TestPoisonClientRegistration:
    +    def test_poison_backs_up_and_removes_client_and_meta(self, tmp_path, monkeypatch):
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +        storage = HermesTokenStorage("srv")
    +        d = tmp_path / "mcp-tokens"
    +        d.mkdir(parents=True)
    +        (d / "srv.json").write_text('{"access_token": "[redacted:secret-assignment]"}')
    +        (d / "srv.client.json").write_text('{"client_id": "dead"}')
    +        (d / "srv.meta.json").write_text('{"token_endpoint": "[redacted:secret-assignment]"}')
    +
    +        removed = storage.poison_client_registration()
    +
    +        assert removed is True
    +        # Client + metadata gone, forcing re-registration on the next flow.
    +        assert not (d / "srv.client.json").exists()
    +        assert not (d / "srv.meta.json").exists()
    +        # Backup of the client file kept for recovery.
    +        assert (d / "srv.client.json.bak").read_text() == '{"client_id": "dead"}'
    +        # Tokens are intentionally preserved.
    +        assert (d / "srv.json").read_text() == '{"access_token": "[redacted:secret-assignment]"}'
    +
    +    def test_poison_noop_when_no_client_file(self, tmp_path, monkeypatch):
    +        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +        storage = HermesTokenStorage("srv")
    +        assert storage.poison_client_registration() is False

  • tests/tools/test_mcp_oauth_manager.py

    @@ -164,3 +164,196 @@ def test_manager_fails_fast_noninteractive_without_cached_tokens(tmp_path, monke
             mgr.get_or_build_provider("linear", "https://mcp.linear.app/mcp", None)
     
         assert mgr._entries["linear"].provider is None
    +
    +
    +# ---------------------------------------------------------------------------
    +# invalid_client auto-heal (GH#36767) — _maybe_flag_poisoned_client
    +# ---------------------------------------------------------------------------
    +
    +import asyncio
    +from types import SimpleNamespace
    +from unittest.mock import MagicMock
    +
    +
    +def _fake_response(status, url, body):
    +    """A minimal stand-in for the httpx.Response the SDK feeds our bridge."""
    +    resp = MagicMock()
    +    resp.status_code = status
    +    resp.request = SimpleNamespace(url=url)
    +
    +    async def _aread():
    +        return body
    +
    +    resp.aread = _aread
    +    return resp
    +
    +
    +def _provider_with_token_endpoint(tmp_path, oauth_config, token_endpoint, monkeypatch):
    +    from tools.mcp_oauth_manager import MCPOAuthManager, reset_manager_for_tests
    +    reset_manager_for_tests()
    +    # Provider construction fails fast in a non-interactive environment with no
    +    # cached tokens (mcp_oauth_manager.py guard). The hermetic test env has no
    +    # TTY, so present an interactive stdin to reach the code under test.
    +    _set_interactive_stdin(monkeypatch)
    +    mgr = MCPOAuthManager()
    +    provider = mgr.get_or_build_provider("srv", "https://mcp.example.com", oauth_config)
    +    provider.context.oauth_metadata = SimpleNamespace(token_endpoint=[redacted:secret-assignment]
    +    provider._initialized = True
    +    return provider
    +
    +
    +def test_invalid_client_at_token_endpoint_poisons(tmp_path, monkeypatch):
    +    """400 invalid_client on the token endpoint deletes the dead client.json."""
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    d = tmp_path / "mcp-tokens"
    +    d.mkdir(parents=True)
    +    (d / "srv.client.json").write_text('{"client_id": "dead"}')
    +    (d / "srv.meta.json").write_text("{}")
    +    provider = _provider_with_token_endpoint(
    +        tmp_path, {}, "https://idp.example.com/oauth/token", monkeypatch
    +    )
    +    resp = _fake_response(
    +        400, "https://idp.example.com/oauth/token", b'{"error":"invalid_client"}'
    +    )
    +
    +    asyncio.run(provider._maybe_flag_poisoned_client(resp))
    +
    +    assert not (d / "srv.client.json").exists()
    +    assert (d / "srv.client.json.bak").exists()
    +    assert provider._initialized is False
    +    assert provider.context.client_info is None
    +
    +
    +def test_invalid_client_at_other_endpoint_is_ignored(tmp_path, monkeypatch):
    +    """An invalid_client body from a non-token endpoint must not poison."""
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    d = tmp_path / "mcp-tokens"
    +    d.mkdir(parents=True)
    +    (d / "srv.client.json").write_text('{"client_id": "live"}')
    +    provider = _provider_with_token_endpoint(
    +        tmp_path, {}, "https://idp.example.com/oauth/token", monkeypatch
    +    )
    +    resp = _fake_response(
    +        400, "https://mcp.example.com/messages", b'{"error":"invalid_client"}'
    +    )
    +
    +    asyncio.run(provider._maybe_flag_poisoned_client(resp))
    +
    +    assert (d / "srv.client.json").exists()
    +    assert provider._initialized is True
    +
    +
    +def test_success_response_is_ignored(tmp_path, monkeypatch):
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    d = tmp_path / "mcp-tokens"
    +    d.mkdir(parents=True)
    +    (d / "srv.client.json").write_text('{"client_id": "live"}')
    +    provider = _provider_with_token_endpoint(
    +        tmp_path, {}, "https://idp.example.com/oauth/token", monkeypatch
    +    )
    +    resp = _fake_response(
    +        200, "https://idp.example.com/oauth/token", b'{"access_token":"x"}'
    +    )
    +
    +    asyncio.run(provider._maybe_flag_poisoned_client(resp))
    +
    +    assert (d / "srv.client.json").exists()
    +    assert provider._initialized is True
    +
    +
    +def test_preregistered_client_is_never_poisoned(tmp_path, monkeypatch):
    +    """A config-supplied client_id is never auto-deleted (re-reg can't help)."""
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    provider = _provider_with_token_endpoint(
    +        tmp_path, {"client_id": "from-config"}, "https://idp.example.com/oauth/token", monkeypatch
    +    )
    +    d = tmp_path / "mcp-tokens"
    +    # _maybe_preregister_client wrote client.json from config during build.
    +    assert (d / "srv.client.json").exists()
    +    resp = _fake_response(
    +        400, "https://idp.example.com/oauth/token", b'{"error":"invalid_client"}'
    +    )
    +
    +    asyncio.run(provider._maybe_flag_poisoned_client(resp))
    +
    +    assert (d / "srv.client.json").exists()
    +    assert provider._initialized is True
    +
    +
    +def test_invalid_client_metadata_does_not_trip(tmp_path, monkeypatch):
    +    """RFC 7591 `invalid_client_metadata` must NOT be mistaken for invalid_client."""
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    d = tmp_path / "mcp-tokens"
    +    d.mkdir(parents=True)
    +    (d / "srv.client.json").write_text('{"client_id": "live"}')
    +    provider = _provider_with_token_endpoint(
    +        tmp_path, {}, "https://idp.example.com/oauth/token", monkeypatch
    +    )
    +    resp = _fake_response(
    +        400, "https://idp.example.com/oauth/token", b'{"error":"invalid_client_metadata"}'
    +    )
    +
    +    asyncio.run(provider._maybe_flag_poisoned_client(resp))
    +
    +    assert (d / "srv.client.json").exists()
    +    assert provider._initialized is True
    +
    +
    +class _FakeMeta:
    +    """Metadata stub usable by both detection and the post-flow persist hook."""
    +
    +    def __init__(self, token_endpoint):
    +        self.token_endpoint = [redacted:secret-assignment]
    +
    +    def model_dump(self, **kwargs):
    +        return {"token_endpoint": self.token_endpoint}
    +
    +
    +def test_bridge_forwards_requests_and_poisons_on_token_endpoint_400(
    +    tmp_path, monkeypatch
    +):
    +    """Drive the REAL async_auth_flow bridge to prove the inserted detection
    +    hook does not break the bidirectional asend() forwarding contract — the
    +    genuinely fragile part. A patched SDK base generator stands in for the
    +    real OAuth flow so we control exactly which response the bridge sees.
    +    """
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    +    token_ep = "[redacted:secret-assignment]"
    +    d = tmp_path / "mcp-tokens"
    +    d.mkdir(parents=True)
    +    (d / "srv.client.json").write_text('{"client_id": "dead"}')
    +
    +    forwarded = []
    +
    +    async def fake_base_flow(self, request):
    +        # Mimic the SDK: yield the request, receive the response, then finish.
    +        forwarded.append(("out", request))
    +        response = yield request
    +        forwarded.append(("in", response))
    +
    +    from mcp.client.auth.oauth2 import OAuthClientProvider
    +    monkeypatch.setattr(OAuthClientProvider, "async_auth_flow", fake_base_flow)
    +
    +    provider = _provider_with_token_endpoint(tmp_path, {}, token_ep, monkeypatch)
    +    provider.context.oauth_metadata = _FakeMeta(token_ep)
    +
    +    sentinel_request = object()
    +    poison_resp = _fake_response(400, token_ep, b'{"error":"invalid_client"}')
    +
    +    async def drive():
    +        gen = provider.async_auth_flow(sentinel_request)
    +        out0 = await gen.__anext__()
    +        assert out0 is sentinel_request  # request forwarded unchanged
    +        try:
    +            await gen.asend(poison_resp)
    +        except StopAsyncIteration:
    +            pass
    +
    +    asyncio.run(drive())
    +
    +    # The poison response reached the inner generator (forwarding intact)...
    +    assert ("in", poison_resp) in forwarded
    +    # ...and the detection hook fired.
    +    assert not (d / "srv.client.json").exists()
    +    assert provider._initialized is False
    +    assert provider.context.client_info is None

  • tools/mcp_oauth.py

    @@ -343,6 +343,41 @@ def remove(self) -> None:
             for p in (self._tokens_path(), self._client_info_path(), self._meta_path()):
                 p.unlink(missing_ok=True)
     
    +    def poison_client_registration(self) -> bool:
    +        """Discard a dead dynamically-registered client so it gets re-created.
    +
    +        Called when the IdP rejects our cached ``client_id`` with
    +        ``invalid_client`` on the token endpoint — proof the server-side
    +        registration is gone (IdP redeploy / DB wipe / rebrand). Deleting
    +        ``client.json`` makes the MCP SDK's ``async_auth_flow`` take the
    +        ``if not client_info`` branch and re-run RFC 7591 dynamic client
    +        registration on the next flow. The stale ``meta.json`` is dropped
    +        too so discovery re-runs against a freshly fetched document.
    +
    +        Tokens are intentionally left in place — the subsequent
    +        re-authorization overwrites them, and keeping them avoids losing a
    +        still-valid refresh token if the re-registration never completes.
    +
    +        A single ``.bak`` copy of the client file is kept for recovery.
    +        Returns True if a client file was present and removed.
    +        """
    +        client_path = self._client_info_path()
    +        if not client_path.exists():
    +            return False
    +        backup = client_path.with_name(client_path.name + ".bak")
    +        try:
    +            backup.write_bytes(client_path.read_bytes())
    +        except OSError as exc:  # non-fatal — proceed with the removal anyway
    +            logger.warning("Could not back up client info at %s: %s", client_path, exc)
    +        client_path.unlink(missing_ok=True)
    +        self._meta_path().unlink(missing_ok=True)
    +        logger.warning(
    +            "MCP OAuth '%s': cached client registration rejected as invalid_client; "
    +            "removed client.json + meta.json (backup at %s) to force re-registration",
    +            self._server_name, backup.name,
    +        )
    +        return True
    +
         def has_cached_tokens(self) -> bool:
             """Return True if we have tokens on disk (may be expired)."""
             return self._tokens_path().exists()

  • tools/mcp_oauth_manager.py

    @@ -36,6 +36,7 @@
     
     import asyncio
     import logging
    +import re
     import threading
     from dataclasses import dataclass, field
     from typing import Any, Optional
    @@ -43,6 +44,26 @@
     logger = logging.getLogger(__name__)
     
     
    +def _same_endpoint(a: str, b: str) -> bool:
    +    """Return True if two URLs target the same endpoint (ignoring query/fragment).
    +
    +    Compares scheme, host (case-insensitive), and path. Used to confirm a
    +    rejected response actually came from the OAuth token endpoint before we
    +    act on an ``invalid_client`` body.
    +    """
    +    from urllib.parse import urlsplit
    +
    +    try:
    +        pa, pb = urlsplit(a), urlsplit(b)
    +    except ValueError:  # pragma: no cover — malformed URL
    +        return False
    +    return (
    +        pa.scheme == pb.scheme
    +        and pa.netloc.lower() == pb.netloc.lower()
    +        and pa.path.rstrip("/") == pb.path.rstrip("/")
    +    )
    +
    +
     # ---------------------------------------------------------------------------
     # Per-server entry
     # ---------------------------------------------------------------------------
    @@ -107,9 +128,21 @@ class HermesMCPOAuthProvider(OAuthClientProvider):
             (``src/utils/auth.ts:1320``, CC-1096 / GH#24317).
             """
     
    -        def __init__(self, *args: Any, server_name: str = "", **kwargs: Any):
    +        def __init__(
    +            self,
    +            *args: Any,
    +            server_name: str = "",
    +            preregistered: bool = False,
    +            **kwargs: Any,
    +        ):
                 super().__init__(*args, **kwargs)
                 self._hermes_server_name = server_name
    +            # When the client_id comes from config.yaml (pre-registered), an
    +            # invalid_client rejection means the *config* is wrong — deleting
    +            # client.json would just be re-seeded from config and re-running
    +            # registration can't help. Only auto-heal dynamically-registered
    +            # clients. See _maybe_flag_poisoned_client.
    +            self._hermes_preregistered = preregistered
     
             async def _initialize(self) -> None:
                 """Load stored tokens + client info AND seed token_expiry_time.
    @@ -284,6 +317,74 @@ def _persist_oauth_metadata_if_changed(self) -> None:
                 ):
                     storage.save_oauth_metadata(meta)
     
    +        async def _maybe_flag_poisoned_client(self, response: Any) -> None:
    +            """Detect a dead client registration and force re-registration.
    +
    +            When the IdP rejects our ``client_id`` with ``invalid_client`` on
    +            the token endpoint (token exchange or refresh), the cached client
    +            registration is provably dead server-side. We delete ``client.json``
    +            (+ stale metadata) so the SDK's next ``async_auth_flow`` takes the
    +            ``if not client_info`` branch and re-runs RFC 7591 dynamic client
    +            registration. This addresses the recurring manual-reset ritual in
    +            GH#36767 for the auto-detectable subset (token-endpoint rejection);
    +            the browser-side "Redirect URI Mismatch" case has no HTTP signal
    +            and is handled by ``hermes mcp reauth``.
    +
    +            Conservative by construction — acts ONLY when all hold:
    +              * status is 400/401,
    +              * the request hit the discovered ``token_endpoint`` (the only
    +                request carrying our ``client_id``), and
    +              * the body carries the ``invalid_client`` error code
    +                (word-boundary match, so RFC 7591's ``invalid_client_metadata``
    +                registration error does not trip it).
    +            Pre-registered (config-supplied) clients are never poisoned.
    +            Fully best-effort: any failure here is swallowed so a detection
    +            miss never breaks the live auth flow.
    +
    +            Covers both the authorization-code token exchange and the
    +            preemptive refresh — but only when ``token_endpoint`` was
    +            discovered (``_initialize`` prefetches it on cold-load). If that
    +            discovery was skipped, the guard returns early and the user falls
    +            back to ``hermes mcp reauth``.
    +            """
    +            try:
    +                if self._hermes_preregistered:
    +                    return
    +                status = getattr(response, "status_code", None)
    +                if status not in (400, 401):
    +                    return
    +                meta = getattr(self.context, "oauth_metadata", None)
    +                token_endpoint = (
    +                    str(meta.token_endpoint)
    +                    if meta is not None and getattr(meta, "token_endpoint", None)
    +                    else None
    +                )
    +                req = getattr(response, "request", None)
    +                req_url = str(req.url) if req is not None else None
    +                if not token_endpoint or not req_url:
    +                    return
    +                if not _same_endpoint(req_url, token_endpoint):
    +                    return
    +                body = await response.aread()
    +                # Word-boundary match: matches `"error":"invalid_client"` but
    +                # not the RFC 7591 registration error `invalid_client_metadata`
    +                # (the trailing `_metadata` removes the right-hand boundary).
    +                if not re.search(rb"\binvalid_client\b", body.lower()):
    +                    return
    +
    +                storage = self.context.storage
    +                from tools.mcp_oauth import HermesTokenStorage
    +                if isinstance(storage, HermesTokenStorage):
    +                    storage.poison_client_registration()
    +                # Drop the in-memory client so the SDK re-registers next flow.
    +                self.context.client_info = None
    +                self._initialized = False
    +            except Exception as exc:  # pragma: no cover — defensive, must not throw
    +                logger.debug(
    +                    "MCP OAuth '%s': invalid_client detection failed (non-fatal): %s",
    +                    self._hermes_server_name, exc,
    +                )
    +
             async def async_auth_flow(self, request):  # type: ignore[override]
                 # Pre-flow hook: ask the manager to refresh from disk if needed.
                 # Any failure here is non-fatal — we just log and proceed with
    @@ -317,6 +418,9 @@ async def async_auth_flow(self, request):  # type: ignore[override]
                     outgoing = await inner.__anext__()
                     while True:
                         incoming = yield outgoing
    +                    # Sniff the response for a dead-client-registration signal
    +                    # before handing it back to the SDK (best-effort, GH#36767).
    +                    await self._maybe_flag_poisoned_client(incoming)
                         outgoing = await inner.asend(incoming)
                 except StopAsyncIteration:
                     # Persist any metadata the SDK discovered lazily during the
    @@ -439,6 +543,7 @@ def _build_provider(
     
             return _HERMES_PROVIDER_CLS(
                 server_name=server_name,
    +            preregistered=bool(cfg.get("client_id")),
                 server_url=entry.server_url,
                 client_metadata=client_metadata,
                 storage=storage,