From 83f2fb0984053ceb2078f3fd488c72140dbd86f5 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:56:06 -0700 Subject: [PATCH] refactor(agent/account_usage,anthropic_credentials,azure_identity_adapter,acp_openai_bridge): outcome table, env lookup helper, compact literals and docs --- agent/account_usage.py | 229 ++++++++++++------------------- agent/acp_openai_bridge.py | 33 ++--- agent/anthropic_credentials.py | 231 ++++++++++++-------------------- agent/azure_identity_adapter.py | 101 +++++--------- 4 files changed, 216 insertions(+), 378 deletions(-) diff --git a/agent/account_usage.py b/agent/account_usage.py index cec3adadf9..14f86b5705 100644 --- a/agent/account_usage.py +++ b/agent/account_usage.py @@ -49,9 +49,7 @@ class AccountUsageSnapshot: def _snapshot(provider: str, source: str, windows: list, details: list, **kw: Any) -> AccountUsageSnapshot: - return AccountUsageSnapshot( - provider=provider, source=source, fetched_at=_utc_now(), windows=tuple(windows), details=tuple(details), **kw - ) + return AccountUsageSnapshot(provider=provider, source=source, fetched_at=_utc_now(), windows=tuple(windows), details=tuple(details), **kw) def _title_case_slug(value: Optional[str]) -> Optional[str]: @@ -87,7 +85,7 @@ def _format_reset(dt: Optional[datetime]) -> str: if hours >= 24: days, hours = divmod(hours, 24) return f"in {days}d {hours}h ({stamp})" - return f"in {hours}h {minutes}m ({stamp})" if hours > 0 else f"in {minutes}m ({stamp})" + return f"in {hours}h {minutes}m ({stamp})" if hours else f"in {minutes}m ({stamp})" def render_account_usage_lines(snapshot: Optional[AccountUsageSnapshot], *, markdown: bool = False) -> list[str]: @@ -100,9 +98,8 @@ def render_account_usage_lines(snapshot: Optional[AccountUsageSnapshot], *, mark if window.used_percent is None: base = f"{window.label}: unavailable" else: - remaining = max(0, round(100 - float(window.used_percent))) - used = max(0, round(float(window.used_percent))) - base = f"{window.label}: {remaining}% remaining ({used}% used)" + used = float(window.used_percent) + base = f"{window.label}: {max(0, round(100 - used))}% remaining ({max(0, round(used))}% used)" if window.reset_at: base += f" • resets {_format_reset(window.reset_at)}" elif window.detail: @@ -127,26 +124,27 @@ def _is_finite_num(v: Any) -> TypeGuard[float]: return _is_num(v) and not isinstance(v, bool) and math.isfinite(v) -def _nous_snapshot(windows: list, details: list, *, source: str, plan: Optional[str] = None) -> AccountUsageSnapshot: - return _snapshot("nous", source, windows, details, title="Nous credits", plan=plan) +def _nous_snapshot(windows: list, details: list, tail: list, *, source: str, plan: Optional[str] = None) -> Optional[AccountUsageSnapshot]: + """Nous snapshot with *tail* lines appended, or None when there is nothing to show.""" + if not windows and not details: + return None + return _snapshot("nous", source, windows, details + tail, title="Nous credits", plan=plan) def build_nous_credits_snapshot(account_info) -> Optional[AccountUsageSnapshot]: - """Map a NousPortalAccountInfo into the /usage snapshot: dollar magnitudes + renewal date + portal - CTA, plus a ``% used`` gauge when the portal supplies ``monthly_credits``. Fail-open → None.""" + """NousPortalAccountInfo → /usage snapshot: dollar magnitudes + renewal date + portal CTA, plus a ``% used`` + gauge when the portal supplies ``monthly_credits``. Fail-open → None.""" try: from hermes_cli.nous_account import nous_portal_topup_url - if account_info is None or not getattr(account_info, "logged_in", False): return None access = getattr(account_info, "paid_service_access_info", None) sub = getattr(account_info, "subscription", None) windows: list[AccountUsageWindow] = [] details: list[str] = [] - # Gauge needs a positive cap AND a finite remaining <= cap (numeric account fields, NOT a - # server *_usd). used = cap - remaining, clamped [0,100] so debt reads 100%. NaN/Inf - # (json.loads accepts bare NaN → "$nan") and remaining > cap (rollover makes the cap a - # meaningless denominator) fall back to the magnitudes lines. + # Gauge needs a positive cap AND a finite remaining <= cap (numeric fields, NOT a server *_usd); used = + # cap - remaining clamped [0,100] so debt reads 100%. NaN/Inf (json.loads accepts bare NaN → "$nan") and + # remaining > cap (rollover makes the cap a meaningless denominator) fall back to the magnitudes lines. if sub is not None: cap = getattr(sub, "monthly_credits", None) sub_remaining = getattr(sub, "credits_remaining", None) @@ -156,10 +154,8 @@ def build_nous_credits_snapshot(account_info) -> Optional[AccountUsageSnapshot]: detail=f"{_fmt_usd(sub_remaining)} of {_fmt_usd(cap)} left", )) if access is not None: - for attr, label in ( - ("subscription_credits_remaining", "Subscription credits"), - ("purchased_credits_remaining", "Top-up credits"), ("total_usable_credits", "Total usable"), - ): + for attr, label in (("subscription_credits_remaining", "Subscription credits"), + ("purchased_credits_remaining", "Top-up credits"), ("total_usable_credits", "Total usable")): value = getattr(access, attr, None) if _is_finite_num(value): details.append(f"{label}: {_fmt_usd(value)}") @@ -172,11 +168,8 @@ def build_nous_credits_snapshot(account_info) -> Optional[AccountUsageSnapshot]: details.append(f"Renews: {period_end}") if getattr(account_info, "paid_service_access", None) is False: details.append(_DEPLETED_LINE) - if not windows and not details: - return None - details += [f"Top up: {nous_portal_topup_url(account_info)}", "(or run /topup)"] - plan = getattr(sub, "plan", None) if sub is not None else None - return _nous_snapshot(windows, details, source="portal-account", plan=plan) + return _nous_snapshot(windows, details, [f"Top up: {nous_portal_topup_url(account_info)}", "(or run /topup)"], + source="portal-account", plan=getattr(sub, "plan", None) if sub is not None else None) except (AttributeError, TypeError): return None @@ -185,7 +178,6 @@ def _nous_logged_in() -> bool: """Cheap local auth-state check: a Nous access token is present. Fail-open False.""" try: from hermes_cli.auth import get_provider_auth_state - tok = (get_provider_auth_state("nous") or {}).get("access_token") return isinstance(tok, str) and bool(tok.strip()) except Exception: @@ -196,7 +188,6 @@ def _fetch_portal_account(timeout: float): """Wall-clock-bounded fresh portal account fetch (raises on any failure/timeout).""" import concurrent.futures from hermes_cli.nous_account import get_nous_portal_account_info - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: return pool.submit(get_nous_portal_account_info, force_fresh=True).result(timeout=timeout) @@ -204,13 +195,12 @@ def _fetch_portal_account(timeout: float): def nous_credits_lines(*, markdown: bool = False, timeout: float = 10.0) -> list[str]: """Rendered Nous-credits /usage lines, or [] when there's nothing to show. - Independent of any live agent (logged-in gate, then a bounded portal fetch); shared by CLI - ``_show_usage`` and the TUI ``session.usage`` RPC. Fail-open: any hiccup or timeout → []. - Dev override: HERMES_DEV_CREDITS_FIXTURE renders from the fixture instead of the portal. + Independent of any live agent (logged-in gate, then a bounded portal fetch); shared by CLI ``_show_usage`` + and the TUI ``session.usage`` RPC. Fail-open: any hiccup or timeout → []. HERMES_DEV_CREDITS_FIXTURE + renders from the fixture instead of the portal. """ try: from agent.credits_tracker import dev_fixture_credits_state - fixture = dev_fixture_credits_state() except Exception: fixture = None @@ -228,8 +218,8 @@ def nous_credits_lines(*, markdown: bool = False, timeout: float = 10.0) -> list def _snapshot_from_credits_state(state) -> Optional[AccountUsageSnapshot]: - """Map a header-shaped CreditsState (dev fixture) to the /usage snapshot, same shape as the portal - path. *_usd strings are display-only; the % comes from CreditsState.used_fraction. Fail-open → None.""" + """Header-shaped CreditsState (dev fixture) → /usage snapshot, same shape as the portal path. *_usd strings + are display-only; the % comes from CreditsState.used_fraction. Fail-open → None.""" try: if state is None: return None @@ -243,26 +233,21 @@ def _snapshot_from_credits_state(state) -> Optional[AccountUsageSnapshot]: label="Subscription", used_percent=max(0.0, min(100.0, uf * 100.0)), detail=f"${sub_usd} of ${cap_usd} left" if sub_usd and cap_usd else None, )) - for value, label in ( - (sub_usd, "Subscription credits"), (getattr(state, "purchased_usd", None), "Top-up credits"), - (getattr(state, "remaining_usd", None), "Total usable"), - ): + for value, label in ((sub_usd, "Subscription credits"), (getattr(state, "purchased_usd", None), "Top-up credits"), + (getattr(state, "remaining_usd", None), "Total usable")): if value: details.append(f"{label}: ${value}") if getattr(state, "paid_access", True) is False: details.append(_DEPLETED_LINE) - if not windows and not details: - return None - details.append("(dev fixture — HERMES_DEV_CREDITS_FIXTURE)") - return _nous_snapshot(windows, details, source="dev-fixture") + return _nous_snapshot(windows, details, ["(dev fixture — HERMES_DEV_CREDITS_FIXTURE)"], source="dev-fixture") except (AttributeError, TypeError): return None @dataclass(frozen=True) class CreditsView: - """Surface-agnostic ``/topup`` balance view: one portal fetch, consumed identically by every money - surface. Fail-open: not logged in / portal unreachable → ``logged_in`` False, ``topup_url`` None.""" + """Surface-agnostic ``/topup`` balance view: one portal fetch, consumed identically by every money surface. + Fail-open: not logged in / portal unreachable → ``logged_in`` False, ``topup_url`` None.""" logged_in: bool balance_lines: tuple[str, ...] = () @@ -272,9 +257,9 @@ class CreditsView: def build_credits_view(*, markdown: bool = False, timeout: float = 10.0) -> CreditsView: - """Build the /topup view: balance block + identity line + top-up URL. Reuses the /usage fetch + - snapshot so numbers match; the balance block drops the trailing top-up/hint lines (/topup has its - own affordance). Fail-open → ``CreditsView(logged_in=False)``.""" + """/topup view: balance block + identity line + top-up URL. Reuses the /usage fetch + snapshot so numbers + match; the balance block drops the trailing top-up/hint lines (/topup has its own affordance). + Fail-open → ``CreditsView(logged_in=False)``.""" not_logged_in = CreditsView(logged_in=False) if not _nous_logged_in(): return not_logged_in @@ -286,7 +271,6 @@ def build_credits_view(*, markdown: bool = False, timeout: float = 10.0) -> Cred if account is None or not getattr(account, "logged_in", False): return not_logged_in from hermes_cli.nous_account import nous_portal_topup_url - balance_lines = [ line for line in render_account_usage_lines(build_nous_credits_snapshot(account), markdown=markdown) @@ -315,16 +299,14 @@ def _codex_backend_urls(base_url: str) -> tuple[str, str, str]: def _resolve_codex_usage_credentials( base_url: Optional[str], api_key: Optional[str], ) -> tuple[str, str, Optional[str]]: - """Codex quota credentials: explicit live-agent creds → native runtime resolver (itself pool-aware) - → direct pool select. Native OAuth stores device-code logins in the pool, so this must not depend - only on the singleton store.""" + """Codex quota credentials: explicit live-agent creds → native runtime resolver (itself pool-aware) → direct + pool select. Native OAuth stores device-code logins in the pool, so the singleton store alone is not enough.""" explicit_key = str(api_key or "").strip() if explicit_key: return explicit_key, str(base_url or "").strip(), None - # Only AuthError is caught so tier 3 can run: a broad except would mask a transient - # refresh/network failure and hand back a DIFFERENT pool account's usage; such errors must - # propagate to the fail-open outer guard. account_id is best-effort: a partial singleton store - # must not sink a usable credential. + # Only AuthError is caught so tier 3 can run: a broad except would mask a transient refresh/network failure + # and hand back a DIFFERENT pool account's usage; such errors must propagate to the fail-open outer guard. + # account_id is best-effort: a partial singleton store must not sink a usable credential. try: creds = resolve_codex_runtime_credentials(refresh_if_expiring=True) account_id: Optional[str] = None @@ -339,7 +321,6 @@ def _resolve_codex_usage_credentials( logger.debug("codex ▸ /usage runtime resolver returned no creds; trying pool", exc_info=True) # Tier 3: pool credentials have no account_id concept → header omitted. from agent.credential_pool import load_pool - entry = load_pool("openai-codex").select() if entry is None: raise RuntimeError("No available openai-codex credential in credential pool") @@ -391,21 +372,17 @@ def _fetch_codex_account_usage( ) -> Optional[AccountUsageSnapshot]: token, resolved_base_url, account_id = _resolve_codex_usage_credentials(base_url, api_key) payload = _get_json(_codex_backend_urls(resolved_base_url)[0], _codex_headers(token, account_id), timeout=15.0) - windows = _usage_windows( - payload.get("rate_limit") or {}, (("primary_window", "Session"), ("secondary_window", "Weekly")), - "used_percent", "reset_at", - ) + windows = _usage_windows(payload.get("rate_limit") or {}, (("primary_window", "Session"), ("secondary_window", "Weekly")), + "used_percent", "reset_at") details: list[str] = [] count = _codex_banked_resets(payload) if count > 0: details.append(f"You have {count} reset{_plural(count)} banked - use /usage reset to activate") - credits = payload.get("credits") or {} - balance = credits.get("balance") - if credits.get("has_credits"): - if _is_num(balance): - details.append(f"Credits balance: ${float(balance):.2f}") - elif credits.get("unlimited"): - details.append("Credits balance: unlimited") + credits, balance = payload.get("credits") or {}, (payload.get("credits") or {}).get("balance") + if credits.get("has_credits") and _is_num(balance): + details.append(f"Credits balance: ${float(balance):.2f}") + elif credits.get("has_credits") and credits.get("unlimited"): + details.append("Credits balance: unlimited") return _snapshot("openai-codex", "usage_api", windows, details, plan=_title_case_slug(payload.get("plan_type"))) @@ -413,7 +390,7 @@ def _fetch_codex_account_usage( class CodexResetRedeemResult: """Outcome of a `/usage reset` attempt against the Codex backend.""" - status: str # reset | nothing_to_reset | no_credit | already_redeemed | not_exhausted | no_credits_banked | unavailable + status: str # reset|nothing_to_reset|no_credit|already_redeemed|not_exhausted|no_credits_banked|unavailable message: str available_count: int = 0 windows_reset: int = 0 @@ -423,8 +400,8 @@ class CodexResetRedeemResult: return self.status == "reset" -# Client-side guard: a window only counts as exhausted when fully used. Below this, redeeming a -# banked reset wastes most of its value → block, point at --force. +# Client-side guard: a window only counts as exhausted when fully used; below this, redeeming a banked reset +# wastes most of its value → block, point at --force. _CODEX_WINDOW_EXHAUSTED_PERCENT = 100.0 @@ -435,31 +412,20 @@ def _unavailable(message: str) -> CodexResetRedeemResult: def _codex_reset_guard(payload: dict, available: int, force: bool) -> Optional[CodexResetRedeemResult]: """Refuse a redemption that would be wasted (no banked credits, or no window fully used and not ``force``).""" if available <= 0: - return CodexResetRedeemResult( - status="no_credits_banked", message="No banked reset credits on this account — nothing to redeem." - ) + return CodexResetRedeemResult(status="no_credits_banked", message="No banked reset credits on this account — nothing to redeem.") rate_limit = payload.get("rate_limit") or {} - used_pcts = [ - float(u) for u in ((rate_limit.get(k) or {}).get("used_percent") for k in ("primary_window", "secondary_window")) - if _is_num(u) - ] + used_pcts = [float(u) for u in ((rate_limit.get(k) or {}).get("used_percent") for k in ("primary_window", "secondary_window")) + if _is_num(u)] worst_used: Optional[float] = max(0.0, *used_pcts) if used_pcts else None if force or (worst_used is not None and worst_used >= _CODEX_WINDOW_EXHAUSTED_PERCENT): return None - usage_note = ( - f"your busiest window is only {worst_used:.0f}% used" - if worst_used is not None - else "your current usage could not be confirmed as exhausted" - ) + usage_note = (f"your busiest window is only {worst_used:.0f}% used" if worst_used is not None + else "your current usage could not be confirmed as exhausted") return CodexResetRedeemResult( - status="not_exhausted", - message=( - f"⚠️ Not redeeming: {usage_note}. A banked reset restores your FULL " - f"5h + weekly limits, so spending it now would waste most of it. " - f"You have {available} reset{_plural(available)} banked. " - f"Use `/usage reset --force` to redeem anyway." - ), - available_count=available, + status="not_exhausted", available_count=available, + message=(f"⚠️ Not redeeming: {usage_note}. A banked reset restores your FULL 5h + weekly limits, so spending it " + f"now would waste most of it. You have {available} reset{_plural(available)} banked. " + f"Use `/usage reset --force` to redeem anyway."), ) @@ -467,47 +433,38 @@ def _codex_reset_outcome(body: dict, available: int) -> CodexResetRedeemResult: """Map the consume response ``code`` to a result (``reset`` also lifts persisted pool cooldowns).""" code = str(body.get("code", "") or "").strip().lower() remaining = max(0, available - 1) - if code == "reset": - # Quota is restored upstream — lift persisted pool cooldowns so the credential isn't - # frozen behind a stale ``last_error_reset_at``. - try: - from hermes_cli.auth import clear_codex_pool_quota_cooldowns - - clear_codex_pool_quota_cooldowns() - except Exception: - logger.debug("Failed to clear Codex pool cooldowns after reset redemption", exc_info=True) - windows_reset = body.get("windows_reset") - return CodexResetRedeemResult( - status="reset", - message=( - f"✅ Reset redeemed — your usage limits have been reset. " - f"{remaining} banked reset{_plural(remaining)} remaining." - ), - available_count=remaining, - windows_reset=int(windows_reset) if _is_num(windows_reset) else 0, - ) outcomes: dict[str, tuple[str, int]] = { - "nothing_to_reset": ( - "Backend reports nothing to reset — your limits aren't exhausted. The credit was NOT spent.", available, - ), + "reset": (f"✅ Reset redeemed — your usage limits have been reset. {remaining} banked reset{_plural(remaining)} remaining.", + remaining), + "nothing_to_reset": ("Backend reports nothing to reset — your limits aren't exhausted. The credit was NOT spent.", available), "no_credit": ("Backend reports no available reset credit on this account.", 0), "already_redeemed": ("This redemption was already processed — no additional credit was spent.", remaining), } - if code in outcomes: - message, count = outcomes[code] - return CodexResetRedeemResult(status=code, message=message, available_count=count) - return _unavailable(f"Unexpected response from the Codex backend: {body!r}") + if code not in outcomes: + return _unavailable(f"Unexpected response from the Codex backend: {body!r}") + windows_reset = 0 + if code == "reset": + # Quota is restored upstream — lift persisted pool cooldowns so the credential isn't frozen behind a + # stale ``last_error_reset_at``. + try: + from hermes_cli.auth import clear_codex_pool_quota_cooldowns + clear_codex_pool_quota_cooldowns() + except Exception: + logger.debug("Failed to clear Codex pool cooldowns after reset redemption", exc_info=True) + raw = body.get("windows_reset") + windows_reset = int(raw) if _is_num(raw) else 0 + message, count = outcomes[code] + return CodexResetRedeemResult(status=code, message=message, available_count=count, windows_reset=windows_reset) def redeem_codex_reset_credit( *, base_url: Optional[str] = None, api_key: Optional[str] = None, force: bool = False, ) -> CodexResetRedeemResult: - """Redeem one banked Codex rate-limit reset credit (`/usage reset`), mirroring the Codex CLI picker: - GET usage → guard (a reset restores the WHOLE 5h + weekly allowance, and the backend's own - ``nothing_to_reset`` guard is less clear) → POST consume with a fresh UUID ``redeem_request_id`` and - no ``credit_id`` (the backend picks the next credit). Never raises: every failure returns a result.""" + """Redeem one banked Codex rate-limit reset credit (`/usage reset`), mirroring the Codex CLI picker: GET usage → + guard (a reset restores the WHOLE 5h + weekly allowance, and the backend's own ``nothing_to_reset`` guard is + less clear) → POST consume with a fresh UUID ``redeem_request_id`` and no ``credit_id`` (the backend picks the + next credit). Never raises: every failure returns a result.""" import uuid - try: token, resolved_base_url, account_id = _resolve_codex_usage_credentials(base_url, api_key) except Exception: @@ -532,11 +489,8 @@ def redeem_codex_reset_credit( except httpx.HTTPStatusError as exc: code = exc.response.status_code if code in (401, 403): - return _unavailable( - "Codex backend rejected the request (HTTP " - f"{code}). Reset credits require ChatGPT-account (OAuth) auth — " - "run `hermes auth` and sign in with your ChatGPT account." - ) + return _unavailable(f"Codex backend rejected the request (HTTP {code}). Reset credits require ChatGPT-account " + "(OAuth) auth — run `hermes auth` and sign in with your ChatGPT account.") return _unavailable(f"Codex backend error (HTTP {code}) — try again shortly.") except Exception as exc: return _unavailable(f"Could not reach the Codex backend: {exc}") @@ -550,20 +504,14 @@ def _fetch_anthropic_account_usage( if not token: return None if not _is_oauth_token(token): - return _snapshot( - "anthropic", "oauth_usage_api", [], [], - unavailable_reason="Anthropic account limits are only available for OAuth-backed Claude accounts.", - ) - headers = { - "Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json", - "anthropic-beta": "oauth-2025-04-20", "User-Agent": "claude-code/2.1.0", - } + return _snapshot("anthropic", "oauth_usage_api", [], [], + unavailable_reason="Anthropic account limits are only available for OAuth-backed Claude accounts.") + headers = {"Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json", + "anthropic-beta": "oauth-2025-04-20", "User-Agent": "claude-code/2.1.0"} payload = _get_json("https://api.anthropic.com/api/oauth/usage", headers, timeout=15.0) windows = _usage_windows( - payload, - (("five_hour", "Current session"), ("seven_day", "Current week"), ("seven_day_opus", "Opus week"), - ("seven_day_sonnet", "Sonnet week")), - "utilization", "resets_at", fraction=True, + payload, (("five_hour", "Current session"), ("seven_day", "Current week"), ("seven_day_opus", "Opus week"), + ("seven_day_sonnet", "Sonnet week")), "utilization", "resets_at", fraction=True, ) details: list[str] = [] extra = payload.get("extra_usage") or {} @@ -585,7 +533,6 @@ def _fetch_openrouter_account_usage(base_url: Optional[str], api_key: Optional[s resp = client.get(f"{normalized}/{path}", headers=headers) resp.raise_for_status() return (resp.json() or {}).get("data") or {} - credits = _data("credits") try: key_data = _data("key") @@ -603,10 +550,8 @@ def _fetch_openrouter_account_usage(base_url: Optional[str], api_key: Optional[s detail_parts = [f"${remaining_value:.2f} of ${limit_value:.2f} remaining"] if limit_reset: detail_parts.append(f"resets {limit_reset}") - windows.append(AccountUsageWindow( - label="API key quota", used_percent=((limit_value - remaining_value) / limit_value) * 100, - detail=" • ".join(detail_parts), - )) + windows.append(AccountUsageWindow(label="API key quota", used_percent=((limit_value - remaining_value) / limit_value) * 100, + detail=" • ".join(detail_parts))) if _is_num(usage): usage_parts = [f"API key usage: ${float(usage):.2f} total"] for key, label in (("usage_daily", "today"), ("usage_weekly", "this week"), ("usage_monthly", "this month")): diff --git a/agent/acp_openai_bridge.py b/agent/acp_openai_bridge.py index 157655c00a..16cf0c7501 100644 --- a/agent/acp_openai_bridge.py +++ b/agent/acp_openai_bridge.py @@ -19,8 +19,7 @@ from openai.types.chat.chat_completion_message_tool_call import ChatCompletionMe TOOL_CALL_BLOCK_RE = re.compile(r"\s*(\{.*?\})\s*", re.DOTALL) TOOL_CALL_JSON_RE = re.compile( - r"\{\s*\"id\"\s*:\s*\"[^\"]+\"\s*,\s*\"type\"\s*:\s*\"function\"\s*,\s*\"function\"\s*:\s*\{.*?\}\s*\}", - re.DOTALL, + r"\{\s*\"id\"\s*:\s*\"[^\"]+\"\s*,\s*\"type\"\s*:\s*\"function\"\s*,\s*\"function\"\s*:\s*\{.*?\}\s*\}", re.DOTALL ) TOOL_CALL_CONTRACT = ( @@ -42,32 +41,23 @@ class StreamChunks(list): def completion_to_stream_chunks(completion: SimpleNamespace) -> StreamChunks: - """Re-shape a one-shot ACP response as OpenAI stream chunks (data chunk + usage chunk). - - Response-level attributes other than choices/usage/model are copied onto the result. - """ + """Re-shape a one-shot ACP response as OpenAI stream chunks (data chunk + usage chunk); response-level + attributes other than choices/usage/model are copied onto the result.""" choice = completion.choices[0] message = choice.message tool_call_deltas = None if message.tool_calls: tool_call_deltas = [ SimpleNamespace( - index=index, - id=getattr(tool_call, "id", None), - type=getattr(tool_call, "type", "function"), - function=SimpleNamespace( - name=getattr(tool_call.function, "name", None), - arguments=getattr(tool_call.function, "arguments", None), - ), + index=index, id=getattr(tool_call, "id", None), type=getattr(tool_call, "type", "function"), + function=SimpleNamespace(name=getattr(tool_call.function, "name", None), + arguments=getattr(tool_call.function, "arguments", None)), ) for index, tool_call in enumerate(message.tool_calls) ] delta = SimpleNamespace( - role="assistant", - content=message.content or None, - tool_calls=tool_call_deltas, - reasoning_content=getattr(message, "reasoning_content", None), - reasoning=getattr(message, "reasoning", None), + role="assistant", content=message.content or None, tool_calls=tool_call_deltas, + reasoning_content=getattr(message, "reasoning_content", None), reasoning=getattr(message, "reasoning", None), ) data_chunk = SimpleNamespace( choices=[SimpleNamespace(index=0, delta=delta, finish_reason=choice.finish_reason)], @@ -144,11 +134,8 @@ def _parse_tool_call(raw_json: str, ordinal: int) -> ChatCompletionMessageToolCa def extract_tool_calls_from_text(text: str) -> tuple[list[ChatCompletionMessageToolCall], str]: - """Pull ```` blocks out of an ACP response. - - Returns ``(tool_calls, cleaned_text)`` with the consumed blocks removed so the assistant message - doesn't show raw JSON. Bare-JSON fallback runs only when no XML block parsed. - """ + """Pull ```` blocks out of an ACP response → ``(tool_calls, cleaned_text)`` with the consumed blocks + removed so the assistant message doesn't show raw JSON. Bare-JSON fallback runs only when no XML block parsed.""" if not isinstance(text, str) or not text.strip(): return [], "" extracted: list[ChatCompletionMessageToolCall] = [] diff --git a/agent/anthropic_credentials.py b/agent/anthropic_credentials.py index 28d6a59677..4a2f91c870 100644 --- a/agent/anthropic_credentials.py +++ b/agent/anthropic_credentials.py @@ -32,15 +32,14 @@ from agent.secret_scope import get_secret as _get_secret logger = logging.getLogger(__name__) _OAUTH_CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e" -# platform.claude.com is the live token host; console.anthropic.com 404s but is kept as a -# fallback. _OAUTH_TOKEN_URL stays for backward-compatible imports. +# platform.claude.com is the live token host; console.anthropic.com 404s but is kept as a fallback. +# _OAUTH_TOKEN_URL stays for backward-compatible imports. _OAUTH_TOKEN_URLS = [ - "https://platform.claude.com/v1/oauth/token", - "https://console.anthropic.com/v1/oauth/token", + "https://platform.claude.com/v1/oauth/token", "https://console.anthropic.com/v1/oauth/token" ] _OAUTH_TOKEN_URL = _OAUTH_TOKEN_URLS[0] -# Anthropic 429s token-endpoint requests whose UA starts with ``claude-code/`` (or Mozilla); -# the real CLI uses bare axios there. Inference (build_anthropic_kwargs) still needs claude-code/. +# Anthropic 429s token-endpoint requests whose UA starts with ``claude-code/`` (or Mozilla); the real CLI uses +# bare axios there. Inference (build_anthropic_kwargs) still needs claude-code/. _OAUTH_TOKEN_USER_AGENT = "axios/1.7.9" _OAUTH_REDIRECT_URI = "https://console.anthropic.com/oauth/code/callback" _OAUTH_SCOPES = "org:create_api_key user:profile user:inference" @@ -52,6 +51,11 @@ def _getenv(name: str, default: str = "") -> str: return val if val is not None else default +def _first_env(*names: str) -> str: + """First non-blank (stripped) value among *names*, else ''.""" + return next((v for v in (_getenv(n).strip() for n in names) if v), "") + + def _is_oauth_token(key: str) -> bool: """True for Anthropic OAuth/setup tokens (sk-ant-*, eyJ JWTs, cc-); False for sk-ant-api* Console keys.""" if not key or key.startswith("sk-ant-api"): @@ -62,8 +66,8 @@ def _is_oauth_token(key: str) -> bool: class CredentialPersistError(RuntimeError): """A rotated single-use credential could not be durably committed. - The refresh POST already spent the old refresh token, so a swallowed write failure - leaves a consumed pair on disk that later replays as invalid_grant. + The refresh POST already spent the old refresh token, so a swallowed write failure leaves a consumed + pair on disk that later replays as invalid_grant. """ def __init__(self, path: Any, cause: BaseException) -> None: @@ -85,9 +89,9 @@ def _load_json_if_exists(path: Path, what: str) -> Optional[Any]: def _atomic_write_private_json(path: Path, payload: Any) -> None: """Write *payload* via a 0o600 O_EXCL temp file + fsync + os.replace. - The token is never briefly umask-readable (write_text + chmod had a TOCTOU window); the - random suffix avoids collisions with concurrent writers and crashed leftovers. The parent - dir's mode is left alone (~/.claude/ is owned by Claude Code). + The token is never briefly umask-readable (write_text + chmod had a TOCTOU window); the random suffix + avoids collisions with concurrent writers and crashed leftovers. The parent dir's mode is left alone + (~/.claude/ is owned by Claude Code). """ path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}") @@ -113,10 +117,9 @@ def _commit_private_json(path: Path, payload: Any, what: str) -> None: raise CredentialPersistError(path, e) from e -# ── Spent-rotation registry ── -# Fingerprints of secrets whose refresh POST succeeded but whose replacement never reached its -# store. Two scopes: process-local (OrderedDict) and a durable sidecar next to the shared -# singleton file so OTHER processes fail closed too. Non-reversible digests; never cleared. +# ── Spent-rotation registry: fingerprints of secrets whose refresh POST succeeded but whose replacement never +# reached its store. Two scopes: process-local (OrderedDict) and a durable sidecar next to the shared singleton +# file so OTHER processes fail closed too. Non-reversible digests; never cleared. _SPENT_ROTATION_LOCK = threading.Lock() _SPENT_ROTATION_FINGERPRINTS: "OrderedDict[str, None]" = OrderedDict() _SPENT_ROTATION_MAX_TRACKED = 64 @@ -146,16 +149,12 @@ def _read_spent_rotation_sidecar(source_path: Optional[Path]) -> set: except (OSError, ValueError): return set() fingerprints = raw.get("fingerprints") if isinstance(raw, dict) else None - if not isinstance(fingerprints, list): - return set() - return {fp for fp in fingerprints if isinstance(fp, str) and fp} + return {fp for fp in fingerprints if isinstance(fp, str) and fp} if isinstance(fingerprints, list) else set() def _append_spent_rotation_sidecar(source_path: Path, fingerprints: list) -> None: - """Merge fingerprints into the sidecar (atomic replace; caller holds the path lock). - - Fail-soft: a sidecar write failure must never mask the process-local verdict. - """ + """Merge fingerprints into the sidecar (atomic replace; caller holds the path lock). Fail-soft: a sidecar + write failure must never mask the process-local verdict.""" sidecar = _spent_rotation_sidecar_path(source_path) try: merged = _read_spent_rotation_sidecar(source_path) @@ -175,16 +174,13 @@ def _append_spent_rotation_sidecar(source_path: Path, fingerprints: list) -> Non def _fingerprint(secret: Any) -> Optional[str]: from agent.credential_persistence import fingerprint_secret_value - value = str(secret or "").strip() return fingerprint_secret_value(value) if value else None def mark_rotation_consumed_uncommitted(*secrets: Any, source_path: Optional[Path] = None) -> None: - """Record the pre-rotation pair of a refresh whose replacement never committed. - - With ``source_path`` the verdict is also persisted to that singleton's sidecar. - """ + """Record the pre-rotation pair of a refresh whose replacement never committed; with ``source_path`` the + verdict is also persisted to that singleton's sidecar.""" recorded = [fp for fp in map(_fingerprint, secrets) if fp] with _SPENT_ROTATION_LOCK: for fingerprint in recorded: @@ -210,24 +206,19 @@ def is_rotation_consumed_uncommitted(secret: Any, *, source_path: Optional[Path] # ── Claude Code credentials (Keychain / ~/.claude/.credentials.json) ── # Only singleton-backed pool sources have a cross-process authority boundary. _SINGLETON_SOURCE_PATHS = { - "claude_code": lambda: claude_code_credentials_path(), - "hermes_pkce": lambda: _get_hermes_oauth_file(), + "claude_code": lambda: claude_code_credentials_path(), "hermes_pkce": lambda: _get_hermes_oauth_file() } def _claude_oauth_record(data: Any, source: str) -> Optional[Dict[str, Any]]: """Normalise a ``{"claudeAiOauth": {...}}`` payload into our credential dict.""" oauth_data = data.get("claudeAiOauth") - if not (oauth_data and isinstance(oauth_data, dict)): - return None - access_token = oauth_data.get("accessToken", "") + access_token = oauth_data.get("accessToken", "") if isinstance(oauth_data, dict) else "" if not access_token: return None return { - "accessToken": access_token, - "refreshToken": oauth_data.get("refreshToken", ""), - "expiresAt": oauth_data.get("expiresAt", 0), - "source": source, + "accessToken": access_token, "refreshToken": oauth_data.get("refreshToken", ""), + "expiresAt": oauth_data.get("expiresAt", 0), "source": source, } @@ -238,8 +229,7 @@ def _read_claude_code_credentials_from_keychain() -> Optional[Dict[str, Any]]: try: result = subprocess.run( ["security", "find-generic-password", "-s", "Claude Code-credentials", "-w"], - capture_output=True, text=True, encoding='utf-8', errors='replace', - timeout=5, stdin=subprocess.DEVNULL, + capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5, stdin=subprocess.DEVNULL, ) except (OSError, subprocess.TimeoutExpired): logger.debug("Keychain: security command not available or timed out") @@ -271,9 +261,9 @@ def _read_claude_code_credentials_from_file() -> Optional[Dict[str, Any]]: def read_claude_code_credentials() -> Optional[Dict[str, Any]]: """Read refreshable Claude Code OAuth credentials (Keychain and/or file). - When both exist: prefer the only non-expired one (Claude Code 2.1.x refreshes one source - but not the other), else the later ``expiresAt`` so a refresh uses the freshest - refreshToken. ~/.claude.json primaryApiKey is deliberately excluded. + When both exist: prefer the only non-expired one (Claude Code 2.1.x refreshes one source but not the + other), else the later ``expiresAt`` so a refresh uses the freshest refreshToken. ~/.claude.json + primaryApiKey is deliberately excluded. """ kc_creds = _read_claude_code_credentials_from_keychain() file_creds = _read_claude_code_credentials_from_file() @@ -301,12 +291,10 @@ def _post_oauth_token( ) -> Dict[str, Any]: """POST to the token endpoints in order; raise the last error if all fail.""" import urllib.request - last_error = None for endpoint in _OAUTH_TOKEN_URLS: req = urllib.request.Request( - endpoint, data=data, method="POST", - headers={"Content-Type": content_type, "User-Agent": user_agent}, + endpoint, data=data, method="POST", headers={"Content-Type": content_type, "User-Agent": user_agent} ) try: with urllib.request.urlopen(req, timeout=timeout) as resp: @@ -329,17 +317,13 @@ def _oauth_token_state(result: Dict[str, Any], *, fallback_refresh_token: str = def refresh_anthropic_oauth_pure(refresh_token: str, *, use_json: bool = False) -> Dict[str, Any]: """Refresh an Anthropic OAuth token without mutating local credential files.""" import urllib.parse - if not refresh_token: raise ValueError("refresh_token is required") payload = {"grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": _OAUTH_CLIENT_ID} - if use_json: - data, content_type = json.dumps(payload).encode(), "application/json" - else: - data, content_type = urllib.parse.urlencode(payload).encode(), "application/x-www-form-urlencoded" - result = _post_oauth_token( - data, content_type=content_type, timeout=10, what="refresh", user_agent=_OAUTH_TOKEN_USER_AGENT - ) + encode, content_type = ((json.dumps, "application/json") if use_json + else (urllib.parse.urlencode, "application/x-www-form-urlencoded")) + result = _post_oauth_token(encode(payload).encode(), content_type=content_type, timeout=10, what="refresh", + user_agent=_OAUTH_TOKEN_USER_AGENT) if not result.get("access_token"): raise ValueError("Anthropic refresh response was missing access_token") return _oauth_token_state(result, fallback_refresh_token=refresh_token) @@ -348,28 +332,22 @@ def refresh_anthropic_oauth_pure(refresh_token: str, *, use_json: bool = False) def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: """Refresh an expired Claude Code OAuth token, returning the new access token. - Refresh tokens are single-use and Claude Code refreshes on its own schedule, so we first - re-read the live sources and adopt an already-rotated token instead of racing it into - ``invalid_grant``. Read, decision, POST and write-back all run under the same path-keyed - cross-process lock the pool uses; without it two profiles can still spend one refresh token. + Refresh tokens are single-use and Claude Code refreshes on its own schedule, so we first re-read the + live sources and adopt an already-rotated token instead of racing it into ``invalid_grant``. Read, + decision, POST and write-back share the pool's path-keyed cross-process lock (else two profiles can + spend one refresh token). """ try: from hermes_cli.auth import AUTH_LOCK_TIMEOUT_SECONDS, _auth_store_lock, env_float - refresh_timeout_seconds = env_float("HERMES_ANTHROPIC_REFRESH_TIMEOUT_SECONDS", 20) lock_timeout_seconds = max(float(AUTH_LOCK_TIMEOUT_SECONDS), float(refresh_timeout_seconds) + 5.0) cred_path = claude_code_credentials_path() with _auth_store_lock(timeout_seconds=lock_timeout_seconds, target_path=cred_path): - # Adopt only a DIFFERENT token with a real future expiry (0/absent expiresAt means - # managed key / unknown expiry). + # Adopt only a DIFFERENT token with a real future expiry (0/absent expiresAt = managed key/unknown). current = read_claude_code_credentials() or {} current_token = current.get("accessToken", "") - if ( - current_token - and current_token != creds.get("accessToken", "") - and (current.get("expiresAt", 0) or 0) > 0 - and is_claude_code_token_valid(current) - ): + if (current_token and current_token != creds.get("accessToken", "") + and (current.get("expiresAt", 0) or 0) > 0 and is_claude_code_token_valid(current)): logger.debug("Adopted Claude Code's already-refreshed OAuth token") return current_token @@ -377,25 +355,20 @@ def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: if not refresh_token: logger.debug("No refresh token available — cannot refresh") return None - # Another process may have spent this token and lost the commit; its sidecar - # verdict is authoritative for the shared source. + # Another process may have spent this token and lost the commit; its sidecar verdict is authoritative. if is_rotation_consumed_uncommitted(refresh_token, source_path=cred_path): - logger.debug( - "Refresh token was already consumed by an uncommitted rotation " - "- refusing to replay it; re-run 'claude setup-token'" - ) + logger.debug("Refresh token was already consumed by an uncommitted rotation " + "- refusing to replay it; re-run 'claude setup-token'") return None try: refreshed = refresh_anthropic_oauth_pure(refresh_token, use_json=False) except Exception as e: logger.debug("Failed to refresh Claude Code token: %s", e) return None - # The POST spent ``refresh_token``; this write is the commit step. On failure, - # fail closed and mark the pre-rotation pair as spent. + # The POST spent ``refresh_token``; this write is the commit step. On failure, fail closed and + # mark the pre-rotation pair as spent. try: - _write_claude_code_credentials( - refreshed["access_token"], refreshed["refresh_token"], refreshed["expires_at_ms"] - ) + _write_claude_code_credentials(refreshed["access_token"], refreshed["refresh_token"], refreshed["expires_at_ms"]) except Exception as e: logger.error( "Anthropic OAuth refresh rotated the single-use token but could not " @@ -419,11 +392,10 @@ def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]: def _write_claude_code_credentials( access_token: str, refresh_token: str, expires_at_ms: int, *, scopes: Optional[list] = None ) -> None: - """Commit refreshed credentials to ~/.claude/.credentials.json. + """Commit refreshed credentials to ~/.claude/.credentials.json; ``CredentialPersistError`` on any failure. - *scopes* (or the previously stored scopes) are persisted because Claude Code >=2.1.81 gates - on ``"user:inference"`` being present. Raises ``CredentialPersistError`` on any failure, - including a corrupt existing file. + *scopes* (or the previously stored scopes) are persisted because Claude Code >=2.1.81 gates on + ``"user:inference"`` being present. A corrupt existing file is a persist failure too. """ cred_path = claude_code_credentials_path() try: @@ -431,9 +403,7 @@ def _write_claude_code_credentials( except (OSError, ValueError) as e: logger.error("Failed to write refreshed credentials to %s: %s", cred_path, e) raise CredentialPersistError(cred_path, e) from e - oauth_data: Dict[str, Any] = { - "accessToken": access_token, "refreshToken": refresh_token, "expiresAt": expires_at_ms, - } + oauth_data: Dict[str, Any] = {"accessToken": access_token, "refreshToken": refresh_token, "expiresAt": expires_at_ms} if scopes is not None: oauth_data["scopes"] = scopes elif "claudeAiOauth" in existing and "scopes" in existing["claudeAiOauth"]: @@ -468,8 +438,8 @@ def _resolve_claude_code_token_from_credentials(creds: Optional[Dict[str, Any]] def _prefer_refreshable_claude_code_token(env_token: str, creds: Optional[Dict[str, Any]]) -> Optional[str]: """Prefer refreshable Claude Code creds over a static env OAuth token. - Hermes historically persisted setup tokens into ANTHROPIC_TOKEN; that static token would - otherwise win before the refreshable credential file is inspected, making refresh impossible. + Hermes historically persisted setup tokens into ANTHROPIC_TOKEN; that static token would otherwise + win before the refreshable credential file is inspected, making refresh impossible. """ if not (env_token and _is_oauth_token(env_token) and isinstance(creds, dict) and creds.get("refreshToken")): return None @@ -483,33 +453,29 @@ def _prefer_refreshable_claude_code_token(env_token: str, creds: Optional[Dict[s def _resolve_anthropic_pool_token() -> Optional[str]: """First available Anthropic OAuth token from credential_pool, read-only. - Enumerates with ``clear_expired=False, refresh=False`` (never ``select()``, which does both) - so diagnostic call sites (account_usage, ``hermes models``) never mutate auth.json or hit - the network; refresh-on-expiry belongs to the API call path's pool recovery. + Enumerates with ``clear_expired=False, refresh=False`` (never ``select()``) so diagnostic call sites + (account_usage, ``hermes models``) never mutate auth.json or hit the network; refresh-on-expiry + belongs to the API call path's pool recovery. """ try: from agent.credential_pool import AUTH_TYPE_OAUTH, load_pool - entries, _pending = load_pool("anthropic")._available_entries(clear_expired=False, refresh=False) except Exception: logger.debug("Failed to read Anthropic credential_pool", exc_info=True) return None for entry in entries: - # access_token may be an explicit null on a persisted entry; a bare None.strip() would - # escape the try/excepts and crash the resolver. + # access_token may be an explicit null on a persisted entry; None.strip() would crash the resolver. token = (getattr(entry, "access_token", None) or "").strip() if getattr(entry, "auth_type", None) != AUTH_TYPE_OAUTH or not token: continue - # load_pool() re-seeds rows from the singleton files, so a spent-but-uncommitted - # rotation (possibly from another process) looks healthy here. + # load_pool() re-seeds rows from the singleton files, so a spent-but-uncommitted rotation + # (possibly from another process) looks healthy here. entry_source_path = spent_rotation_source_path(getattr(entry, "source", None)) if any( is_rotation_consumed_uncommitted(secret, source_path=entry_source_path) for secret in (token, getattr(entry, "refresh_token", None)) ): - logger.debug( - "Skipping Anthropic pool entry %s: rotated-but-uncommitted credential", getattr(entry, "id", "?") - ) + logger.debug("Skipping Anthropic pool entry %s: rotated-but-uncommitted credential", getattr(entry, "id", "?")) continue return token return None @@ -518,31 +484,22 @@ def _resolve_anthropic_pool_token() -> Optional[str]: def resolve_anthropic_token() -> Optional[str]: """Resolve an Anthropic token from all sources in priority order (see module docstring).""" _read_creds = functools.cache(read_claude_code_credentials) # read the file at most once per resolve - for env_var in ("ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"): - token = _getenv(env_var).strip() - if token: - return _prefer_refreshable_claude_code_token(token, _read_creds()) or token - # An explicit API key must not be shadowed by auto-discovered OAuth creds. - api_key = _getenv("ANTHROPIC_API_KEY").strip() + token = _first_env("ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN") + if token: + return _prefer_refreshable_claude_code_token(token, _read_creds()) or token + api_key = _first_env("ANTHROPIC_API_KEY") # an explicit API key must not be shadowed by discovered OAuth creds if api_key: return api_key return _resolve_claude_code_token_from_credentials(_read_creds()) or _resolve_anthropic_pool_token() def run_oauth_setup_token() -> Optional[str]: - """Run 'claude setup-token' interactively; return the resulting token or None. - - Raises FileNotFoundError if the 'claude' CLI is not installed. - """ + """Run 'claude setup-token' interactively; the resulting token or None. FileNotFoundError if no 'claude' CLI.""" import shutil - claude_path = shutil.which("claude") if not claude_path: - raise FileNotFoundError( - "The 'claude' CLI is not installed. Install it with: npm install -g @anthropic-ai/claude-code" - ) - # Interactive: stdin/stdout/stderr inherited so the user can complete the OAuth prompt; - # the TUI-EOF stdin concern does not apply here. noqa: subprocess-stdin + raise FileNotFoundError("The 'claude' CLI is not installed. Install it with: npm install -g @anthropic-ai/claude-code") + # Interactive: stdio inherited so the user can complete the OAuth prompt. noqa: subprocess-stdin try: subprocess.run([claude_path, "setup-token"]) except (KeyboardInterrupt, EOFError): @@ -550,11 +507,7 @@ def run_oauth_setup_token() -> Optional[str]: creds = read_claude_code_credentials() if creds and is_claude_code_token_valid(creds): return creds["accessToken"] - for env_var in ("CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_TOKEN"): - val = _getenv(env_var).strip() - if val: - return val - return None + return _first_env("CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_TOKEN") or None # ── Hermes-native PKCE OAuth flow (~/.hermes/.anthropic_oauth.json); mirrors Claude Code / pi-ai / OpenCode ── @@ -565,13 +518,10 @@ def _get_hermes_oauth_file() -> Path: def _root_hermes_oauth_file() -> Optional[Path]: - """Global-root ``.anthropic_oauth.json`` inside a named profile; None in classic mode. - - Used to commit a rotation of a grant the profile borrowed via the pool's root fallback. - """ + """Global-root ``.anthropic_oauth.json`` inside a named profile (None in classic mode); used to commit a + rotation of a grant the profile borrowed via the pool's root fallback.""" try: from hermes_constants import get_default_hermes_root - root = get_default_hermes_root() if root.resolve(strict=False) == get_hermes_home().resolve(strict=False): return None @@ -591,31 +541,20 @@ def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: """Run Hermes-native OAuth PKCE flow and return credential state.""" import webbrowser from urllib.parse import urlencode - verifier, challenge = _generate_pkce() oauth_state = secrets.token_urlsafe(32) params = { - "code": "true", - "client_id": _OAUTH_CLIENT_ID, - "response_type": "code", - "redirect_uri": _OAUTH_REDIRECT_URI, - "scope": _OAUTH_SCOPES, - "code_challenge": challenge, - "code_challenge_method": "S256", - "state": oauth_state, + "code": "true", "client_id": _OAUTH_CLIENT_ID, "response_type": "code", "redirect_uri": _OAUTH_REDIRECT_URI, + "scope": _OAUTH_SCOPES, "code_challenge": challenge, "code_challenge_method": "S256", "state": oauth_state, } auth_url = f"https://claude.ai/oauth/authorize?{urlencode(params)}" print("\n".join([ - "", - "Authorize Hermes with your Claude Pro/Max subscription.", - "", + "", "Authorize Hermes with your Claude Pro/Max subscription.", "", "╭─ Claude Pro/Max Authorization ────────────────────╮", "│ │", "│ Open this link in your browser: │", "╰───────────────────────────────────────────────────╯", - "", - f" {auth_url}", - "", + "", f" {auth_url}", "", ])) try: from hermes_cli.auth import _can_open_graphical_browser as _can_open_gui @@ -625,7 +564,6 @@ def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: with contextlib.suppress(Exception): webbrowser.open(auth_url) print(" (Browser opened automatically)") - print("\nAfter authorizing, you'll see a code. Paste it below.\n") try: auth_code = input("Authorization code: ").strip() @@ -641,12 +579,8 @@ def run_hermes_oauth_login_pure() -> Optional[Dict[str, Any]]: return None try: exchange_data = json.dumps({ - "grant_type": "authorization_code", - "client_id": _OAUTH_CLIENT_ID, - "code": code, - "state": received_state, - "redirect_uri": _OAUTH_REDIRECT_URI, - "code_verifier": verifier, + "grant_type": "authorization_code", "client_id": _OAUTH_CLIENT_ID, "code": code, "state": received_state, + "redirect_uri": _OAUTH_REDIRECT_URI, "code_verifier": verifier, }).encode() result = _post_oauth_token(exchange_data, content_type="application/json", timeout=15, what="exchange") except Exception as e: @@ -667,12 +601,11 @@ def read_hermes_oauth_credentials() -> Optional[Dict[str, Any]]: def _write_hermes_oauth_credentials( access_token: str, refresh_token: Optional[str], expires_at_ms: Optional[int], *, target: Optional[Path] = None ) -> None: - """Commit refreshed hermes_pkce tokens to ~/.hermes/.anthropic_oauth.json. + """Commit refreshed hermes_pkce tokens to ~/.hermes/.anthropic_oauth.json (``CredentialPersistError`` on failure). - ``target`` lets a named profile commit a grant it BORROWED from the global root back to - the ROOT singleton instead of forking a copy under its own HERMES_HOME; without this - write-through the next ``load_pool()`` re-seeds the stale (consumed) pair from the file - over the rotated pool entry. Raises ``CredentialPersistError`` when the pair does not land. + ``target`` lets a named profile commit a grant it BORROWED from the global root back to the ROOT + singleton instead of forking a copy under its own HERMES_HOME; without this write-through the next + ``load_pool()`` re-seeds the stale (consumed) pair from the file over the rotated pool entry. """ _commit_private_json( target if target is not None else _get_hermes_oauth_file(), diff --git a/agent/azure_identity_adapter.py b/agent/azure_identity_adapter.py index 11f7172bd3..72ef39984f 100644 --- a/agent/azure_identity_adapter.py +++ b/agent/azure_identity_adapter.py @@ -20,9 +20,8 @@ from typing import Any, Callable, Dict, Optional logger = logging.getLogger(__name__) -# Microsoft-documented Foundry inference scope for ALL endpoint shapes. The older -# ``https://cognitiveservices.azure.com/.default`` is an ARM control-plane scope rejected for -# inference by newer resources; override via ``model.entra.scope`` if required. +# Microsoft-documented Foundry inference scope for ALL endpoint shapes. The older cognitiveservices.azure.com +# scope is an ARM control-plane scope rejected for inference by newer resources; override via ``model.entra.scope``. SCOPE_AI_AZURE_DEFAULT = "https://ai.azure.com/.default" _AZURE_IDENTITY_FEATURE = "provider.azure_identity" @@ -69,12 +68,9 @@ def reset_credential_cache() -> None: @dataclass(frozen=True) class EntraIdentityConfig: - """Hermes-managed Entra knobs; everything else (tenant, SP secret, federated token file, authority, - ``AZURE_CLIENT_ID``...) flows through azure-identity's standard ``AZURE_*`` env vars. - - ``exclude_interactive_browser`` keeps probes non-interactive; the setup wizard never writes it. - Frozen so it is hashable for ``lru_cache`` and picklable across multiprocessing workers. - """ + """Hermes-managed Entra knobs; everything else (tenant, SP secret, federated token file, authority...) flows + through azure-identity's standard ``AZURE_*`` env vars. ``exclude_interactive_browser`` keeps probes + non-interactive (the setup wizard never writes it). Frozen: hashable for ``lru_cache``, picklable for workers.""" scope: str = SCOPE_AI_AZURE_DEFAULT exclude_interactive_browser: bool = True @@ -99,9 +95,8 @@ def build_credential(config: EntraIdentityConfig) -> Any: """Cached ``DefaultAzureCredential``. ``maxsize=1`` is intentional: a process uses one ``model.entra.*`` block at a time. Only Hermes knobs are passed as kwargs; the rest comes from ``AZURE_*`` env vars.""" ai = _require_azure_identity() - kwargs: Dict[str, Any] = {} - if not config.exclude_interactive_browser: # SDK default already excludes the browser - kwargs["exclude_interactive_browser_credential"] = False + # SDK default already excludes the browser; only pass the kwarg when opting in. + kwargs = {} if config.exclude_interactive_browser else {"exclude_interactive_browser_credential": False} return ai.DefaultAzureCredential(**kwargs) @@ -116,8 +111,7 @@ def _install_failure(allow_install: bool) -> Optional[Dict[str, Any]]: if has_azure_identity_installed(): return None if not allow_install: - return {"error": "azure-identity not installed", - "hint": "pip install azure-identity (or rely on lazy install at first use)"} + return {"error": "azure-identity not installed", "hint": "pip install azure-identity (or rely on lazy install at first use)"} try: _require_azure_identity() except ImportError as exc: @@ -128,11 +122,9 @@ def _install_failure(allow_install: bool) -> Optional[Dict[str, Any]]: def build_token_provider(scope: Optional[str] = None, *, config: Optional[EntraIdentityConfig] = None, base_url: Optional[str] = None, exclude_interactive_browser: bool = True, ) -> Callable[[], str]: - """Zero-arg callable minting a fresh Entra bearer JWT — pass as ``OpenAI(api_key=...)``. - - Scope precedence: ``config.scope`` > ``scope`` kwarg > default. ``base_url`` is unused (back-compat). - Not picklable: ship the ``EntraIdentityConfig`` and rebuild in the worker. - """ + """Zero-arg callable minting a fresh Entra bearer JWT — pass as ``OpenAI(api_key=...)``. Scope precedence: + ``config.scope`` > ``scope`` kwarg > default; ``base_url`` is unused (back-compat). Not picklable: ship the + ``EntraIdentityConfig`` and rebuild in the worker.""" ai = _require_azure_identity() config = _resolve_config(config, scope, exclude_interactive_browser=exclude_interactive_browser) return ai.get_bearer_token_provider(build_credential(config), config.scope) @@ -157,11 +149,9 @@ def _probe_token(config: EntraIdentityConfig, timeout_seconds: float) -> Optiona def has_azure_identity_credentials(scope: Optional[str] = None, *, config: Optional[EntraIdentityConfig] = None, timeout_seconds: float = 10.0, allow_install: bool = True, **overrides: Any) -> bool: - """Timeout-bounded probe: can the chain mint a token now? Never raises. - - ``allow_install=False`` makes it a strict "is installed?" check for hot paths (CLI startup) where pip - must never run. NOT used by ``is_provider_configured()`` (structural, no mint). - """ + """Timeout-bounded probe: can the chain mint a token now? Never raises. ``allow_install=False`` makes it a + strict "is installed?" check for hot paths (CLI startup) where pip must never run. NOT used by + ``is_provider_configured()`` (structural, no mint).""" failure = _install_failure(allow_install) if failure is not None: if "exc" in failure: @@ -185,15 +175,10 @@ def _scoped_env(name: str) -> str: """Credential-bearing env read via the profile secret scope so a multiplexed profile never reports another profile's env-bridged credentials; unscoped CLI probes fall back to plain env.""" try: - from agent.secret_scope import UnscopedSecretError, get_secret - - try: - return (get_secret(name) or "").strip() - except UnscopedSecretError: - pass - except Exception: - pass - return _env(name) + from agent.secret_scope import get_secret + return (get_secret(name) or "").strip() + except Exception: # UnscopedSecretError, import failure, or any scope error + return _env(name) # (label, predicate) for env-var-driven credential sources, in chain order. @@ -208,11 +193,9 @@ _ENV_SOURCE_CHECKS = ( def describe_active_credential(config: Optional[EntraIdentityConfig] = None, *, scope: Optional[str] = None, timeout_seconds: float = 10.0, allow_install: bool = True, **overrides: Any) -> Dict[str, Any]: - """Doctor / preflight diagnostics. Never raises; ``{"ok": False, "error": ...}`` on failure. - - azure-identity hides the winning inner credential, so this reports a coarse picture (env sources, - token expiry) rather than a class name; ``AZURE_LOG_LEVEL=DEBUG`` shows the chain. - """ + """Doctor / preflight diagnostics. Never raises; ``{"ok": False, "error": ...}`` on failure. azure-identity + hides the winning inner credential, so this reports a coarse picture (env sources, token expiry) rather + than a class name; ``AZURE_LOG_LEVEL=DEBUG`` shows the chain.""" info: Dict[str, Any] = {"ok": False} failure = _install_failure(allow_install) if failure is not None: @@ -226,11 +209,9 @@ def describe_active_credential(config: Optional[EntraIdentityConfig] = None, *, result = _probe_token(config, timeout_seconds) if result is None: info["error"] = f"Token probe timed out after {timeout_seconds:.0f}s" - info["hint"] = ( - "DefaultAzureCredential can be slow when the token service is unreachable " - "or when az login state is stale. Try `az login` or set " - "AZURE_CLIENT_ID / AZURE_TENANT_ID / AZURE_CLIENT_SECRET." - ) + info["hint"] = ("DefaultAzureCredential can be slow when the token service is unreachable " + "or when az login state is stale. Try `az login` or set " + "AZURE_CLIENT_ID / AZURE_TENANT_ID / AZURE_CLIENT_SECRET.") return info if "error" in result: info["error"] = result["error"] @@ -251,12 +232,9 @@ def is_token_provider(value: Any) -> bool: def materialize_bearer_for_http(value: Any) -> str: - """Mint a fresh Bearer JWT for a manual HTTP request (calls the provider once). - - Only for sites building ``Authorization`` outside the OpenAI SDK; the Anthropic SDK can't take a - callable, so :func:`build_bearer_http_client` calls this from an httpx hook. ``ValueError`` on an - unusable value or empty token. - """ + """Mint a fresh Bearer JWT for a manual HTTP request (calls the provider once). Only for sites building + ``Authorization`` outside the OpenAI SDK; the Anthropic SDK can't take a callable, so + :func:`build_bearer_http_client` calls this from an httpx hook. ``ValueError`` on an unusable value/empty token.""" if is_token_provider(value): token = value() if not isinstance(token, str) or not token: @@ -273,12 +251,10 @@ def _strip_auth_headers(request: Any) -> None: def build_bearer_http_client(token_provider: Callable[[], str], **httpx_kwargs: Any) -> Any: - """``httpx.Client`` minting a fresh Entra bearer JWT per outbound request. - - The Anthropic SDK computes ``Authorization`` once at construction, so per-request refresh needs a - ``request`` hook: mint (cheap — azure-identity caches), strip pre-set auth headers, set - ``Authorization: Bearer``. ``httpx_kwargs`` are forwarded verbatim (``timeout``, ``transport``...). - """ + """``httpx.Client`` minting a fresh Entra bearer JWT per outbound request. The Anthropic SDK computes + ``Authorization`` once at construction, so per-request refresh needs a ``request`` hook: mint (cheap — + azure-identity caches), strip pre-set auth headers, set ``Authorization: Bearer``. ``httpx_kwargs`` are + forwarded verbatim (``timeout``, ``transport``...).""" if not is_token_provider(token_provider): raise ValueError("build_bearer_http_client requires a zero-arg callable token provider") import httpx @@ -287,15 +263,12 @@ def build_bearer_http_client(token_provider: Callable[[], str], **httpx_kwargs: try: token = materialize_bearer_for_http(token_provider) except ValueError as exc: - # Chain exhausted / az login expired: strip ALL auth headers (incl. the anthropic_adapter - # placeholder sentinel) so Azure returns a clean "missing auth" 401 and the sentinel never - # reaches upstream logs. WARNING so the misconfiguration is visible at default levels. - logger.warning( - "Bearer hook: Entra ID token provider returned empty (%s) " - "— stripping Authorization headers. Azure will respond 401. " - "Run `hermes doctor` or `az login` to recover.", - exc, - ) + # Chain exhausted / az login expired: strip ALL auth headers (incl. the anthropic_adapter placeholder + # sentinel) so Azure returns a clean "missing auth" 401 and the sentinel never reaches upstream logs. + # WARNING so the misconfiguration is visible at default levels. + logger.warning("Bearer hook: Entra ID token provider returned empty (%s) " + "— stripping Authorization headers. Azure will respond 401. " + "Run `hermes doctor` or `az login` to recover.", exc) _strip_auth_headers(request) return _strip_auth_headers(request)