OAuth token rotation persistence and cross-process sync

Fixes on record

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

• #15111 fix(nous-oauth): preserve obtained_at in pool + actionable message on RT reuse

  merged to main

  files: agent/credential_pool.py, hermes_cli/auth.py, tests/agent/test_credential_pool.py, tests/hermes_cli/test_auth_nous_provider.py

  Show the diff

  • agent/credential_pool.py

    @@ -1056,6 +1056,18 @@ def _is_suppressed(_p, _s):  # type: ignore[misc]
                         "inference_base_url": state.get("inference_base_url"),
                         "agent_key": state.get("agent_key"),
                         "agent_key_expires_at": state.get("agent_key_expires_at"),
    +                    # Carry the mint/refresh timestamps into the pool so
    +                    # freshness-sensitive consumers (self-heal hooks, pool
    +                    # pruning by age) can distinguish just-minted credentials
    +                    # from stale ones.  Without these, fresh device_code
    +                    # entries get obtained_at=None and look older than they
    +                    # are (#15099).
    +                    "obtained_at": state.get("obtained_at"),
    +                    "expires_in": state.get("expires_in"),
    +                    "agent_key_id": state.get("agent_key_id"),
    +                    "agent_key_expires_in": state.get("agent_key_expires_in"),
    +                    "agent_key_reused": state.get("agent_key_reused"),
    +                    "agent_key_obtained_at": state.get("agent_key_obtained_at"),
                         "tls": state.get("tls") if isinstance(state.get("tls"), dict) else None,
                         "label": seeded_label,
                     },

  • hermes_cli/auth.py

    @@ -1889,6 +1889,28 @@ def _refresh_access_token(
         code = str(error_payload.get("error", "invalid_grant"))
         description = str(error_payload.get("error_description") or "Refresh token exchange failed")
         relogin = code in {"invalid_grant", "invalid_token"}
    +
    +    # Detect the OAuth 2.1 "refresh token reuse" signal from the Nous portal
    +    # server and surface an actionable message.  This fires when an external
    +    # process (health-check script, monitoring tool, custom self-heal hook)
    +    # called POST /api/oauth/token with Hermes's refresh_token without
    +    # persisting the rotated token back to auth.json — the server then
    +    # retires the original RT, Hermes's next refresh uses it, and the whole
    +    # session chain gets revoked as a token-theft signal (#15099).
    +    lowered = description.lower()
    +    if "reuse" in lowered or "reuse detected" in lowered:
    +        description = (
    +            "Nous Portal detected refresh-token reuse and revoked this session.\n"
    +            "This usually means an external process (monitoring script, "
    +            "custom self-heal hook, or another Hermes install sharing "
    +            "~/.hermes/auth.json) called POST /api/oauth/token with Hermes's "
    +            "refresh token without persisting the rotated token back.\n"
    +            "Nous refresh tokens are single-use — only Hermes may call the "
    +            "refresh endpoint. For health checks, use `hermes auth status` "
    +            "instead.\n"
    +            "Re-authenticate with: hermes auth add nous"
    +        )
    +
         raise AuthError(description, provider="nous", code=code, relogin_required=relogin)
     
     

  • tests/agent/test_credential_pool.py

    @@ -1102,3 +1102,72 @@ def test_load_pool_does_not_seed_qwen_oauth_when_no_token(tmp_path, monkeypatch)
     
         assert not pool.has_credentials()
         assert pool.entries() == []
    +
    +
    +def test_nous_seed_from_singletons_preserves_obtained_at_timestamps(tmp_path, monkeypatch):
    +    """Regression test for #15099 secondary issue.
    +
    +    When ``_seed_from_singletons`` materialises a device_code pool entry from
    +    the ``providers.nous`` singleton, it must carry the mint/refresh
    +    timestamps (``obtained_at``, ``agent_key_obtained_at``, ``expires_in``,
    +    etc.) into the pool entry.  Without them, freshness-sensitive consumers
    +    (self-heal hooks, pool pruning by age) treat just-minted credentials as
    +    older than they actually are and evict them.
    +    """
    +    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
    +    _write_auth_store(
    +        tmp_path,
    +        {
    +            "version": 1,
    +            "providers": {
    +                "nous": {
    +                    "access_token": "at_XXXXXXXX",
    +                    "refresh_token": "rt_YYYYYYYY",
    +                    "client_id": "hermes-cli",
    +                    "portal_base_url": "https://portal.nousresearch.com",
    +                    "inference_base_url": "https://inference.nousresearch.com/v1",
    +                    "token_type": "Bearer",
    +                    "scope": "openid profile",
    +                    "obtained_at": "2026-04-24T10:00:00+00:00",
    +                    "expires_at": "2026-04-24T11:00:00+00:00",
    +                    "expires_in": 3600,
    +                    "agent_key": "sk-nous-AAAA",
    +                    "agent_key_id": "ak_123",
    +                    "agent_key_expires_at": "2026-04-25T10:00:00+00:00",
    +                    "agent_key_expires_in": 86400,
    +                    "agent_key_reused": False,
    +                    "agent_key_obtained_at": "2026-04-24T10:00:05+00:00",
    +                    "tls": {"insecure": False, "ca_bundle": None},
    +                },
    +            },
    +        },
    +    )
    +
    +    from agent.credential_pool import load_pool
    +
    +    pool = load_pool("nous")
    +    entries = pool.entries()
    +
    +    device_entries = [e for e in entries if e.source == "device_code"]
    +    assert len(device_entries) == 1, f"expected single device_code entry; got {len(device_entries)}"
    +    e = device_entries[0]
    +
    +    # Direct dataclass fields — must survive the singleton → pool copy.
    +    assert e.access_token == "at_XXXXXXXX"
    +    assert e.refresh_token == "rt_YYYYYYYY"
    +    assert e.expires_at == "2026-04-24T11:00:00+00:00"
    +    assert e.agent_key == "sk-nous-AAAA"
    +    assert e.agent_key_expires_at == "2026-04-25T10:00:00+00:00"
    +
    +    # Extra fields — this is what regressed.  These must be carried through
    +    # via ``extra`` dict or __getattr__, NOT silently dropped.
    +    assert e.obtained_at == "2026-04-24T10:00:00+00:00", (
    +        f"obtained_at was dropped during seed; got {e.obtained_at!r}. This breaks "
    +        f"downstream pool-freshness consumers (#15099)."
    +    )
    +    assert e.agent_key_obtained_at == "2026-04-24T10:00:05+00:00"
    +    assert e.expires_in == 3600
    +    assert e.agent_key_id == "ak_123"
    +    assert e.agent_key_expires_in == 86400
    +    assert e.agent_key_reused is False
    +

  • tests/hermes_cli/test_auth_nous_provider.py

    @@ -732,3 +732,83 @@ def test_persist_nous_credentials_no_label_uses_auto_derived(tmp_path, monkeypat
         # No "label" key embedded in providers.nous when the caller didn't supply one.
         payload = json.loads((hermes_home / "auth.json").read_text())
         assert "label" not in payload["providers"]["nous"]
    +
    +
    +def test_refresh_token_reuse_detection_surfaces_actionable_message():
    +    """Regression for #15099.
    +
    +    When the Nous Portal server returns ``invalid_grant`` with
    +    ``error_description`` containing "reuse detected", Hermes must surface an
    +    actionable message explaining that an external process consumed the
    +    refresh token.  The default opaque "Refresh token reuse detected; please
    +    re-authenticate" string led users to report this as a Hermes persistence
    +    bug when the true cause is external RT consumption (monitoring scripts,
    +    custom self-heal hooks).
    +    """
    +    from hermes_cli.auth import _refresh_access_token
    +
    +    class _FakeResponse:
    +        status_code = 400
    +
    +        def json(self):
    +            return {
    +                "error": "invalid_grant",
    +                "error_description": "Refresh token reuse detected; please re-authenticate",
    +            }
    +
    +    class _FakeClient:
    +        def post(self, *args, **kwargs):
    +            return _FakeResponse()
    +
    +    with pytest.raises(AuthError) as exc_info:
    +        _refresh_access_token(
    +            client=_FakeClient(),
    +            portal_base_url="https://portal.nousresearch.com",
    +            client_id="hermes-cli",
    +            refresh_token="[redacted:secret-assignment]",
    +        )
    +
    +    message = str(exc_info.value)
    +    assert "refresh-token reuse" in message.lower() or "refresh token reuse" in message.lower()
    +    # The message must mention the external-process cause and give next steps.
    +    assert "external process" in message.lower() or "monitoring script" in message.lower()
    +    assert "hermes auth add nous" in message.lower()
    +    # Must still be classified as invalid_grant + relogin_required.
    +    assert exc_info.value.code == "invalid_grant"
    +    assert exc_info.value.relogin_required is True
    +
    +
    +def test_refresh_non_reuse_error_keeps_original_description():
    +    """Non-reuse invalid_grant errors must keep their original description untouched.
    +
    +    Only the "reuse detected" signature should trigger the actionable message;
    +    generic ``invalid_grant: Refresh session has been revoked`` (the
    +    downstream consequence) keeps its original text so we don't overwrite
    +    useful server context for unrelated failure modes.
    +    """
    +    from hermes_cli.auth import _refresh_access_token
    +
    +    class _FakeResponse:
    +        status_code = 400
    +
    +        def json(self):
    +            return {
    +                "error": "invalid_grant",
    +                "error_description": "Refresh session has been revoked",
    +            }
    +
    +    class _FakeClient:
    +        def post(self, *args, **kwargs):
    +            return _FakeResponse()
    +
    +    with pytest.raises(AuthError) as exc_info:
    +        _refresh_access_token(
    +            client=_FakeClient(),
    +            portal_base_url="https://portal.nousresearch.com",
    +            client_id="hermes-cli",
    +            refresh_token="[redacted:secret-assignment]",
    +        )
    +
    +    assert "Refresh session has been revoked" in str(exc_info.value)
    +    # Must not have been rewritten with the reuse message.
    +    assert "external process" not in str(exc_info.value).lower()

• #46614 fix(auth): resolve xAI OAuth credentials across profiles + write rotated tokens back to root

  merged to main

  files: hermes_cli/auth.py, scripts/release.py, tests/hermes_cli/test_xai_oauth_profile_auth.py, tests/hermes_cli/test_xai_oauth_writethrough.py

  Show the diff

  • hermes_cli/auth.py

    @@ -1079,8 +1079,13 @@ def _load_auth_store(auth_file: Optional[Path] = None) -> Dict[str, Any]:
         return {"version": AUTH_STORE_VERSION, "providers": {}}
     
     
    -def _save_auth_store(auth_store: Dict[str, Any]) -> Path:
    -    auth_file = _auth_file_path()
    +def _save_auth_store(auth_store: Dict[str, Any], target_path: Optional[Path] = None) -> Path:
    +    # target_path=None preserves the existing contract (write the active
    +    # store at _auth_file_path()). An explicit path lets callers persist a
    +    # specific store — e.g. the global-root write-through for rotating xAI
    +    # OAuth grants (#43589) — reusing this function's atomic O_EXCL + 0o600
    +    # write so the root auth.json gets the same TOCTOU-safe treatment.
    +    auth_file = target_path if target_path is not None else _auth_file_path()
         auth_file.parent.mkdir(parents=True, exist_ok=True)
         # Tighten parent dir to 0o700 so siblings can't traverse to creds.
         # No-op on Windows (POSIX mode bits not enforced); ignore failures.
    @@ -3886,13 +3891,64 @@ def _entry_usable(entry: Dict[str, Any]) -> bool:
     # xAI Grok OAuth — tokens stored in ~/.hermes/auth.json
     # =============================================================================
     
    +def _xai_oauth_state_from_store(auth_store: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    +    """Return usable xAI OAuth state from provider state or credential pool."""
    +    state = _load_provider_state(auth_store, "xai-oauth")
    +    tokens = [redacted:secret-assignment]"tokens") if isinstance(state, dict) else None
    +    if isinstance(tokens, dict):
    +        access_token = [redacted:secret-assignment]"access_token", "") or "").strip()
    +        refresh_token = [redacted:secret-assignment]"refresh_token", "") or "").strip()
    +        if access_token and refresh_token:
    +            return state
    +
    +    credential_pool = [redacted:secret-assignment]"credential_pool")
    +    entries = (
    +        credential_pool.get("xai-oauth")
    +        if isinstance(credential_pool, dict)
    +        else None
    +    )
    +    if isinstance(entries, list):
    +        for entry in entries:
    +            if not isinstance(entry, dict):
    +                continue
    +            access_token = [redacted:secret-assignment]"access_token", "") or "").strip()
    +            refresh_token = [redacted:secret-assignment]"refresh_token", "") or "").strip()
    +            if not access_token or not refresh_token:
    +                continue
    +            merged = dict(state or {})
    +            merged["tokens"] = {
    +                "access_token": access_token,
    +                "refresh_token": refresh_token,
    +                "token_type": str(entry.get("token_type") or "Bearer"),
    +            }
    +            if entry.get("last_refresh"):
    +                merged["last_refresh"] = entry.get("last_refresh")
    +            merged.setdefault("auth_mode", "oauth_pkce")
    +            return merged
    +
    +    return state if isinstance(state, dict) else None
    +
    +
    +def _xai_oauth_state_has_usable_tokens(state: Optional[Dict[str, Any]]) -> bool:
    +    tokens = [redacted:secret-assignment]"tokens") if isinstance(state, dict) else None
    +    return (
    +        isinstance(tokens, dict)
    +        and bool(str(tokens.get("access_token", "") or "").strip())
    +        and bool(str(tokens.get("refresh_token", "") or "").strip())
    +    )
    +
    +
     def _read_xai_oauth_tokens(*, _lock: bool = True) -> Dict[str, Any]:
         if _lock:
             with _auth_store_lock():
                 auth_store = _load_auth_store()
         else:
             auth_store = _load_auth_store()
    -    state = _load_provider_state(auth_store, "xai-oauth")
    +    state = _xai_oauth_state_from_store(auth_store)
    +    if not _xai_oauth_state_has_usable_tokens(state):
    +        global_state = _xai_oauth_state_from_store(_load_global_auth_store())
    +        if _xai_oauth_state_has_usable_tokens(global_state):
    +            state = global_state
         if not state:
             raise AuthError(
                 "No xAI OAuth credentials stored. Select xAI Grok OAuth (SuperGrok / Premium+) in `hermes model`.",
    @@ -3932,6 +3988,62 @@ def _read_xai_oauth_tokens(*, _lock: bool = True) -> Dict[str, Any]:
         }
     
     
    +def _profile_has_own_xai_oauth_state(auth_store: Dict[str, Any]) -> bool:
    +    """True when this store has its OWN ``providers.xai-oauth`` block.
    +
    +    Distinguishes a profile that genuinely shadows the root xAI grant from
    +    one that only *reads* root via ``_load_provider_state``'s fallback. Only
    +    the latter needs the refresh write-through below.
    +    """
    +    providers = auth_store.get("providers")
    +    return isinstance(providers, dict) and isinstance(providers.get("xai-oauth"), dict)
    +
    +
    +def _write_through_xai_oauth_to_global_root(state: Dict[str, Any]) -> None:
    +    """Persist a rotated xAI OAuth ``state`` into the global-root auth.json.
    +
    +    Best-effort write-through for the multi-profile rotation hazard (#43589):
    +    xAI rotates the refresh_token on every refresh, so when a profile session
    +    refreshes a grant it resolved from the root fallback, the rotated chain
    +    must land back in root. Otherwise root keeps a now-revoked refresh token
    +    and every other profile reading the stale root grant dies with
    +    ``invalid_grant`` once its access token expires.
    +
    +    Only updates ``providers.xai-oauth`` in the root store; never touches the
    +    profile store (the caller already saved that). Swallows all errors — a
    +    failed write-through degrades to the pre-existing behavior (root stale),
    +    it must never break the profile's own successful save.
    +    """
    +    global_path = _global_auth_file_path()
    +    if global_path is None:
    +        # Classic mode (profile == root); the profile save already hit root.
    +        return
    +    # Seat belt: under pytest, refuse to write the real user's
    +    # ~/.hermes/auth.json even when HERMES_HOME points at a profile path
    +    # (mirrors the read-side guard in _load_global_auth_store). Uses the
    +    # unmodified HOME env, not Path.home() which fixtures may monkeypatch.
    +    if os.environ.get("PYTEST_CURRENT_TEST"):
    +        real_home_env = os.environ.get("HOME", "")
    +        if real_home_env:
    +            real_root = Path(real_home_env) / ".hermes" / "auth.json"
    +            try:
    +                if global_path.resolve(strict=False) == real_root.resolve(strict=False):
    +                    return
    +            except Exception:
    +                return
    +    try:
    +        if global_path.exists():
    +            global_store = _load_auth_store(global_path)
    +        else:
    +            global_store = {}
    +        if not isinstance(global_store, dict):
    +            return
    +        _store_provider_state(global_store, "xai-oauth", dict(state), set_active=False)
    +        _save_auth_store(global_store, global_path)
    +    except Exception as exc:  # pragma: no cover - best effort
    +        logger.debug("xAI OAuth: write-through to global root failed: %s", exc)
    +
    +
     def _save_xai_oauth_tokens(
         tokens: [redacted:secret-assignment] Any],
         *,
    @@ -3943,6 +4055,11 @@ def _save_xai_oauth_tokens(
             last_refresh = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
         with _auth_store_lock():
             auth_store = _load_auth_store()
    +        # A profile that lacks its own xai-oauth block is reading the root
    +        # grant through _load_provider_state's fallback. When such a profile
    +        # refreshes the (rotating) grant, we must write the rotated chain back
    +        # to root too, or root is left holding a revoked refresh token (#43589).
    +        write_through_to_root = not _profile_has_own_xai_oauth_state(auth_store)
             state = _load_provider_state(auth_store, "xai-oauth") or {}
             state["tokens"] = tokens
             state["last_refresh"] = last_refresh
    @@ -3953,6 +4070,8 @@ def _save_xai_oauth_tokens(
                 state["redirect_uri"] = redirect_uri
             _save_provider_state(auth_store, "xai-oauth", state)
             _save_auth_store(auth_store)
    +        if write_through_to_root:
    +            _write_through_xai_oauth_to_global_root(state)
     
     
     def _xai_access_token_is_expiring(access_token: str, skew_seconds: int = 0) -> bool:

  • scripts/release.py

    @@ -1541,6 +1541,7 @@
         "[redacted:email]": "dschnurbusch",  # PR #42373 salvage (archive compressed conversation lineages)
         "[redacted:email]": "bcsmith528",  # PR #20589 salvage (register_slack_action_handler plugin API)
         "[redacted:email]": "liuhao1024",  # PR #45494 salvage (claim session slot before auto-resume task; #45456)
    +    "[redacted:email]": "capt-marbles",  # PR #38440 salvage (resolve xAI OAuth credentials across profiles; #43589)
     }
     
     

  • tests/hermes_cli/test_xai_oauth_profile_auth.py

    @@ -0,0 +1,79 @@
    +"""Regression tests for xAI OAuth auth resolution in profile/cron contexts."""
    +
    +import pytest
    +
    +from hermes_cli import auth
    +from hermes_cli.auth import AuthError
    +
    +
    +def test_read_xai_oauth_tokens_uses_credential_pool_when_provider_tokens_empty(monkeypatch):
    +    """Profile auth can have fresh pool tokens while singleton provider state is empty.
    +
    +    This mirrors profiled cron after re-auth/credential-pool sync: the xAI
    +    OAuth credential is usable, but `providers.xai-oauth.tokens` may be empty
    +    or stale. Treating that as missing auth makes cron keep failing after the
    +    user has successfully re-authenticated.
    +    """
    +    store = {
    +        "providers": {"xai-oauth": {"tokens": {}, "last_auth_error": {}}},
    +        "credential_pool": {
    +            "xai-oauth": [
    +                {
    +                    "access_token": "pool-access",
    +                    "refresh_token": "pool-refresh",
    +                    "token_type": "Bearer",
    +                    "last_refresh": "2026-06-03T19:00:00Z",
    +                }
    +            ]
    +        },
    +    }
    +    monkeypatch.setattr(auth, "_load_auth_store", lambda: store)
    +    monkeypatch.setattr(auth, "_load_global_auth_store", lambda: {})
    +
    +    resolved = auth._read_xai_oauth_tokens(_lock=False)
    +
    +    assert resolved["tokens"]["access_token"] == "pool-access"
    +    assert resolved["tokens"]["refresh_token"] == "pool-refresh"
    +    assert resolved["tokens"]["token_type"] == "Bearer"
    +    assert resolved["last_refresh"] == "2026-06-03T19:00:00Z"
    +
    +
    +def test_read_xai_oauth_tokens_uses_global_store_when_profile_state_empty(monkeypatch):
    +    """A profile/cron process should see root xAI auth after user re-auths there."""
    +    profile_store = {"providers": {"xai-oauth": {"tokens": {}}}}
    +    global_store = {
    +        "providers": {
    +            "xai-oauth": {
    +                "tokens": {
    +                    "access_token": "global-access",
    +                    "refresh_token": "global-refresh",
    +                    "token_type": "Bearer",
    +                },
    +                "last_refresh": "2026-06-03T19:05:00Z",
    +            }
    +        }
    +    }
    +    monkeypatch.setattr(auth, "_load_auth_store", lambda: profile_store)
    +    monkeypatch.setattr(auth, "_load_global_auth_store", lambda: global_store)
    +
    +    resolved = auth._read_xai_oauth_tokens(_lock=False)
    +
    +    assert resolved["tokens"]["access_token"] == "global-access"
    +    assert resolved["tokens"]["refresh_token"] == "global-refresh"
    +    assert resolved["last_refresh"] == "2026-06-03T19:05:00Z"
    +
    +
    +def test_read_xai_oauth_tokens_still_requires_usable_tokens(monkeypatch):
    +    """Fallback should not hide genuinely broken xAI auth state."""
    +    store = {
    +        "providers": {"xai-oauth": {"tokens": {}}},
    +        "credential_pool": {"xai-oauth": [{"access_token": "", "refresh_token": ""}]},
    +    }
    +    monkeypatch.setattr(auth, "_load_auth_store", lambda: store)
    +    monkeypatch.setattr(auth, "_load_global_auth_store", lambda: {})
    +
    +    with pytest.raises(AuthError) as exc:
    +        auth._read_xai_oauth_tokens(_lock=False)
    +
    +    assert exc.value.code == "xai_auth_missing_access_token"
    +    assert exc.value.relogin_required is True

  • tests/hermes_cli/test_xai_oauth_writethrough.py

    @@ -0,0 +1,169 @@
    +"""Regression tests for xAI OAuth refresh write-through to the global root.
    +
    +Companion to ``test_xai_oauth_profile_auth.py``. That file covers the READ
    +fallback (profile -> credential pool -> global root). These cover the WRITE
    +side: when a profile that has no own ``providers.xai-oauth`` block refreshes
    +the (rotating) grant it resolved from the root fallback, the rotated tokens
    +must be written back to the global root too. Otherwise root keeps a revoked
    +refresh token and every other profile reading root's stale grant dies with
    +``invalid_grant`` once its access token expires (issue #43589).
    +
    +The tests drive the real ``_save_xai_oauth_tokens`` against real on-disk auth
    +stores (profile + root under ``tmp_path``) rather than mocking the save
    +boundary, so they exercise the actual atomic write path.
    +"""
    +
    +import json
    +
    +import pytest
    +
    +from hermes_cli import auth
    +
    +
    +def _write_store(path, store):
    +    path.parent.mkdir(parents=True, exist_ok=True)
    +    path.write_text(json.dumps(store), encoding="utf-8")
    +
    +
    +def _read_store(path):
    +    return json.loads(path.read_text(encoding="utf-8"))
    +
    +
    [redacted:email]
    +def profile_and_root(tmp_path, monkeypatch):
    +    """Wire a profile auth store + a distinct global-root auth store on disk.
    +
    +    Returns (profile_path, root_path). The pytest seat belt in
    +    ``_write_through_xai_oauth_to_global_root`` only refuses the *real* user's
    +    ``$HOME/.hermes/auth.json``; a tmp_path root is allowed, so we point HOME
    +    away from the tmp root to keep the guard from tripping on these fixtures.
    +    """
    +    profile_path = tmp_path / "profiles" / "work" / "auth.json"
    +    root_path = tmp_path / "root" / "auth.json"
    +
    +    monkeypatch.setattr(auth, "_auth_file_path", lambda: profile_path)
    +    monkeypatch.setattr(auth, "_global_auth_file_path", lambda: root_path)
    +    # Keep the pytest write seat belt from matching our tmp root.
    +    monkeypatch.setenv("HOME", str(tmp_path / "not-the-root"))
    +    return profile_path, root_path
    +
    +
    +def test_refresh_writes_through_to_root_when_profile_has_no_own_state(profile_and_root):
    +    """Profile reading root's grant must push rotated tokens back to root."""
    +    profile_path, root_path = profile_and_root
    +    # Profile has NO own xai-oauth block (reads root via fallback).
    +    _write_store(profile_path, {"version": 1, "providers": {}})
    +    _write_store(
    +        root_path,
    +        {
    +            "version": 1,
    +            "providers": {
    +                "xai-oauth": {
    +                    "tokens": {
    +                        "access_token": "old-access",
    +                        "refresh_token": "old-refresh",
    +                    }
    +                }
    +            },
    +        },
    +    )
    +
    +    rotated = {
    +        "access_token": "new-access",
    +        "refresh_token": "new-refresh",
    +        "token_type": "Bearer",
    +    }
    +    auth._save_xai_oauth_tokens(rotated)
    +
    +    # Profile got the rotated chain (existing behavior).
    +    profile = _read_store(profile_path)
    +    assert profile["providers"]["xai-oauth"]["tokens"]["refresh_token"] == "new-refresh"
    +
    +    # AND the global root no longer holds the revoked refresh token (#43589).
    +    root = _read_store(root_path)
    +    assert root["providers"]["xai-oauth"]["tokens"]["access_token"] == "new-access"
    +    assert root["providers"]["xai-oauth"]["tokens"]["refresh_token"] == "new-refresh"
    +
    +
    +def test_refresh_does_not_touch_root_when_profile_has_own_state(profile_and_root):
    +    """A profile that genuinely shadows root must NOT clobber the root grant."""
    +    profile_path, root_path = profile_and_root
    +    # Profile has its OWN xai-oauth block: it shadows root legitimately.
    +    _write_store(
    +        profile_path,
    +        {
    +            "version": 1,
    +            "providers": {
    +                "xai-oauth": {
    +                    "tokens": {
    +                        "access_token": "profile-old",
    +                        "refresh_token": "profile-old-refresh",
    +                    }
    +                }
    +            },
    +        },
    +    )
    +    _write_store(
    +        root_path,
    +        {
    +            "version": 1,
    +            "providers": {
    +                "xai-oauth": {
    +                    "tokens": {
    +                        "access_token": "root-untouched",
    +                        "refresh_token": "root-untouched-refresh",
    +                    }
    +                }
    +            },
    +        },
    +    )
    +
    +    auth._save_xai_oauth_tokens(
    +        {"access_token": "profile-new", "refresh_token": "profile-new-refresh"}
    +    )
    +
    +    profile = _read_store(profile_path)
    +    assert profile["providers"]["xai-oauth"]["tokens"]["refresh_token"] == "profile-new-refresh"
    +
    +    # Root is a separate grant chain — must be left exactly as-is.
    +    root = _read_store(root_path)
    +    assert root["providers"]["xai-oauth"]["tokens"]["access_token"] == "root-untouched"
    +    assert root["providers"]["xai-oauth"]["tokens"]["refresh_token"] == "root-untouched-refresh"
    +
    +
    +def test_write_through_is_noop_in_classic_mode(tmp_path, monkeypatch):
    +    """Classic mode (profile == root) already saves to root; no double write."""
    +    profile_path = tmp_path / "auth.json"
    +    monkeypatch.setattr(auth, "_auth_file_path", lambda: profile_path)
    +    # Classic mode: _global_auth_file_path returns None.
    +    monkeypatch.setattr(auth, "_global_auth_file_path", lambda: None)
    +    _write_store(profile_path, {"version": 1, "providers": {}})
    +
    +    # Should not raise and should persist to the single store.
    +    auth._save_xai_oauth_tokens(
    +        {"access_token": "a", "refresh_token": "r"}
    +    )
    +    store = _read_store(profile_path)
    +    assert store["providers"]["xai-oauth"]["tokens"]["refresh_token"] == "r"
    +
    +
    +def test_write_through_failure_does_not_break_profile_save(profile_and_root, monkeypatch):
    +    """A failed root write-through must not break the profile's own save."""
    +    profile_path, root_path = profile_and_root
    +    _write_store(profile_path, {"version": 1, "providers": {}})
    +    _write_store(root_path, {"version": 1, "providers": {}})
    +
    +    # Make the root write blow up; the profile save must still succeed.
    +    real_save = auth._save_auth_store
    +
    +    def _exploding_save(store, target_path=None):
    +        if target_path is not None and target_path == root_path:
    +            raise OSError("simulated root write failure")
    +        return real_save(store, target_path)
    +
    +    monkeypatch.setattr(auth, "_save_auth_store", _exploding_save)
    +
    +    auth._save_xai_oauth_tokens({"access_token": "a", "refresh_token": "r"})
    +
    +    profile = _read_store(profile_path)
    +    assert profile["providers"]["xai-oauth"]["tokens"]["refresh_token"] == "r"

• #52760 fix(auth): write rotated Codex/xAI pool grant through to global root (#48415)

  merged to main

  files: agent/credential_pool.py, tests/agent/test_credential_pool_oauth_writethrough.py