fix(mcp): surface dead proxy state (#1786)

## Description

When the configured Headroom proxy is down, the MCP server can still
start cleanly and return successful-looking no-op compression or zeroed
stats. That hides the real failure from the client and makes it look
like Headroom is working while compression has stopped. This change
makes proxy-backed MCP tool paths surface unreachable-proxy state
explicitly instead of silently degrading.

Closes #881

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Detect unreachable configured proxy state before returning
proxy-backed MCP tool results.
- Report proxy-unreachable status for compression and stats instead of
presenting no-op output as healthy.
- Preserve local MCP behavior when proxy checking is disabled or a
local-only tool path is intended.
- Keep the short `/livez` health probe isolated from the shared proxy
client used by retrieval and stats calls.

## Testing

- [x] Unit tests pass (`uv run pytest tests/test_ccr_mcp_server.py
tests/test_provider_registry.py -q`)
- [x] Linting passes (`uv run ruff check headroom/ccr/mcp_server.py
headroom/providers/registry.py tests/test_ccr_mcp_server.py
tests/test_provider_registry.py`; `uv run ruff format --check
headroom/ccr/mcp_server.py headroom/providers/registry.py
tests/test_ccr_mcp_server.py tests/test_provider_registry.py`)
- [ ] Type checking passes (`uv run mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
collected 26 items

tests\test_ccr_mcp_server.py ...s..........                              [ 53%]
tests\test_provider_registry.py ............                             [100%]

======================== 25 passed, 1 skipped in 6.64s ========================
All checks passed!
4 files already formatted
```

## Real Behavior Proof

- Environment: Windows, focused local MCP tests.
- Exact command / steps: Run `uv run pytest
.tmp\headroom_t45_regression.py -q` in the base and head worktrees, then
run `uv run pytest tests/test_ccr_mcp_server.py
tests/test_provider_registry.py -q`, `uv run ruff check
headroom/ccr/mcp_server.py headroom/providers/registry.py
tests/test_ccr_mcp_server.py tests/test_provider_registry.py`, and `uv
run ruff format --check headroom/ccr/mcp_server.py
headroom/providers/registry.py tests/test_ccr_mcp_server.py
tests/test_provider_registry.py` in the head worktree.
- Observed result: `base: KeyError: 'proxy'` on the new
proxy-unreachable assertions, `head: .tmp\headroom_t45_regression.py
.... [100%]`, broader head suite `25 passed, 1 skipped in 6.64s`, and
the proxy health probe regression preserved the shared proxy client used
by retrieval and stats.
- Not tested: The reporter's bundled macOS runtime, live Claude Desktop
MCP logs, and the full test suite.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Additional Notes

Documentation and changelog are left unchecked for now; the behavior
change is an error-surfacing fix for existing MCP tools and Headroom
generates changelog entries from conventional commits.

headroom/ccr/mcp_server.py

@@ -253,6 +253,23 @@ def _read_shared_events(window_seconds: int = SESSION_WINDOW_SECONDS) -> list[di
     return events
 
 
+def _build_proxy_unreachable_payload(
+    *,
+    proxy_url: str,
+    error: str,
+    http_status: int | None = None,
+) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "url": proxy_url,
+        "status": "unreachable",
+        "error": error,
+        "warning": f"Configured proxy {proxy_url} is unreachable ({error}).",
+    }
+    if http_status is not None:
+        payload["http_status"] = http_status
+    return payload
+
+
 @dataclass
 class SessionStats:
     """Track compression statistics for the current MCP session."""
@@ -529,6 +546,58 @@ class HeadroomMCPServer:
         result: dict[str, Any] = response.json()
         return result
 
+    async def _probe_proxy_unreachable(self) -> dict[str, Any] | None:
+        """Return explicit proxy-unreachable state when the configured proxy is down."""
+        if not self.check_proxy or not HTTPX_AVAILABLE:
+            return None
+        try:
+            async with httpx.AsyncClient(timeout=5.0) as client:
+                response = await client.get(f"{self.proxy_url}/livez")
+        except Exception as exc:
+            return _build_proxy_unreachable_payload(
+                proxy_url=self.proxy_url,
+                error=f"{type(exc).__name__}: {exc}",
+            )
+        if response.status_code != 200:
+            detail = None
+            try:
+                payload = response.json()
+            except Exception:
+                payload = None
+            if isinstance(payload, dict):
+                detail = payload.get("status")
+            if detail is None:
+                detail = response.text.strip() or None
+            error = f"HTTP {response.status_code}"
+            if detail:
+                error = f"{error} ({detail})"
+            return _build_proxy_unreachable_payload(
+                proxy_url=self.proxy_url,
+                error=error,
+                http_status=response.status_code,
+            )
+        try:
+            payload = response.json()
+        except Exception as exc:
+            return _build_proxy_unreachable_payload(
+                proxy_url=self.proxy_url,
+                error=f"invalid /livez payload: {type(exc).__name__}: {exc}",
+                http_status=response.status_code,
+            )
+        if not isinstance(payload, dict):
+            return _build_proxy_unreachable_payload(
+                proxy_url=self.proxy_url,
+                error="invalid /livez payload",
+                http_status=response.status_code,
+            )
+        if payload.get("status") != "healthy" or payload.get("alive") is not True:
+            return _build_proxy_unreachable_payload(
+                proxy_url=self.proxy_url,
+                error=f"proxy reported {payload.get('status', 'unhealthy')}",
+                http_status=response.status_code,
+            )
+        return None
+
     def _setup_handlers(self) -> None:
         """Register all MCP tool handlers."""
 
@@ -693,6 +762,11 @@ class HeadroomMCPServer:
         except Exception:
             logger.debug("durable savings recording failed", exc_info=True)
 
+        proxy_status = await self._probe_proxy_unreachable()
+        if proxy_status:
+            result["proxy"] = proxy_status
+            result["warning"] = proxy_status["warning"]
+
         return [TextContent(type="text", text=json.dumps(result, indent=2))]
 
     def _record_savings(self, result: dict[str, Any]) -> None:
@@ -807,6 +881,11 @@ class HeadroomMCPServer:
                 proxy_stats = self._extract_proxy_stats(proxy_data)
                 if proxy_stats:
                     stats["proxy"] = proxy_stats
+            else:
+                proxy_status = await self._probe_proxy_unreachable()
+                if proxy_status:
+                    stats["proxy"] = proxy_status
+                    stats["warning"] = proxy_status["warning"]
 
         return [TextContent(type="text", text=json.dumps(stats, indent=2))]
 

headroom/providers/registry.py

@@ -93,6 +93,21 @@ def _normalize_api_url(url: str | None, *, default: str) -> str:
     return normalized
 
 
+def _log_backend_init_failure(
+    logger: logging.Logger,
+    *,
+    backend: str,
+    provider: str,
+    exc: Exception,
+) -> None:
+    logger.error(
+        "backend initialization failed: backend=%s provider=%s error=%s",
+        backend,
+        provider,
+        exc,
+    )
+
+
 def resolve_api_overrides(
     *,
     anthropic_api_url: str | None,
@@ -160,6 +175,7 @@ def create_proxy_backend(
 
     if backend == "anyllm" or backend.startswith("anyllm-"):
         provider = anyllm_provider
+        backend_name = "anyllm" if backend == "anyllm" else backend
         try:
             backend_cls = anyllm_backend_cls or _load_anyllm_backend()
             instance = cast("Backend", backend_cls(provider=provider, api_base=openai_api_url))
@@ -169,7 +185,12 @@ def create_proxy_backend(
             logger.warning("any-llm backend not available: %s", exc)
             return None
         except Exception as exc:  # pragma: no cover - defensive logging
-            logger.error("Failed to initialize any-llm backend: %s", exc)
+            _log_backend_init_failure(
+                logger,
+                backend=backend_name,
+                provider=provider,
+                exc=exc,
+            )
             return None
 
     normalized_backend = backend if backend.startswith("litellm-") else f"litellm-{backend}"
@@ -192,7 +213,12 @@ def create_proxy_backend(
         logger.warning("LiteLLM backend not available: %s", exc)
         return None
     except Exception as exc:  # pragma: no cover - defensive logging
-        logger.error("Failed to initialize LiteLLM backend: %s", exc)
+        _log_backend_init_failure(
+            logger,
+            backend=normalized_backend,
+            provider=provider,
+            exc=exc,
+        )
         return None
 
 

tests/test_ccr_mcp_server.py

@@ -92,6 +92,91 @@ def test_compress_savings_percent_tracks_token_counts(fresh_store) -> None:
         assert result["savings_percent"] > 0.0
 
 
+def test_mcp_compress_surfaces_unreachable_proxy(fresh_store) -> None:
+    server = mcp_server.HeadroomMCPServer(
+        proxy_url="http://127.0.0.1:9",
+        check_proxy=True,
+    )
+
+    response = asyncio.run(server._handle_compress({"content": "dead proxy check"}))
+    payload = json.loads(response[0].kwargs["text"])
+
+    assert payload["proxy"]["status"] == "unreachable"
+    assert payload["proxy"]["url"] == "http://127.0.0.1:9"
+    assert "unreachable" in payload["warning"].lower()
+
+
+def test_mcp_stats_surfaces_unreachable_proxy() -> None:
+    server = mcp_server.HeadroomMCPServer(
+        proxy_url="http://127.0.0.1:9",
+        check_proxy=True,
+    )
+
+    response = asyncio.run(server._handle_stats())
+    payload = json.loads(response[0].kwargs["text"])
+
+    assert payload["proxy"]["status"] == "unreachable"
+    assert payload["proxy"]["url"] == "http://127.0.0.1:9"
+    assert "unreachable" in payload["warning"].lower()
+
+
+def test_mcp_proxy_probe_preserves_shared_proxy_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    class ProbeResponse:
+        status_code = 200
+        text = ""
+
+        @staticmethod
+        def json() -> dict[str, object]:
+            return {"status": "healthy", "alive": True}
+
+    class ProbeClient:
+        def __init__(self, *, timeout: float) -> None:
+            seen["timeout"] = timeout
+
+        async def __aenter__(self) -> ProbeClient:
+            return self
+
+        async def __aexit__(self, *_args: object) -> None:
+            seen["closed"] = True
+
+        async def get(self, url: str) -> ProbeResponse:
+            seen["url"] = url
+            return ProbeResponse()
+
+    seen: dict[str, object] = {}
+    shared_client = object()
+    monkeypatch.setattr(mcp_server.httpx, "AsyncClient", ProbeClient)
+
+    server = mcp_server.HeadroomMCPServer(
+        proxy_url="http://127.0.0.1:8765",
+        check_proxy=True,
+    )
+    server._http_client = shared_client  # type: ignore[assignment]
+
+    result = asyncio.run(server._probe_proxy_unreachable())
+
+    assert result is None
+    assert seen == {
+        "timeout": 5.0,
+        "url": "http://127.0.0.1:8765/livez",
+        "closed": True,
+    }
+    assert server._http_client is shared_client
+
+
+def test_mcp_local_mode_still_works_without_proxy_checking(fresh_store) -> None:
+    server = mcp_server.HeadroomMCPServer(
+        proxy_url="http://127.0.0.1:9",
+        check_proxy=False,
+    )
+
+    response = asyncio.run(server._handle_compress({"content": "local mode stays available"}))
+    payload = json.loads(response[0].kwargs["text"])
+
+    assert "proxy" not in payload
+    assert "warning" not in payload or "unreachable" not in payload["warning"].lower()
+
+
 def test_mcp_retrieve_returns_full_content(fresh_store) -> None:
     """Retrieval is by hash: a stored, unexpired entry always returns its full
     original content (never empty, never a spurious "not found")."""

tests/test_provider_registry.py

@@ -128,6 +128,26 @@ def test_create_proxy_backend_handles_missing_litellm_backend(caplog) -> None:
     assert "LiteLLM backend not available" in caplog.text
 
 
+def test_create_proxy_backend_logs_structured_failure_details(caplog) -> None:
+    logger = logging.getLogger("test")
+
+    with caplog.at_level(logging.ERROR):
+        missing = create_proxy_backend(
+            backend="bedrock",
+            anyllm_provider="ignored",
+            bedrock_region="us-east-1",
+            logger=logger,
+            litellm_backend_cls=lambda provider, region, profile_name=None: (_ for _ in ()).throw(
+                RuntimeError("boom")
+            ),
+        )
+
+    assert missing is None
+    assert "backend initialization failed: backend=litellm-bedrock provider=bedrock error=boom" in (
+        caplog.text
+    )
+
+
 def test_proxy_provider_runtime_loaders_cache_backend_types(monkeypatch) -> None:
     import headroom.providers.registry as registry