fix(models): memoize the Copilot ACP session probe; source it from the provider profile

`/model <x>` onto copilot-acp validates through `models_validate._static_catalog`, which
reads `provider_model_ids` with no disk cache. After the session probe landed, every such
switch spawned `copilot --acp`, ran the handshake, and killed it (1-3 s; up to the 15 s
probe timeout when the CLI is installed but the session stalls). The GitHub-API tier that
path used before sat behind a 5-minute in-memory memo; the ACP tier now has the same memo,
and it remembers failures too so a broken CLI is not re-spawned per switch.

The probe itself moves to `CopilotACPProfile.fetch_models` — the slot that already said
"model listing is handled by the ACP subprocess" and returned None — so hermes_cli/models.py
no longer hand-builds `CopilotACPClient` kwargs that `profile.create_client` owns.
Discovery failures are logged at debug instead of swallowed.

Tests: the two picker wiring tests collapse into one parametrized contract; a new test
proves three consecutive switch validations pay one probe and a failed probe is not retried
(fails when the memo read is removed).
This commit is contained in:
kshitijk4poor
2026-09-12 11:56:05 +05:30
committed by kshitij
parent 0cd897286a
commit d5ceff958d
4 changed files with 100 additions and 68 deletions

View File

@@ -1235,24 +1235,34 @@ def _codex_catalog(normalized: str, force_refresh: bool) -> list[str]:
return get_codex_model_ids(access_token=access_token)
def _copilot_catalog(normalized: str, force_refresh: bool) -> Optional[list[str]]:
if normalized == "copilot-acp":
try:
from agent.copilot_acp_client import CopilotACPClient
from hermes_cli.auth import resolve_external_process_provider_credentials
_COPILOT_ACP_SESSION_MEMO_TTL = 300.0 # 5 min, same as the GitHub catalog memo; SWR disk cache handles the rest
_copilot_acp_session_memo: Optional[tuple[float, Optional[list[str]]]] = None
credentials = resolve_external_process_provider_credentials("copilot-acp")
if str(credentials.get("base_url") or "").startswith("acp://"):
live = CopilotACPClient(
api_key=credentials.get("api_key"),
base_url=credentials.get("base_url"),
command=credentials.get("command"),
args=credentials.get("args"),
).list_models()
if live:
return live
except Exception:
pass
def _copilot_acp_session_models(force_refresh: bool) -> Optional[list[str]]:
"""Enabled models from a signed-in ``copilot --acp`` session, memoized for a few minutes —
successes AND failures. Model-switch validation (``models_validate._static_catalog``) reads
this uncached on every ``/model`` switch, and each miss is a CLI spawn + handshake (up to the
probe timeout), so without the memo every switch paid a subprocess."""
global _copilot_acp_session_memo
now = time.monotonic()
memo = _copilot_acp_session_memo
if not force_refresh and memo is not None and now - memo[0] < _COPILOT_ACP_SESSION_MEMO_TTL:
return memo[1]
from providers import get_provider_profile
try:
live = get_provider_profile("copilot-acp").fetch_models() or None
except Exception:
logger.debug("copilot-acp session model discovery failed", exc_info=True)
live = None
_copilot_acp_session_memo = (now, live)
return live
def _copilot_catalog(normalized: str, force_refresh: bool) -> Optional[list[str]]:
if normalized == "copilot-acp" and (live := _copilot_acp_session_models(force_refresh)):
return live
try:
live = _fetch_github_models(_resolve_copilot_catalog_api_key())
if live:

View File

@@ -21,6 +21,26 @@ class CopilotACPProfile(ProviderProfile):
return CopilotACPClient(**client_kwargs)
def fetch_models(
self, *, api_key: str | None = None, base_url: str | None = None, timeout: float = 15.0
) -> list[str] | None:
"""Enabled models advertised by a short-lived signed-in ACP session (``session/new``).
The CLI may keep its login in an OS credential store with no token Hermes can reuse, so
the session is the only source that reflects the account's enablement. ``api_key`` /
``base_url`` are ignored: the subprocess owns auth. None when the CLI is missing, refuses
``--acp``, or the probe fails/times out — callers fall back to their next source.
"""
from hermes_cli.auth import resolve_external_process_provider_credentials
creds = resolve_external_process_provider_credentials(self.name)
if not str(creds.get("base_url") or "").startswith("acp://"):
return None
client = self.create_client(
api_key=creds.get("api_key"), base_url=creds.get("base_url"),
command=creds.get("command"), args=creds.get("args"))
return client.list_models(timeout_seconds=timeout) or None
copilot_acp = CopilotACPProfile(
name="copilot-acp", aliases=("github-copilot-acp", "copilot-acp-agent"),
@@ -28,7 +48,6 @@ copilot_acp = CopilotACPProfile(
env_vars=(), # Managed by ACP subprocess
base_url="acp://copilot", # ACP internal scheme
auth_type="external_process",
supports_model_listing=False, # model listing is handled by the ACP subprocess
# How to launch the CLI; env var names predate this profile (formerly hardcoded in
# hermes_cli/auth.py), so existing setups keep working.
process_command="copilot",

View File

@@ -7,6 +7,7 @@ import pytest
from hermes_cli.model_switch import list_authenticated_providers
from hermes_cli import model_switch_providers
from hermes_cli import models
from hermes_cli.models import provider_model_ids
@@ -86,53 +87,55 @@ def test_copilot_acp_hidden_when_executable_missing(monkeypatch, _no_other_copil
"copilot-acp must stay hidden when no executable resolves"
def test_copilot_acp_catalog_comes_from_authenticated_session_without_token():
session_models = ["auto", "gpt-5.6-sol", "claude-sonnet-5"]
credentials = {
"api_key": "copilot-acp",
"base_url": "acp://copilot",
"command": "copilot",
"args": ["--acp", "--stdio"],
}
with patch(
"hermes_cli.auth.resolve_external_process_provider_credentials",
return_value=credentials,
), patch(
"agent.copilot_acp_client.CopilotACPClient.list_models",
return_value=session_models,
) as list_models, patch(
"hermes_cli.models._resolve_copilot_catalog_api_key",
return_value="",
), patch(
"hermes_cli.models._fetch_github_models",
return_value=[],
) as github_models:
assert provider_model_ids("copilot-acp", force_refresh=True) == session_models
list_models.assert_called_once_with()
github_models.assert_not_called()
_ACP_CREDS = {"api_key": "copilot-acp", "base_url": "acp://copilot", "command": "copilot", "args": ["--acp", "--stdio"]}
def test_copilot_acp_catalog_falls_back_when_session_probe_fails():
credentials = {
"api_key": "copilot-acp",
"base_url": "acp://copilot",
"command": "copilot",
"args": ["--acp", "--stdio"],
}
@pytest.fixture()
def _fresh_acp_memo(monkeypatch):
monkeypatch.setattr(models, "_copilot_acp_session_memo", None)
with patch(
"hermes_cli.auth.resolve_external_process_provider_credentials",
return_value=credentials,
), patch(
"agent.copilot_acp_client.CopilotACPClient.list_models",
side_effect=TimeoutError("probe timeout"),
), patch(
"hermes_cli.models._resolve_copilot_catalog_api_key",
return_value="catalog-token",
), patch(
"hermes_cli.models._fetch_github_models",
return_value=["api-fallback-model"],
):
assert provider_model_ids("copilot-acp", force_refresh=True) == ["api-fallback-model"]
@pytest.mark.parametrize(
("session_probe", "github_token", "github_models", "expected"),
[
# Signed-in session, no GitHub token anywhere: the session list wins, the API is never asked.
({"return_value": ["auto", "gpt-5.6-sol", "claude-sonnet-5"]}, "", [], ["auto", "gpt-5.6-sol", "claude-sonnet-5"]),
# Session probe fails: token-based GitHub discovery is still the next source.
({"side_effect": TimeoutError("probe timeout")}, "catalog-token", ["api-fallback-model"], ["api-fallback-model"]),
],
ids=["session-wins-without-token", "github-fallback-when-probe-fails"],
)
def test_copilot_acp_catalog_prefers_authenticated_session(
_fresh_acp_memo, session_probe, github_token, github_models, expected):
with patch("hermes_cli.auth.resolve_external_process_provider_credentials", return_value=_ACP_CREDS), \
patch("agent.copilot_acp_client.CopilotACPClient.list_models", **session_probe) as list_models, \
patch("hermes_cli.models._resolve_copilot_catalog_api_key", return_value=github_token), \
patch("hermes_cli.models._fetch_github_models", return_value=github_models) as github:
assert provider_model_ids("copilot-acp", force_refresh=True) == expected
list_models.assert_called_once()
assert github.called is bool(github_token)
def test_copilot_acp_session_probe_is_memoized_across_model_switch_validation(_fresh_acp_memo):
"""``/model`` validation reads the catalog uncached on every switch; each miss spawns the CLI.
A run of switches must pay one probe, and a failed probe must not be retried per switch."""
from hermes_cli.models_validate import validate_requested_model
with patch("hermes_cli.auth.resolve_external_process_provider_credentials", return_value=_ACP_CREDS), \
patch("agent.copilot_acp_client.CopilotACPClient.list_models", return_value=["gpt-5.6-terra"]) as list_models, \
patch("hermes_cli.models._resolve_copilot_catalog_api_key", return_value=""), \
patch("hermes_cli.models._fetch_github_models", return_value=[]):
for _ in range(3):
verdict = validate_requested_model("gpt-5.6-terra", "copilot-acp", api_key="copilot-acp", base_url="acp://copilot")
assert verdict["accepted"] and verdict["recognized"]
assert list_models.call_count == 1
models._copilot_acp_session_memo = None
with patch("hermes_cli.auth.resolve_external_process_provider_credentials", return_value=_ACP_CREDS), \
patch("agent.copilot_acp_client.CopilotACPClient.list_models", side_effect=RuntimeError("not signed in")) as list_models, \
patch("hermes_cli.models._resolve_copilot_catalog_api_key", return_value=""), \
patch("hermes_cli.models._fetch_github_models", return_value=[]):
for _ in range(3):
provider_model_ids("copilot-acp")
assert list_models.call_count == 1

View File

@@ -213,12 +213,12 @@ class TestModelPickerBaseUrlIntegration:
def test_profiles_without_model_listing_never_hit_the_network():
"""SDK/subprocess-backed profiles (bedrock, vertex, copilot-acp) still carry a base_url the
generic fetch_models would happily GET ``/models`` against; the flag must short-circuit first."""
"""SDK-backed profiles (bedrock, vertex) still carry a base_url the generic fetch_models
would happily GET ``/models`` against; the flag must short-circuit first."""
from providers import list_providers
flagged = [p for p in list_providers() if not p.supports_model_listing]
assert {p.name for p in flagged} >= {"bedrock", "vertex", "copilot-acp"}
assert {p.name for p in flagged} >= {"bedrock", "vertex"}
with patch("hermes_cli.urllib_security.open_credentialed_url") as opener:
for profile in flagged:
assert profile.fetch_models(api_key="k", base_url=profile.base_url) is None, profile.name