mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
refactor(copilot): move credential logic behind generic ProviderProfile seams
GitHub->Copilot token exchange (hermes_cli/copilot_auth.py) moves into the in-tree copilot plugin (plugins/model-providers/copilot/credentials.py), and core's Copilot credential branches become generic profile hooks the plugin fills: resolve_api_key_secret, runtime_base_url, prepare_pool_entry, seed_pool_credentials, remove_credential_source, refresh_runtime_credentials, is_stale_credential_error, request_rejection_hint, external_auth_evidence and the ambient_credentials field. Behaviour-preserving (gh auth token fallback kept).
This commit is contained in:
1 parent
0c771d7a5b
commit
2e30ce41c9
19 files changed
+506
-374
No files matched your search
+16
-21
@@ -2170,12 +2170,12 @@ def _resolve_api_key_provider() -> tuple[Optional[OpenAI], Optional[str]]:
|
||||
if not is_provider_explicitly_configured("anthropic"):
|
||||
continue
|
||||
return _try_anthropic()
|
||||
if provider_id == "copilot":
|
||||
# Explicit-config gate: ambient gh-CLI credentials must not silently become aux fallback (#114740).
|
||||
with contextlib.suppress(ImportError):
|
||||
from hermes_cli.auth import is_provider_explicitly_configured
|
||||
if not is_provider_explicitly_configured("copilot"):
|
||||
continue
|
||||
from providers import get_provider_profile
|
||||
if getattr(get_provider_profile(provider_id), "ambient_credentials", False):
|
||||
# Explicit-config gate: ambient credentials (another tool's login) must not silently become aux fallback (#114740).
|
||||
from hermes_cli.auth import is_provider_explicitly_configured
|
||||
if not is_provider_explicitly_configured(provider_id):
|
||||
continue
|
||||
model = _get_aux_model_for_provider(provider_id) or None
|
||||
if model is None:
|
||||
continue # skip provider if we don't know a valid aux model
|
||||
@@ -2238,7 +2238,9 @@ def _endpoint_default_headers(
|
||||
if base_url_host_matches(base_url, "api.kimi.com"):
|
||||
headers: dict = {"User-Agent": "claude-code/0.1.0"}
|
||||
elif base_url_host_matches(base_url, "githubcopilot.com"):
|
||||
from hermes_cli.copilot_auth import copilot_request_headers
|
||||
from providers import get_provider_profile
|
||||
get_provider_profile("copilot") # loads the plugin package that owns these credentials
|
||||
from plugins.model_providers.copilot.credentials import copilot_request_headers
|
||||
headers = dict(copilot_request_headers(is_agent_turn=True, is_vision=is_vision))
|
||||
elif base_url_host_matches(base_url, "integrate.api.nvidia.com"):
|
||||
headers = dict(build_nvidia_nim_headers(base_url))
|
||||
@@ -3818,16 +3820,6 @@ def _creds_have_api_key(creds: dict[str, Any]) -> bool:
|
||||
return bool(str(creds.get("api_key", "") or "").strip())
|
||||
|
||||
|
||||
def _refresh_copilot_credentials() -> bool:
|
||||
from hermes_cli.copilot_auth import _jwt_cache, _token_fingerprint, exchange_copilot_token, resolve_copilot_token
|
||||
raw_token, _source = resolve_copilot_token()
|
||||
if not str(raw_token or "").strip():
|
||||
return False
|
||||
_jwt_cache.pop(_token_fingerprint(raw_token), None)
|
||||
exchange_copilot_token(raw_token)
|
||||
return True
|
||||
|
||||
|
||||
def _refresh_codex_credentials() -> bool:
|
||||
from hermes_cli.auth import resolve_codex_runtime_credentials
|
||||
return _creds_have_api_key(resolve_codex_runtime_credentials(force_refresh=True))
|
||||
@@ -3877,7 +3869,7 @@ def _refresh_vertex_credentials() -> bool:
|
||||
|
||||
# Each refresher returns True when a usable credential exists; the caller then evicts cached clients.
|
||||
_CREDENTIAL_REFRESHERS: dict[str, Callable[..., bool]] = {
|
||||
"copilot": _refresh_copilot_credentials, "openai-codex": _refresh_codex_credentials,
|
||||
"openai-codex": _refresh_codex_credentials,
|
||||
"nous": _refresh_nous_credentials, "anthropic": _refresh_anthropic_credentials,
|
||||
"xai-oauth": _refresh_xai_oauth_credentials, "vertex": _refresh_vertex_credentials,
|
||||
}
|
||||
@@ -3888,7 +3880,11 @@ def _refresh_provider_credentials(provider: str, *, failed_api_key: str = "") ->
|
||||
normalized = _normalize_aux_provider(provider)
|
||||
refresher = _CREDENTIAL_REFRESHERS.get(normalized)
|
||||
if refresher is None:
|
||||
return False
|
||||
from providers import get_provider_profile
|
||||
profile = get_provider_profile(normalized)
|
||||
if profile is None:
|
||||
return False
|
||||
refresher = lambda: profile.refresh_runtime_credentials(reason="auxiliary_unauthorized")
|
||||
try:
|
||||
if not (refresher(failed_api_key) if normalized == "anthropic" else refresher()):
|
||||
return False
|
||||
@@ -5289,9 +5285,8 @@ def _resolve_api_key_branch(req: _ResolveRequest, pconfig: Any, resolve_creds: C
|
||||
if not api_key and is_actual_local_base_url(raw_base_url):
|
||||
api_key = ACTUAL_LOCAL_NOAUTH_PLACEHOLDER
|
||||
if not api_key:
|
||||
tried_sources = list(pconfig.api_key_env_vars) + (["gh auth token"] if provider == "copilot" else [])
|
||||
logger.debug("resolve_provider_client: provider %s has no API key configured (tried: %s)",
|
||||
provider, ", ".join(tried_sources))
|
||||
provider, ", ".join(pconfig.api_key_env_vars))
|
||||
return None, None
|
||||
base_url = _to_openai_base_url(raw_base_url)
|
||||
# Explicit base_url override: a fallback_model/custom_providers entry pointing a built-in name elsewhere.
|
||||
|
||||
+16
-69
@@ -451,7 +451,9 @@ class ClientLifecycleMixin:
|
||||
request_kwargs["max_retries"] = 0
|
||||
is_copilot = base_url_host_matches(str(request_kwargs.get("base_url", "")), "githubcopilot.com")
|
||||
if is_copilot and self._api_kwargs_have_image_parts(api_kwargs or {}):
|
||||
from hermes_cli.copilot_auth import copilot_request_headers
|
||||
from providers import get_provider_profile
|
||||
get_provider_profile("copilot") # loads the plugin package that owns these credentials
|
||||
from plugins.model_providers.copilot.credentials import copilot_request_headers
|
||||
request_kwargs["default_headers"] = copilot_request_headers(is_agent_turn=True, is_vision=True)
|
||||
cached, stale = self._checkout_request_slot(_OPENAI_SLOT, request_kwargs)
|
||||
if cached is not None:
|
||||
@@ -806,77 +808,22 @@ class ClientLifecycleMixin:
|
||||
logger.info("Vertex AI OAuth token refreshed")
|
||||
return ok
|
||||
|
||||
def _apply_copilot_token(self, token: str, enterprise_base_url: Any, *, reason: str) -> bool:
|
||||
def _try_refresh_provider_runtime_credentials(self, *, reason: str) -> bool:
|
||||
"""Re-mint the live credential through the provider profile's ``refresh_runtime_credentials``
|
||||
and rebuild the shared client (caller enforces the single-shot guard). ``reason``:
|
||||
``unauthorized`` (401) or ``stale`` (a non-auth error the profile reads as a stale credential)."""
|
||||
from agent.conversation_loop import _credential_refresh_profile
|
||||
profile = _credential_refresh_profile(self)
|
||||
refreshed = profile.refresh_runtime_credentials(reason=reason) if profile else None
|
||||
if not refreshed:
|
||||
return False
|
||||
token, base_url = refreshed
|
||||
self.api_key = token
|
||||
if enterprise_base_url:
|
||||
self.base_url = enterprise_base_url.rstrip("/")
|
||||
if base_url:
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self._sync_client_kwargs_credentials()
|
||||
self._apply_client_headers_for_base_url(str(self.base_url or ""))
|
||||
return self._replace_primary_openai_client(reason=reason)
|
||||
|
||||
def _try_refresh_copilot_client_credentials(self) -> bool:
|
||||
"""Refresh Copilot credentials and rebuild the shared OpenAI client (caller enforces the single-shot guard).
|
||||
|
||||
The raw GitHub token is stable; the short-TTL *exchanged* IDE token is what expires mid-turn (``401 IDE
|
||||
token expired``), so force a fresh exchange rather than re-resolving the raw token.
|
||||
"""
|
||||
if not self._is_copilot_provider():
|
||||
return False
|
||||
try:
|
||||
from hermes_cli.copilot_auth import resolve_copilot_token, get_copilot_api_token, evict_cached_exchanged_token
|
||||
new_token, token_source = resolve_copilot_token()
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot credential refresh failed: %s", exc)
|
||||
return False
|
||||
new_token, enterprise_base_url = (new_token.strip() if isinstance(new_token, str) else ""), None
|
||||
if not new_token:
|
||||
return False
|
||||
# Fall back to the raw token only if the exchange itself is unavailable.
|
||||
try:
|
||||
evict_cached_exchanged_token(new_token)
|
||||
api_token, exchanged_base_url = get_copilot_api_token(new_token)
|
||||
if isinstance(api_token, str) and api_token.strip():
|
||||
new_token, enterprise_base_url = api_token.strip(), exchanged_base_url
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot 401 re-exchange failed, using resolved token: %s", exc)
|
||||
ok = self._apply_copilot_token(new_token, enterprise_base_url, reason="copilot_credential_refresh")
|
||||
if ok:
|
||||
logger.info("Copilot credentials refreshed from %s", token_source)
|
||||
return ok
|
||||
|
||||
def _try_recover_stale_copilot_credential(self) -> bool:
|
||||
"""Force a fresh Copilot token exchange + client rebuild after a 400 (single-shot, caller-guarded).
|
||||
|
||||
Copilot surfaces a stale credential as ``400 model_not_available_for_integrator`` / ``model_not_supported``
|
||||
— typically a raw ``ghu_`` token cached when the startup exchange degraded (restricted integrator allowlist).
|
||||
"""
|
||||
if not self._is_copilot_provider():
|
||||
return False
|
||||
try:
|
||||
from hermes_cli.copilot_auth import resolve_copilot_token, get_copilot_api_token, evict_cached_exchanged_token
|
||||
raw_token, token_source = resolve_copilot_token()
|
||||
if not isinstance(raw_token, str) or not raw_token.strip():
|
||||
return False
|
||||
raw_token = raw_token.strip()
|
||||
# Evict the cached (possibly degraded/raw) exchanged token so the exchange hits the network.
|
||||
evict_cached_exchanged_token(raw_token)
|
||||
api_token, enterprise_base_url = get_copilot_api_token(raw_token)
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot stale-credential recovery failed: %s", exc)
|
||||
return False
|
||||
if not isinstance(api_token, str) or not api_token.strip():
|
||||
return False
|
||||
# Exchange STILL degraded to the raw token: a rebuild won't help — don't burn the single-shot retry.
|
||||
if api_token == raw_token and not enterprise_base_url:
|
||||
logger.warning(
|
||||
"Copilot stale-credential recovery: exchange still degraded to raw token; skipping retry "
|
||||
"(network/exchange endpoint unavailable)."
|
||||
)
|
||||
return False
|
||||
ok = self._apply_copilot_token(api_token.strip(), enterprise_base_url, reason="copilot_stale_credential_recovery")
|
||||
if ok:
|
||||
logger.info("Copilot credentials re-exchanged after stale-credential 400 (source=%s)", token_source)
|
||||
return ok
|
||||
return self._replace_primary_openai_client(reason=f"{profile.name}_credential_{reason}")
|
||||
|
||||
def _try_refresh_anthropic_client_credentials(self) -> bool:
|
||||
# Only native Anthropic rotates OAuth tokens; other anthropic_messages providers (MiniMax, Alibaba, ...)
|
||||
|
||||
+16
-24
@@ -362,32 +362,24 @@ def _apply_active_turn_redirect(agent: Any, messages: list[dict[str, Any]], text
|
||||
agent._stream_needs_break = True
|
||||
|
||||
|
||||
def _is_copilot_provider(agent: Any) -> bool:
|
||||
"""Delegate to ``AIAgent._is_copilot_provider``; the fallback keeps the ``github-copilot`` /
|
||||
``github`` aliases so credential recovery is not skipped for them."""
|
||||
try:
|
||||
return bool(agent._is_copilot_provider())
|
||||
except Exception:
|
||||
return (getattr(agent, "provider", "") or "").strip().lower() in {
|
||||
"copilot",
|
||||
"github-copilot",
|
||||
"github",
|
||||
}
|
||||
def _credential_refresh_profile(agent: Any) -> Any:
|
||||
"""The profile owning the live credential's runtime refresh: the agent's provider by name, else
|
||||
(a ``custom`` route at a provider's host) the registered profile whose hostname the base URL
|
||||
targets. Only profiles that override ``refresh_runtime_credentials`` qualify."""
|
||||
from providers import get_provider_profile, list_providers
|
||||
from providers.base import ProviderProfile
|
||||
|
||||
def _refreshes(profile: Any) -> bool:
|
||||
return type(profile).refresh_runtime_credentials is not ProviderProfile.refresh_runtime_credentials
|
||||
|
||||
def _is_stale_copilot_credential_error(status_code: Optional[int], error_message: str) -> bool:
|
||||
"""Detect a Copilot 400 that is really a STALE / DEGRADED credential (status 400 AND an
|
||||
integrator/model-not-supported marker, so a wrong model name never triggers the
|
||||
single-shot re-exchange). Caller enforces scoping/guard."""
|
||||
lowered = (error_message or "").lower()
|
||||
if status_code != 400 and "error code: 400" not in lowered:
|
||||
return False
|
||||
return any(marker in lowered for marker in (
|
||||
"model_not_available_for_integrator",
|
||||
"not available for integrator",
|
||||
"model_not_supported",
|
||||
"the requested model is not supported",
|
||||
))
|
||||
named = get_provider_profile((getattr(agent, "provider", "") or "").strip().lower())
|
||||
if named is not None and _refreshes(named):
|
||||
return named
|
||||
base_url = str(getattr(agent, "base_url", "") or "")
|
||||
if not base_url:
|
||||
return None
|
||||
return next((p for p in list_providers()
|
||||
if _refreshes(p) and p.get_hostname() and base_url_host_matches(base_url, p.get_hostname())), None)
|
||||
|
||||
|
||||
def _pressure_with_real_floor(compressor: Any, rough_tokens: int) -> int:
|
||||
|
||||
+29
-75
@@ -2632,77 +2632,33 @@ def _seed_nous_singleton(seed: _Seeder, auth_store: dict[str, Any]) -> None:
|
||||
})
|
||||
|
||||
|
||||
# Warn once per token per process when Copilot exchange degrades to raw token (#114740).
|
||||
_COPILOT_RAW_DEGRADATION_WARNED: set[str] = set()
|
||||
|
||||
|
||||
def _warn_copilot_raw_degradation_once(token: str) -> None:
|
||||
"""WARN once per token per process when Copilot exchange degrades to raw token (#114740)."""
|
||||
fingerprint = fingerprint_secret_value(token) or "unknown"
|
||||
if fingerprint in _COPILOT_RAW_DEGRADATION_WARNED:
|
||||
def _seed_profile_singleton(seed: _Seeder) -> None:
|
||||
"""Rows a provider profile seeds itself (``seed_pool_credentials``): ambient/env tokens it
|
||||
validates or exchanges. The profile gates its own work on suppression and explicit config."""
|
||||
from providers import get_provider_profile
|
||||
profile = get_provider_profile(seed.provider)
|
||||
if profile is None:
|
||||
return
|
||||
_COPILOT_RAW_DEGRADATION_WARNED.add(fingerprint)
|
||||
logger.warning(
|
||||
"Copilot token exchange degraded to RAW token (exchange "
|
||||
"unavailable); enterprise-only models may 400 with "
|
||||
"model_not_available_for_integrator until exchange recovers."
|
||||
)
|
||||
|
||||
|
||||
def _reset_copilot_raw_degradation_warned() -> None:
|
||||
"""Clear the degradation warning cache (for test isolation)."""
|
||||
_COPILOT_RAW_DEGRADATION_WARNED.clear()
|
||||
|
||||
|
||||
def _seed_copilot_singleton(seed: _Seeder) -> None:
|
||||
# Copilot tokens are resolved dynamically via `gh auth token` or env vars
|
||||
# (COPILOT_GITHUB_TOKEN / GH_TOKEN); they don't live in the auth store.
|
||||
try:
|
||||
from hermes_cli.copilot_auth import (
|
||||
COPILOT_ENV_VARS,
|
||||
resolve_copilot_token,
|
||||
get_copilot_api_token,
|
||||
)
|
||||
# All-sources gate BEFORE any work: resolve_copilot_token() shells out
|
||||
# and the exchange retries 3x with backoff (~35s worst case); a user
|
||||
# who suppressed every copilot source must not pay that on every pool
|
||||
# load. The source space here matches credential_sources._remove_copilot_gh.
|
||||
copilot_sources = ["gh_cli"] + [f"env:{v}" for v in COPILOT_ENV_VARS]
|
||||
if all(seed.is_suppressed(seed.provider, s) for s in copilot_sources):
|
||||
return
|
||||
token, source = resolve_copilot_token()
|
||||
if not token:
|
||||
return
|
||||
# Exact match: a substring test would classify GH_TOKEN/GITHUB_TOKEN
|
||||
# as gh_cli and bypass a user's per-env-var suppression.
|
||||
source_name = "gh_cli" if source == "gh auth token" else f"env:{source}"
|
||||
# Per-source gate BEFORE the (~35s worst case) network exchange.
|
||||
if seed.is_suppressed(seed.provider, source_name):
|
||||
return
|
||||
from hermes_cli.auth import is_provider_explicitly_configured
|
||||
if not is_provider_explicitly_configured(seed.provider):
|
||||
# Copilot is only discovered here (ambient gh CLI login), not selected anywhere: no
|
||||
# model will be routed to it, so the network exchange — and its degradation warning on
|
||||
# every pool load — buys nothing (#114740). Seed the raw token; the load that follows
|
||||
# the user selecting copilot re-seeds and exchanges.
|
||||
api_token, enterprise_base_url = token, None
|
||||
else:
|
||||
api_token, enterprise_base_url = get_copilot_api_token(token)
|
||||
# get_copilot_api_token falls back to the RAW token when the exchange
|
||||
# fails; the Copilot API then routes it to the fallback
|
||||
# "copilot-language-server" integrator whose allowlist omits
|
||||
# enterprise-only models -> HTTP 400 on every turn. Surface it once.
|
||||
if api_token == token and not enterprise_base_url:
|
||||
_warn_copilot_raw_degradation_once(token)
|
||||
pconfig = PROVIDER_REGISTRY.get(seed.provider)
|
||||
seed.upsert(source_name, {
|
||||
from hermes_cli.auth import is_provider_explicitly_configured
|
||||
rows = profile.seed_pool_credentials(
|
||||
is_suppressed=lambda source: seed.is_suppressed(seed.provider, source),
|
||||
explicitly_configured=lambda: is_provider_explicitly_configured(seed.provider))
|
||||
pconfig = PROVIDER_REGISTRY.get(seed.provider)
|
||||
for row in rows or ():
|
||||
seed.upsert(row["source"], {
|
||||
"auth_type": AUTH_TYPE_API_KEY,
|
||||
"access_token": api_token,
|
||||
"base_url": enterprise_base_url or (pconfig.inference_base_url if pconfig else ""),
|
||||
"label": source,
|
||||
"access_token": row["access_token"],
|
||||
"base_url": row.get("base_url") or (pconfig.inference_base_url if pconfig else ""),
|
||||
"label": row.get("label") or row["source"],
|
||||
})
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot token seed failed: %s", exc)
|
||||
|
||||
|
||||
def _profile_seeds_pool(provider: str) -> bool:
|
||||
"""Whether the provider's profile owns pool seeding (the generic env loop must not overwrite it)."""
|
||||
from providers import get_provider_profile
|
||||
from providers.base import ProviderProfile
|
||||
profile = get_provider_profile(provider)
|
||||
return profile is not None and type(profile).seed_pool_credentials is not ProviderProfile.seed_pool_credentials
|
||||
|
||||
|
||||
def _seed_qwen_singleton(seed: _Seeder) -> None:
|
||||
@@ -2789,8 +2745,6 @@ def _seed_from_singletons(provider: str, entries: list[PooledCredential]) -> tup
|
||||
_seed_anthropic_singletons(seed)
|
||||
elif provider == "nous":
|
||||
_seed_nous_singleton(seed, auth_store)
|
||||
elif provider == "copilot":
|
||||
_seed_copilot_singleton(seed)
|
||||
elif provider == "qwen-oauth":
|
||||
_seed_qwen_singleton(seed)
|
||||
elif provider == "minimax-oauth":
|
||||
@@ -2801,6 +2755,8 @@ def _seed_from_singletons(provider: str, entries: list[PooledCredential]) -> tup
|
||||
if provider == "openai-codex" and seed.is_suppressed(provider, "device_code"):
|
||||
return seed.result
|
||||
_seed_tokens_singleton(seed, auth_store)
|
||||
else:
|
||||
_seed_profile_singleton(seed)
|
||||
return seed.result
|
||||
|
||||
|
||||
@@ -2903,11 +2859,9 @@ def _env_key_var_candidates(env_vars: list[str], entries: list[PooledCredential]
|
||||
|
||||
def _seed_from_env(provider: str, entries: list[PooledCredential]) -> tuple[bool, set[str]]:
|
||||
seed = _Seeder(provider, entries)
|
||||
# Copilot's singleton branch exchanges the raw ghu_ OAuth token for the
|
||||
# api token via `get_copilot_api_token`; the generic loop would re-read
|
||||
# COPILOT_GITHUB_TOKEN and overwrite it with the RAW token, causing 400s
|
||||
# ("not available for integrator copilot-language-server").
|
||||
if provider == "copilot":
|
||||
# A profile that seeds its own rows may exchange the raw env token for a runtime one; the
|
||||
# generic loop would re-read the same env var and overwrite it with the RAW token.
|
||||
if _profile_seeds_pool(provider):
|
||||
return seed.result
|
||||
|
||||
if provider == "openrouter":
|
||||
|
||||
+25
-27
@@ -88,8 +88,29 @@ class RemovalStep:
|
||||
|
||||
|
||||
def find_removal_step(provider: str, source: str) -> Optional[RemovalStep]:
|
||||
"""First matching RemovalStep, or None (``manual``: nothing external to clean)."""
|
||||
return next((step for step in _REGISTRY if step.matches(provider, source)), None)
|
||||
"""The provider profile's own step when it claims *source*, else the first matching registry
|
||||
step, or None (``manual``: nothing external to clean)."""
|
||||
return _profile_removal_step(provider, source) or next(
|
||||
(step for step in _REGISTRY if step.matches(provider, source)), None)
|
||||
|
||||
|
||||
def _profile_removal_step(provider: str, source: str) -> Optional[RemovalStep]:
|
||||
"""``ProviderProfile.remove_credential_source``: a provider seeding one token under several
|
||||
sources suppresses every variant, or the duplicates resurrect the entry."""
|
||||
from providers import get_provider_profile
|
||||
profile = get_provider_profile(provider)
|
||||
claimed = profile.remove_credential_source(source) if profile else None
|
||||
if claimed is None:
|
||||
return None
|
||||
sources, hints = claimed
|
||||
|
||||
def remove_fn(provider: str, removed) -> RemovalResult:
|
||||
from hermes_cli.auth import suppress_credential_source
|
||||
for variant in sources:
|
||||
suppress_credential_source(provider, variant)
|
||||
return RemovalResult(hints=list(hints))
|
||||
|
||||
return RemovalStep(provider=provider, source_id=source, remove_fn=remove_fn)
|
||||
|
||||
|
||||
def _remove_env_source(provider: str, removed) -> RemovalResult:
|
||||
@@ -196,22 +217,6 @@ def _remove_codex_device_code(provider: str, removed) -> RemovalResult:
|
||||
return result
|
||||
|
||||
|
||||
def _remove_copilot_gh(provider: str, removed) -> RemovalResult:
|
||||
"""The same Copilot token is seeded as gh_cli AND env:<VAR> rows, so suppress
|
||||
every variant or the duplicates resurrect the entry. gh CLI and shell state
|
||||
are left untouched."""
|
||||
from hermes_cli.auth import suppress_credential_source
|
||||
|
||||
suppress_credential_source(provider, "gh_cli")
|
||||
for env_var in ("COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN"):
|
||||
suppress_credential_source(provider, f"env:{env_var}")
|
||||
return RemovalResult(hints=[
|
||||
"Suppressed all copilot token sources (gh_cli + env vars) — they will not be re-seeded.",
|
||||
"Note: Your gh CLI / shell environment is unchanged.",
|
||||
"Run `hermes auth add copilot` to re-enable if needed.",
|
||||
])
|
||||
|
||||
|
||||
def _suppress_only(*hints: str) -> Callable[..., RemovalResult]:
|
||||
"""remove_fn for sources whose backing file/config belongs to another tool
|
||||
(Claude Code, Qwen CLI, config.yaml): never delete it, just suppress and
|
||||
@@ -221,16 +226,9 @@ def _suppress_only(*hints: str) -> Callable[..., RemovalResult]:
|
||||
return remove_fn
|
||||
|
||||
|
||||
# ORDER MATTERS — ``find_removal_step`` returns the first match. Provider-
|
||||
# specific steps precede the generic ``env:*`` step so copilot's ``env:GH_TOKEN``
|
||||
# takes the copilot path (no .env edits) rather than the generic env-var removal.
|
||||
# ORDER MATTERS — ``find_removal_step`` returns the first match (after a provider profile's own
|
||||
# ``remove_credential_source`` claim, so a profile's ``env:*`` rows skip the generic .env removal).
|
||||
_REGISTRY: list[RemovalStep] = [
|
||||
RemovalStep(
|
||||
provider="copilot", source_id="gh_cli",
|
||||
match_fn=lambda src: src == "gh_cli" or src.startswith("env:"),
|
||||
remove_fn=_remove_copilot_gh,
|
||||
description="gh auth token / COPILOT_GITHUB_TOKEN / GH_TOKEN",
|
||||
),
|
||||
RemovalStep(
|
||||
provider="*", source_id="env:",
|
||||
match_fn=lambda src: src.startswith("env:"),
|
||||
|
||||
@@ -1023,7 +1023,7 @@ def classify_api_error(
|
||||
The credential is never included in the returned context."""
|
||||
from hermes_cli.anon_auth import is_anonymous_request
|
||||
status_code = _extract_status_code(error)
|
||||
# Copilot/GitHub Models RateLimitError may not set .status_code; force 429.
|
||||
# Some SDK RateLimitErrors (OpenAI-compatible relays) may not set .status_code; force 429.
|
||||
if status_code is None and type(error).__name__ == "RateLimitError":
|
||||
status_code = 429
|
||||
body = _extract_error_body(error)
|
||||
|
||||
+13
-15
@@ -272,13 +272,11 @@ def settle_unrecovered_error(
|
||||
current_turn_user_idx: Any = None,
|
||||
) -> UnrecoveredErrorVerdict:
|
||||
"""Decide the fate of an API error that every recovery chain declined: local validation /
|
||||
non-retryable client errors (Copilot stale-credential self-heal first, then fallback, then a
|
||||
non-retryable client errors (provider stale-credential self-heal first, then fallback, then a
|
||||
terminal result), max-retries exhaustion (primary transport recovery -> fallback -> terminal
|
||||
result), else the interruptible error backoff. ``FailoverReason.billing`` (402) is deliberately
|
||||
treated as non-retryable (#31273)."""
|
||||
from agent.conversation_loop import (
|
||||
_arm_fallback_restart, _is_copilot_provider, _is_stale_copilot_credential_error
|
||||
)
|
||||
from agent.conversation_loop import _arm_fallback_restart, _credential_refresh_profile
|
||||
|
||||
def _verdict(action: str, result: Optional[dict[str, Any]] = None) -> UnrecoveredErrorVerdict:
|
||||
return UnrecoveredErrorVerdict(
|
||||
@@ -321,21 +319,21 @@ def settle_unrecovered_error(
|
||||
# slug is dead for this account, so record it before the fallback walk runs (#106475).
|
||||
from agent.fallback_cooldown import _mark_entitlement_rejected_model
|
||||
_mark_entitlement_rejected_model(agent, api_error)
|
||||
# Copilot self-heal BEFORE fallback: a stale credential yields a 400
|
||||
# ``model_not_available_for_integrator`` / ``model_not_supported``, not a 401.
|
||||
# Fresh token + client rebuild, one retry, SAME provider.
|
||||
# Provider self-heal BEFORE fallback: some providers surface a stale credential as a
|
||||
# non-auth error (the profile's ``is_stale_credential_error``), not a 401. Fresh
|
||||
# credential + client rebuild, one retry, SAME provider.
|
||||
if (
|
||||
_is_copilot_provider(agent)
|
||||
and not _retry.copilot_stale_cred_retry_attempted
|
||||
and _is_stale_copilot_credential_error(
|
||||
status_code, str(getattr(api_error, "message", "") or api_error)
|
||||
not _retry.provider_stale_refresh_attempted
|
||||
and (profile := _credential_refresh_profile(agent)) is not None
|
||||
and profile.is_stale_credential_error(
|
||||
status_code=status_code, message=str(getattr(api_error, "message", "") or api_error)
|
||||
)
|
||||
):
|
||||
_retry.copilot_stale_cred_retry_attempted = True
|
||||
if agent._try_recover_stale_copilot_credential():
|
||||
_retry.provider_stale_refresh_attempted = True
|
||||
if agent._try_refresh_provider_runtime_credentials(reason="stale"):
|
||||
agent._buffer_vprint(
|
||||
"🔐 Copilot credential re-exchanged after "
|
||||
"model_not_available 400. Retrying request..."
|
||||
f"🔐 {profile.display_name or profile.name} credential renewed after a "
|
||||
"stale-credential error. Retrying request..."
|
||||
)
|
||||
retry_count = 0
|
||||
return _verdict("continue")
|
||||
|
||||
+13
-18
@@ -28,7 +28,6 @@ from agent.model_metadata import (
|
||||
)
|
||||
from agent.turn_failure_copy import site_copy, stamp_failure
|
||||
from agent.turn_retry_state import TurnRetryState
|
||||
from utils import base_url_host_matches
|
||||
|
||||
logger = logging.getLogger("agent.conversation_loop")
|
||||
|
||||
@@ -39,14 +38,6 @@ _RETRY_HINT = " 💡 Try /new to start a fresh conversation, or /compress to r
|
||||
# overflow (request ≈ window) never lands below half.
|
||||
_UNEXPLAINED_REJECTION_FRACTION = 0.5
|
||||
|
||||
_GITHUB_MODELS_HINT = (
|
||||
" 💡 GitHub Models free tier (models.inference.ai.azure.com) caps every",
|
||||
" request at ~8K tokens. Hermes' system prompt + tool schemas baseline",
|
||||
" exceeds that floor, so this endpoint cannot run an agentic loop.",
|
||||
" Use the `copilot` provider with a Copilot subscription token (`hermes",
|
||||
" setup` → GitHub Copilot), or pick any other provider.",
|
||||
)
|
||||
|
||||
_MINIMAX_ANTHROPIC_URLS = ("https://api.minimax.io/anthropic", "https://api.minimaxi.com/anthropic")
|
||||
|
||||
|
||||
@@ -460,6 +451,14 @@ def _recover_context_length(st: _Recovery, _retry: TurnRetryState, error_msg: st
|
||||
)
|
||||
|
||||
|
||||
def _request_rejection_hint(status_code: Optional[int], base_url: str) -> tuple[str, ...]:
|
||||
"""Every registered profile's ``request_rejection_hint`` for this endpoint, whatever route reached
|
||||
it (a free tier is commonly reached as ``custom``)."""
|
||||
from providers import list_providers
|
||||
return tuple(line for profile in list_providers()
|
||||
for line in profile.request_rejection_hint(status_code=status_code, base_url=base_url))
|
||||
|
||||
|
||||
def recover_from_overflow(
|
||||
agent: Any, api_error: Exception, classified: Any, _retry: TurnRetryState, *,
|
||||
status_code: Optional[int], error_msg: str, wrapped_output_cap_budget: Optional[int],
|
||||
@@ -469,7 +468,7 @@ def recover_from_overflow(
|
||||
effective_task_id: Any,
|
||||
) -> OverflowVerdict:
|
||||
"""413 payload-too-large and context-length recovery (compress + retry, output-cap
|
||||
clamp, provider-reported context limit, GitHub Models free-tier hint). Order is
|
||||
clamp, provider-reported context limit, endpoint hard-cap hints). Order is
|
||||
load-bearing: 413 is checked BEFORE the generic 4xx handler, and context-length
|
||||
errors (incl. relay-wrapped output-cap 429s) BEFORE non-retryable client errors.
|
||||
Compression progress is scored in payload BYTES for 413 (never the byte-blind token
|
||||
@@ -483,14 +482,10 @@ def recover_from_overflow(
|
||||
overflow_reason=getattr(classified.reason, "value", None),
|
||||
)
|
||||
|
||||
# GitHub Models free tier caps requests at 8K tokens, under the system prompt +
|
||||
# tool schema floor; compression can't help, so say so.
|
||||
if (
|
||||
status_code == 413
|
||||
and isinstance(agent.base_url, str)
|
||||
and base_url_host_matches(agent.base_url, "models.inference.ai.azure.com")
|
||||
):
|
||||
for line in _GITHUB_MODELS_HINT:
|
||||
# An endpoint whose hard cap no recovery can meet (a provider profile's
|
||||
# ``request_rejection_hint``, whatever route reached it): say so.
|
||||
if isinstance(agent.base_url, str):
|
||||
for line in _request_rejection_hint(status_code, agent.base_url):
|
||||
agent._vprint(f"{agent.log_prefix}{line}", force=True, diagnostic=True)
|
||||
|
||||
if classified.reason == FailoverReason.payload_too_large:
|
||||
|
||||
@@ -401,10 +401,10 @@ def _print_anthropic_401_diagnostics(agent: Any, key: Any) -> None:
|
||||
def _refresh_credentials_after_401(
|
||||
agent: Any, api_error: Exception, _retry: TurnRetryState, status_code: Optional[int]
|
||||
) -> bool:
|
||||
"""Per-provider one-shot credential refresh on 401 (codex/xai, vertex, nous, copilot,
|
||||
anthropic), printing user-facing diagnostics when the nous/anthropic refresh fails.
|
||||
Returns True when a refresh succeeded and the call should be retried."""
|
||||
from agent.conversation_loop import _is_copilot_provider
|
||||
"""Per-provider one-shot credential refresh on 401 (codex/xai, vertex, nous, a profile's
|
||||
``refresh_runtime_credentials``, anthropic), printing user-facing diagnostics when the
|
||||
nous/anthropic refresh fails. Returns True when a refresh succeeded and the call should be retried."""
|
||||
from agent.conversation_loop import _credential_refresh_profile
|
||||
|
||||
if status_code != 401:
|
||||
return False
|
||||
@@ -433,10 +433,11 @@ def _refresh_credentials_after_401(
|
||||
agent._buffer_vprint("🔐 Nous agent key refreshed after 401. Retrying request...")
|
||||
return True
|
||||
_print_nous_401_diagnostics(agent, api_error)
|
||||
if _is_copilot_provider(agent) and not _retry.copilot_auth_retry_attempted:
|
||||
_retry.copilot_auth_retry_attempted = True
|
||||
if agent._try_refresh_copilot_client_credentials():
|
||||
agent._buffer_vprint("🔐 Copilot credentials refreshed after 401. Retrying request...")
|
||||
if not _retry.provider_auth_refresh_attempted and (profile := _credential_refresh_profile(agent)) is not None:
|
||||
_retry.provider_auth_refresh_attempted = True
|
||||
if agent._try_refresh_provider_runtime_credentials(reason="unauthorized"):
|
||||
agent._buffer_vprint(
|
||||
f"🔐 {profile.display_name or profile.name} credentials refreshed after 401. Retrying request...")
|
||||
return True
|
||||
if (
|
||||
agent.api_mode == "anthropic_messages"
|
||||
|
||||
@@ -23,10 +23,10 @@ class TurnRetryState:
|
||||
# refusal, and one route re-read after a wrong-host refusal (``anon_on_paid_host``).
|
||||
welcome_model_switch_attempted: bool = False
|
||||
welcome_route_heal_attempted: bool = False
|
||||
copilot_auth_retry_attempted: bool = False
|
||||
# Copilot surfaces a stale credential as a 400 ``model_not_available_for_integrator``
|
||||
# / ``model_not_supported``, not a 401 — separate guard from the 401 one.
|
||||
copilot_stale_cred_retry_attempted: bool = False
|
||||
# Provider-profile runtime refresh (``refresh_runtime_credentials``): one 401 refresh, and a
|
||||
# separate one for a non-auth error the profile reads as a stale credential.
|
||||
provider_auth_refresh_attempted: bool = False
|
||||
provider_stale_refresh_attempted: bool = False
|
||||
vertex_auth_retry_attempted: bool = False
|
||||
|
||||
# Format / payload recovery guards
|
||||
|
||||
+29
-77
@@ -376,19 +376,10 @@ def _model_level_key_env(provider_id: str) -> str:
|
||||
|
||||
def _resolve_api_key_provider_secret(provider_id: str, pconfig: ProviderConfig) -> tuple[str, str]:
|
||||
"""Resolve an API-key provider's token and indicate where it came from."""
|
||||
if provider_id == "copilot":
|
||||
# The dedicated copilot auth module does proper token validation/exchange.
|
||||
try:
|
||||
from hermes_cli.copilot_auth import resolve_copilot_token, get_copilot_api_token
|
||||
token, source = resolve_copilot_token()
|
||||
if token:
|
||||
api_token, _base_url = get_copilot_api_token(token)
|
||||
return api_token, source
|
||||
except ValueError as exc:
|
||||
logger.warning("Copilot token validation failed: %s", exc)
|
||||
except Exception:
|
||||
pass
|
||||
return "", ""
|
||||
profile = _credential_profile(provider_id)
|
||||
owned = profile.resolve_api_key_secret() if profile else None
|
||||
if owned is not None: # provider-owned resolution (token validation / exchange) is final
|
||||
return owned
|
||||
|
||||
# Prefer ~/.hermes/.env over os.environ so a deliberate key rotation in .env isn't shadowed by
|
||||
# a stale shell export inherited from a parent process (Codex CLI, test runners, etc.).
|
||||
@@ -1596,11 +1587,17 @@ def _config_model_provider() -> tuple[Any, Optional[str]]:
|
||||
return None, None
|
||||
|
||||
|
||||
# API-key providers never auto-selected from env: GitHub tokens are commonly present for repo/tool
|
||||
# access and must not hijack inference; LM Studio is a local server whose availability isn't
|
||||
# implied by LM_API_KEY (may be offline; no-auth setup uses a placeholder). Both need an explicit
|
||||
# choice.
|
||||
_NO_AUTO_DETECT_PROVIDERS = frozenset({"copilot", "lmstudio"})
|
||||
# API-key providers never auto-selected from env: LM Studio is a local server whose availability
|
||||
# isn't implied by LM_API_KEY (may be offline; no-auth setup uses a placeholder), and a profile with
|
||||
# ``ambient_credentials`` reads tokens commonly present for other purposes (repo/tool access) that
|
||||
# must not hijack inference. Both need an explicit choice.
|
||||
_NO_AUTO_DETECT_PROVIDERS = frozenset({"lmstudio"})
|
||||
|
||||
|
||||
def _credential_profile(provider_id: str) -> Any:
|
||||
"""The provider's registered profile (credential seams), or None."""
|
||||
from providers import get_provider_profile
|
||||
return get_provider_profile(provider_id)
|
||||
|
||||
|
||||
def _env_key_auto_detected(
|
||||
@@ -1610,6 +1607,8 @@ def _env_key_auto_detected(
|
||||
for pid, pconfig in PROVIDER_REGISTRY.items():
|
||||
if pconfig.auth_type != "api_key" or pid in _NO_AUTO_DETECT_PROVIDERS:
|
||||
continue
|
||||
if getattr(_credential_profile(pid), "ambient_credentials", False):
|
||||
continue
|
||||
for env_var in pconfig.api_key_env_vars:
|
||||
if has_usable_secret(scoped_key_env(env_var)):
|
||||
if oauth_active and oauth_active != pid:
|
||||
@@ -2080,51 +2079,18 @@ def _external_process_auth_evidence(provider_id: str, resolved_command: Optional
|
||||
"""Best-effort POSITIVE evidence ``(verified, source)`` that an external-process CLI is authed.
|
||||
|
||||
False means "not verifiable from here", NOT "signed out". Subprocess-free (spawning the CLI from
|
||||
status endpoints/pickers re-creates the cold-start stall copilot_auth.py avoids). Generic evidence
|
||||
status endpoints/pickers re-creates a cold-start stall). Generic evidence
|
||||
for any external-process profile is its binary resolving: the subprocess owns real auth and Hermes
|
||||
has nothing else to inspect, so out-of-tree ACP rows pass credential-gated surfaces (Desktop
|
||||
``explicit_only`` picker) like the bundled one, whose CLI additionally exposes readable token stores."""
|
||||
if provider_id == "copilot-acp":
|
||||
return _copilot_acp_auth_evidence()
|
||||
``explicit_only`` picker); a profile whose CLI exposes readable token stores answers
|
||||
``external_auth_evidence()`` instead."""
|
||||
evidence = _credential_profile(provider_id)
|
||||
evidence = evidence.external_auth_evidence() if evidence else None
|
||||
if evidence is not None:
|
||||
return evidence
|
||||
return (True, f"command: {resolved_command}") if resolved_command else (False, None)
|
||||
|
||||
|
||||
def _copilot_acp_auth_evidence() -> tuple[bool, Optional[str]]:
|
||||
"""Copilot CLI token stores readable without spawning it (env tokens, plaintext config, hosts.json)."""
|
||||
# 1. Supported env tokens — the same vars the Copilot CLI itself honors.
|
||||
try:
|
||||
from hermes_cli.copilot_auth import COPILOT_ENV_VARS, validate_copilot_token
|
||||
for env_var in COPILOT_ENV_VARS:
|
||||
val = os.getenv(env_var, "").strip()
|
||||
if val and validate_copilot_token(val)[0]:
|
||||
return True, f"env: {env_var}"
|
||||
except Exception as exc:
|
||||
logger.debug("copilot-acp env token evidence check failed: %s", exc)
|
||||
# 2. The Copilot CLI's own plaintext token store (written by `copilot login` when no OS keychain
|
||||
# is available). The file is JSONC — strip //-comment lines before parsing.
|
||||
try:
|
||||
cli_config = os.path.expanduser("~/.copilot/config.json")
|
||||
if os.path.isfile(cli_config):
|
||||
with open(cli_config, "r", encoding="utf-8-sig", errors="ignore") as fh:
|
||||
raw = "\n".join(
|
||||
line for line in fh.read().splitlines() if not line.lstrip().startswith("//"))
|
||||
tokens = (json.loads(raw) if raw.strip() else {}).get("copilotTokens")
|
||||
if isinstance(tokens, dict) and any(
|
||||
isinstance(v, str) and v.strip() for v in tokens.values()):
|
||||
return True, "~/.copilot/config.json"
|
||||
except Exception as exc:
|
||||
logger.debug("copilot-acp CLI config evidence check failed: %s", exc)
|
||||
# 3. Known on-disk GitHub Copilot credential stores (the same files models.py fingerprints).
|
||||
for cred_path in ("~/.config/github-copilot/hosts.json", "~/.config/github-copilot/apps.json"):
|
||||
try:
|
||||
expanded = os.path.expanduser(cred_path)
|
||||
if os.path.isfile(expanded) and os.path.getsize(expanded) > 2:
|
||||
return True, cred_path
|
||||
except OSError:
|
||||
continue
|
||||
return False, None
|
||||
|
||||
|
||||
def _external_process_spec(
|
||||
pconfig: ProviderConfig) -> tuple[str, list[str], str, Optional[str], tuple[str, ...]]:
|
||||
"""``(command, args, base_url, resolved_command, command_env_vars)`` for an ACP provider.
|
||||
@@ -2257,29 +2223,12 @@ def _default_api_key_base_url(api_key: str, default: str, env_url: str) -> str:
|
||||
return env_url.rstrip("/") if env_url else default
|
||||
|
||||
|
||||
def _copilot_runtime_base_url(api_key: str, default: str, env_url: str) -> str:
|
||||
"""Copilot's API base comes from the token-exchange response (endpoints.api, proxy-ep fallback),
|
||||
authoritative for Enterprise / proxied accounts; falls back to the registry default."""
|
||||
base_url = _default_api_key_base_url(api_key, default, env_url)
|
||||
try:
|
||||
from hermes_cli.copilot_auth import resolve_copilot_token, get_copilot_api_token
|
||||
raw_token, _ = resolve_copilot_token()
|
||||
if raw_token:
|
||||
resolved = (get_copilot_api_token(raw_token)[1] or "").strip()
|
||||
if resolved:
|
||||
base_url = resolved
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot base URL resolution fell back to default: %s", exc)
|
||||
return base_url
|
||||
|
||||
|
||||
# Providers whose runtime base URL is not simply env-override-or-registry-default:
|
||||
# ``(api_key, registry_default, env_override) -> base_url``.
|
||||
_API_KEY_BASE_URL_RESOLVERS: dict[str, Callable[[str, str, str], str]] = {
|
||||
"kimi-coding": _resolve_kimi_base_url,
|
||||
"kimi-coding-cn": _resolve_kimi_base_url,
|
||||
"zai": _resolve_zai_base_url,
|
||||
"copilot": _copilot_runtime_base_url,
|
||||
"lmstudio": lambda *a: _normalize_lmstudio_runtime_base_url(_default_api_key_base_url(*a)),
|
||||
"actual": lambda *a: normalize_actual_base_url(_default_api_key_base_url(*a))}
|
||||
|
||||
@@ -2301,9 +2250,12 @@ def resolve_api_key_provider_credentials(provider_id: str) -> dict[str, Any]:
|
||||
|
||||
env_url = _provider_env_base_url(pconfig)
|
||||
resolve_url = _API_KEY_BASE_URL_RESOLVERS.get(provider_id, _default_api_key_base_url)
|
||||
base_url = resolve_url(api_key, pconfig.inference_base_url, env_url)
|
||||
profile = _credential_profile(provider_id)
|
||||
base_url = (profile.runtime_base_url(api_key=api_key, default=pconfig.inference_base_url, env_url=env_url)
|
||||
if profile else None)
|
||||
base_url = resolve_url(api_key, pconfig.inference_base_url, env_url) if base_url is None else base_url
|
||||
# An API-key provider must never hand back an empty base URL (a set-but-empty
|
||||
# COPILOT_API_BASE_URL or similar env override otherwise wedges chat inference).
|
||||
# <PROVIDER>_BASE_URL env override otherwise wedges chat inference).
|
||||
if not _nonempty_str(base_url):
|
||||
base_url = pconfig.inference_base_url
|
||||
|
||||
|
||||
@@ -518,7 +518,9 @@ def _copilot_obtain_token() -> bool:
|
||||
return False
|
||||
if choice == "1":
|
||||
try:
|
||||
from hermes_cli.copilot_auth import copilot_device_code_login
|
||||
from providers import get_provider_profile
|
||||
get_provider_profile("copilot") # loads the plugin package that owns these credentials
|
||||
from plugins.model_providers.copilot.credentials import copilot_device_code_login
|
||||
token = copilot_device_code_login()
|
||||
if not token:
|
||||
print(" Login cancelled or failed.")
|
||||
@@ -539,7 +541,9 @@ def _copilot_obtain_token() -> bool:
|
||||
return False
|
||||
# Validate token type
|
||||
with contextlib.suppress(ImportError):
|
||||
from hermes_cli.copilot_auth import validate_copilot_token
|
||||
from providers import get_provider_profile
|
||||
get_provider_profile("copilot") # loads the plugin package that owns these credentials
|
||||
from plugins.model_providers.copilot.credentials import validate_copilot_token
|
||||
valid, msg = validate_copilot_token(new_key)
|
||||
if not valid:
|
||||
print(f" ✗ {msg}")
|
||||
|
||||
@@ -1298,7 +1298,9 @@ def resolve_fast_mode_overrides(
|
||||
def _first_exchangeable_copilot_token(raw_tokens) -> str:
|
||||
"""Exchange stored GitHub tokens in order; the first that validates AND exchanges wins (every
|
||||
entry is tried so a later valid token survives an earlier malformed one)."""
|
||||
from hermes_cli.copilot_auth import exchange_copilot_token, validate_copilot_token
|
||||
from providers import get_provider_profile
|
||||
get_provider_profile("copilot") # loads the plugin package that owns these credentials
|
||||
from plugins.model_providers.copilot.credentials import exchange_copilot_token, validate_copilot_token
|
||||
|
||||
for raw in raw_tokens:
|
||||
raw = str(raw or "").strip()
|
||||
@@ -2276,7 +2278,9 @@ def _payload_items(payload: Any) -> list[dict[str, Any]]:
|
||||
def copilot_default_headers(*, is_agent_turn: bool = True) -> dict[str, str]:
|
||||
"""Standard headers for Copilot API requests."""
|
||||
try:
|
||||
from hermes_cli.copilot_auth import copilot_request_headers
|
||||
from providers import get_provider_profile
|
||||
get_provider_profile("copilot") # loads the plugin package that owns these credentials
|
||||
from plugins.model_providers.copilot.credentials import copilot_request_headers
|
||||
return copilot_request_headers(is_agent_turn=is_agent_turn)
|
||||
except ImportError:
|
||||
return {
|
||||
|
||||
@@ -665,27 +665,6 @@ def _refresh_nous_pool_entry(pool: CredentialPool, entry: Any, pool_api_key: str
|
||||
return entry, pool_api_key
|
||||
|
||||
|
||||
def _exchange_copilot_pool_entry(entry: Any, pool_api_key: str) -> str:
|
||||
"""Exchange a copilot pool entry that still carries the RAW GitHub token.
|
||||
|
||||
The seeder skips the exchange while copilot is merely discovered (ambient gh login, not in
|
||||
config); here copilot IS the runtime target (`/model copilot/… --session`, `--provider copilot`,
|
||||
delegation/cron overrides), and a raw token routes to the language-server integrator whose
|
||||
allowlist omits enterprise-only models (400 model_not_available_for_integrator)."""
|
||||
from hermes_cli.copilot_auth import get_copilot_api_token, validate_copilot_token
|
||||
if not pool_api_key or not validate_copilot_token(pool_api_key)[0]:
|
||||
return pool_api_key # already an exchanged API token
|
||||
api_token, enterprise_base_url = get_copilot_api_token(pool_api_key)
|
||||
if api_token == pool_api_key and not enterprise_base_url:
|
||||
from agent.credential_pool import _warn_copilot_raw_degradation_once
|
||||
_warn_copilot_raw_degradation_once(pool_api_key)
|
||||
return pool_api_key
|
||||
entry.access_token = api_token
|
||||
if enterprise_base_url:
|
||||
entry.base_url = enterprise_base_url
|
||||
return api_token
|
||||
|
||||
|
||||
def _resolve_from_pool(provider: str, requested_provider: str, model_cfg: dict[str, Any], explicit_api_key, explicit_base_url,
|
||||
target_model) -> Optional[dict[str, Any]]:
|
||||
"""Runtime from the provider's credential pool, or None to continue down the ladder."""
|
||||
@@ -703,8 +682,10 @@ def _resolve_from_pool(provider: str, requested_provider: str, model_cfg: dict[s
|
||||
pool_api_key = _pool_entry_api_key(entry)
|
||||
if provider == "nous":
|
||||
entry, pool_api_key = _refresh_nous_pool_entry(pool, entry, pool_api_key)
|
||||
elif provider == "copilot":
|
||||
pool_api_key = _exchange_copilot_pool_entry(entry, pool_api_key)
|
||||
else:
|
||||
from providers import get_provider_profile
|
||||
if (profile := get_provider_profile(provider)) is not None:
|
||||
pool_api_key = profile.prepare_pool_entry(entry, pool_api_key)
|
||||
if not has_usable_secret(pool_api_key):
|
||||
return None
|
||||
if pool_api_key and credential_pool_matches_provider(pool, provider, base_url=_pool_entry_base_url(entry)):
|
||||
@@ -897,7 +878,7 @@ def _anthropic_env_runtime(requested_provider: str, model_cfg: dict[str, Any], t
|
||||
|
||||
|
||||
def _api_key_provider_runtime(provider, pconfig, requested_provider, model_cfg, target_model) -> dict[str, Any]:
|
||||
"""Registry ``api_key`` providers (z.ai/GLM, Kimi, MiniMax, copilot, …) from env/config."""
|
||||
"""Registry ``api_key`` providers (z.ai/GLM, Kimi, MiniMax, …) from env/config."""
|
||||
creds = resolve_api_key_provider_credentials(provider)
|
||||
# Actual Computer: a loopback model_cfg base_url selects the daemon's no-auth local API; inject
|
||||
# the placeholder BEFORE the usable-secret gate (mirrors the env-driven path).
|
||||
|
||||
@@ -148,8 +148,7 @@ def _warm_gateway_module() -> None:
|
||||
"""
|
||||
for mod in (
|
||||
"hermes_cli.gateway",
|
||||
"hermes_cli.auth", # provider auth state → copilot_auth → subprocess
|
||||
"hermes_cli.copilot_auth",
|
||||
"hermes_cli.auth", # provider auth state
|
||||
"hermes_cli.runtime_provider",
|
||||
"hermes_cli.skin_engine", # resolve_skin() config + skin engine init
|
||||
"hermes_cli.inventory", # provider catalogs + models.dev cache
|
||||
|
||||
@@ -15,6 +15,12 @@ from providers.base import ProviderProfile
|
||||
class CopilotACPProfile(ProviderProfile):
|
||||
"""GitHub Copilot ACP — external process, no REST models endpoint."""
|
||||
|
||||
def external_auth_evidence(self) -> tuple[bool, str | None]:
|
||||
"""The Copilot CLI's token stores, read without spawning it."""
|
||||
from plugins.model_providers.copilot.credentials import copilot_cli_auth_evidence
|
||||
|
||||
return copilot_cli_auth_evidence()
|
||||
|
||||
def create_client(self, **client_kwargs: Any) -> Any:
|
||||
"""Build the ACP stdio shim rather than an HTTP client."""
|
||||
from agent.copilot_acp_client import CopilotACPClient
|
||||
|
||||
@@ -10,8 +10,10 @@ from typing import Any
|
||||
from providers import register_provider
|
||||
from providers.base import ProviderProfile
|
||||
|
||||
from .credentials import CopilotCredentialsMixin
|
||||
|
||||
class CopilotProfile(ProviderProfile):
|
||||
|
||||
class CopilotProfile(CopilotCredentialsMixin, ProviderProfile):
|
||||
"""GitHub Copilot / GitHub Models — editor headers + reasoning."""
|
||||
|
||||
def build_api_kwargs_extras(
|
||||
@@ -38,7 +40,7 @@ class CopilotProfile(ProviderProfile):
|
||||
copilot = CopilotProfile(
|
||||
name="copilot", aliases=("github-copilot", "github-models", "github-model", "github"),
|
||||
env_vars=("COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN"), base_url="https://api.githubcopilot.com",
|
||||
auth_type="copilot",
|
||||
auth_type="copilot", ambient_credentials=True,
|
||||
)
|
||||
|
||||
register_provider(copilot)
|
||||
@@ -1,5 +1,6 @@
|
||||
"""GitHub Copilot authentication utilities (credential order matches the Copilot CLI:
|
||||
COPILOT_GITHUB_TOKEN, GH_TOKEN, GITHUB_TOKEN, then ``gh auth token``)."""
|
||||
"""GitHub Copilot credentials: token resolution (order matches the Copilot CLI: COPILOT_GITHUB_TOKEN,
|
||||
GH_TOKEN, GITHUB_TOKEN, then ``gh auth token``), device-code login, the GitHub→Copilot token exchange,
|
||||
and :class:`CopilotCredentialsMixin`, which fills core's generic ``ProviderProfile`` credential seams."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -16,7 +17,7 @@ import time
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
from hermes_cli._subprocess_compat import IS_WINDOWS, windows_hide_flags
|
||||
from utils import atomic_json_write
|
||||
@@ -509,3 +510,262 @@ def copilot_request_headers(
|
||||
if is_vision:
|
||||
headers["Copilot-Vision-Request"] = "true"
|
||||
return headers
|
||||
|
||||
|
||||
# ── ProviderProfile credential seams ────────────────────────────────────────────────────────────
|
||||
|
||||
# Warn once per token per process when the exchange degrades to the raw token (#114740).
|
||||
_RAW_DEGRADATION_WARNED: set[str] = set()
|
||||
|
||||
|
||||
def _warn_raw_degradation_once(token: str) -> None:
|
||||
"""WARN once per token per process when Copilot exchange degrades to raw token (#114740)."""
|
||||
from agent.credential_persistence import fingerprint_secret_value
|
||||
fingerprint = fingerprint_secret_value(token) or "unknown"
|
||||
if fingerprint in _RAW_DEGRADATION_WARNED:
|
||||
return
|
||||
_RAW_DEGRADATION_WARNED.add(fingerprint)
|
||||
logger.warning(
|
||||
"Copilot token exchange degraded to RAW token (exchange "
|
||||
"unavailable); enterprise-only models may 400 with "
|
||||
"model_not_available_for_integrator until exchange recovers."
|
||||
)
|
||||
|
||||
|
||||
def _reset_raw_degradation_warned() -> None:
|
||||
"""Clear the degradation warning cache (for test isolation)."""
|
||||
_RAW_DEGRADATION_WARNED.clear()
|
||||
|
||||
|
||||
# Every pool source the same token can be seeded under (gh_cli + one per env var).
|
||||
_POOL_SOURCES = ("gh_cli",) + tuple(f"env:{v}" for v in COPILOT_ENV_VARS)
|
||||
|
||||
_STALE_CREDENTIAL_MARKERS = (
|
||||
"model_not_available_for_integrator",
|
||||
"not available for integrator",
|
||||
"model_not_supported",
|
||||
"the requested model is not supported",
|
||||
)
|
||||
|
||||
_GITHUB_MODELS_FREE_TIER_HOST = "models.inference.ai.azure.com"
|
||||
_GITHUB_MODELS_HINT = (
|
||||
" 💡 GitHub Models free tier (models.inference.ai.azure.com) caps every",
|
||||
" request at ~8K tokens. Hermes' system prompt + tool schemas baseline",
|
||||
" exceeds that floor, so this endpoint cannot run an agentic loop.",
|
||||
" Use the `copilot` provider with a Copilot subscription token (`hermes",
|
||||
" setup` → GitHub Copilot), or pick any other provider.",
|
||||
)
|
||||
|
||||
|
||||
def _refresh_exchanged_token() -> Optional[tuple[str, Optional[str]]]:
|
||||
"""401 on the main client: the raw GitHub token is stable, the short-TTL *exchanged* IDE token is
|
||||
what expires mid-turn (``401 IDE token expired``), so force a fresh exchange; fall back to the
|
||||
raw token only if the exchange itself is unavailable."""
|
||||
try:
|
||||
new_token, token_source = resolve_copilot_token()
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot credential refresh failed: %s", exc)
|
||||
return None
|
||||
new_token, enterprise_base_url = (new_token.strip() if isinstance(new_token, str) else ""), None
|
||||
if not new_token:
|
||||
return None
|
||||
try:
|
||||
evict_cached_exchanged_token(new_token)
|
||||
api_token, exchanged_base_url = get_copilot_api_token(new_token)
|
||||
if isinstance(api_token, str) and api_token.strip():
|
||||
new_token, enterprise_base_url = api_token.strip(), exchanged_base_url
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot 401 re-exchange failed, using resolved token: %s", exc)
|
||||
logger.info("Copilot credentials refreshed from %s", token_source)
|
||||
return new_token, enterprise_base_url
|
||||
|
||||
|
||||
def _reexchange_stale_token() -> Optional[tuple[str, Optional[str]]]:
|
||||
"""Stale-credential 400 (``model_not_available_for_integrator``): typically a raw ``ghu_`` token
|
||||
cached when the startup exchange degraded. Evict + re-exchange; None when still degraded (a
|
||||
rebuild would not help, so the caller's single-shot retry is not burned)."""
|
||||
try:
|
||||
raw_token, token_source = resolve_copilot_token()
|
||||
if not isinstance(raw_token, str) or not raw_token.strip():
|
||||
return None
|
||||
raw_token = raw_token.strip()
|
||||
evict_cached_exchanged_token(raw_token)
|
||||
api_token, enterprise_base_url = get_copilot_api_token(raw_token)
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot stale-credential recovery failed: %s", exc)
|
||||
return None
|
||||
if not isinstance(api_token, str) or not api_token.strip():
|
||||
return None
|
||||
if api_token == raw_token and not enterprise_base_url:
|
||||
logger.warning(
|
||||
"Copilot stale-credential recovery: exchange still degraded to raw token; skipping retry "
|
||||
"(network/exchange endpoint unavailable)."
|
||||
)
|
||||
return None
|
||||
logger.info("Copilot credentials re-exchanged after stale-credential 400 (source=%s)", token_source)
|
||||
return api_token.strip(), enterprise_base_url
|
||||
|
||||
|
||||
def _revalidate_in_memory_token() -> Optional[tuple[str, Optional[str]]]:
|
||||
"""Auxiliary-client 401: drop only the in-process JWT and exchange again (a persisted, still-valid
|
||||
JWT and the negative cache are honoured); raises when the exchange fails."""
|
||||
raw_token, _source = resolve_copilot_token()
|
||||
if not str(raw_token or "").strip():
|
||||
return None
|
||||
_jwt_cache.pop(_token_fingerprint(raw_token), None)
|
||||
api_token, _expires_at, base_url = exchange_copilot_token(raw_token)
|
||||
return api_token, base_url
|
||||
|
||||
|
||||
_RUNTIME_REFRESHERS = {
|
||||
"unauthorized": _refresh_exchanged_token,
|
||||
"stale": _reexchange_stale_token,
|
||||
"auxiliary_unauthorized": _revalidate_in_memory_token,
|
||||
}
|
||||
|
||||
|
||||
class CopilotCredentialsMixin:
|
||||
"""Fills the generic credential seams of ``providers.base.ProviderProfile`` for Copilot."""
|
||||
|
||||
def resolve_api_key_secret(self) -> tuple[str, str]:
|
||||
try:
|
||||
token, source = resolve_copilot_token()
|
||||
if token:
|
||||
api_token, _base_url = get_copilot_api_token(token)
|
||||
return api_token, source
|
||||
except ValueError as exc:
|
||||
logger.warning("Copilot token validation failed: %s", exc)
|
||||
except Exception:
|
||||
pass
|
||||
return "", ""
|
||||
|
||||
def runtime_base_url(self, *, api_key: str, default: str, env_url: str) -> str:
|
||||
"""The API base comes from the token-exchange response (endpoints.api, proxy-ep fallback),
|
||||
authoritative for Enterprise / proxied accounts; falls back to the registry default."""
|
||||
base_url = env_url.rstrip("/") if env_url else default
|
||||
try:
|
||||
raw_token, _ = resolve_copilot_token()
|
||||
if raw_token:
|
||||
resolved = (get_copilot_api_token(raw_token)[1] or "").strip()
|
||||
if resolved:
|
||||
base_url = resolved
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot base URL resolution fell back to default: %s", exc)
|
||||
return base_url
|
||||
|
||||
def prepare_pool_entry(self, entry: Any, api_key: str) -> str:
|
||||
"""A pool row may still carry the RAW GitHub token: the seeder skips the exchange while
|
||||
copilot is merely discovered (ambient gh login, not in config). Here copilot IS the runtime
|
||||
target, and a raw token routes to the language-server integrator whose allowlist omits
|
||||
enterprise-only models (400 model_not_available_for_integrator)."""
|
||||
if not api_key or not validate_copilot_token(api_key)[0]:
|
||||
return api_key # already an exchanged API token
|
||||
api_token, enterprise_base_url = get_copilot_api_token(api_key)
|
||||
if api_token == api_key and not enterprise_base_url:
|
||||
_warn_raw_degradation_once(api_key)
|
||||
return api_key
|
||||
entry.access_token = api_token
|
||||
if enterprise_base_url:
|
||||
entry.base_url = enterprise_base_url
|
||||
return api_token
|
||||
|
||||
def seed_pool_credentials(self, *, is_suppressed: Callable[[str], bool],
|
||||
explicitly_configured: Callable[[], bool]) -> list[dict]:
|
||||
# Tokens are resolved dynamically via env vars or `gh auth token`; they don't live in the
|
||||
# auth store. All-sources gate BEFORE any work: resolve_copilot_token() shells out and the
|
||||
# exchange retries 3x with backoff (~35s worst case); a user who suppressed every copilot
|
||||
# source must not pay that on every pool load.
|
||||
try:
|
||||
if all(is_suppressed(s) for s in _POOL_SOURCES):
|
||||
return []
|
||||
token, source = resolve_copilot_token()
|
||||
if not token:
|
||||
return []
|
||||
# Exact match: a substring test would classify GH_TOKEN/GITHUB_TOKEN as gh_cli and
|
||||
# bypass a user's per-env-var suppression.
|
||||
source_name = "gh_cli" if source == "gh auth token" else f"env:{source}"
|
||||
if is_suppressed(source_name): # per-source gate BEFORE the network exchange
|
||||
return []
|
||||
if not explicitly_configured():
|
||||
# Only discovered here (ambient gh CLI login), not selected anywhere: no model is
|
||||
# routed to it, so the exchange — and its degradation warning on every pool load —
|
||||
# buys nothing (#114740). Seed the raw token; selecting copilot re-seeds and exchanges.
|
||||
api_token, enterprise_base_url = token, None
|
||||
else:
|
||||
api_token, enterprise_base_url = get_copilot_api_token(token)
|
||||
# A failed exchange falls back to the RAW token, which the Copilot API routes to the
|
||||
# "copilot-language-server" integrator whose allowlist omits enterprise-only models
|
||||
# -> HTTP 400 on every turn. Surface it once.
|
||||
if api_token == token and not enterprise_base_url:
|
||||
_warn_raw_degradation_once(token)
|
||||
return [{"source": source_name, "access_token": api_token, "base_url": enterprise_base_url,
|
||||
"label": source}]
|
||||
except Exception as exc:
|
||||
logger.debug("Copilot token seed failed: %s", exc)
|
||||
return []
|
||||
|
||||
def remove_credential_source(self, source: str) -> Optional[tuple[tuple[str, ...], tuple[str, ...]]]:
|
||||
"""The same token is seeded as gh_cli AND env:<VAR> rows, so suppress every variant or the
|
||||
duplicates resurrect the entry. gh CLI and shell state are left untouched."""
|
||||
if source != "gh_cli" and not source.startswith("env:"):
|
||||
return None
|
||||
return _POOL_SOURCES, (
|
||||
"Suppressed all copilot token sources (gh_cli + env vars) — they will not be re-seeded.",
|
||||
"Note: Your gh CLI / shell environment is unchanged.",
|
||||
"Run `hermes auth add copilot` to re-enable if needed.",
|
||||
)
|
||||
|
||||
def refresh_runtime_credentials(self, *, reason: str) -> Optional[tuple[str, Optional[str]]]:
|
||||
refresher = _RUNTIME_REFRESHERS.get(reason)
|
||||
return refresher() if refresher else None
|
||||
|
||||
def is_stale_credential_error(self, *, status_code: Optional[int], message: str) -> bool:
|
||||
"""A 400 that is really a STALE / DEGRADED credential (status 400 AND an integrator /
|
||||
model-not-supported marker, so a wrong model name never triggers the re-exchange)."""
|
||||
lowered = (message or "").lower()
|
||||
if status_code != 400 and "error code: 400" not in lowered:
|
||||
return False
|
||||
return any(marker in lowered for marker in _STALE_CREDENTIAL_MARKERS)
|
||||
|
||||
def request_rejection_hint(self, *, status_code: Optional[int], base_url: str) -> tuple[str, ...]:
|
||||
"""GitHub Models free tier caps requests at 8K tokens, under the system prompt + tool schema
|
||||
floor; compression can't help, so say so."""
|
||||
from utils import base_url_host_matches
|
||||
if status_code == 413 and base_url_host_matches(base_url, _GITHUB_MODELS_FREE_TIER_HOST):
|
||||
return _GITHUB_MODELS_HINT
|
||||
return ()
|
||||
|
||||
|
||||
def copilot_cli_auth_evidence() -> tuple[bool, Optional[str]]:
|
||||
"""Copilot CLI token stores readable without spawning it (env tokens, plaintext config, hosts.json)."""
|
||||
# 1. Supported env tokens — the same vars the Copilot CLI itself honors.
|
||||
try:
|
||||
for env_var in COPILOT_ENV_VARS:
|
||||
val = os.getenv(env_var, "").strip()
|
||||
if val and validate_copilot_token(val)[0]:
|
||||
return True, f"env: {env_var}"
|
||||
except Exception as exc:
|
||||
logger.debug("copilot-acp env token evidence check failed: %s", exc)
|
||||
# 2. The Copilot CLI's own plaintext token store (written by `copilot login` when no OS keychain
|
||||
# is available). The file is JSONC — strip //-comment lines before parsing.
|
||||
try:
|
||||
cli_config = os.path.expanduser("~/.copilot/config.json")
|
||||
if os.path.isfile(cli_config):
|
||||
with open(cli_config, "r", encoding="utf-8-sig", errors="ignore") as fh:
|
||||
raw = "\n".join(
|
||||
line for line in fh.read().splitlines() if not line.lstrip().startswith("//"))
|
||||
tokens = (json.loads(raw) if raw.strip() else {}).get("copilotTokens")
|
||||
if isinstance(tokens, dict) and any(
|
||||
isinstance(v, str) and v.strip() for v in tokens.values()):
|
||||
return True, "~/.copilot/config.json"
|
||||
except Exception as exc:
|
||||
logger.debug("copilot-acp CLI config evidence check failed: %s", exc)
|
||||
# 3. Known on-disk GitHub Copilot credential stores (the same files models.py fingerprints).
|
||||
for cred_path in ("~/.config/github-copilot/hosts.json", "~/.config/github-copilot/apps.json"):
|
||||
try:
|
||||
expanded = os.path.expanduser(cred_path)
|
||||
if os.path.isfile(expanded) and os.path.getsize(expanded) > 2:
|
||||
return True, cred_path
|
||||
except OSError:
|
||||
continue
|
||||
return False, None
|
||||
@@ -536,3 +536,47 @@ class ProviderProfile:
|
||||
except Exception as exc:
|
||||
logger.debug("fetch_models(%s): %s", self.name, exc)
|
||||
return None
|
||||
|
||||
# ── Lane L2: provider-owned credential seams ──────────────────────────────
|
||||
# Credentials are commonly ambient (another tool's login, a token exported for other purposes):
|
||||
# never auto-detected as the inference provider, and the auxiliary fallback chain skips the
|
||||
# provider unless the user explicitly configured it.
|
||||
ambient_credentials: bool = field(default=False, kw_only=True)
|
||||
|
||||
def resolve_api_key_secret(self) -> tuple[str, str] | None:
|
||||
"""Provider-owned ``(runtime_api_key, source)`` resolution; None = generic env/pool ladder."""
|
||||
return None
|
||||
|
||||
def runtime_base_url(self, *, api_key: str, default: str, env_url: str) -> str | None:
|
||||
"""Runtime base URL derived from the credential (e.g. an account endpoint); None = env or default."""
|
||||
return None
|
||||
|
||||
def prepare_pool_entry(self, entry: Any, api_key: str) -> str:
|
||||
"""Turn the selected pool row's secret into the runtime key (may update ``entry``); default as-is."""
|
||||
return api_key
|
||||
|
||||
def seed_pool_credentials(self, *, is_suppressed: Callable[[str], bool],
|
||||
explicitly_configured: Callable[[], bool]) -> list[dict] | None:
|
||||
"""Pool rows ``{source, access_token, base_url, label}`` this provider seeds itself; None = generic env seeding."""
|
||||
return None
|
||||
|
||||
def remove_credential_source(self, source: str) -> tuple[tuple[str, ...], tuple[str, ...]] | None:
|
||||
"""``hermes auth remove`` of *source*: ``(sources_to_suppress, hints)``; None = generic removal."""
|
||||
return None
|
||||
|
||||
def refresh_runtime_credentials(self, *, reason: str) -> tuple[str, str | None] | None:
|
||||
"""Re-mint the live client's ``(api_key, base_url or None)`` on ``reason`` (``unauthorized`` |
|
||||
``stale`` | ``auxiliary_unauthorized``); None = nothing fresher. Error path only, I/O allowed."""
|
||||
return None
|
||||
|
||||
def is_stale_credential_error(self, *, status_code: int | None, message: str) -> bool:
|
||||
"""True when a non-auth error really means the credential went stale (pure, no I/O)."""
|
||||
return False
|
||||
|
||||
def request_rejection_hint(self, *, status_code: int | None, base_url: str) -> tuple[str, ...]:
|
||||
"""User-facing lines explaining an endpoint's hard rejection of *any* route (pure); default none."""
|
||||
return ()
|
||||
|
||||
def external_auth_evidence(self) -> tuple[bool, str | None] | None:
|
||||
"""Subprocess-free ``(verified, source)`` login evidence for an external CLI; None = binary resolves."""
|
||||
return None
|
||||
Reference in new issue
Block a user