From bc5628298a8be7c09122b8cc95f9816b258fa643 Mon Sep 17 00:00:00 2001
From: Mark Angler <19395192+MarkAngler@users.noreply.github.com>
Date: Fri, 10 Jul 2026 10:44:28 -0400
Subject: [PATCH 1/2] databricks: bucket UC model-services by
 supported_api_types
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Model discovery bucketed `system.ai.*` ids into claude/codex/gemini/oss by
matching substrings in the model name. Names don't say which API dialect a
model speaks, so the guess misfired three ways:

- `gpt-oss-120b`/`gpt-oss-20b` matched `"gpt-" in id` and were bucketed as
  OpenAI Responses models. They only speak `mlflow/v1/chat/completions`, so pi
  registered them under `databricks-openai` pointing at /ai-gateway/codex/v1 —
  a route that rejects them. Claude's web_search MCP aimed at the same route.
- `claude-fable-5` matched none of `claude-{opus,sonnet,haiku}-` and vanished.
- `claude_models` is a `{family: id}` map, so only the newest model per family
  reached agents that render a full model list. A workspace serving 11 Claude
  models showed pi 3.

Every model-service entry already carries `supported_api_types`; the listing
fetched it and threw it away. Bucket by that instead, with precedence — a
Claude model also advertises mlflow chat-completions and would otherwise land
in two buckets. Workspaces whose payload predates the field fall back to the
name rules, with the codex rule tightened to require a version digit after
`gpt-` so gpt-oss is excluded there too.

Also:
- Sort every bucket with `model_version_sort_key`, replacing a lexicographic
  reverse sort that ranked `opus-4-8` above `opus-4-10`, and leaving
  `codex_models[0]` as the newest rather than the alphabetically first.
- Return `claude_ids` (every Anthropic-dialect id) alongside the family map;
  persist it as `claude_model_ids` and register it in pi's models.json.
  opencode deliberately keeps reading the family map — it takes `anthropic[0]`
  as its default, which widening would silently move onto claude-fable-5.
- `check_gateway_endpoint("codex")` now delegates to `codex.default_model`.
  A non-empty list wasn't enough: codex drops ids that don't parse as
  `gpt-<version>`, so a gpt-oss-only workspace reported codex available and
  then failed at launch with "No models available for codex".

State keys are additive; no STATE_VERSION bump (a mismatch discards every
workspace's state, including managed_configs and provider_services).

Refs #204

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/ucode/agents/__init__.py |   6 +-
 src/ucode/agents/claude.py   |  12 +-
 src/ucode/agents/pi.py       |  22 ++-
 src/ucode/cli.py             |  27 ++-
 src/ucode/databricks.py      | 256 ++++++++++++++++++++--------
 tests/test_agent_claude.py   |  13 +-
 tests/test_agent_pi.py       |  64 +++++--
 tests/test_agents_init.py    |   9 +-
 tests/test_cli.py            |  38 +++--
 tests/test_databricks.py     | 321 +++++++++++++++++++++++++++++++----
 tests/test_e2e_uc.py         |  36 +++-
 11 files changed, 634 insertions(+), 170 deletions(-)

diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py
index f44b91b..d7360d3 100644
--- a/src/ucode/agents/__init__.py
+++ b/src/ucode/agents/__init__.py
@@ -328,7 +328,11 @@ def check_gateway_endpoint(state: dict, tool: str) -> bool:
     if tool == "opencode":
         return bool(state.get("opencode_models"))
     if tool == "codex":
-        return bool(state.get("codex_models"))
+        # A non-empty list isn't enough: codex pins the model id verbatim and
+        # `default_model` drops any id that doesn't parse as `gpt-<version>`.
+        # Testing the list alone reports codex available and then fails at
+        # launch with "No models available for codex".
+        return codex.default_model(state) is not None
     if tool == "gemini":
         return bool(state.get("gemini_models"))
     if tool == "copilot":
diff --git a/src/ucode/agents/claude.py b/src/ucode/agents/claude.py
index 422e763..7ef8b66 100644
--- a/src/ucode/agents/claude.py
+++ b/src/ucode/agents/claude.py
@@ -24,6 +24,7 @@
     build_auth_shell_command,
     build_tool_base_url,
     get_databricks_token,
+    is_responses_model_id,
 )
 from ucode.launcher import exec_or_spawn
 from ucode.state import mark_tool_managed, save_state
@@ -51,15 +52,20 @@ def is_update_available() -> tuple[str, str] | None:
 def _resolve_web_search_model(state: dict) -> str | None:
     """Pick the model the web_search MCP server should call. Prefers an
     explicit override in state, otherwise the first endpoint discovered as
-    Responses-API-capable. Returns None if no GPT endpoint is available —
-    callers should skip the MCP wiring in that case."""
+    Responses-API-capable. Returns None if no such endpoint is available —
+    callers should skip the MCP wiring in that case.
+
+    The MCP server POSTs to `/ai-gateway/codex/v1/responses`, which only
+    forwards models that speak the Responses dialect. `codex_models` holds
+    exactly those, so the guard below is a backstop against a discovery
+    regression silently repointing the MCP at a chat-completions model."""
     override = state.get("web_search_model")
     if isinstance(override, str) and override.strip():
         return override.strip()
     codex_models = state.get("codex_models") or []
     if isinstance(codex_models, list) and codex_models:
         first = codex_models[0]
-        if isinstance(first, str) and first.strip():
+        if isinstance(first, str) and first.strip() and is_responses_model_id(first.strip()):
             return first.strip()
     return None
 
diff --git a/src/ucode/agents/pi.py b/src/ucode/agents/pi.py
index e7c1760..7a481cf 100644
--- a/src/ucode/agents/pi.py
+++ b/src/ucode/agents/pi.py
@@ -83,7 +83,7 @@ def is_update_available() -> tuple[str, str] | None:
 
 def _resolve_model_selector(
     model: str,
-    claude_models: dict[str, str],
+    claude_ids: list[str],
     codex_models: list[str],
     gemini_models: list[str],
 ) -> str:
@@ -91,7 +91,7 @@ def _resolve_model_selector(
     for name in PROVIDER_NAMES:
         if model.startswith(f"{name}/"):
             return model
-    if model in claude_models.values():
+    if model in claude_ids:
         return f"databricks-claude/{model}"
     if model in codex_models:
         return f"databricks-openai/{model}"
@@ -104,18 +104,23 @@ def render_overlay(
     model: str,
     token: str,
     pi_base_urls: dict[str, str],
-    claude_models: dict[str, str],
+    claude_ids: list[str],
     codex_models: list[str],
     gemini_models: list[str],
 ) -> tuple[dict, list[list[str]]]:
-    """Return (overlay, managed_key_paths) for ~/.pi/agent/models.json."""
+    """Return (overlay, managed_key_paths) for ~/.pi/agent/models.json.
+
+    ``claude_ids`` is every Anthropic-dialect model the workspace serves, not
+    just the newest per family — Pi has its own model picker, so it should see
+    the full list.
+    """
     providers: dict = {}
     keys: list[list[str]] = [["model"]]
     # Pi expands header values that match an env var name. Our UA contains
     # `/` and a space so it can never collide — safe to pass as a literal.
     ua_headers = {"User-Agent": f"ucode/{ucode_version()} pi/{agent_version('pi')}"}
 
-    claude_ids = sorted(set(claude_models.values()))
+    claude_ids = sorted(set(claude_ids))
     if claude_ids:
         providers["databricks-claude"] = {
             "baseUrl": pi_base_urls["claude"],
@@ -151,7 +156,7 @@ def render_overlay(
         }
         keys.append(["providers", "databricks-gemini"])
     overlay: dict = {
-        "model": _resolve_model_selector(model, claude_models, codex_models, gemini_models),
+        "model": _resolve_model_selector(model, claude_ids, codex_models, gemini_models),
     }
     if providers:
         overlay["providers"] = providers
@@ -171,11 +176,14 @@ def write_tool_config(
             state["workspace"], state.get("profile"), force_refresh=force_refresh
         )
     pi_base_urls = state.get("base_urls", {}).get("pi") or build_pi_base_urls(state["workspace"])
+    # State written before claude_model_ids existed only carries the family map;
+    # fall back to its values so a stale state still renders a usable config.
+    claude_ids = state.get("claude_model_ids") or list((state.get("claude_models") or {}).values())
     overlay, managed_keys = render_overlay(
         model,
         token,
         pi_base_urls,
-        state.get("claude_models") or {},
+        claude_ids,
         state.get("codex_models") or [],
         state.get("gemini_models") or [],
     )
diff --git a/src/ucode/cli.py b/src/ucode/cli.py
index b2f23be..14238ad 100644
--- a/src/ucode/cli.py
+++ b/src/ucode/cli.py
@@ -280,6 +280,7 @@ def configure_shared_state(
     codex_reason: str | None = None
     oss_reason: str | None = None
     claude_models = {}
+    claude_model_ids: list[str] = []
     gemini_models = []
     codex_models = []
     oss_models = []
@@ -300,25 +301,32 @@ def configure_shared_state(
         # empty (workspace without UC model-services, or the listing failed), fall
         # back to the per-family AI Gateway listing for that family only.
         with spinner("Fetching available models..."):
-            ms_claude, ms_codex, ms_gemini, ms_oss, ms_reason = discover_model_services(
-                workspace, token
-            )
+            discovered = discover_model_services(workspace, token)
+            ms_reason = discovered.reason
             if want_claude:
-                claude_models, claude_reason = ms_claude, ms_reason
-                if not claude_models:
-                    claude_models, claude_reason = discover_claude_models(workspace, token)
+                claude_models = discovered.claude_models
+                claude_model_ids = discovered.claude_ids
+                claude_reason = ms_reason
+                if not claude_model_ids:
+                    claude_models, claude_model_ids, claude_reason = discover_claude_models(
+                        workspace, token
+                    )
             if want_gemini:
-                gemini_models, gemini_reason = ms_gemini, ms_reason
+                gemini_models, gemini_reason = discovered.gemini_models, ms_reason
                 if not gemini_models:
                     gemini_models, gemini_reason = discover_gemini_models(workspace, token)
             if want_codex:
-                codex_models, codex_reason = ms_codex, ms_reason
+                codex_models, codex_reason = discovered.codex_models, ms_reason
                 if not codex_models:
                     codex_models, codex_reason = discover_codex_models(workspace, token)
             if want_oss:
-                oss_models, oss_reason = ms_oss, ms_reason
+                oss_models, oss_reason = discovered.oss_models, ms_reason
     opencode_models: dict[str, list[str]] = {}
     if claude_models:
+        # Deliberately the family map, not claude_model_ids: opencode picks
+        # `anthropic[0]` as its default, and the family map is ordered
+        # opus → sonnet → haiku. Widening this to every Claude id would silently
+        # move opencode's default onto whichever id sorts first.
         opencode_models["anthropic"] = list(claude_models.values())
     if gemini_models:
         opencode_models["gemini"] = gemini_models
@@ -350,6 +358,7 @@ def configure_shared_state(
     else:
         if want_claude:
             state["claude_models"] = claude_models
+            state["claude_model_ids"] = claude_model_ids
         if want_gemini:
             state["gemini_models"] = gemini_models
         if want_codex:
diff --git a/src/ucode/databricks.py b/src/ucode/databricks.py
index 9f89c6b..9a09fbf 100644
--- a/src/ucode/databricks.py
+++ b/src/ucode/databricks.py
@@ -22,6 +22,7 @@
 from concurrent.futures import (
     TimeoutError as FutureTimeoutError,
 )
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Literal, cast, overload
 from urllib import error as urllib_error
@@ -1096,10 +1097,57 @@ def build_auth_shell_command(
 # Databricks-managed foundation models under `system.ai`.
 _MODEL_SERVICE_REQUIRED_PREFIX = "system.ai."
 
-# Supported OSS chat families, matched by name substring. Add an entry to
-# support a new family.
+# API dialects a model-service advertises in its `supported_api_types`. A model
+# can speak several — every Claude model also advertises mlflow chat-completions
+# — so buckets are assigned by precedence rather than by sweeping each dialect
+# independently, or a Claude model would land in both `claude` and `oss`.
+_ANTHROPIC_API_TYPE = "anthropic/v1/messages"
+_RESPONSES_API_TYPE = "openai/v1/responses"
+_GEMINI_API_TYPE = "gemini/v1/generateContent"
+_MLFLOW_CHAT_API_TYPE = "mlflow/v1/chat/completions"
+
+# Name rules, used only for workspaces whose model-services payload predates
+# `supported_api_types`. The codex rule requires a version digit: `gpt-oss-120b`
+# is an open-weight chat model served on the mlflow route, not a Responses
+# endpoint, and a bare `gpt-` substring wrongly claims it. (Mirrors
+# `codex._GPT_RE`; duplicated so databricks.py stays free of agent imports.)
+_CLAUDE_NAME_FRAGMENT = "claude-"
+_GEMINI_NAME_FRAGMENT = "gemini-"
+_GPT_VERSIONED_RE = re.compile(r"gpt-\d")
+
+# Supported OSS chat families, matched by name substring. This is an allowlist of
+# families vetted against the agents that serve them (see `_MODEL_TOKEN_LIMITS`),
+# not a list of every mlflow-capable model — qwen, llama and gemma all speak the
+# same dialect and are deliberately excluded. Add an entry to support a family.
 _OSS_MODEL_FAMILIES = ("kimi-", "glm-")
 
+
+@dataclass(frozen=True)
+class DiscoveredModels:
+    """Workspace model ids bucketed by the API dialect each agent speaks.
+
+    ``claude_models`` maps ``opus``/``sonnet``/``haiku`` to the newest id in that
+    family, for agents that select by family (Claude Code's
+    ``ANTHROPIC_DEFAULT_*_MODEL`` env vars). ``claude_ids`` is every
+    Anthropic-dialect id — including models outside those three families, e.g.
+    ``claude-fable-5`` — for agents like Pi that register a full model list.
+    It is grouped by family and newest-first within each, so no entry is
+    meaningfully "the default"; read ``claude_models`` for that.
+
+    ``codex_models`` and ``gemini_models`` are newest first: consumers that need
+    a single default take entry ``[0]``.
+
+    ``reason`` is None on success, else it explains why the buckets are empty.
+    """
+
+    claude_models: dict[str, str] = field(default_factory=dict)
+    claude_ids: list[str] = field(default_factory=list)
+    codex_models: list[str] = field(default_factory=list)
+    gemini_models: list[str] = field(default_factory=list)
+    oss_models: list[str] = field(default_factory=list)
+    reason: str | None = None
+
+
 # Per-family token limits (context window + max output tokens). These are a
 # property of the model + its `/ai-gateway/mlflow/v1` route (the gateway rejects
 # requests whose output exceeds the cap), not of any one agent — so every agent
@@ -1125,11 +1173,13 @@ def model_token_limits(model_id: str) -> dict[str, int] | None:
     return None
 
 
-def _model_service_id(service: dict) -> str | None:
-    """Extract the `system.ai.<model-name>` id from one model-service entry.
+def _model_service_entry(service: dict) -> tuple[str, list[str]] | None:
+    """Extract `(system.ai.<model-name>, supported_api_types)` from one entry.
 
     Returns None for services in any other schema, so user/internal model
-    services don't leak into the family buckets."""
+    services don't leak into the family buckets. The api-type list is empty when
+    the workspace's API predates `supported_api_types`; callers read that as
+    "capabilities unknown" and fall back to name rules."""
     name = service.get("name")
     if not isinstance(name, str):
         return None
@@ -1138,7 +1188,11 @@ def _model_service_id(service: dict) -> str | None:
         name = name[len(_MODEL_SERVICE_NAME_PREFIX) :]
     if not name.startswith(_MODEL_SERVICE_REQUIRED_PREFIX):
         return None
-    return name or None
+    if not name:
+        return None
+    raw = service.get("supported_api_types")
+    api_types = [t for t in raw if isinstance(t, str)] if isinstance(raw, list) else []
+    return name, api_types
 
 
 # The model-services metastore listing REQUIRES a bounded `page_size`:
@@ -1174,17 +1228,18 @@ def list_model_services(
     *,
     page_size: int = _MODEL_SERVICES_PAGE_SIZE,
     max_pages: int = 100,
-) -> tuple[list[str], str | None]:
-    """List all `system.ai.*` model ids via the UC model-services API.
+) -> tuple[dict[str, list[str]], str | None]:
+    """List all `system.ai.*` model services and the API dialects each speaks.
 
     Pages through ``/api/2.1/unity-catalog/model-services`` (metastore scope)
-    with a bounded ``page_size`` (the endpoint 499s without one) and returns the
-    de-duplicated, sorted list of ``system.ai.<model-name>`` ids. Returns
-    (ids, reason); reason is None on success, otherwise it describes why the
-    list is empty (HTTP/network error or no services).
+    with a bounded ``page_size`` (the endpoint 499s without one) and returns
+    (services, reason). ``services`` maps each ``system.ai.<model-name>`` id to
+    its ``supported_api_types``, sorted by id; the list is empty for a workspace
+    whose API predates that field. ``reason`` is None on success, otherwise it
+    describes why the mapping is empty (HTTP/network error or no services).
     """
     hostname = workspace_hostname(workspace)
-    ids: list[str] = []
+    services: dict[str, list[str]] = {}
     page_token: str | None = None
     seen_tokens: set[str] = set()
     last_reason: str | None = None
@@ -1202,9 +1257,13 @@ def list_model_services(
         data = cast(dict, payload) if isinstance(payload, dict) else {}
         for service in data.get("model_services", []):
             if isinstance(service, dict):
-                model_id = _model_service_id(service)
-                if model_id:
-                    ids.append(model_id)
+                entry = _model_service_entry(service)
+                if entry:
+                    model_id, api_types = entry
+                    # A duplicate id across pages keeps whichever entry actually
+                    # advertised capabilities.
+                    if api_types or model_id not in services:
+                        services[model_id] = api_types
         page_token = data.get("next_page_token") or None
         if not page_token:
             last_reason = None
@@ -1213,60 +1272,108 @@ def list_model_services(
             break
         seen_tokens.add(page_token)
 
-    deduped = sorted(set(ids))
-    if deduped:
-        return deduped, None
-    return [], last_reason or "model-services listing returned no models"
+    if services:
+        return dict(sorted(services.items())), None
+    return {}, last_reason or "model-services listing returned no models"
 
 
-def discover_model_services(
-    workspace: str, token: str
-) -> tuple[dict[str, str], list[str], list[str], list[str], str | None]:
-    """Discover models via UC model-services and bucket them by family name.
+def is_responses_model_id(model_id: str) -> bool:
+    """True when ``model_id`` names a GPT model that speaks the Responses dialect.
+
+    A name-shaped backstop for callers that hold a bare id and no capability
+    record. `gpt-oss-*` is an open-weight chat model served on the mlflow route,
+    so a bare `gpt-` substring is not sufficient — a version digit is required.
+    """
+    return bool(_GPT_VERSIONED_RE.search(model_id))
 
-    Returns (claude_models, codex_models, gemini_models, oss_models, reason):
 
-    - ``claude_models`` maps ``opus``/``sonnet``/``haiku`` to the newest
-      matching ``system.ai.claude-*`` id (mirrors ``discover_claude_models``).
-    - ``codex_models`` is the list of ``system.ai.*gpt-*`` ids.
-    - ``gemini_models`` is the list of ``system.ai.*gemini-*`` ids, newest first.
-    - ``oss_models`` is the list of OSS-model ``system.ai.*`` ids.
+def _speaks(api_types: list[str], api_type: str, name_matches: bool) -> bool:
+    """True when a model advertises ``api_type``.
 
-    ``reason`` is None on success, else explains why nothing was found. Family
-    bucketing is by name substring because the model-services API does not
-    expose per-model API dialects.
+    Falls back to ``name_matches`` for entries whose payload carries no
+    ``supported_api_types`` (older workspaces), so discovery never degrades
+    below the name-substring behavior it replaces.
     """
-    ids, reason = list_model_services(workspace, token)
-    if not ids:
-        return {}, [], [], [], reason
+    if api_types:
+        return api_type in api_types
+    return name_matches
+
 
-    claude_models: dict[str, str] = {}
+def _claude_family_map(claude_ids: list[str]) -> dict[str, str]:
+    """Map ``opus``/``sonnet``/``haiku`` to the newest id in that family.
+
+    Ids outside those three families (e.g. ``claude-fable-5``) belong to no
+    family and are reachable only through the full id list — Claude Code selects
+    by family env var and has no slot for them.
+    """
+    result: dict[str, str] = {}
     for family in ("opus", "sonnet", "haiku"):
         candidates = sorted(
-            [m for m in ids if f"claude-{family}-" in m],
-            reverse=True,
+            [m for m in claude_ids if f"claude-{family}-" in m],
+            key=model_version_sort_key,
         )
         if candidates:
-            claude_models[family] = candidates[0]
+            result[family] = candidates[0]
+    return result
+
+
+def _is_oss_model(model_id: str, api_types: list[str]) -> bool:
+    """True for an allowlisted OSS chat family served on the mlflow route."""
+    if not any(family in model_id for family in _OSS_MODEL_FAMILIES):
+        return False
+    return _speaks(api_types, _MLFLOW_CHAT_API_TYPE, True)
 
-    codex_models = [m for m in ids if "gpt-" in m]
-    gemini_models = sorted([m for m in ids if "gemini-" in m], key=model_version_sort_key)
 
-    oss_models = [m for m in ids if any(family in m for family in _OSS_MODEL_FAMILIES)]
+def discover_model_services(workspace: str, token: str) -> DiscoveredModels:
+    """Discover models via UC model-services and bucket them by API dialect.
 
-    if not (claude_models or codex_models or gemini_models or oss_models):
-        sample = ", ".join(ids[:5])
-        return (
-            {},
-            [],
-            [],
-            [],
-            (
+    Each service advertises the dialects it speaks in ``supported_api_types``;
+    buckets are assigned by precedence because a model can speak several (every
+    Claude model also serves mlflow chat-completions). Workspaces whose API
+    predates that field fall back to matching the model name.
+    """
+    services, reason = list_model_services(workspace, token)
+    if not services:
+        return DiscoveredModels(reason=reason)
+
+    claude_ids: list[str] = []
+    codex_models: list[str] = []
+    gemini_models: list[str] = []
+    oss_models: list[str] = []
+    for model_id, api_types in services.items():
+        if _speaks(api_types, _ANTHROPIC_API_TYPE, _CLAUDE_NAME_FRAGMENT in model_id):
+            claude_ids.append(model_id)
+        elif _speaks(api_types, _RESPONSES_API_TYPE, bool(_GPT_VERSIONED_RE.search(model_id))):
+            codex_models.append(model_id)
+        elif _speaks(api_types, _GEMINI_API_TYPE, _GEMINI_NAME_FRAGMENT in model_id):
+            gemini_models.append(model_id)
+        elif _is_oss_model(model_id, api_types):
+            oss_models.append(model_id)
+
+    # `model_version_sort_key` groups by the non-numeric name prefix, then orders
+    # by version descending. For codex/gemini every id shares a prefix, so [0] is
+    # the newest — which is what their consumers take. Claude ids span several
+    # prefixes (opus/sonnet/haiku/fable), so this groups by family rather than
+    # ranking globally; callers wanting a default read `claude_models`.
+    claude_ids.sort(key=model_version_sort_key)
+    codex_models.sort(key=model_version_sort_key)
+    gemini_models.sort(key=model_version_sort_key)
+
+    if not (claude_ids or codex_models or gemini_models or oss_models):
+        sample = ", ".join(list(services)[:5])
+        return DiscoveredModels(
+            reason=(
                 "model-services returned model ids but none matched "
                 f"claude/gpt/gemini/oss families (got: {sample})"
-            ),
+            )
         )
-    return claude_models, codex_models, gemini_models, oss_models, None
+    return DiscoveredModels(
+        claude_models=_claude_family_map(claude_ids),
+        claude_ids=claude_ids,
+        codex_models=codex_models,
+        gemini_models=gemini_models,
+        oss_models=oss_models,
+    )
 
 
 # --- MCP services (parallel to model services) -----------------------------
@@ -1777,17 +1884,21 @@ def collect_pair(has_fn, pair):
     return sorted(pairs), None
 
 
-def discover_claude_models(workspace: str, token: str) -> tuple[dict[str, str], str | None]:
-    """Discover Claude families on this workspace's AI Gateway.
+def discover_claude_models(
+    workspace: str, token: str
+) -> tuple[dict[str, str], list[str], str | None]:
+    """Discover Claude models on this workspace's AI Gateway.
 
-    Returns (models_by_family, reason). reason is None on success; otherwise it
-    describes why the dict is empty (HTTP error, network error, or no models
+    Returns (models_by_family, claude_ids, reason). ``claude_ids`` is every
+    Claude model the gateway serves, newest first, including ids outside the
+    opus/sonnet/haiku families. reason is None on success; otherwise it
+    describes why the result is empty (HTTP error, network error, or no models
     matching the expected naming convention).
     """
     hostname = workspace_hostname(workspace)
     payload, reason = _http_get_json(f"https://{hostname}/ai-gateway/anthropic/v1/models", token)
     if payload is None:
-        return {}, reason
+        return {}, [], reason
 
     data = cast(dict, payload) if isinstance(payload, dict) else {}
     raw_ids = [
@@ -1796,28 +1907,25 @@ def discover_claude_models(workspace: str, token: str) -> tuple[dict[str, str],
         if isinstance(m.get("id"), str) and not m["id"].endswith("-anthropic")
     ]
 
-    result: dict[str, str] = {}
-    for family, key in [("opus", "opus"), ("sonnet", "sonnet"), ("haiku", "haiku")]:
-        candidates = sorted(
-            [m for m in raw_ids if f"databricks-claude-{family}-" in m],
-            reverse=True,
-        )
-        if candidates:
-            result[key] = candidates[0]
-    if result:
-        return result, None
+    claude_ids = sorted(
+        [m for m in raw_ids if _CLAUDE_NAME_FRAGMENT in m], key=model_version_sort_key
+    )
+    result = _claude_family_map(claude_ids)
+    if claude_ids:
+        return result, claude_ids, None
     if not raw_ids:
-        return {}, "AI Gateway returned no Claude model ids"
+        return {}, [], "AI Gateway returned no Claude model ids"
     sample = ", ".join(raw_ids[:5])
-    return {}, (
-        "AI Gateway returned model ids but none matched "
-        f"`databricks-claude-{{opus,sonnet,haiku}}-*` (got: {sample})"
+    return (
+        {},
+        [],
+        (f"AI Gateway returned model ids but none matched `databricks-claude-*` (got: {sample})"),
     )
 
 
 def fetch_ai_gateway_claude_models(workspace: str, token: str) -> dict[str, str]:
-    """Backwards-compatible wrapper that discards the diagnostic reason."""
-    models, _ = discover_claude_models(workspace, token)
+    """Backwards-compatible wrapper that discards the ids and diagnostic reason."""
+    models, _ids, _reason = discover_claude_models(workspace, token)
     return models
 
 
diff --git a/tests/test_agent_claude.py b/tests/test_agent_claude.py
index b2d3a27..8641bcc 100644
--- a/tests/test_agent_claude.py
+++ b/tests/test_agent_claude.py
@@ -247,15 +247,22 @@ def test_uses_explicit_override(self):
         assert claude._resolve_web_search_model({"web_search_model": "explicit"}) == "explicit"
 
     def test_falls_back_to_first_codex_model(self):
-        state = {"codex_models": ["m1", "m2"]}
-        assert claude._resolve_web_search_model(state) == "m1"
+        state = {"codex_models": ["system.ai.gpt-5-6", "system.ai.gpt-5-2-codex"]}
+        assert claude._resolve_web_search_model(state) == "system.ai.gpt-5-6"
 
     def test_returns_none_when_no_codex_models(self):
         assert claude._resolve_web_search_model({}) is None
         assert claude._resolve_web_search_model({"codex_models": []}) is None
 
+    def test_ignores_codex_model_that_is_not_responses_capable(self):
+        # The MCP POSTs to /ai-gateway/codex/v1/responses. If a discovery
+        # regression ever leaks a chat-completions model (gpt-oss-*) into
+        # codex_models, skip the MCP rather than wire it to a route that 400s.
+        state = {"codex_models": ["system.ai.gpt-oss-120b"]}
+        assert claude._resolve_web_search_model(state) is None
+
     def test_override_wins_over_codex_models(self):
-        state = {"web_search_model": "winner", "codex_models": ["loser"]}
+        state = {"web_search_model": "winner", "codex_models": ["system.ai.gpt-5-6"]}
         assert claude._resolve_web_search_model(state) == "winner"
 
 
diff --git a/tests/test_agent_pi.py b/tests/test_agent_pi.py
index 0afc5fb..00a3bac 100644
--- a/tests/test_agent_pi.py
+++ b/tests/test_agent_pi.py
@@ -23,7 +23,7 @@ def _base_urls() -> dict[str, str]:
 def _empty() -> dict:
     """No-models input bundle for render_overlay."""
     return {
-        "claude_models": {},
+        "claude_ids": [],
         "codex_models": [],
         "gemini_models": [],
     }
@@ -36,7 +36,7 @@ def _overlay(model: str, token: str = "tok", **kwargs):
         model,
         token,
         _base_urls(),
-        bundle["claude_models"],
+        bundle["claude_ids"],
         bundle["codex_models"],
         bundle["gemini_models"],
     )
@@ -64,7 +64,7 @@ def test_no_providers_when_no_models(self):
         assert "providers" not in overlay
 
     def test_claude_provider_uses_anthropic_messages(self):
-        overlay, _ = _overlay("claude-sonnet", claude_models={"sonnet": "claude-sonnet"})
+        overlay, _ = _overlay("claude-sonnet", claude_ids=["claude-sonnet"])
         provider = overlay["providers"]["databricks-claude"]
         assert provider["api"] == "anthropic-messages"
         assert provider["baseUrl"] == f"{WS}/ai-gateway/anthropic"
@@ -84,7 +84,7 @@ def test_gemini_provider_uses_google_generative_ai(self):
     def test_all_three_providers_when_all_present(self):
         overlay, _ = _overlay(
             "claude-sonnet",
-            claude_models={"sonnet": "claude-sonnet"},
+            claude_ids=["claude-sonnet"],
             codex_models=["gpt-5"],
             gemini_models=["gemini-2"],
         )
@@ -101,7 +101,7 @@ def test_user_agent_set_on_all_three_providers(self, monkeypatch):
         monkeypatch.setattr(pi, "agent_version", lambda binary: "0.74.0")
         overlay, _ = _overlay(
             "claude-sonnet",
-            claude_models={"sonnet": "claude-sonnet"},
+            claude_ids=["claude-sonnet"],
             codex_models=["gpt-5"],
             gemini_models=["gemini-2"],
         )
@@ -115,7 +115,7 @@ def test_claude_disables_eager_tool_input_streaming(self):
         # Gateway's Anthropic translator rejects per-tool
         # `eager_input_streaming`; this flag makes pi send the legacy beta
         # header instead.
-        overlay, _ = _overlay("claude-sonnet", claude_models={"sonnet": "claude-sonnet"})
+        overlay, _ = _overlay("claude-sonnet", claude_ids=["claude-sonnet"])
         compat = overlay["providers"]["databricks-claude"]["compat"]
         assert compat["supportsEagerToolInputStreaming"] is False
 
@@ -132,15 +132,13 @@ def test_openai_and_gemini_have_no_compat_flags(self):
 
 class TestRenderOverlayAuthAndModels:
     def test_token_in_api_key(self):
-        overlay, _ = _overlay(
-            "claude-sonnet", token="mytoken", claude_models={"sonnet": "claude-sonnet"}
-        )
+        overlay, _ = _overlay("claude-sonnet", token="mytoken", claude_ids=["claude-sonnet"])
         assert overlay["providers"]["databricks-claude"]["apiKey"] == "mytoken"
 
     def test_auth_header_flag_set_on_all_providers(self):
         overlay, _ = _overlay(
             "claude-sonnet",
-            claude_models={"sonnet": "claude-sonnet"},
+            claude_ids=["claude-sonnet"],
             codex_models=["gpt-5"],
             gemini_models=["gemini-2"],
         )
@@ -148,10 +146,12 @@ def test_auth_header_flag_set_on_all_providers(self):
             assert overlay["providers"][name]["authHeader"] is True
 
     def test_claude_models_listed(self):
-        claude_models = {"opus": "claude-opus", "sonnet": "claude-sonnet"}
-        overlay, _ = _overlay("claude-sonnet", claude_models=claude_models)
+        # Pi has its own picker, so it registers every Claude id the workspace
+        # serves — not just the newest of each opus/sonnet/haiku family.
+        claude_ids = ["claude-opus-4-8", "claude-opus-4-6", "claude-sonnet-5", "claude-fable-5"]
+        overlay, _ = _overlay("claude-sonnet-5", claude_ids=claude_ids)
         ids = {m["id"] for m in overlay["providers"]["databricks-claude"]["models"]}
-        assert ids == {"claude-opus", "claude-sonnet"}
+        assert ids == set(claude_ids)
 
     def test_openai_models_listed(self):
         overlay, _ = _overlay("gpt-5", codex_models=["gpt-5", "gpt-5-mini"])
@@ -172,7 +172,7 @@ def test_managed_keys_include_model(self):
     def test_managed_keys_include_each_provider_present(self):
         _, keys = _overlay(
             "claude-sonnet",
-            claude_models={"sonnet": "claude-sonnet"},
+            claude_ids=["claude-sonnet"],
             codex_models=["gpt-5"],
             gemini_models=["gemini-2"],
         )
@@ -182,7 +182,7 @@ def test_managed_keys_include_each_provider_present(self):
 
 class TestRenderOverlayModelSelector:
     def test_prefixes_claude_model(self):
-        overlay, _ = _overlay("claude-sonnet", claude_models={"sonnet": "claude-sonnet"})
+        overlay, _ = _overlay("claude-sonnet", claude_ids=["claude-sonnet"])
         assert overlay["model"] == "databricks-claude/claude-sonnet"
 
     def test_prefixes_openai_model(self):
@@ -196,7 +196,7 @@ def test_prefixes_gemini_model(self):
     def test_preserves_already_prefixed_model(self):
         overlay, _ = _overlay(
             "databricks-claude/claude-sonnet",
-            claude_models={"sonnet": "claude-sonnet"},
+            claude_ids=["claude-sonnet"],
         )
         assert overlay["model"] == "databricks-claude/claude-sonnet"
 
@@ -281,6 +281,7 @@ def _state(self, **overrides) -> dict:
             "workspace": WS,
             "base_urls": {"pi": _base_urls()},
             "claude_models": {"sonnet": "claude-sonnet"},
+            "claude_model_ids": ["claude-sonnet"],
             "codex_models": [],
             "gemini_models": [],
             "managed_configs": {},
@@ -288,6 +289,37 @@ def _state(self, **overrides) -> dict:
         state.update(overrides)
         return state
 
+    def test_registers_every_claude_id(self, tmp_path, monkeypatch):
+        pi_mod, config_file, _, _ = self._setup(tmp_path, monkeypatch)
+        state = self._state(
+            claude_models={"sonnet": "claude-sonnet-5"},
+            claude_model_ids=["claude-sonnet-5", "claude-opus-4-6", "claude-fable-5"],
+        )
+        with (
+            patch("ucode.agents.pi.get_databricks_token", return_value="tok"),
+            patch("ucode.agents.pi.save_state"),
+        ):
+            pi_mod.write_tool_config(state, "claude-sonnet-5")
+
+        config = json.loads(config_file.read_text())
+        ids = {m["id"] for m in config["providers"]["databricks-claude"]["models"]}
+        assert ids == {"claude-sonnet-5", "claude-opus-4-6", "claude-fable-5"}
+
+    def test_falls_back_to_family_map_on_stale_state(self, tmp_path, monkeypatch):
+        # State written before claude_model_ids existed still renders a config.
+        pi_mod, config_file, _, _ = self._setup(tmp_path, monkeypatch)
+        state = self._state(claude_models={"sonnet": "claude-sonnet"})
+        state.pop("claude_model_ids")
+        with (
+            patch("ucode.agents.pi.get_databricks_token", return_value="tok"),
+            patch("ucode.agents.pi.save_state"),
+        ):
+            pi_mod.write_tool_config(state, "claude-sonnet")
+
+        config = json.loads(config_file.read_text())
+        ids = {m["id"] for m in config["providers"]["databricks-claude"]["models"]}
+        assert ids == {"claude-sonnet"}
+
     def test_stale_managed_providers_removed_before_merge(self, tmp_path, monkeypatch):
         pi_mod, config_file, _, _ = self._setup(tmp_path, monkeypatch)
 
diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py
index 576fc22..08b8c0b 100644
--- a/tests/test_agents_init.py
+++ b/tests/test_agents_init.py
@@ -93,7 +93,14 @@ def test_claude_unavailable_when_no_models(self):
         assert check_gateway_endpoint({}, "claude") is False
 
     def test_codex_available(self):
-        assert check_gateway_endpoint({"codex_models": ["model-a"]}, "codex") is True
+        assert check_gateway_endpoint({"codex_models": ["system.ai.gpt-5"]}, "codex") is True
+
+    def test_codex_unavailable_when_no_model_parses_as_gpt(self):
+        # A non-empty list isn't enough: codex pins the id verbatim and
+        # default_model drops anything that isn't `gpt-<version>`. Reporting
+        # available here would fail at launch instead of at configure time.
+        state = {"codex_models": ["system.ai.gpt-oss-120b", "system.ai.gpt-oss-20b"]}
+        assert check_gateway_endpoint(state, "codex") is False
 
     def test_gemini_available(self):
         assert check_gateway_endpoint({"gemini_models": ["gemini-2"]}, "gemini") is True
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 3c5d404..39e95c7 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -10,6 +10,7 @@
 from typer.testing import CliRunner
 
 from ucode.cli import app
+from ucode.databricks import DiscoveredModels
 
 _ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
 
@@ -1045,8 +1046,8 @@ def _stub_deps(monkeypatch, *, pat_token, existing_state=None):
         monkeypatch.setattr(cli_mod, "find_profile_name_for_host", lambda w: None)
         monkeypatch.setattr(cli_mod, "get_databricks_token", lambda w, p: "token")
         monkeypatch.setattr(cli_mod, "ensure_ai_gateway_v2", lambda w, t: None)
-        monkeypatch.setattr(cli_mod, "discover_model_services", lambda w, t: ({}, [], [], [], None))
-        monkeypatch.setattr(cli_mod, "discover_claude_models", lambda w, t: ({}, None))
+        monkeypatch.setattr(cli_mod, "discover_model_services", lambda w, t: DiscoveredModels())
+        monkeypatch.setattr(cli_mod, "discover_claude_models", lambda w, t: ({}, [], None))
         monkeypatch.setattr(cli_mod, "discover_gemini_models", lambda w, t: ([], None))
         monkeypatch.setattr(cli_mod, "discover_codex_models", lambda w, t: ([], None))
         monkeypatch.setattr(cli_mod, "build_shared_base_urls", lambda w: {})
@@ -1117,24 +1118,30 @@ def test_uc_models_used_without_legacy_fallback(self, monkeypatch):
         monkeypatch.setattr(
             cli_mod,
             "discover_model_services",
-            lambda w, t: (
-                {"opus": "system.ai.claude-opus-4-8"},
-                ["system.ai.gpt-5"],
-                [],
-                [],
-                None,
+            lambda w, t: DiscoveredModels(
+                claude_models={"opus": "system.ai.claude-opus-4-8"},
+                claude_ids=["system.ai.claude-opus-4-8", "system.ai.claude-fable-5"],
+                codex_models=["system.ai.gpt-5"],
             ),
         )
         legacy_called: list[str] = []
         monkeypatch.setattr(
             cli_mod,
             "discover_claude_models",
-            lambda w, t: legacy_called.append("claude") or ({}, None),
+            lambda w, t: legacy_called.append("claude") or ({}, [], None),
         )
 
         state = cli_mod.configure_shared_state(self.WS, profile="DEFAULT")
 
         assert state["claude_models"] == {"opus": "system.ai.claude-opus-4-8"}
+        # The full id list is persisted for agents that register every model.
+        assert state["claude_model_ids"] == [
+            "system.ai.claude-opus-4-8",
+            "system.ai.claude-fable-5",
+        ]
+        # opencode takes anthropic[0] as its default, so it stays on the
+        # family map — widening it here would move the default onto fable.
+        assert state["opencode_models"]["anthropic"] == ["system.ai.claude-opus-4-8"]
         assert state["codex_models"] == ["system.ai.gpt-5"]
         assert legacy_called == []
         assert "uc_enabled" not in state
@@ -1143,13 +1150,16 @@ def test_falls_back_to_legacy_when_uc_empty(self, monkeypatch):
         # No UC model-services: each family falls back to the legacy listing.
         cli_mod, *_ = self._stub_deps(monkeypatch, pat_token="dapi-pat")
         monkeypatch.setattr(
-            cli_mod, "discover_model_services", lambda w, t: ({}, [], [], [], "no model services")
+            cli_mod,
+            "discover_model_services",
+            lambda w, t: DiscoveredModels(reason="no model services"),
         )
         monkeypatch.setattr(
             cli_mod,
             "discover_claude_models",
             lambda w, t: (
                 {"opus": "databricks-claude-opus-4-8", "sonnet": "databricks-claude-sonnet-4-6"},
+                ["databricks-claude-opus-4-8", "databricks-claude-sonnet-4-6"],
                 None,
             ),
         )
@@ -1160,6 +1170,10 @@ def test_falls_back_to_legacy_when_uc_empty(self, monkeypatch):
             "opus": "databricks-claude-opus-4-8",
             "sonnet": "databricks-claude-sonnet-4-6",
         }
+        assert state["claude_model_ids"] == [
+            "databricks-claude-opus-4-8",
+            "databricks-claude-sonnet-4-6",
+        ]
 
 
 class TestConfigureSkipValidate:
@@ -1221,8 +1235,8 @@ def _stub_external_deps(monkeypatch):
         monkeypatch.setattr(cli_mod, "find_profile_name_for_host", lambda w: None)
         monkeypatch.setattr(cli_mod, "get_databricks_token", lambda w, p: "token")
         monkeypatch.setattr(cli_mod, "ensure_ai_gateway_v2", lambda w, t: None)
-        monkeypatch.setattr(cli_mod, "discover_model_services", lambda w, t: ({}, [], [], [], None))
-        monkeypatch.setattr(cli_mod, "discover_claude_models", lambda w, t: ({}, None))
+        monkeypatch.setattr(cli_mod, "discover_model_services", lambda w, t: DiscoveredModels())
+        monkeypatch.setattr(cli_mod, "discover_claude_models", lambda w, t: ({}, [], None))
         monkeypatch.setattr(cli_mod, "discover_gemini_models", lambda w, t: ([], None))
         monkeypatch.setattr(cli_mod, "discover_codex_models", lambda w, t: ([], None))
         monkeypatch.setattr(cli_mod, "build_shared_base_urls", lambda w: {})
diff --git a/tests/test_databricks.py b/tests/test_databricks.py
index f59aa1d..f0d8c9a 100644
--- a/tests/test_databricks.py
+++ b/tests/test_databricks.py
@@ -129,15 +129,58 @@ def test_selects_opus_4_8_when_advertised(self, monkeypatch):
         }
         monkeypatch.setattr(db_mod, "_http_get_json", lambda url, token: (payload, None))
 
-        models, reason = db_mod.discover_claude_models(WS, "token")
+        models, ids, reason = db_mod.discover_claude_models(WS, "token")
 
         assert reason is None
         assert models["opus"] == "databricks-claude-opus-4-8"
+        assert set(ids) == {
+            "databricks-claude-opus-4-7",
+            "databricks-claude-opus-4-8",
+            "databricks-claude-sonnet-4-6",
+        }
+
+    def test_two_digit_minor_version_beats_single_digit(self, monkeypatch):
+        payload = {
+            "data": [
+                {"id": "databricks-claude-opus-4-8"},
+                {"id": "databricks-claude-opus-4-10"},
+            ]
+        }
+        monkeypatch.setattr(db_mod, "_http_get_json", lambda url, token: (payload, None))
+
+        models, _ids, _reason = db_mod.discover_claude_models(WS, "token")
+
+        assert models["opus"] == "databricks-claude-opus-4-10"
+
+    def test_familyless_model_reported_in_ids_only(self, monkeypatch):
+        # A Claude model outside opus/sonnet/haiku has no family env var, but
+        # agents with their own picker (pi) must still see it.
+        payload = {"data": [{"id": "databricks-claude-fable-5"}]}
+        monkeypatch.setattr(db_mod, "_http_get_json", lambda url, token: (payload, None))
+
+        models, ids, reason = db_mod.discover_claude_models(WS, "token")
+
+        assert reason is None
+        assert models == {}
+        assert ids == ["databricks-claude-fable-5"]
+
+
+ANTHROPIC = "anthropic/v1/messages"
+RESPONSES = "openai/v1/responses"
+GEMINI = "gemini/v1/generateContent"
+MLFLOW_CHAT = "mlflow/v1/chat/completions"
 
 
-def _model_service(model_id: str) -> dict:
-    """A model-services entry whose `name` strips to `model_id`."""
-    return {"name": f"model-services/{model_id}"}
+def _model_service(model_id: str, api_types: list[str] | None = None) -> dict:
+    """A model-services entry whose `name` strips to `model_id`.
+
+    Omitting `api_types` models a workspace whose API predates
+    `supported_api_types`, exercising the name-rule fallback.
+    """
+    entry: dict = {"name": f"model-services/{model_id}"}
+    if api_types is not None:
+        entry["supported_api_types"] = api_types
+    return entry
 
 
 class TestModelTokenLimits:
@@ -157,7 +200,130 @@ def test_uncapped_model_returns_none(self):
         assert db_mod.model_token_limits("system.ai.kimi-k2-7-code") is None
 
 
+class TestDiscoverModelServicesByCapability:
+    """Bucketing driven by each service's advertised `supported_api_types`."""
+
+    def test_buckets_by_advertised_api_type(self, monkeypatch):
+        payload = {
+            "model_services": [
+                _model_service("system.ai.claude-opus-4-8", [MLFLOW_CHAT, ANTHROPIC]),
+                _model_service("system.ai.claude-sonnet-5", [MLFLOW_CHAT, ANTHROPIC]),
+                _model_service("system.ai.gpt-5-6", [RESPONSES]),
+                _model_service("system.ai.gemini-3-5-flash", [GEMINI]),
+                _model_service("system.ai.glm-5-2", [MLFLOW_CHAT]),
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        assert found.reason is None
+        # Grouped by family prefix, newest-first within each family.
+        assert found.claude_ids == ["system.ai.claude-opus-4-8", "system.ai.claude-sonnet-5"]
+        assert found.codex_models == ["system.ai.gpt-5-6"]
+        assert found.gemini_models == ["system.ai.gemini-3-5-flash"]
+        assert found.oss_models == ["system.ai.glm-5-2"]
+
+    def test_claude_ids_newest_first_within_a_family(self, monkeypatch):
+        payload = {
+            "model_services": [
+                _model_service("system.ai.claude-opus-4-6", [ANTHROPIC]),
+                _model_service("system.ai.claude-opus-4-8", [ANTHROPIC]),
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        assert found.claude_ids == ["system.ai.claude-opus-4-8", "system.ai.claude-opus-4-6"]
+
+    def test_gpt_oss_is_not_a_codex_model(self, monkeypatch):
+        # gpt-oss-* only speaks mlflow chat-completions. Bucketing it as a
+        # Responses model points pi's databricks-openai provider (and claude's
+        # web_search MCP) at /ai-gateway/codex/v1, which rejects it.
+        payload = {
+            "model_services": [
+                _model_service("system.ai.gpt-oss-120b", [MLFLOW_CHAT]),
+                _model_service("system.ai.gpt-oss-20b", [MLFLOW_CHAT]),
+                _model_service("system.ai.claude-opus-4-8", [MLFLOW_CHAT, ANTHROPIC]),
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        assert found.codex_models == []
+        # Not allowlisted for opencode either, so it lands nowhere.
+        assert found.oss_models == []
+        assert found.claude_ids == ["system.ai.claude-opus-4-8"]
+
+    def test_claude_model_is_not_also_bucketed_as_oss(self, monkeypatch):
+        # Every Claude model advertises mlflow chat-completions too; precedence
+        # must keep it out of the oss bucket.
+        payload = {
+            "model_services": [
+                _model_service("system.ai.claude-opus-4-8", [MLFLOW_CHAT, ANTHROPIC]),
+                _model_service("system.ai.glm-5-2", [MLFLOW_CHAT]),
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        assert found.oss_models == ["system.ai.glm-5-2"]
+        assert found.claude_ids == ["system.ai.claude-opus-4-8"]
+
+    def test_returns_every_claude_id_and_newest_per_family(self, monkeypatch):
+        # claude-fable-5 belongs to no opus/sonnet/haiku family, so it is
+        # reachable only through the full id list.
+        payload = {
+            "model_services": [
+                _model_service(f"system.ai.{m}", [ANTHROPIC])
+                for m in (
+                    "claude-fable-5",
+                    "claude-haiku-4-5",
+                    "claude-opus-4-6",
+                    "claude-opus-4-8",
+                    "claude-sonnet-4-5",
+                    "claude-sonnet-5",
+                )
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        assert len(found.claude_ids) == 6
+        assert "system.ai.claude-fable-5" in found.claude_ids
+        assert found.claude_models == {
+            "opus": "system.ai.claude-opus-4-8",
+            "sonnet": "system.ai.claude-sonnet-5",
+            "haiku": "system.ai.claude-haiku-4-5",
+        }
+
+    def test_responses_model_included_regardless_of_name(self, monkeypatch):
+        # Capability wins over the name rule when the payload advertises one.
+        payload = {"model_services": [_model_service("system.ai.o4-turbo", [RESPONSES])]}
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        assert db_mod.discover_model_services(WS, "token").codex_models == ["system.ai.o4-turbo"]
+
+
 class TestDiscoverModelServices:
+    """Name-rule fallback: workspaces whose payload omits supported_api_types."""
+
     def test_buckets_families_by_name(self, monkeypatch):
         payload = {
             "model_services": [
@@ -176,19 +342,85 @@ def test_buckets_families_by_name(self, monkeypatch):
             db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
         )
 
-        claude, codex, gemini, oss, reason = db_mod.discover_model_services(WS, "token")
+        found = db_mod.discover_model_services(WS, "token")
 
-        assert reason is None
+        assert found.reason is None
         # Newest opus wins; sonnet bucketed; haiku absent.
-        assert claude == {
+        assert found.claude_models == {
             "opus": "system.ai.claude-opus-4-8",
             "sonnet": "system.ai.claude-sonnet-4-6",
         }
-        assert codex == ["system.ai.gpt-5"]
+        assert found.codex_models == ["system.ai.gpt-5"]
         # Gemini ordered newest-first via the shared sort key.
-        assert gemini[0] == "system.ai.gemini-3-5-flash"
+        assert found.gemini_models[0] == "system.ai.gemini-3-5-flash"
         # kimi and glm are the allowlisted OSS families; llama is not.
-        assert oss == ["system.ai.glm-5-2", "system.ai.kimi-k2-7-code"]
+        assert found.oss_models == ["system.ai.glm-5-2", "system.ai.kimi-k2-7-code"]
+
+    def test_gpt_oss_excluded_without_capability_data(self, monkeypatch):
+        # The name rule requires a version digit after `gpt-`, so a workspace
+        # with no supported_api_types still keeps gpt-oss out of codex.
+        payload = {
+            "model_services": [
+                _model_service("system.ai.gpt-oss-120b"),
+                _model_service("system.ai.gpt-5-6"),
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        assert found.codex_models == ["system.ai.gpt-5-6"]
+
+    def test_codex_models_sorted_newest_first(self, monkeypatch):
+        payload = {
+            "model_services": [
+                _model_service("system.ai.gpt-4-1"),
+                _model_service("system.ai.gpt-5-6"),
+                _model_service("system.ai.gpt-5-2-codex"),
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        # Alphabetical order would put gpt-4-1 first; pi and copilot take [0].
+        assert found.codex_models[0] == "system.ai.gpt-5-6"
+
+    def test_two_digit_minor_version_beats_single_digit(self, monkeypatch):
+        # Lexicographic reverse-sort ranks `4-8` above `4-10`.
+        payload = {
+            "model_services": [
+                _model_service("system.ai.claude-opus-4-8"),
+                _model_service("system.ai.claude-opus-4-10"),
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        assert found.claude_models["opus"] == "system.ai.claude-opus-4-10"
+
+    def test_single_segment_version_buckets_into_family(self, monkeypatch):
+        # `claude-sonnet-5` has no dotted minor; it must still be a sonnet.
+        payload = {
+            "model_services": [
+                _model_service("system.ai.claude-sonnet-4-6"),
+                _model_service("system.ai.claude-sonnet-5"),
+            ]
+        }
+        monkeypatch.setattr(
+            db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
+        )
+
+        found = db_mod.discover_model_services(WS, "token")
+
+        assert found.claude_models["sonnet"] == "system.ai.claude-sonnet-5"
 
     def test_oss_allowlist_drops_unsupported_families(self, monkeypatch):
         # Only kimi/glm are allowlisted; other families are dropped.
@@ -206,11 +438,11 @@ def test_oss_allowlist_drops_unsupported_families(self, monkeypatch):
             db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
         )
 
-        claude, codex, gemini, oss, reason = db_mod.discover_model_services(WS, "token")
+        found = db_mod.discover_model_services(WS, "token")
 
-        assert reason is None
-        assert (claude, codex, gemini) == ({}, [], [])
-        assert oss == ["system.ai.glm-5-2", "system.ai.kimi-k2-7-code"]
+        assert found.reason is None
+        assert (found.claude_models, found.codex_models, found.gemini_models) == ({}, [], [])
+        assert found.oss_models == ["system.ai.glm-5-2", "system.ai.kimi-k2-7-code"]
 
     def test_paginates_via_next_page_token(self, monkeypatch):
         pages = {
@@ -231,21 +463,23 @@ def fake_get(url, token, timeout=10):
 
         monkeypatch.setattr(db_mod, "_http_get_json", fake_get)
 
-        claude, codex, _, _, reason = db_mod.discover_model_services(WS, "token")
+        found = db_mod.discover_model_services(WS, "token")
 
-        assert reason is None
-        assert codex == ["system.ai.gpt-5"]
-        assert claude == {"opus": "system.ai.claude-opus-4-8"}
+        assert found.reason is None
+        assert found.codex_models == ["system.ai.gpt-5"]
+        assert found.claude_models == {"opus": "system.ai.claude-opus-4-8"}
 
     def test_http_failure_returns_reason(self, monkeypatch):
         monkeypatch.setattr(
             db_mod, "_http_get_json", lambda url, token, timeout=10: (None, "HTTP 500 Server Error")
         )
 
-        claude, codex, gemini, oss, reason = db_mod.discover_model_services(WS, "token")
+        found = db_mod.discover_model_services(WS, "token")
 
-        assert (claude, codex, gemini, oss) == ({}, [], [], [])
-        assert reason == "HTTP 500 Server Error"
+        assert found.claude_models == {}
+        assert (found.claude_ids, found.codex_models, found.gemini_models) == ([], [], [])
+        assert found.oss_models == []
+        assert found.reason == "HTTP 500 Server Error"
 
     def test_no_matching_families_reports_sample(self, monkeypatch):
         payload = {"model_services": [_model_service("system.ai.llama-4-maverick")]}
@@ -253,10 +487,12 @@ def test_no_matching_families_reports_sample(self, monkeypatch):
             db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
         )
 
-        claude, codex, gemini, oss, reason = db_mod.discover_model_services(WS, "token")
+        found = db_mod.discover_model_services(WS, "token")
 
-        assert (claude, codex, gemini, oss) == ({}, [], [], [])
-        assert reason is not None and "llama-4-maverick" in reason
+        assert found.claude_models == {}
+        assert (found.claude_ids, found.codex_models, found.gemini_models) == ([], [], [])
+        assert found.oss_models == []
+        assert found.reason is not None and "llama-4-maverick" in found.reason
 
     def test_ignores_non_system_ai_schemas(self, monkeypatch):
         # The metastore listing returns services from every schema; only
@@ -274,13 +510,13 @@ def test_ignores_non_system_ai_schemas(self, monkeypatch):
             db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None)
         )
 
-        claude, codex, gemini, oss, reason = db_mod.discover_model_services(WS, "token")
+        found = db_mod.discover_model_services(WS, "token")
 
-        assert reason is None
-        assert codex == ["system.ai.gpt-5"]
-        assert claude == {}  # temp.erni.claude-* must not be bucketed
-        assert gemini == []
-        assert oss == []
+        assert found.reason is None
+        assert found.codex_models == ["system.ai.gpt-5"]
+        assert found.claude_models == {}  # temp.erni.claude-* must not be bucketed
+        assert found.gemini_models == []
+        assert found.oss_models == []
 
     def test_requests_bounded_page_size(self, monkeypatch):
         # The endpoint 499s without a bounded page_size, so every request must
@@ -289,16 +525,31 @@ def test_requests_bounded_page_size(self, monkeypatch):
 
         def fake_get(url, token, timeout=10):
             urls.append(url)
-            return {"model_services": [_model_service("system.ai.gpt-5")]}, None
+            return {"model_services": [_model_service("system.ai.gpt-5", [RESPONSES])]}, None
 
         monkeypatch.setattr(db_mod, "_http_get_json", fake_get)
 
-        ids, reason = db_mod.list_model_services(WS, "token")
+        services, reason = db_mod.list_model_services(WS, "token")
 
-        assert ids == ["system.ai.gpt-5"]
+        assert services == {"system.ai.gpt-5": [RESPONSES]}
         assert reason is None
         assert all("page_size=" in u for u in urls)
 
+    def test_missing_api_types_yields_empty_capability_list(self, monkeypatch):
+        monkeypatch.setattr(
+            db_mod,
+            "_http_get_json",
+            lambda url, token, timeout=10: (
+                {"model_services": [_model_service("system.ai.x")]},
+                None,
+            ),
+        )
+
+        services, reason = db_mod.list_model_services(WS, "token")
+
+        assert services == {"system.ai.x": []}
+        assert reason is None
+
     def test_retries_page_before_giving_up(self, monkeypatch):
         payload = {"model_services": [_model_service("system.ai.gpt-5")]}
         calls = {"n": 0}
@@ -311,10 +562,10 @@ def flaky_get(url, token, timeout=10):
 
         monkeypatch.setattr(db_mod, "_http_get_json", flaky_get)
 
-        ids, reason = db_mod.list_model_services(WS, "token")
+        services, reason = db_mod.list_model_services(WS, "token")
 
         assert reason is None
-        assert ids == ["system.ai.gpt-5"]
+        assert list(services) == ["system.ai.gpt-5"]
         assert calls["n"] == 3  # two failures, third succeeds
 
 
diff --git a/tests/test_e2e_uc.py b/tests/test_e2e_uc.py
index c716dcb..69b9daa 100644
--- a/tests/test_e2e_uc.py
+++ b/tests/test_e2e_uc.py
@@ -17,18 +17,20 @@
 from ucode.cli import configure_shared_state
 from ucode.databricks import (
     discover_model_services,
+    is_responses_model_id,
     list_mcp_services,
 )
 from ucode.state import load_state
 
 
 def _has_uc_models(workspace: str, token: str) -> bool:
-    claude, codex, gemini, oss, _reason = discover_model_services(workspace, token)
-    return bool(claude or codex or gemini or oss)
+    found = discover_model_services(workspace, token)
+    return bool(found.claude_ids or found.codex_models or found.gemini_models or found.oss_models)
 
 
 def _all_resolved_model_ids(state: dict) -> list[str]:
     ids: list[str] = list((state.get("claude_models") or {}).values())
+    ids += state.get("claude_model_ids") or []
     ids += state.get("codex_models") or []
     ids += state.get("gemini_models") or []
     ids += state.get("oss_models") or []
@@ -43,18 +45,19 @@ def _all_resolved_model_ids(state: dict) -> list[str]:
 
 class TestDiscoverModelServicesE2E:
     def test_returns_only_system_ai_models(self, e2e_workspace, e2e_token):
-        claude, codex, gemini, oss, reason = discover_model_services(e2e_workspace, e2e_token)
-        if not (claude or codex or gemini or oss):
-            pytest.skip(f"No system.ai.* model services on workspace: {reason}")
+        found = discover_model_services(e2e_workspace, e2e_token)
+        if not _has_uc_models(e2e_workspace, e2e_token):
+            pytest.skip(f"No system.ai.* model services on workspace: {found.reason}")
         non_system = sorted(
             {
                 m
                 for m in _all_resolved_model_ids(
                     {
-                        "claude_models": claude,
-                        "codex_models": codex,
-                        "gemini_models": gemini,
-                        "oss_models": oss,
+                        "claude_models": found.claude_models,
+                        "claude_model_ids": found.claude_ids,
+                        "codex_models": found.codex_models,
+                        "gemini_models": found.gemini_models,
+                        "oss_models": found.oss_models,
                     }
                 )
                 if not m.startswith("system.ai.")
@@ -62,6 +65,21 @@ def test_returns_only_system_ai_models(self, e2e_workspace, e2e_token):
         )
         assert not non_system, f"Non-system.ai entries leaked through: {non_system[:5]}"
 
+    def test_codex_bucket_holds_only_responses_models(self, e2e_workspace, e2e_token):
+        # `gpt-oss-*` speaks mlflow chat-completions, not the Responses dialect
+        # codex and pi's databricks-openai provider route to.
+        found = discover_model_services(e2e_workspace, e2e_token)
+        if not found.codex_models:
+            pytest.skip("No Responses-capable models on this workspace.")
+        bad = [m for m in found.codex_models if not is_responses_model_id(m)]
+        assert not bad, f"Non-Responses models bucketed as codex: {bad}"
+
+    def test_claude_family_map_is_a_subset_of_claude_ids(self, e2e_workspace, e2e_token):
+        found = discover_model_services(e2e_workspace, e2e_token)
+        if not found.claude_ids:
+            pytest.skip("No Claude models on this workspace.")
+        assert set(found.claude_models.values()) <= set(found.claude_ids)
+
 
 class TestListMcpServicesE2E:
     def test_returns_only_system_ai_mcp_services(self, e2e_workspace, e2e_token):

From 73093d7ac691895abb2b4f2fec1d202e4a4f13dd Mon Sep 17 00:00:00 2001
From: Mark Angler <19395192+MarkAngler@users.noreply.github.com>
Date: Fri, 10 Jul 2026 10:44:28 -0400
Subject: [PATCH 2/2] pi: seed settings.json instead of clobbering the user's
 model choice
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Pi persists the model picked in its own selector to settings.json (via
`setDefaultModelAndProvider`). ucode overwrote `defaultProvider`/`defaultModel`
on every launch *and* every 30 minutes from the token-refresh thread, so the
choice could never survive a relaunch — the "ucode just overwrites them" half
of #204.

`_write_settings` now only writes when there is nothing to preserve or when
what is there is ucode-authored residue that no longer resolves:

- nothing pinned yet — seed it, so Pi's `findInitialModel` can't fall through
  to an env-key-backed provider (e.g. HF_TOKEN exposing huggingface). This is
  the behavior the pin existed for, and it is unchanged.
- pinned to a ucode provider we no longer register (`databricks-openai` on a
  workspace with no Responses models, or a legacy provider name) — repair.
- pinned to a ucode provider we do register, but to a model it no longer
  offers — repair.

A pin naming a provider ucode does not manage is left alone: Pi writes
defaultProvider only on an explicit user selection, so it is a deliberate
choice, not fall-through.

The refresh thread now passes `update_settings=False`. It still rewrites
models.json every 30 minutes — that rotates the bearer token baked into the
file, which is load-bearing — but no longer resets the model.

Refs #204

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/ucode/agents/pi.py | 58 ++++++++++++++++++++++-----
 tests/test_agent_pi.py | 90 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 139 insertions(+), 9 deletions(-)

diff --git a/src/ucode/agents/pi.py b/src/ucode/agents/pi.py
index 7a481cf..c28d624 100644
--- a/src/ucode/agents/pi.py
+++ b/src/ucode/agents/pi.py
@@ -169,7 +169,14 @@ def write_tool_config(
     token: str | None = None,
     *,
     force_refresh: bool = False,
+    update_settings: bool = True,
 ) -> tuple[dict, str]:
+    """Write Pi's models.json, and seed settings.json unless told not to.
+
+    ``update_settings=False`` is used by the background token refresh: it must
+    rewrite models.json to rotate the baked-in bearer, but must not touch the
+    model the user picked in Pi's own model selector.
+    """
     backup_existing_file(PI_CONFIG_PATH, PI_BACKUP_PATH)
     if token is None:
         token = get_databricks_token(
@@ -194,21 +201,47 @@ def write_tool_config(
             providers.pop(stale, None)
     merged = deep_merge_dict(existing, overlay)
     write_json_file(PI_CONFIG_PATH, merged)
-    _write_settings(overlay["model"])
+    if update_settings:
+        _write_settings(overlay["model"], overlay.get("providers") or {})
     state = mark_tool_managed(state, "pi", managed_keys)
     save_state(state)
     return state, token
 
 
-def _write_settings(model_selector: str) -> None:
-    # Pin defaultProvider/defaultModel in settings.json so Pi doesn't fall
-    # through to an env-key-backed provider (e.g. HF_TOKEN exposing
-    # huggingface) in `findInitialModel` when no --model is passed.
+def _settings_need_repair(existing: dict, providers: dict) -> bool:
+    """True when ucode should (re)write Pi's defaultProvider/defaultModel.
+
+    Pi persists the user's model-selector choice into settings.json via
+    `setDefaultModelAndProvider`, so an existing pin is a user preference and
+    must survive. We only write when there is nothing to preserve, or when what
+    is there is ucode-authored residue that no longer resolves:
+
+    - nothing pinned yet — seed it, so Pi's `findInitialModel` can't fall
+      through to an env-key-backed provider (e.g. HF_TOKEN exposing huggingface)
+    - pinned to a ucode provider we no longer register (e.g. `databricks-openai`
+      on a workspace with no Responses models, or a `LEGACY_PROVIDER_NAMES` entry)
+    - pinned to a ucode provider we do register, but to a model it no longer offers
+
+    A pin naming a provider ucode does not manage is the user's own; leave it.
+    """
+    provider = existing.get("defaultProvider")
+    model_id = existing.get("defaultModel")
+    if not isinstance(provider, str) or not isinstance(model_id, str) or not model_id:
+        return True
+    if provider in providers:
+        offered = {m["id"] for m in providers[provider].get("models") or []}
+        return model_id not in offered
+    return provider in (*PROVIDER_NAMES, *LEGACY_PROVIDER_NAMES)
+
+
+def _write_settings(model_selector: str, providers: dict) -> None:
     provider, _, model_id = model_selector.partition("/")
     if not model_id:
         return
-    backup_existing_file(PI_SETTINGS_PATH, PI_SETTINGS_BACKUP_PATH)
     existing = read_json_safe(PI_SETTINGS_PATH)
+    if not _settings_need_repair(existing, providers):
+        return
+    backup_existing_file(PI_SETTINGS_PATH, PI_SETTINGS_BACKUP_PATH)
     merged = deep_merge_dict(existing, {"defaultProvider": provider, "defaultModel": model_id})
     write_json_file(PI_SETTINGS_PATH, merged)
 
@@ -226,18 +259,25 @@ def default_model(state: dict) -> str | None:
     return gemini_models[0] if gemini_models else None
 
 
-def _refresh_token_once(state: dict, *, force_refresh: bool = False) -> str:
+def _refresh_token_once(
+    state: dict, *, force_refresh: bool = False, update_settings: bool = True
+) -> str:
     model = default_model(state)
     if not model:
         raise RuntimeError("No Pi model is available on this workspace.")
-    _, token = write_tool_config(state, model, force_refresh=force_refresh)
+    _, token = write_tool_config(
+        state, model, force_refresh=force_refresh, update_settings=update_settings
+    )
     return token
 
 
 def _refresh_forever(state: dict, stop_event: threading.Event) -> None:
     while not stop_event.wait(TOKEN_REFRESH_INTERVAL_SECONDS):
         try:
-            _refresh_token_once(state, force_refresh=True)
+            # Rotate the bearer baked into models.json, but leave settings.json
+            # alone: mid-session the user may have switched models in Pi's
+            # selector, and Pi persists that choice there.
+            _refresh_token_once(state, force_refresh=True, update_settings=False)
         except RuntimeError:
             continue
 
diff --git a/tests/test_agent_pi.py b/tests/test_agent_pi.py
index 00a3bac..467184c 100644
--- a/tests/test_agent_pi.py
+++ b/tests/test_agent_pi.py
@@ -423,6 +423,96 @@ def test_pre_existing_settings_are_backed_up_before_first_write(self, tmp_path,
         assert merged["defaultProvider"] == "databricks-claude"
         assert merged["theme"] == "Default Dark"
 
+    def _write(self, pi_mod, state, model="claude-sonnet", **kwargs):
+        with (
+            patch("ucode.agents.pi.get_databricks_token", return_value="tok"),
+            patch("ucode.agents.pi.save_state"),
+        ):
+            pi_mod.write_tool_config(state, model, token="tok", **kwargs)
+
+    def test_preserves_user_model_choice(self, tmp_path, monkeypatch):
+        # Pi writes settings.json when the user picks a model in its selector.
+        # ucode must not reset that on the next launch.
+        pi_mod, _, settings_file, _ = self._setup(tmp_path, monkeypatch)
+        settings_file.write_text(
+            json.dumps({"defaultProvider": "databricks-claude", "defaultModel": "claude-opus"}),
+            encoding="utf-8",
+        )
+        state = self._state(claude_model_ids=["claude-sonnet", "claude-opus"])
+
+        self._write(pi_mod, state)
+
+        settings = json.loads(settings_file.read_text())
+        assert settings["defaultModel"] == "claude-opus"
+
+    def test_repairs_pin_to_provider_no_longer_registered(self, tmp_path, monkeypatch):
+        # `databricks-openai` disappears on a workspace with no Responses models;
+        # a settings pin naming it would leave Pi unable to resolve a model.
+        pi_mod, _, settings_file, _ = self._setup(tmp_path, monkeypatch)
+        settings_file.write_text(
+            json.dumps(
+                {"defaultProvider": "databricks-openai", "defaultModel": "system.ai.gpt-oss-120b"}
+            ),
+            encoding="utf-8",
+        )
+
+        self._write(pi_mod, self._state())
+
+        settings = json.loads(settings_file.read_text())
+        assert settings["defaultProvider"] == "databricks-claude"
+        assert settings["defaultModel"] == "claude-sonnet"
+
+    def test_repairs_pin_to_model_no_longer_offered(self, tmp_path, monkeypatch):
+        pi_mod, _, settings_file, _ = self._setup(tmp_path, monkeypatch)
+        settings_file.write_text(
+            json.dumps({"defaultProvider": "databricks-claude", "defaultModel": "claude-retired"}),
+            encoding="utf-8",
+        )
+
+        self._write(pi_mod, self._state())
+
+        assert json.loads(settings_file.read_text())["defaultModel"] == "claude-sonnet"
+
+    def test_leaves_a_non_ucode_provider_pin_alone(self, tmp_path, monkeypatch):
+        # Pi only writes defaultProvider on an explicit user selection, so a
+        # provider ucode doesn't manage is a deliberate choice.
+        pi_mod, _, settings_file, _ = self._setup(tmp_path, monkeypatch)
+        original = {"defaultProvider": "openrouter", "defaultModel": "some/model"}
+        settings_file.write_text(json.dumps(original), encoding="utf-8")
+
+        self._write(pi_mod, self._state())
+
+        assert json.loads(settings_file.read_text()) == original
+
+    def test_refresh_path_does_not_write_settings(self, tmp_path, monkeypatch):
+        pi_mod, _, settings_file, _ = self._setup(tmp_path, monkeypatch)
+
+        self._write(pi_mod, self._state(), update_settings=False)
+
+        assert not settings_file.exists()
+
+    def test_refresh_thread_rotates_token_without_touching_settings(self, tmp_path, monkeypatch):
+        pi_mod, _, _, _ = self._setup(tmp_path, monkeypatch)
+        calls: list[dict] = []
+        monkeypatch.setattr(
+            pi_mod,
+            "write_tool_config",
+            lambda state, model, **kwargs: (calls.append(kwargs), (state, "tok"))[1],
+        )
+        monkeypatch.setattr(pi_mod, "TOKEN_REFRESH_INTERVAL_SECONDS", 0)
+
+        class _FiresOnce:
+            def __init__(self):
+                self.waits = 0
+
+            def wait(self, _timeout):
+                self.waits += 1
+                return self.waits > 1  # run the body once, then exit the loop
+
+        pi_mod._refresh_forever(self._state(), _FiresOnce())
+
+        assert calls == [{"force_refresh": True, "update_settings": False}]
+
 
 class TestValidateAllToolsPiRollback:
     def test_failed_pi_validation_rolls_back_settings(self, tmp_path, monkeypatch):