Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions run_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -4505,6 +4505,16 @@ def _swap_credential(self, entry) -> None:
runtime_key = getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
runtime_base = getattr(entry, "runtime_base_url", None) or getattr(entry, "base_url", None) or self.base_url

# For Copilot, refresh both token and base_url from a live exchange to avoid
# reusing a stale public endpoint with a valid Enterprise credential (#61746).
if self.provider == "copilot":
try:
from hermes_cli.copilot_auth import get_copilot_api_token
runtime_key, runtime_base = get_copilot_api_token(entry.access_token or "")
except Exception:
# Fall back to the existing entry values if exchange fails.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

entry.access_token is normally already the exchanged Copilot API token (agent/credential_pool.py:1931-1949). Re-exchange failure is swallowed by get_copilot_api_token() and returns (same_token, None) (hermes_cli/copilot_auth.py:415-434), so this assignment clears the selected endpoint instead of reaching the except fallback. Preserve/derive the selected endpoint for exchanged tokens.

pass

if self.api_mode == "anthropic_messages":
from agent.anthropic_adapter import build_anthropic_client, _is_oauth_token

Expand Down
106 changes: 106 additions & 0 deletions tests/agent/test_credential_pool_routing.py
Original file line number Diff line number Diff line change
Expand Up @@ -235,3 +235,109 @@ def test_no_pool_returns_false(self):
)
assert recovered is False
assert has_retried is False


# ---------------------------------------------------------------------------
# 6. Copilot credential rotation refreshes both token and base_url
# ---------------------------------------------------------------------------


class TestCopilotCredentialRefresh:
"""Copilot-specific: _swap_credential fetches fresh token+base_url together."""

def _make_copilot_entry(self, base_url: str) -> MagicMock:
"""Create a mock pool entry for Copilot."""
entry = MagicMock()
entry.access_token = "gho_test_token"
entry.runtime_api_key = None
entry.runtime_base_url = None
entry.base_url = base_url
return entry

@patch("hermes_cli.copilot_auth.get_copilot_api_token")
def test_copilot_swap_refreshes_token_and_base_url(self, mock_exchange):
"""When provider is copilot, _swap_credential calls live token exchange."""
from run_agent import AIAgent

# Mock live exchange returning Enterprise endpoint
mock_exchange.return_value = ("exchanged_jwt", "https://api.enterprise.githubcopilot.com")

with patch.object(AIAgent, "__init__", lambda self, **kw: None):
agent = AIAgent()
agent.provider = "copilot"
agent.api_mode = "openai"
agent._client_kwargs = {}
agent.base_url = "https://api.githubcopilot.com" # initial public endpoint
agent._replace_primary_openai_client = MagicMock()

# Pool entry with stale public endpoint
stale_entry = self._make_copilot_entry("https://api.githubcopilot.com")

# Swap credential - should trigger live exchange
agent._swap_credential(stale_entry)

# Verify live exchange was called with the entry's raw token
mock_exchange.assert_called_once_with("gho_test_token")

# Verify the client was updated with fresh Enterprise endpoint
assert agent.api_key == "exchanged_jwt"
assert agent.base_url == "https://api.enterprise.githubcopilot.com"
assert agent._client_kwargs["api_key"] == "exchanged_jwt"
assert agent._client_kwargs["base_url"] == "https://api.enterprise.githubcopilot.com"

@patch("hermes_cli.copilot_auth.get_copilot_api_token")
def test_copilot_swap_falls_back_on_exchange_failure(self, mock_exchange):
"""If Copilot exchange fails, _swap_credential falls back to entry values."""
from run_agent import AIAgent

# Simulate exchange failure
mock_exchange.side_effect = ValueError("network error")

with patch.object(AIAgent, "__init__", lambda self, **kw: None):
agent = AIAgent()
agent.provider = "copilot"
agent.api_mode = "openai"
agent._client_kwargs = {}
agent.base_url = "https://api.githubcopilot.com"
agent._replace_primary_openai_client = MagicMock()

# Pool entry
entry = self._make_copilot_entry("https://api.githubcopilot.com")

# Swap credential - should attempt exchange, fall back on failure
agent._swap_credential(entry)

# Verify exchange was attempted
mock_exchange.assert_called_once_with("gho_test_token")

# Verify fallback to entry values occurred
assert agent.api_key == "gho_test_token" # from entry.access_token
assert agent.base_url == "https://api.githubcopilot.com"

def test_non_copilot_swap_does_not_call_exchange(self):
"""For non-Copilot providers, _swap_credential uses entry values directly."""
from run_agent import AIAgent

with patch.object(AIAgent, "__init__", lambda self, **kw: None):
agent = AIAgent()
agent.provider = "openai" # Not Copilot
agent.api_mode = "openai"
agent._client_kwargs = {}
agent.base_url = "https://api.openai.com"
agent._replace_primary_openai_client = MagicMock()

# Entry
entry = MagicMock()
entry.access_token = "sk_test"
entry.runtime_api_key = None
entry.runtime_base_url = None
entry.base_url = "https://api.openai.com"

# Swap - should NOT call Copilot exchange
with patch("hermes_cli.copilot_auth.get_copilot_api_token") as mock_exchange:
agent._swap_credential(entry)
mock_exchange.assert_not_called()

# Verify entry values were used
assert agent.api_key == "sk_test"
assert agent.base_url == "https://api.openai.com"
Loading