Skip to content
Merged
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
126 changes: 126 additions & 0 deletions acp_adapter/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,110 @@

logger = logging.getLogger(__name__)


def _named_custom_provider_catalogs() -> list[tuple[str, str, list[tuple[str, str]]]]:
"""Return ``(slug, label, [(model_id, description), ...])`` for named endpoints.

Covers both the v12 ``providers:`` mapping and the legacy
``custom_providers:`` list. These endpoints never appear in canonical
provider enumeration, so without this the ACP model selector hides every
named endpoint that the TUI ``/model`` picker already renders (#47039
implemented named-endpoint rows for the TUI surface only).

Model lists come from the entry's declared models (``default_model`` +
``models``), refreshed from the endpoint's live ``/models`` listing when a
credential is available and ``discover_models`` is not disabled. Declared
models are kept even when live discovery fails — some OpenAI-compatible
endpoints (e.g. Bedrock Mantle Responses) expose no ``/models`` route at
all yet serve the declared models fine.

Slugs use the ``custom:<name>`` shape that ``parse_model_input`` and
``resolve_runtime_provider`` already resolve, so encoded choice ids
(``custom:<name>:<model>``) round-trip through ``set_session_model``
unchanged.
"""
try:
from hermes_cli.config import (
get_compatible_custom_providers,
is_provider_enabled,
load_config,
)
from hermes_cli.models import fetch_api_models
except ImportError:
return []

try:
cfg = load_config()
entries = get_compatible_custom_providers(cfg)
except Exception:
logger.debug("Could not load named custom providers", exc_info=True)
return []

# ``get_compatible_custom_providers`` drops the ``enabled`` flag during
# normalization, so collect explicitly disabled provider keys from the
# raw config and skip their entries below.
disabled_keys: set[str] = set()
raw_providers = cfg.get("providers") if isinstance(cfg, dict) else None
if isinstance(raw_providers, dict):
for raw_key, raw_entry in raw_providers.items():
if isinstance(raw_entry, dict) and not is_provider_enabled(raw_entry):
disabled_keys.add(str(raw_key).strip().lower())

catalogs: list[tuple[str, str, list[tuple[str, str]]]] = []
for entry in entries:
if not isinstance(entry, dict):
continue
provider_key = str(entry.get("provider_key", "") or "").strip()
if provider_key.lower() in disabled_keys:
continue
name = str(entry.get("name", "") or "").strip()
base_url = str(entry.get("base_url", "") or "").strip()
if not name or not base_url:
continue
slug_source = provider_key or name
slug = "custom:" + slug_source.strip().lower().replace(" ", "-")

api_key = str(entry.get("api_key", "") or "").strip()
if not api_key:
key_env = str(entry.get("key_env", "") or "").strip()
api_key = os.environ.get(key_env, "").strip() if key_env else ""

declared: list[str] = []
default_model = str(entry.get("model", "") or "").strip()
if default_model:
declared.append(default_model)
models_cfg = entry.get("models")
if isinstance(models_cfg, dict):
for mid in models_cfg:
mid = str(mid or "").strip()
if mid and mid not in declared:
declared.append(mid)

if not api_key and not declared:
# No credential to discover with and nothing declared:
# not addressable from the selector.
continue

model_ids = list(declared)
discover = entry.get("discover_models", True)
if isinstance(discover, str):
discover = discover.lower() not in {"false", "no", "0"}
if discover and api_key:
try:
live = fetch_api_models(
api_key, base_url, api_mode=entry.get("api_mode")
)
except Exception:
live = None
if live:
model_ids = declared + [m for m in live if m not in declared]

if not model_ids:
continue
catalogs.append((slug, name, [(mid, "") for mid in model_ids]))

return catalogs

try:
from hermes_cli import __version__ as HERMES_VERSION
except Exception:
Expand Down Expand Up @@ -618,6 +722,28 @@ def _build_model_state(self, state: SessionState) -> SessionModelState | None:
)
seen_ids.add(choice_id)

# Named user-defined endpoints (providers: / custom_providers:)
# are invisible to canonical provider enumeration — append them
# so editor clients can select them like the TUI /model picker.
for named_slug, named_label, named_catalog in _named_custom_provider_catalogs():
for named_model, named_desc in named_catalog:
named_choice = self._encode_model_choice(named_slug, named_model)
if not named_choice or named_choice in seen_ids:
continue
named_parts = [f"Provider: {named_label}"]
if named_desc:
named_parts.append(str(named_desc).strip())
if named_slug == normalized_provider and named_model == model:
named_parts.append("current")
available_models.append(
ModelInfo(
model_id=named_choice,
name=named_model,
description=" • ".join(part for part in named_parts if part),
)
)
seen_ids.add(named_choice)

current_model_id = self._encode_model_choice(normalized_provider, model)
if current_model_id and current_model_id not in seen_ids:
available_models.insert(
Expand Down
203 changes: 203 additions & 0 deletions tests/acp/test_named_provider_catalogs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,203 @@
"""Tests for named user-defined provider entries in the ACP model selector.

Named endpoints from the ``providers:`` mapping (and legacy
``custom_providers:`` list) are invisible to canonical provider enumeration,
so ``_build_model_state`` must append them explicitly for ACP clients to
offer them — the TUI ``/model`` picker already renders these entries
(#47039 implemented named endpoints for the TUI surface only).
"""

from types import SimpleNamespace
from unittest.mock import patch

import pytest

from acp_adapter.server import HermesACPAgent, _named_custom_provider_catalogs
from acp_adapter.session import SessionManager
from acp.schema import SessionModelState


MANTLE_URL = "https://bedrock-mantle.us-east-1.api.aws/openai/v1"


def _cfg(providers=None, custom_providers=None):
cfg = {}
if providers is not None:
cfg["providers"] = providers
if custom_providers is not None:
cfg["custom_providers"] = custom_providers
return cfg


class TestNamedCustomProviderCatalogs:
def test_declared_default_model_survives_failed_discovery(self, monkeypatch):
"""Endpoints without a /models route keep their declared models."""
monkeypatch.setenv("BEDROCK_MANTLE_API_KEY", "test-key")
cfg = _cfg(
providers={
"bedrock-mantle": {
"name": "AWS Bedrock Mantle",
"base_url": MANTLE_URL,
"key_env": "BEDROCK_MANTLE_API_KEY",
"api_mode": "codex_responses",
"default_model": "openai.gpt-5.5",
}
}
)
with patch("hermes_cli.config.load_config", return_value=cfg), patch(
"hermes_cli.models.fetch_api_models", return_value=None
):
catalogs = _named_custom_provider_catalogs()

assert catalogs == [
(
"custom:bedrock-mantle",
"AWS Bedrock Mantle",
[("openai.gpt-5.5", "")],
)
]

def test_live_discovery_extends_declared_models(self, monkeypatch):
monkeypatch.setenv("SOME_KEY", "k")
cfg = _cfg(
providers={
"relay": {
"name": "Relay",
"base_url": "https://relay.example/v1",
"key_env": "SOME_KEY",
"default_model": "model-a",
}
}
)
with patch("hermes_cli.config.load_config", return_value=cfg), patch(
"hermes_cli.models.fetch_api_models",
return_value=["model-a", "model-b"],
):
catalogs = _named_custom_provider_catalogs()

assert len(catalogs) == 1
slug, label, models = catalogs[0]
assert slug == "custom:relay"
assert [m for m, _ in models] == ["model-a", "model-b"]

def test_declared_models_dict_included(self, monkeypatch):
monkeypatch.setenv("SOME_KEY", "k")
cfg = _cfg(
providers={
"relay": {
"name": "Relay",
"base_url": "https://relay.example/v1",
"key_env": "SOME_KEY",
"default_model": "model-a",
"models": {"model-b": {}, "model-c": {}},
}
}
)
with patch("hermes_cli.config.load_config", return_value=cfg), patch(
"hermes_cli.models.fetch_api_models", return_value=None
):
catalogs = _named_custom_provider_catalogs()

assert [m for m, _ in catalogs[0][2]] == ["model-a", "model-b", "model-c"]

def test_disabled_provider_skipped(self, monkeypatch):
monkeypatch.setenv("SOME_KEY", "k")
cfg = _cfg(
providers={
"off": {
"name": "Disabled Endpoint",
"base_url": "https://off.example/v1",
"key_env": "SOME_KEY",
"default_model": "m",
"enabled": False,
}
}
)
with patch("hermes_cli.config.load_config", return_value=cfg), patch(
"hermes_cli.models.fetch_api_models", return_value=None
):
assert _named_custom_provider_catalogs() == []

def test_no_credential_and_no_declared_models_skipped(self, monkeypatch):
monkeypatch.delenv("MISSING_KEY", raising=False)
cfg = _cfg(
providers={
"bare": {
"name": "Bare",
"base_url": "https://bare.example/v1",
"key_env": "MISSING_KEY",
}
}
)
with patch("hermes_cli.config.load_config", return_value=cfg), patch(
"hermes_cli.models.fetch_api_models", return_value=None
):
assert _named_custom_provider_catalogs() == []

def test_legacy_custom_providers_list_included(self, monkeypatch):
monkeypatch.setenv("SOME_KEY", "k")
cfg = _cfg(
custom_providers=[
{
"name": "Legacy Endpoint",
"base_url": "https://legacy.example/v1",
"key_env": "SOME_KEY",
"model": "legacy-model",
}
]
)
with patch("hermes_cli.config.load_config", return_value=cfg), patch(
"hermes_cli.models.fetch_api_models", return_value=None
):
catalogs = _named_custom_provider_catalogs()

assert catalogs == [
("custom:legacy-endpoint", "Legacy Endpoint", [("legacy-model", "")])
]


class TestModelStateIncludesNamedProviders:
@pytest.mark.asyncio
async def test_named_provider_models_appear_in_model_state(self):
manager = SessionManager(
agent_factory=lambda: SimpleNamespace(
model="gpt-5.4", provider="openai-codex"
)
)
acp_agent = HermesACPAgent(session_manager=manager)

with patch(
"hermes_cli.models.curated_models_for_provider",
return_value=[("gpt-5.4", "recommended")],
), patch(
"acp_adapter.server._named_custom_provider_catalogs",
return_value=[
(
"custom:bedrock-mantle",
"AWS Bedrock Mantle",
[("openai.gpt-5.5", "")],
)
],
):
resp = await acp_agent.new_session(cwd="/tmp")

assert isinstance(resp.models, SessionModelState)
ids = [m.model_id for m in resp.models.available_models]
# Current provider's models come first, named endpoints after.
assert ids[0] == "openai-codex:gpt-5.4"
assert "custom:bedrock-mantle:openai.gpt-5.5" in ids
named = next(
m
for m in resp.models.available_models
if m.model_id == "custom:bedrock-mantle:openai.gpt-5.5"
)
assert "AWS Bedrock Mantle" in (named.description or "")

def test_selector_choice_id_round_trips_through_parse_model_input(self):
"""The encoded choice id must resolve back to the named provider."""
from hermes_cli.models import parse_model_input

choice_id = "custom:bedrock-mantle:openai.gpt-5.5"
provider, model = parse_model_input(choice_id, "bedrock")
assert provider == "custom:bedrock-mantle"
assert model == "openai.gpt-5.5"
Loading