Skip to content

fix(auth): share Codex OAuth pool across profiles #34121

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
vishutdhar wants to merge 1 commit into from
Closed

fix(auth): share Codex OAuth pool across profiles #34121

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 61 additions & 13 deletions agent/credential_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ def _load_config_safe() -> Optional[dict]:
AUTH_TYPE_API_KEY = "api_key"

SOURCE_MANUAL = "manual"
CODEX_SHARED_POOL_SOURCES = {"device_code", "manual:device_code"}

STRATEGY_FILL_FIRST = "fill_first"
STRATEGY_ROUND_ROBIN = "round_robin"
Expand Down Expand Up @@ -496,7 +497,7 @@ def _sync_anthropic_entry_from_credentials_file(self, entry: PooledCredential) -
return entry

def _sync_codex_entry_from_auth_store(self, entry: PooledCredential) -> PooledCredential:
"""Sync a Codex device_code pool entry from auth.json if tokens differ.
"""Sync a Codex device-code-backed pool entry from auth.json if tokens differ.

When a Codex OAuth access token expires (or the ChatGPT account hits
its 5h/weekly quota), the pool entry gets marked ``STATUS_EXHAUSTED``
Expand All @@ -508,13 +509,53 @@ def _sync_codex_entry_from_auth_store(self, entry: PooledCredential) -> PooledCr
though fresh credentials are sitting on disk — and every request
fails with "no available entries (all exhausted or empty)".

Mirrors the Nous/Anthropic resync paths above. Only applies to
device_code-sourced entries; env/API-key-sourced entries have no
auth.json shadow to sync from.
In profile mode, ``credential_pool.openai-codex`` is shared from the
global-root auth store because Codex refresh tokens are single-use.
Prefer that shared pool copy, then fall back to the singleton provider
state for legacy stores that have not materialised a pool entry yet.
"""
if self.provider != "openai-codex" or entry.source != "device_code":
if (
self.provider != "openai-codex"
or entry.auth_type != AUTH_TYPE_OAUTH
or entry.source not in CODEX_SHARED_POOL_SOURCES
):
return entry
try:
shared_entries = read_credential_pool("openai-codex")
shared_payload = None
for payload in shared_entries:
if not isinstance(payload, dict):
continue
if payload.get("id") == entry.id:
shared_payload = payload
break
if shared_payload is None:
for payload in shared_entries:
if not isinstance(payload, dict):
continue
if payload.get("source") == entry.source:
shared_payload = payload
break
if isinstance(shared_payload, dict):
shared_entry = PooledCredential.from_dict("openai-codex", shared_payload)
if shared_entry.access_token and (
shared_entry.access_token != (entry.access_token or "")
or (
shared_entry.refresh_token
and shared_entry.refresh_token != (entry.refresh_token or "")
)
or shared_entry.last_status != entry.last_status
or shared_entry.last_error_code != entry.last_error_code
or shared_entry.last_error_reset_at != entry.last_error_reset_at
):
logger.debug(
"Pool entry %s: syncing Codex tokens from shared credential pool",
entry.id,
)
self._replace_entry(entry, shared_entry)
return shared_entry
if entry.source != "device_code":
return entry
with _auth_store_lock():
auth_store = _load_auth_store()
state = _load_provider_state(auth_store, "openai-codex")
Expand Down Expand Up @@ -818,6 +859,8 @@ def _refresh_entry(self, entry: PooledCredential, *, force: bool) -> Optional[Po
synced = self._sync_codex_entry_from_auth_store(entry)
if synced is not entry:
entry = synced
if not self._entry_needs_refresh(entry):
return entry
refreshed = auth_mod.refresh_codex_oauth_pure(
entry.access_token,
entry.refresh_token,
Expand Down Expand Up @@ -978,7 +1021,14 @@ def _refresh_entry(self, entry: PooledCredential, *, force: bool) -> Optional[Po
# if they have rotated since.
if self.provider == "openai-codex":
synced = self._sync_codex_entry_from_auth_store(entry)
if synced.refresh_token != entry.refresh_token:
if (
synced is not entry
and (
synced.refresh_token != entry.refresh_token
or synced.access_token != entry.access_token
or not self._entry_needs_refresh(synced)
)
):
logger.debug(
"Codex OAuth refresh failed but auth.json has newer tokens — adopting"
)
Expand Down Expand Up @@ -1180,14 +1230,12 @@ def _available_entries(self, *, clear_expired: bool = False, refresh: bool = Fal
if synced is not entry:
entry = synced
cleared_any = True
# For openai-codex entries, same pattern: the user may have
# re-authed via `hermes model` / `hermes auth` after a 429/401,
# leaving fresh tokens on disk while the pool entry is still
# frozen behind last_error_reset_at (can be hours in the
# future for ChatGPT weekly windows).
# For openai-codex entries, sync before status checks even when
# they are not exhausted yet. In profile mode the global-root pool
# is the single source of truth for single-use refresh tokens.
if (self.provider == "openai-codex"
and entry.source == "device_code"
and entry.last_status == STATUS_EXHAUSTED):
and entry.auth_type == AUTH_TYPE_OAUTH
and entry.source in CODEX_SHARED_POOL_SOURCES):
synced = self._sync_codex_entry_from_auth_store(entry)
if synced is not entry:
entry = synced
Expand Down
162 changes: 133 additions & 29 deletions hermes_cli/auth.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@
CODEX_OAUTH_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
CODEX_OAUTH_TOKEN_URL = "https://auth.openai.com/oauth/token"
CODEX_ACCESS_TOKEN_REFRESH_SKEW_SECONDS = 120
SHARED_CREDENTIAL_POOL_PROVIDERS = frozenset({"openai-codex"})
XAI_OAUTH_ISSUER = "https://auth.x.ai"
XAI_OAUTH_DISCOVERY_URL = f"{XAI_OAUTH_ISSUER}/.well-known/openid-configuration"
XAI_OAUTH_CLIENT_ID = "b1a00492-073a-47ea-816f-4c329264a828"
Expand Down Expand Up @@ -869,15 +870,19 @@ def _oauth_trace(event: str, *, sequence_id: Optional[str] = None, **fields: Any
# Auth Store — persistence layer for ~/.hermes/auth.json
# =============================================================================

def _auth_file_path() -> Path:
path = get_hermes_home() / "auth.json"
def _assert_test_safe_auth_path(path: Path) -> Path:
# Seat belt: if pytest is running and HERMES_HOME resolves to the real
# user's auth store, refuse rather than silently corrupt it. This catches
# tests that forgot to monkeypatch HERMES_HOME, tests invoked without the
# hermetic conftest, or sandbox escapes via threads/subprocesses. In
# production (no PYTEST_CURRENT_TEST) this is a single dict lookup.
if os.environ.get("PYTEST_CURRENT_TEST"):
real_home_auth = (Path.home() / ".hermes" / "auth.json").resolve(strict=False)
real_home_env = os.environ.get("HOME", "")
real_home_auth = (
Path(real_home_env) / ".hermes" / "auth.json"
if real_home_env
else Path.home() / ".hermes" / "auth.json"
).resolve(strict=False)
try:
resolved = path.resolve(strict=False)
except Exception:
Expand All @@ -891,6 +896,10 @@ def _auth_file_path() -> Path:
return path


def _auth_file_path() -> Path:
return _assert_test_safe_auth_path(get_hermes_home() / "auth.json")


def _global_auth_file_path() -> Optional[Path]:
"""Return the global-root auth.json when the process is in profile mode.

Expand Down Expand Up @@ -963,6 +972,7 @@ def _auth_lock_path() -> Path:


_auth_lock_holder = threading.local()
_global_auth_lock_holder = threading.local()


@contextmanager
Expand Down Expand Up @@ -1056,6 +1066,24 @@ def _auth_store_lock(timeout_seconds: float = AUTH_LOCK_TIMEOUT_SECONDS):
yield


@contextmanager
def _global_auth_store_lock(timeout_seconds: float = AUTH_LOCK_TIMEOUT_SECONDS):
"""Cross-process lock for global-root auth.json writes in profile mode."""
global_auth_file = _global_auth_file_path()
if global_auth_file is None:
with _auth_store_lock(timeout_seconds=timeout_seconds):
yield
return
global_auth_file = _assert_test_safe_auth_path(global_auth_file)
with _file_lock(
global_auth_file.with_suffix(".lock"),
_global_auth_lock_holder,
timeout_seconds,
"Timed out waiting for global auth store lock",
):
yield


def _load_auth_store(auth_file: Optional[Path] = None) -> Dict[str, Any]:
auth_file = auth_file or _auth_file_path()
if not auth_file.exists():
Expand Down Expand Up @@ -1096,8 +1124,8 @@ 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], auth_file: Optional[Path] = None) -> Path:
auth_file = _assert_test_safe_auth_path(auth_file or _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.
Expand Down Expand Up @@ -1214,18 +1242,14 @@ def get_auth_provider_display_name(provider_id: str) -> str:
def read_credential_pool(provider_id: Optional[str] = None) -> Dict[str, Any]:
"""Return the persisted credential pool, or one provider slice.

In profile mode, the profile's credential pool is authoritative. If a
provider has no entries in the profile, entries from the global-root
``auth.json`` are used as a read-only fallback — so workers spawned in a
profile can see providers that were only authenticated at global scope.
In profile mode, most providers keep profile-local pools and use the
global-root ``auth.json`` as a read-only fallback only when the profile has
no entries. Providers in ``SHARED_CREDENTIAL_POOL_PROVIDERS`` use the
global-root pool first because their OAuth refresh tokens are single-use
and cannot be safely duplicated across profile-local pools.

Profile entries always win: the global fallback only applies per-provider
when the profile has zero entries for that provider. Once the user runs
``hermes auth add <provider>`` inside the profile, profile entries
fully shadow global for that provider on the next read.

Writes always go to the profile (``write_credential_pool`` is unchanged).
See issue #18594 follow-up.
Writes for shared providers go to the global-root auth store in profile
mode. Writes for all other providers remain profile-local.
"""
auth_store = _load_auth_store()
pool = auth_store.get("credential_pool")
Expand All @@ -1243,13 +1267,21 @@ def read_credential_pool(provider_id: Optional[str] = None) -> Dict[str, Any]:
for gp_key, gp_entries in global_pool.items():
if not isinstance(gp_entries, list) or not gp_entries:
continue
if gp_key in SHARED_CREDENTIAL_POOL_PROVIDERS:
merged[gp_key] = list(gp_entries)
continue
# Per-provider shadowing: profile wins whenever it has ANY entries.
existing = merged.get(gp_key)
if isinstance(existing, list) and existing:
continue
merged[gp_key] = list(gp_entries)
return merged

if provider_id in SHARED_CREDENTIAL_POOL_PROVIDERS:
global_entries = global_pool.get(provider_id)
if isinstance(global_entries, list) and global_entries:
return list(global_entries)

provider_entries = pool.get(provider_id)
if isinstance(provider_entries, list) and provider_entries:
return list(provider_entries)
Expand All @@ -1265,8 +1297,14 @@ def write_credential_pool(provider_id: str, entries: List[Dict[str, Any]]) -> Pa
credentials. Callers may pass raw dictionaries, so sanitize here even when
``PooledCredential.to_dict()`` already did the same work upstream.
"""
with _auth_store_lock():
auth_store = _load_auth_store()
shared_auth_file = (
_global_auth_file_path()
if provider_id in SHARED_CREDENTIAL_POOL_PROVIDERS
else None
)
lock = _global_auth_store_lock if shared_auth_file is not None else _auth_store_lock
with lock():
auth_store = _load_auth_store(shared_auth_file)
pool = auth_store.get("credential_pool")
if not isinstance(pool, dict):
pool = {}
Expand All @@ -1276,7 +1314,7 @@ def write_credential_pool(provider_id: str, entries: List[Dict[str, Any]]) -> Pa
if isinstance(entry, dict) else entry
for entry in entries
]
return _save_auth_store(auth_store)
return _save_auth_store(auth_store, auth_file=shared_auth_file)


def suppress_credential_source(provider_id: str, source: str) -> None:
Expand Down Expand Up @@ -3378,10 +3416,78 @@ def _sync_codex_pool_entries(
entry["last_error_reset_at"] = None


def _clone_pool_entries(entries: Any) -> List[Any]:
if not isinstance(entries, list):
return []
return [dict(entry) if isinstance(entry, dict) else entry for entry in entries]


def _append_codex_device_code_pool_entry(
entries: List[Any],
tokens: Dict[str, str],
last_refresh: Optional[str],
) -> None:
access_token = tokens.get("access_token")
if not access_token:
return
entry = {
"id": uuid.uuid4().hex[:6],
"source": "device_code",
"auth_type": "oauth",
"priority": 0,
"label": "device_code",
"access_token": access_token,
"base_url": DEFAULT_CODEX_BASE_URL,
"last_status": None,
"last_status_at": None,
"last_error_code": None,
"last_error_reason": None,
"last_error_message": None,
"last_error_reset_at": None,
}
refresh_token = tokens.get("refresh_token")
if refresh_token:
entry["refresh_token"] = refresh_token
if last_refresh:
entry["last_refresh"] = last_refresh
entries.append(entry)


def _sync_shared_codex_pool_entries(
tokens: Dict[str, str],
last_refresh: Optional[str],
profile_entries: List[Any],
) -> None:
"""Mirror Codex device-code entries into the global-root pool in profile mode."""
global_auth_file = _global_auth_file_path()
if global_auth_file is None:
return
with _global_auth_store_lock():
auth_store = _load_auth_store(global_auth_file)
pool = auth_store.get("credential_pool")
if not isinstance(pool, dict):
pool = {}
auth_store["credential_pool"] = pool
if not isinstance(pool.get("openai-codex"), list) and profile_entries:
pool["openai-codex"] = _clone_pool_entries(profile_entries)
_sync_codex_pool_entries(auth_store, tokens, last_refresh)
entries = pool.get("openai-codex")
if not isinstance(entries, list):
entries = []
pool["openai-codex"] = entries
if (
not any(isinstance(entry, dict) and entry.get("source") == "device_code" for entry in entries)
and not is_source_suppressed("openai-codex", "device_code")
):
_append_codex_device_code_pool_entry(entries, tokens, last_refresh)
_save_auth_store(auth_store, auth_file=global_auth_file)


def _save_codex_tokens(tokens: Dict[str, str], last_refresh: str = None) -> None:
"""Save Codex OAuth tokens to Hermes auth store (~/.hermes/auth.json)."""
if last_refresh is None:
last_refresh = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
profile_entries: List[Any] = []
with _auth_store_lock():
auth_store = _load_auth_store()
state = _load_provider_state(auth_store, "openai-codex") or {}
Expand All @@ -3390,7 +3496,11 @@ def _save_codex_tokens(tokens: Dict[str, str], last_refresh: str = None) -> None
state["auth_mode"] = "chatgpt"
_save_provider_state(auth_store, "openai-codex", state)
_sync_codex_pool_entries(auth_store, tokens, last_refresh)
profile_entries = _clone_pool_entries(
(auth_store.get("credential_pool") or {}).get("openai-codex")
)
_save_auth_store(auth_store)
_sync_shared_codex_pool_entries(tokens, last_refresh, profile_entries)


def refresh_codex_oauth_pure(
Expand Down Expand Up @@ -3653,21 +3763,15 @@ def _pool_codex_access_token() -> str:
"""Return the most-recent usable access_token from the openai-codex pool.

Used as a fallback by ``resolve_codex_runtime_credentials`` when the
singleton has no creds. Reads ``credential_pool.openai-codex`` entries
directly from auth.json and picks the first non-empty access_token,
singleton has no creds. Reads ``credential_pool.openai-codex`` through
``read_credential_pool`` so profile-mode workers use the shared global
Codex pool, then picks the first non-empty access_token,
preferring entries that are not currently in an exhaustion cooldown.
Returns ``""`` when no usable entry is found (caller handles by raising
the original AuthError).
"""
try:
with _auth_store_lock():
auth_store = _load_auth_store()
pool = auth_store.get("credential_pool")
if not isinstance(pool, dict):
return ""
entries = pool.get("openai-codex")
if not isinstance(entries, list):
return ""
entries = read_credential_pool("openai-codex")

def _entry_usable(entry: Dict[str, Any]) -> bool:
if not isinstance(entry, dict):
Expand Down
Loading