feat(proxy): cc-switch reconciler — keep Headroom in the request path alongside cc-switch (#1030)

## Description

[cc-switch](https://github.com/farion1231/cc-switch) is a desktop
provider manager for Claude Code and other coding agents; when a Claude
Code provider is selected, it writes that provider's endpoint and token
into `~/.claude/settings.json`.

This PR adds an opt-in reconciler so Headroom can stay in Claude Code's
request path when cc-switch rewrites that file during provider switches.

The reconciler captures third-party Anthropic-compatible upstream URLs,
points Claude back at the local Headroom proxy, and leaves
official/empty OAuth settings direct unless explicitly opted in. This
update also hardens the watcher so rapid settings rewrites that share
the same float-second mtime are still detected.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [x] 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

- Added an opt-in `HEADROOM_CC_SWITCH_RECONCILE=1` watcher for cc-switch
direct-injection mode.
- Added loopback-only `GET/PUT /admin/upstream` runtime upstream
inspection and override endpoints.
- Preserved token/model settings while rewriting only
`env.ANTHROPIC_BASE_URL` back to the local Headroom proxy.
- Switched reconciler change detection from float-second `st_mtime` to
nanosecond `st_mtime_ns` so rapid provider switches are not missed.
- Added pytest coverage for capture/rewrite behavior, official-provider
defaults, route-official opt-in, loop safety, enabled flags, and the
same-float-mtime provider-switch case.

## Testing

- [ ] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
python -m pytest tests/test_proxy/test_cc_switch_reconciler.py
12 passed in 0.17s

python -m ruff check .
All checks passed!

python -m mypy headroom
Success: no issues found in 359 source files

python -m pytest
7 failed, 5996 passed, 492 skipped, 5814 warnings in 396.41s
```

## Real Behavior Proof

- Environment: macOS, branch `feat/cc-switch-reconciler`, Python 3.13.3.
- Exact command / steps: Ran the focused reconciler pytest file, full
repository ruff check, full `mypy headroom`, and full pytest from the
local PR branch.
- Observed result: All 12 reconciler tests passed, including the rapid
provider-switch case where two writes share the same float mtime but
differ by nanoseconds. Full `ruff check .` and `mypy headroom` passed.
Full pytest completed with 7 failures outside the cc-switch reconciler
test file.
- Not tested: Live cc-switch plus Claude Code end-to-end switch.

## 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
- [x] 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
- [ ] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A.

## Additional Notes

Documentation and CHANGELOG updates are not included in this PR. The
reconciler remains opt-in and off by default.

---------

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

headroom/proxy/cc_switch_reconciler.py

@@ -0,0 +1,192 @@
+"""cc-switch reconciler: keep Headroom in the request path without fighting cc-switch.
+
+Background
+----------
+cc-switch (https://github.com/farion1231/cc-switch) in its default *direct
+injection* mode rewrites the **entire** ``~/.claude/settings.json`` every time
+the user switches provider (atomic overwrite -- see
+``src-tauri/src/services/provider/live.rs``). It writes the selected provider's
+real endpoint + token, e.g.::
+
+    {"env": {"ANTHROPIC_BASE_URL": "https://api.deepseek.com/anthropic",
+              "ANTHROPIC_AUTH_TOKEN": "sk-..."}}
+
+Claude Code re-reads that file, so a running session immediately follows the
+switch. The problem: that overwrite blows away any ``ANTHROPIC_BASE_URL`` that
+points at Headroom.
+
+What this does
+--------------
+A lightweight in-process watcher (poll-based, robust against atomic renames):
+
+1. Detects cc-switch's overwrite.
+2. **Captures** the real provider endpoint and sets it as Headroom's upstream
+   (runtime, no restart -- ``HeadroomProxy.ANTHROPIC_API_URL`` is a class attr
+   read per request).
+3. **Rewrites only** ``env.ANTHROPIC_BASE_URL`` back to Headroom's local URL,
+   leaving the token / model / everything else untouched.
+
+Result: ``Claude -> Headroom (compress) -> selected provider``. cc-switch never
+knows; it is the *trigger*, not a coordination partner. The token rides in the
+request (Claude -> Headroom -> upstream, passed through verbatim); Headroom
+never reads or stores it.
+
+Safety
+------
+- Loop-safe: once ``base_url`` already equals Headroom's URL, it is left alone.
+- Official / empty env (``{"env": {}}``, what cc-switch writes for "Claude
+  Official") is **left direct** by default -- subscription OAuth through a custom
+  base URL is fragile, so v1 does not route it through Headroom. Set
+  ``HEADROOM_CC_SWITCH_ROUTE_OFFICIAL=1`` to route official through Headroom too.
+- Gated entirely behind ``HEADROOM_CC_SWITCH_RECONCILE=1`` -- off by default, so
+  it never affects users who do not opt in.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+from collections.abc import Callable
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+_POLL_INTERVAL_S = 0.3
+
+
+def _settings_path() -> Path:
+    base = os.environ.get("CLAUDE_CONFIG_DIR") or os.path.join(os.path.expanduser("~"), ".claude")
+    return Path(base).expanduser() / "settings.json"
+
+
+def reconciler_enabled() -> bool:
+    return os.environ.get("HEADROOM_CC_SWITCH_RECONCILE", "").strip().lower() in (
+        "1",
+        "true",
+        "yes",
+        "on",
+    )
+
+
+def _route_official() -> bool:
+    return os.environ.get("HEADROOM_CC_SWITCH_ROUTE_OFFICIAL", "").strip().lower() in (
+        "1",
+        "true",
+        "yes",
+        "on",
+    )
+
+
+class CCSwitchReconciler:
+    """Polls Claude settings.json and keeps Headroom in the path (see module docstring)."""
+
+    def __init__(
+        self,
+        *,
+        proxy_url: str,
+        default_upstream: str,
+        set_upstream: Callable[[str], None],
+        path: Path | None = None,
+    ) -> None:
+        self.proxy_url = proxy_url.rstrip("/")
+        self.default_upstream = default_upstream
+        self._set_upstream = set_upstream
+        self.path = path or _settings_path()
+        self.current_upstream: str | None = None
+        self._task: asyncio.Task | None = None
+        self._last_mtime_ns: int | None = None
+
+    async def start(self) -> None:
+        if self._task is not None:
+            return
+        logger.info("cc-switch reconciler: watching %s -> proxy %s", self.path, self.proxy_url)
+        self._task = asyncio.create_task(self._loop())
+
+    async def stop(self) -> None:
+        if self._task is None:
+            return
+        self._task.cancel()
+        try:
+            await self._task
+        except (asyncio.CancelledError, Exception):  # noqa: BLE001
+            pass
+        self._task = None
+
+    async def _loop(self) -> None:
+        while True:
+            try:
+                self.tick()
+            except Exception as exc:  # noqa: BLE001 - watcher must never die
+                logger.debug("cc-switch reconciler tick error: %s", exc)
+            await asyncio.sleep(_POLL_INTERVAL_S)
+
+    # Synchronous core (also directly unit-testable).
+    def tick(self) -> bool:
+        """One reconcile pass. Returns True if it rewrote settings.json."""
+        try:
+            mtime_ns = self.path.stat().st_mtime_ns
+        except FileNotFoundError:
+            return False
+        if mtime_ns == self._last_mtime_ns:
+            return False
+
+        try:
+            data = json.loads(self.path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, OSError):
+            # Transient read/parse failure (e.g. caught mid atomic-replace, or
+            # cc-switch wrote partial JSON). Do NOT consume this mtime — leave
+            # _last_mtime_ns untouched so the next tick retries instead of
+            # treating the broken state as already-processed.
+            return False
+        # Read succeeded: now it is safe to mark this mtime processed.
+        self._last_mtime_ns = mtime_ns
+        if not isinstance(data, dict):
+            return False
+        env = data.get("env")
+        env = dict(env) if isinstance(env, dict) else {}
+        url = env.get("ANTHROPIC_BASE_URL")
+        # A non-string base_url (number/list from a hand-edited file) would
+        # raise on .rstrip() below and spam the watcher loop; treat as empty.
+        if not isinstance(url, str):
+            url = ""
+
+        # Empty / official: cc-switch wrote {"env": {}} (Claude Official, OAuth).
+        if not url:
+            if _route_official():
+                self.current_upstream = self.default_upstream
+                self._set_upstream(self.default_upstream)
+                env["ANTHROPIC_BASE_URL"] = self.proxy_url
+                data["env"] = env
+                self._atomic_write(data)
+                logger.info("cc-switch reconciler: official -> route via Headroom")
+                return True
+            return False  # leave official direct (default, safe for OAuth)
+
+        # Already pointing at us: nothing to do (loop guard).
+        if url.rstrip("/") == self.proxy_url:
+            return False
+
+        # Third-party / custom endpoint: capture it as upstream, point Claude at us.
+        self.current_upstream = url
+        self._set_upstream(url)
+        env["ANTHROPIC_BASE_URL"] = self.proxy_url
+        data["env"] = env
+        self._atomic_write(data)
+        logger.info(
+            "cc-switch reconciler: captured upstream=%s, base_url -> %s", url, self.proxy_url
+        )
+        return True
+
+    def _atomic_write(self, data: dict) -> None:
+        # Per-process temp name: multiple Headroom processes reconciling the
+        # same settings.json must not clobber each other's temp file.
+        tmp = self.path.with_name(f"{self.path.name}.{os.getpid()}.hrtmp")
+        tmp.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
+        os.replace(tmp, self.path)
+        # Skip the mtime bump caused by our own write so we don't re-process it.
+        try:
+            self._last_mtime_ns = self.path.stat().st_mtime_ns
+        except OSError:
+            pass

headroom/proxy/server.py

@@ -1672,6 +1672,32 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
     config = config or ProxyConfig()
     proxy = HeadroomProxy(config)
 
+    # cc-switch reconciler (opt-in: HEADROOM_CC_SWITCH_RECONCILE=1).
+    # Keeps Headroom in the request path while cc-switch overwrites
+    # ~/.claude/settings.json on every provider switch. See
+    # headroom/proxy/cc_switch_reconciler.py for the full rationale.
+    from headroom.proxy.cc_switch_reconciler import (
+        CCSwitchReconciler,
+        reconciler_enabled,
+    )
+
+    _cc_reconciler: CCSwitchReconciler | None = None
+    if reconciler_enabled():
+        _cc_proxy_port = config.port if hasattr(config, "port") else 8787
+
+        def _set_anthropic_upstream(url: str) -> None:
+            from headroom.providers.registry import _normalize_api_url
+
+            HeadroomProxy.ANTHROPIC_API_URL = _normalize_api_url(
+                url, default=DEFAULT_ANTHROPIC_API_URL
+            )
+
+        _cc_reconciler = CCSwitchReconciler(
+            proxy_url=f"http://127.0.0.1:{_cc_proxy_port}",
+            default_upstream=DEFAULT_ANTHROPIC_API_URL,
+            set_upstream=_set_anthropic_upstream,
+        )
+
     # Telemetry beacon (anonymous aggregate stats).
     # With uvicorn workers > 1, each worker runs the lifespan independently.
     # We must ensure only ONE beacon runs across all workers — otherwise each
@@ -1771,6 +1797,15 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
                 else:
                     logger.debug("Beacon: skipping (another worker owns the lock)")
 
+                # Only the beacon-lock owner runs the reconciler. With
+                # uvicorn workers > 1 each worker runs this lifespan; without
+                # this guard every worker would watch + rewrite settings.json
+                # concurrently and each process would hold its own
+                # HeadroomProxy.ANTHROPIC_API_URL, so workers could disagree on
+                # the upstream. Single-owner mirrors the beacon's reasoning.
+                if _cc_reconciler is not None and _beacon_is_owner[0]:
+                    await _cc_reconciler.start()
+
                 app.state.ready = True
                 yield
             except Exception as exc:
@@ -1779,6 +1814,8 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
         finally:
             app.state.ready = False
             # Shutdown
+            if _cc_reconciler is not None:
+                await _cc_reconciler.stop()
             if _beacon_is_owner[0]:
                 await _beacon.stop()
                 _release_beacon_lock()
@@ -2239,6 +2276,21 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
     )
     from headroom.proxy.loopback_guard import require_loopback as _require_loopback
 
+    @app.get("/admin/upstream", dependencies=[Depends(_require_loopback)])
+    async def get_upstream():
+        """Current Anthropic upstream + cc-switch reconciler state (loopback-only).
+
+        Read-only. The upstream is mutated only via the in-process cc-switch
+        reconciler (driven by ~/.claude/settings.json) — there is deliberately
+        no HTTP write route, so a local process cannot redirect credential-
+        bearing traffic to an arbitrary URL through this surface.
+        """
+        return {
+            "anthropic": HeadroomProxy.ANTHROPIC_API_URL,
+            "cc_switch_reconcile": _cc_reconciler is not None,
+            "captured_upstream": getattr(_cc_reconciler, "current_upstream", None),
+        }
+
     @app.get("/debug/tasks", dependencies=[Depends(_require_loopback)])
     async def debug_tasks(stack: bool = False):
         """Enumerate running asyncio tasks.

tests/test_proxy/test_cc_switch_reconciler.py

@@ -0,0 +1,164 @@
+"""Tests for the cc-switch reconciler.
+
+The reconciler keeps Headroom in the request path while cc-switch overwrites
+``~/.claude/settings.json`` on every provider switch. See
+``headroom/proxy/cc_switch_reconciler.py``.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+
+import pytest
+
+from headroom.proxy.cc_switch_reconciler import CCSwitchReconciler
+
+PROXY = "http://127.0.0.1:8787"
+DEFAULT = "https://api.anthropic.com"
+
+
+def _make(tmp_path):
+    captured: list[str] = []
+    sf = tmp_path / "settings.json"
+    r = CCSwitchReconciler(
+        proxy_url=PROXY,
+        default_upstream=DEFAULT,
+        set_upstream=captured.append,
+        path=sf,
+    )
+    return r, sf, captured
+
+
+def _write(sf, obj):
+    sf.write_text(json.dumps(obj))
+    os.utime(sf, None)
+
+
+def test_third_party_captured_and_base_url_rewritten(tmp_path):
+    r, sf, captured = _make(tmp_path)
+    _write(
+        sf,
+        {
+            "env": {
+                "ANTHROPIC_BASE_URL": "https://api.deepseek.com/anthropic",
+                "ANTHROPIC_AUTH_TOKEN": "sk-x",
+                "ANTHROPIC_MODEL": "deepseek",
+            }
+        },
+    )
+    assert r.tick() is True
+    env = json.loads(sf.read_text())["env"]
+    # base_url repointed to Headroom; token + model preserved verbatim.
+    assert env["ANTHROPIC_BASE_URL"] == PROXY
+    assert env["ANTHROPIC_AUTH_TOKEN"] == "sk-x"
+    assert env["ANTHROPIC_MODEL"] == "deepseek"
+    # Real endpoint captured as the upstream.
+    assert captured[-1] == "https://api.deepseek.com/anthropic"
+    assert r.current_upstream == "https://api.deepseek.com/anthropic"
+
+
+def test_no_rewrite_loop(tmp_path):
+    r, sf, _ = _make(tmp_path)
+    _write(sf, {"env": {"ANTHROPIC_BASE_URL": "https://api.deepseek.com/anthropic"}})
+    assert r.tick() is True
+    # Already pointing at Headroom now -> must be a no-op (no infinite loop).
+    assert r.tick() is False
+
+
+def test_switching_provider_recaptures(tmp_path):
+    r, sf, captured = _make(tmp_path)
+    _write(
+        sf,
+        {
+            "env": {
+                "ANTHROPIC_BASE_URL": "https://api.deepseek.com/anthropic",
+                "ANTHROPIC_AUTH_TOKEN": "sk-d",
+            }
+        },
+    )
+    assert r.tick() is True
+    _write(
+        sf,
+        {
+            "env": {
+                "ANTHROPIC_BASE_URL": "https://api.kimi.com/anthropic",
+                "ANTHROPIC_AUTH_TOKEN": "sk-k",
+            }
+        },
+    )
+    assert r.tick() is True
+    assert captured[-1] == "https://api.kimi.com/anthropic"
+    assert json.loads(sf.read_text())["env"]["ANTHROPIC_AUTH_TOKEN"] == "sk-k"
+
+
+def test_same_float_mtime_provider_switch_recaptures(tmp_path):
+    r, sf, captured = _make(tmp_path)
+    base_ns = 1_700_000_000_000_000_000
+    _write(sf, {"env": {"ANTHROPIC_BASE_URL": "https://api.deepseek.com/anthropic"}})
+    os.utime(sf, ns=(base_ns, base_ns))
+    assert r.tick() is True
+
+    _write(sf, {"env": {"ANTHROPIC_BASE_URL": "https://api.kimi.com/anthropic"}})
+    os.utime(sf, ns=(base_ns + 1, base_ns + 1))
+    assert sf.stat().st_mtime == float(base_ns / 1_000_000_000)
+    assert r.tick() is True
+    assert captured[-1] == "https://api.kimi.com/anthropic"
+
+
+def test_official_left_direct_by_default(tmp_path, monkeypatch):
+    monkeypatch.delenv("HEADROOM_CC_SWITCH_ROUTE_OFFICIAL", raising=False)
+    r, sf, _ = _make(tmp_path)
+    _write(sf, {"env": {}})
+    # Empty env = "Claude Official" (OAuth). Default: leave it direct.
+    assert r.tick() is False
+    assert json.loads(sf.read_text())["env"] == {}
+
+
+def test_official_routed_when_opted_in(tmp_path, monkeypatch):
+    monkeypatch.setenv("HEADROOM_CC_SWITCH_ROUTE_OFFICIAL", "1")
+    r, sf, captured = _make(tmp_path)
+    _write(sf, {"env": {}})
+    assert r.tick() is True
+    assert json.loads(sf.read_text())["env"]["ANTHROPIC_BASE_URL"] == PROXY
+    assert captured[-1] == DEFAULT
+
+
+def test_missing_file_is_noop(tmp_path):
+    r, _, _ = _make(tmp_path)  # path does not exist yet
+    assert r.tick() is False
+
+
+def test_non_string_base_url_does_not_crash(tmp_path):
+    r, sf, captured = _make(tmp_path)
+    # A hand-edited / malformed file with a non-string base_url must not raise
+    # (would otherwise blow up on .rstrip() and spam the watcher loop).
+    _write(sf, {"env": {"ANTHROPIC_BASE_URL": 1234}})
+    assert r.tick() is False  # treated as empty -> left direct
+    assert captured == []
+
+
+def test_transient_invalid_json_retries_next_tick(tmp_path):
+    r, sf, captured = _make(tmp_path)
+    # Mid-write garbage: read/parse fails, mtime must NOT be consumed.
+    sf.write_text("{not valid json")
+    os.utime(sf, ns=(1_700_000_000_000_000_000, 1_700_000_000_000_000_000))
+    assert r.tick() is False
+    assert r._last_mtime_ns is None  # broken state not marked processed
+
+    # File repaired at the SAME mtime: next tick must still process it.
+    sf.write_text(json.dumps({"env": {"ANTHROPIC_BASE_URL": "https://api.deepseek.com/anthropic"}}))
+    os.utime(sf, ns=(1_700_000_000_000_000_000, 1_700_000_000_000_000_000))
+    assert r.tick() is True
+    assert captured[-1] == "https://api.deepseek.com/anthropic"
+
+
+@pytest.mark.parametrize(
+    "val,expected",
+    [("1", True), ("true", True), ("on", True), ("0", False), ("", False)],
+)
+def test_enabled_flag(monkeypatch, val, expected):
+    from headroom.proxy.cc_switch_reconciler import reconciler_enabled
+
+    monkeypatch.setenv("HEADROOM_CC_SWITCH_RECONCILE", val)
+    assert reconciler_enabled() is expected