Skip Claude models the user cannot call when picking defaults

src/ucode/cli.py

@@ -6,7 +6,7 @@
 import os
 import shutil
 import subprocess
-from collections.abc import Iterator
+from collections.abc import Collection, Iterator
 from contextlib import contextmanager, redirect_stdout
 from enum import StrEnum
 from importlib import metadata
@@ -53,6 +53,7 @@
     ensure_custom_oauth_cli_token,
 )
 from ucode.databricks import (
+    CLAUDE_ALL_DENIED_REASON,
     SKILLS_MCP_MIN_DATABRICKS_CLI_VERSION,
     apply_pat_environment,
     build_shared_base_urls,
@@ -625,14 +626,7 @@ def configure_shared_state(
     else:
         print_warning(f"Model service: {model_service_probe.detail}")
 
-    want_claude = (
-        fetch_all
-        or "claude" in tools
-        or "opencode" in tools
-        or "kilo" in tools
-        or "copilot" in tools
-        or "pi" in tools
-    )
+    want_claude = fetch_all or _tools_want_claude(tools)
     want_gemini = (
         fetch_all or "gemini" in tools or "opencode" in tools or "kilo" in tools or "pi" in tools
     )
@@ -672,11 +666,16 @@ def configure_shared_state(
         # family only.
         with spinner("Fetching available models..."):
             ms_claude, ms_codex, ms_gemini, ms_oss, ms_reason = discover_model_services(
-                workspace, token
+                workspace, token, probe_claude=want_claude
             )
+            claude_denied = ms_reason == CLAUDE_ALL_DENIED_REASON
+            if claude_denied:
+                ms_reason = None  # applies to Claude only
             if want_claude:
                 claude_models, claude_reason = ms_claude, ms_reason
-                if not claude_models:
+                if claude_denied:
+                    claude_reason = CLAUDE_ALL_DENIED_REASON
+                elif not claude_models:
                     claude_models, claude_reason = discover_claude_models(workspace, token)
             if want_gemini:
                 gemini_models, gemini_reason = ms_gemini, ms_reason
@@ -1208,6 +1207,14 @@ def _status_default_model(tool: str, state: dict, models: list[str]) -> str | No
     return models[0] if models and tool in ("gemini", "opencode", "kilo", "copilot", "pi") else None
 
 
+_CLAUDE_MODEL_CONSUMERS = ("claude", "opencode", "kilo", "copilot", "pi")
+
+
+def _tools_want_claude(tools: Collection[str] | None) -> bool:
+    """True when any selected tool consumes the discovered Claude models (None = all tools)."""
+    return tools is None or any(t in tools for t in _CLAUDE_MODEL_CONSUMERS)
+
+
 def _live_status_model_state(state: dict, tools: set[str]) -> tuple[dict, str]:
     """Return a fresh, read-only model inventory and its freshness label."""
     workspace = state.get("workspace")
@@ -1224,7 +1231,11 @@ def _live_status_model_state(state: dict, tools: set[str]) -> tuple[dict, str]:
         with spinner("Refreshing live workspace models..."):
             token = get_databricks_token(workspace, profile)
             claude_models, codex_models, gemini_models, oss_models, shared_reason = (
-                discover_model_services(workspace, token)
+                discover_model_services(
+                    workspace,
+                    token,
+                    probe_claude=False,  # probes are billed; configure does them, not status
+                )
             )
             reasons: dict[str, str | None] = {}
             if not claude_models:

src/ucode/databricks.py

@@ -5,6 +5,7 @@
 
 import configparser
 import functools
+import hashlib
 import json
 import logging
 import logging.handlers
@@ -21,6 +22,7 @@
 import threading
 import time
 from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from datetime import datetime
 from decimal import Decimal, InvalidOperation
@@ -2163,6 +2165,7 @@ def clear_model_services_cache() -> None:
     """Forget cached model-service listings (used by tests, and after a workspace switch)."""
     _MODEL_SERVICES_CACHE.clear()
     _MODEL_PROVIDER_SERVICES_CACHE.clear()
+    _CLAUDE_CALLABLE_CACHE.clear()
 
 
 def has_cached_model_provider_services(workspace: str, parent: str | None = None) -> bool:
@@ -2325,8 +2328,96 @@ def discover_claude_models_unbucketed(workspace: str, token: str) -> tuple[list[
     return [m for m in ids if "claude-" in m.lower()], None
 
 
+# Per-process memo of `_claude_model_is_callable`, keyed by (workspace, token fingerprint, model id)
+# so a profile/PAT switch on the same workspace never reuses another identity's answer. Only
+# definite answers are stored; a probe that could not tell is never cached.
+_CLAUDE_CALLABLE_CACHE: dict[tuple[str, str, str], bool] = {}
+
+
+_CLAUDE_PROBE_TIMEOUT = 10
+_CLAUDE_PROBE_MAX_CANDIDATES = 2  # newest N per family; a deeper fallback is left unprobed
+_CLAUDE_PROBE_DEADLINE = 15.0  # overall seconds across all families
+
+
+def _token_fingerprint(token: str) -> str:
+    return hashlib.sha256(token.encode()).hexdigest()[:16]
+
+
+def clear_claude_callable_cache() -> None:
+    _CLAUDE_CALLABLE_CACHE.clear()
+
+
+def _claude_model_is_callable(
+    workspace: str, token: str, model_id: str, *, timeout: int = _CLAUDE_PROBE_TIMEOUT
+) -> bool:
+    """False only when the gateway answers a 1-token request with an EXECUTE-denial 403.
+
+    The model-services listing does not say which services the caller may invoke, so a model the
+    user lacks EXECUTE on still shows up. Any other outcome (success, OAuth-scope 403, IP-ACL /
+    budget / policy 403, 5xx, network error) returns True: an inconclusive probe must keep
+    today's behavior.
+    """
+    key = (workspace, _token_fingerprint(token), model_id)
+    cached = _CLAUDE_CALLABLE_CACHE.get(key)
+    if cached is not None:
+        return cached
+    _payload, reason = _http_post_json(
+        f"{workspace_origin(workspace)}/ai-gateway/anthropic/v1/messages",
+        token,
+        {"model": model_id, "max_tokens": 1, "messages": [{"role": "user", "content": "hi"}]},
+        timeout=timeout,
+    )
+    if reason is None:
+        callable_ = True
+    elif _looks_like_execute_denial(reason):
+        callable_ = False
+    else:
+        return True
+    _CLAUDE_CALLABLE_CACHE[key] = callable_
+    return callable_
+
+
+def _drop_uncallable_claude_models(
+    workspace: str, token: str, candidates: dict[str, list[str]]
+) -> dict[str, str]:
+    """Pick the newest callable model per Claude family; drop families with none.
+
+    Families are probed in parallel, newest candidate first, stopping at the first callable one.
+    Probing is bounded: at most ``_CLAUDE_PROBE_MAX_CANDIDATES`` candidates per family and an
+    overall ``_CLAUDE_PROBE_DEADLINE``. Anything not probed in time is inconclusive and kept, so the
+    result matches the unprobed path. A family whose probed candidates were all definitively
+    EXECUTE-denied is omitted rather than defaulted to a model we know is denied.
+    """
+    if not candidates:
+        return {}
+    deadline = time.monotonic() + _CLAUDE_PROBE_DEADLINE
+
+    def pick(models: list[str]) -> str | None:
+        for model in models[:_CLAUDE_PROBE_MAX_CANDIDATES]:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return model  # out of time: inconclusive, keep
+            if _claude_model_is_callable(
+                workspace, token, model, timeout=max(1, min(_CLAUDE_PROBE_TIMEOUT, int(remaining)))
+            ):
+                return model
+        if len(models) > _CLAUDE_PROBE_MAX_CANDIDATES:
+            return models[_CLAUDE_PROBE_MAX_CANDIDATES]  # beyond the probe cap: unprobed, keep
+        return None
+
+    with ThreadPoolExecutor(max_workers=len(candidates)) as pool:
+        picks = dict(zip(candidates, pool.map(pick, candidates.values()), strict=True))
+    return {family: model for family, model in picks.items() if model}
+
+
+CLAUDE_ALL_DENIED_REASON = (
+    "you don't have permission to use any Claude model on this workspace "
+    "(EXECUTE is denied on every Claude model service)"
+)
+
+
 def discover_model_services(
-    workspace: str, token: str
+    workspace: str, token: str, *, probe_claude: bool = False
 ) -> tuple[dict[str, str], list[str], list[str], list[str], str | None]:
     """Discover models via UC model-services and bucket them by family name.
 
@@ -2339,6 +2430,12 @@ def discover_model_services(
     - ``gemini_models`` is the list of ``system.ai.*gemini-*`` ids, newest first.
     - ``oss_models`` is the list of OSS-model ``system.ai.*`` ids.
 
+    ``probe_claude`` gates the billed 1-token callability probes used to skip Claude models the
+    user cannot call; opt-in, pass True only when a Claude agent is wanted. A family whose candidates are all
+    denied is omitted. When every Claude family is denied, ``reason`` is ``CLAUDE_ALL_DENIED_REASON``
+    (even if other families were found) so callers skip the unprobed ``discover_claude_models``
+    fallback, which would list the same denied models.
+
     ``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.
@@ -2347,21 +2444,28 @@ def discover_model_services(
     if not ids:
         return {}, [], [], [], reason
 
-    claude_models: dict[str, str] = {}
+    claude_candidates: dict[str, list[str]] = {}
     for family in ANTHROPIC_FAMILIES:
         candidates = sorted(
             [m for m in ids if f"claude-{family}-" in m],
             reverse=True,
         )
         if candidates:
-            claude_models[family] = candidates[0]
+            claude_candidates[family] = candidates
+    if probe_claude:
+        claude_models = _drop_uncallable_claude_models(workspace, token, claude_candidates)
+    else:
+        claude_models = {family: models[0] for family, models in claude_candidates.items()}
+    claude_denied = bool(probe_claude and claude_candidates and not claude_models)
 
     codex_models = sorted([m for m in ids if "gpt-" in m], key=model_version_sort_key)
     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)]
 
     if not (claude_models or codex_models or gemini_models or oss_models):
+        if claude_denied:
+            return {}, [], [], [], CLAUDE_ALL_DENIED_REASON
         sample = ", ".join(ids[:5])
         return (
             {},
@@ -2373,7 +2477,13 @@ def discover_model_services(
                 f"claude/gpt/gemini/oss families (got: {sample})"
             ),
         )
-    return claude_models, codex_models, gemini_models, oss_models, None
+    return (
+        claude_models,
+        codex_models,
+        gemini_models,
+        oss_models,
+        CLAUDE_ALL_DENIED_REASON if claude_denied else None,
+    )
 
 
 # --- Managed coding-agent config (admin-authored, developer-read) -----------
@@ -3692,6 +3802,21 @@ def _looks_like_permission_failure(reason: str) -> bool:
     return "HTTP 403" in reason
 
 
+def _looks_like_execute_denial(reason: str) -> bool:
+    """True only for the gateway's model-service EXECUTE denial.
+
+    Real wording: "User does not have permission to use model service 'system.ai.claude-opus-5-5'".
+    Other 403s (IP ACL, budget, policy, OAuth scope) say nothing about this model, so they are
+    "unknown" and must not drop a family.
+    """
+    lowered = reason.lower()
+    return (
+        "http 403" in lowered
+        and "permission to use model service" in lowered
+        and not _looks_like_scope_failure(reason)
+    )
+
+
 CODING_AGENT_RECOMMEND_MODEL_PATH = "/api/ai-gateway/v2/coding-agent-configs:recommendModel"
 
 

tests/conftest.py

@@ -92,6 +92,9 @@ def reject_privileged_write(path, _desired_text):
     # The model-services listing is memoized for the life of the process, so without this a cached
     # result would leak into the next test and make a stubbed listing look like it was never called.
     databricks_mod.clear_model_services_cache()
+    # The Claude callability probe POSTs to the gateway; keep tests offline unless they opt in.
+    monkeypatch.setattr(databricks_mod, "_claude_model_is_callable", lambda *_a, **_k: True)
+    databricks_mod.clear_claude_callable_cache()
     databricks_mod.clear_workspace_org_id_cache()
     # The resolved Databricks CLI path/discovery is cached for the life of the
     # process; a real resolution (or a prior test's patched one) must not leak

tests/test_claude_search_discovery.py

@@ -47,7 +47,7 @@ def search_setup(tmp_path, monkeypatch):
     monkeypatch.setattr(
         cli,
         "discover_model_services",
-        lambda *_: ({"opus": CLAUDE_MODEL}, [SEARCH_MODEL], [], [], None),
+        lambda *_, **_kw: ({"opus": CLAUDE_MODEL}, [SEARCH_MODEL], [], [], None),
     )
     monkeypatch.setattr(claude, "refresh_managed_config", lambda *_: SimpleNamespace(manifest=None))
 
@@ -76,7 +76,7 @@ def test_claude_only_setup_registers_search_without_prior_codex_setup(
         monkeypatch.setattr(
             cli,
             "discover_model_services",
-            lambda *_: ({"opus": CLAUDE_MODEL}, [], [], [], None),
+            lambda *_, **_kw: ({"opus": CLAUDE_MODEL}, [], [], [], None),
         )
         monkeypatch.setattr(cli, "discover_codex_models", lambda *_: ([SEARCH_MODEL], None))
     else:
@@ -110,7 +110,7 @@ def test_claude_only_setup_without_search_models_preserves_existing_search(
     monkeypatch.setattr(
         cli,
         "discover_model_services",
-        lambda *_: ({"opus": CLAUDE_MODEL}, [], [], [], None),
+        lambda *_, **_kw: ({"opus": CLAUDE_MODEL}, [], [], [], None),
     )
     monkeypatch.setattr(cli, "discover_codex_models", lambda *_: ([], "No available GPT models"))