diff --git a/src/ucode/cli.py b/src/ucode/cli.py index 624c18524..407513ead 100644 --- a/src/ucode/cli.py +++ b/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: diff --git a/src/ucode/databricks.py b/src/ucode/databricks.py index 6286a0276..f98fea711 100644 --- a/src/ucode/databricks.py +++ b/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,14 +2444,19 @@ 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) @@ -2362,6 +2464,8 @@ def discover_model_services( 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" diff --git a/tests/conftest.py b/tests/conftest.py index 3b8ff90f2..85e216486 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/test_claude_search_discovery.py b/tests/test_claude_search_discovery.py index f83f00d10..ffb51161d 100644 --- a/tests/test_claude_search_discovery.py +++ b/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")) diff --git a/tests/test_cli.py b/tests/test_cli.py index c7953a1e3..fc01459bb 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -5121,7 +5121,9 @@ def _stub_deps(monkeypatch, *, pat_token, existing_state=None): monkeypatch.setattr( cli_mod, "probe_unity_gateway_capabilities", lambda w, t: MODEL_SERVICE_PROBE ) - monkeypatch.setattr(cli_mod, "discover_model_services", lambda w, t: ({}, [], [], [], None)) + monkeypatch.setattr( + cli_mod, "discover_model_services", lambda w, t, **_kw: ({}, [], [], [], None) + ) 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)) @@ -5143,6 +5145,40 @@ def test_use_pat_exports_bearer_and_skips_login(self, monkeypatch): assert state["use_pat"] is True assert saved and saved[-1]["use_pat"] is True + def test_configure_passes_probe_claude_by_wanted_tools(self, monkeypatch): + cli_mod, *_ = self._stub_deps(monkeypatch, pat_token="dapi-pat") + seen: list[bool] = [] + + def fake_ms(w, t, *, probe_claude): + seen.append(probe_claude) + return {}, [], [], [], None + + monkeypatch.setattr(cli_mod, "discover_model_services", fake_ms) + cli_mod.configure_shared_state(self.WS, profile="DEFAULT", tools=["codex"]) + cli_mod.configure_shared_state(self.WS, profile="DEFAULT", tools=["claude"]) + cli_mod.configure_shared_state(self.WS, profile="DEFAULT") + assert seen == [False, True, True] + + def test_all_claude_denied_skips_unprobed_fallback(self, monkeypatch): + cli_mod, *_ = self._stub_deps(monkeypatch, pat_token="dapi-pat") + monkeypatch.setattr( + cli_mod, + "discover_model_services", + lambda w, t, **_kw: ({}, ["gpt-5-5"], [], [], db_mod.CLAUDE_ALL_DENIED_REASON), + ) + + def boom(w, t): + raise AssertionError("must not list denied Claude models unprobed") + + monkeypatch.setattr(cli_mod, "discover_claude_models", boom) + state = cli_mod.configure_shared_state(self.WS, profile="DEFAULT", tools=["claude"]) + + assert state["claude_models"] == {} + reasons = state["_discovery_reasons"] + assert reasons["claude"] == db_mod.CLAUDE_ALL_DENIED_REASON + assert "permission" in reasons["claude"] + assert reasons["codex"] is None + def test_happy_path_prints_success_without_model_service_detail(self, monkeypatch, capsys): cli_mod, *_ = self._stub_deps(monkeypatch, pat_token="dapi-pat") @@ -5281,7 +5317,7 @@ def test_uc_models_used_without_legacy_fallback(self, monkeypatch): monkeypatch.setattr( cli_mod, "discover_model_services", - lambda w, t: ( + lambda w, t, **_kw: ( {"opus": "system.ai.claude-opus-4-8"}, ["system.ai.gpt-5"], [], @@ -5308,7 +5344,7 @@ def test_codex_only_configure_persists_discovered_oss_models(self, monkeypatch): monkeypatch.setattr( cli_mod, "discover_model_services", - lambda w, t: ( + lambda w, t, **_kw: ( {}, ["system.ai.gpt-5-6-sol"], [], @@ -5331,7 +5367,7 @@ def _stub_with_fable(self, monkeypatch): monkeypatch.setattr( cli_mod, "discover_model_services", - lambda w, t: ( + lambda w, t, **_kw: ( {"fable": "system.ai.claude-fable-5", "opus": "system.ai.claude-opus-4-8"}, [], [], @@ -5370,7 +5406,7 @@ def test_legacy_fable_toggle_is_discarded(self, monkeypatch, legacy_enabled): monkeypatch.setattr( cli_mod, "discover_model_services", - lambda w, t: ({"fable": "system.ai.claude-fable-5"}, [], [], [], None), + lambda w, t, **_kw: ({"fable": "system.ai.claude-fable-5"}, [], [], [], None), ) state = cli_mod.configure_shared_state(self.WS, profile="DEFAULT") @@ -5416,12 +5452,14 @@ 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, **_kw: ({}, [], [], [], "no model services"), ) monkeypatch.setattr( cli_mod, "discover_claude_models", - lambda w, t: ( + lambda w, t, **_kw: ( {"opus": "databricks-claude-opus-4-8", "sonnet": "databricks-claude-sonnet-4-6"}, None, ), @@ -5479,7 +5517,9 @@ def _stub_external_deps(monkeypatch): monkeypatch.setattr( cli_mod, "probe_unity_gateway_capabilities", lambda w, t: MODEL_SERVICE_PROBE ) - monkeypatch.setattr(cli_mod, "discover_model_services", lambda w, t: ({}, [], [], [], None)) + monkeypatch.setattr( + cli_mod, "discover_model_services", lambda w, t, **_kw: ({}, [], [], [], None) + ) 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)) @@ -6513,6 +6553,30 @@ def test_redirect_rebinds_stdout_without_touching_the_descriptor(self, capfd): assert "Gateway status" in captured.err +class TestToolsWantClaude: + def test_only_claude_consumers_trigger_probe(self): + assert cli_mod._tools_want_claude(None) + assert cli_mod._tools_want_claude({"claude"}) + assert cli_mod._tools_want_claude({"codex", "pi"}) + assert not cli_mod._tools_want_claude({"codex"}) + assert not cli_mod._tools_want_claude({"gemini", "codex"}) + + @pytest.mark.parametrize("tools", [{"codex"}, {"claude"}, {"claude", "codex"}]) + def test_status_never_probes(self, tools): + state = {**MINIMAL_STATE, "profile": "p"} + with ( + patch("ucode.cli.get_databricks_token", return_value="token"), + patch( + "ucode.cli.discover_model_services", + return_value=({}, ["c"], [], [], None), + ) as ms, + patch("ucode.cli.discover_claude_models", return_value=({}, None)), + patch("ucode.cli.discover_gemini_models", return_value=([], None)), + ): + cli_mod._live_status_model_state(state, tools) + assert ms.call_args.kwargs["probe_claude"] is False + + class TestLaunchWithManagedFilePicker: PICKER_MODELS = {"opus": "system.ai.claude-opus-5-5", "haiku": "system.ai.claude-haiku-4-5"} diff --git a/tests/test_databricks.py b/tests/test_databricks.py index d2b569ea6..e574ff845 100644 --- a/tests/test_databricks.py +++ b/tests/test_databricks.py @@ -520,6 +520,151 @@ def test_uncapped_model_returns_none(self): assert db_mod.model_token_limits("system.ai.kimi-k2-7-code") is None +# conftest stubs the probe for offline tests; keep the real one for the probe tests below. +_REAL_CLAUDE_PROBE = db_mod._claude_model_is_callable + +_DENIED = ( + "HTTP 403 Forbidden: User does not have permission to use model service " + "'system.ai.claude-opus-5-5'" +) + + +class TestUncallableClaudeModels: + def _listing(self, monkeypatch, ids): + payload = {"model_services": [_model_service(i) for i in ids]} + monkeypatch.setattr( + db_mod, "_http_get_json", lambda url, token, timeout=10: (payload, None) + ) + + def _post(self, monkeypatch, denied, error=None): + calls = [] + + def fake(url, token, payload, *, timeout=10, headers=None): + calls.append(payload["model"]) + if payload["model"] in denied: + return None, _DENIED + return ({"id": "msg"}, None) if error is None else (None, error) + + monkeypatch.setattr(db_mod, "_http_post_json", fake) + monkeypatch.setattr(db_mod, "_claude_model_is_callable", _REAL_CLAUDE_PROBE) + return calls + + def test_denied_model_excluded_and_not_default(self, monkeypatch): + self._listing( + monkeypatch, + [ + "system.ai.claude-opus-5-5", + "system.ai.claude-opus-4-8", + "system.ai.claude-sonnet-4-6", + ], + ) + self._post(monkeypatch, {"system.ai.claude-opus-5-5"}) + claude, *_ = db_mod.discover_model_services(WS, "token", probe_claude=True) + assert claude == { + "opus": "system.ai.claude-opus-4-8", + "sonnet": "system.ai.claude-sonnet-4-6", + } + + def test_family_with_no_callable_model_is_dropped(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5", "system.ai.claude-sonnet-4-6"]) + self._post(monkeypatch, {"system.ai.claude-opus-5-5"}) + claude, *_ = db_mod.discover_model_services(WS, "token", probe_claude=True) + assert claude == {"sonnet": "system.ai.claude-sonnet-4-6"} + from ucode.agents import claude as claude_agent + + assert ( + claude_agent.default_model({"claude_models": claude}) == "system.ai.claude-sonnet-4-6" + ) + + def test_accessible_models_unaffected_and_cached(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5", "system.ai.claude-opus-4-8"]) + calls = self._post(monkeypatch, set()) + claude, *_ = db_mod.discover_model_services(WS, "token", probe_claude=True) + assert claude == {"opus": "system.ai.claude-opus-5-5"} + assert calls == ["system.ai.claude-opus-5-5"] + assert db_mod._claude_model_is_callable(WS, "token", "system.ai.claude-opus-5-5") + assert calls == ["system.ai.claude-opus-5-5"] + + @pytest.mark.parametrize( + "error", + [ + "HTTP 500 Internal Server Error", + "network error: timed out", + "HTTP 429 Too Many", + "HTTP 403 Forbidden: Source IP is blocked by an IP access list", + "HTTP 403 Forbidden: Budget limit exceeded for this user", + "HTTP 403 Forbidden: OAuth token does not have required scopes", + ], + ) + def test_probe_failure_keeps_current_behavior(self, monkeypatch, error): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5", "system.ai.claude-opus-4-8"]) + self._post(monkeypatch, set(), error=error) + claude, *_ = db_mod.discover_model_services(WS, "token", probe_claude=True) + assert claude == {"opus": "system.ai.claude-opus-5-5"} + + def test_cache_is_scoped_to_identity(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5"]) + calls = self._post(monkeypatch, set()) + assert db_mod._claude_model_is_callable(WS, "token-a", "system.ai.claude-opus-5-5") + assert len(calls) == 1 + # Same workspace, different identity: must re-probe, and may get a different answer. + self._post(monkeypatch, {"system.ai.claude-opus-5-5"}) + assert db_mod._claude_model_is_callable(WS, "token-a", "system.ai.claude-opus-5-5") + assert not db_mod._claude_model_is_callable(WS, "token-b", "system.ai.claude-opus-5-5") + + def test_probe_skipped_when_claude_not_wanted(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5", "system.ai.gpt-5-5"]) + calls = self._post(monkeypatch, {"system.ai.claude-opus-5-5"}) + claude, codex, *_ = db_mod.discover_model_services(WS, "token", probe_claude=False) + assert calls == [] + assert claude == {"opus": "system.ai.claude-opus-5-5"} + assert codex == ["system.ai.gpt-5-5"] + + def test_all_denied_omits_families(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5", "system.ai.claude-sonnet-4-6"]) + self._post(monkeypatch, {"system.ai.claude-opus-5-5", "system.ai.claude-sonnet-4-6"}) + claude, *_, reason = db_mod.discover_model_services(WS, "token", probe_claude=True) + assert claude == {} + assert reason == db_mod.CLAUDE_ALL_DENIED_REASON + + def test_all_denied_flagged_even_when_other_families_exist(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5", "system.ai.gpt-5-5"]) + self._post(monkeypatch, {"system.ai.claude-opus-5-5"}) + claude, codex, *_, reason = db_mod.discover_model_services(WS, "token", probe_claude=True) + assert claude == {} + assert codex == ["system.ai.gpt-5-5"] + assert reason == db_mod.CLAUDE_ALL_DENIED_REASON + + def test_unprobed_listing_never_flags_denied(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5"]) + *_, reason = db_mod.discover_model_services(WS, "token") + assert reason is None + + def test_probing_is_opt_in(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5"]) + calls = self._post(monkeypatch, {"system.ai.claude-opus-5-5"}) + claude, *_ = db_mod.discover_model_services(WS, "token") + assert calls == [] + assert claude == {"opus": "system.ai.claude-opus-5-5"} + + def test_probe_capped_per_family(self, monkeypatch): + models = [f"system.ai.claude-opus-{v}" for v in ("5-5", "5-4", "5-3", "5-2")] + self._listing(monkeypatch, models) + calls = self._post(monkeypatch, set(models)) + claude, *_ = db_mod.discover_model_services(WS, "token", probe_claude=True) + assert len(calls) == 2 + # Candidates past the cap are unprobed (inconclusive), so the next one is kept. + assert claude == {"opus": "system.ai.claude-opus-5-3"} + + def test_deadline_keeps_remaining_candidates(self, monkeypatch): + self._listing(monkeypatch, ["system.ai.claude-opus-5-5", "system.ai.claude-opus-4-8"]) + calls = self._post(monkeypatch, {"system.ai.claude-opus-5-5"}) + monkeypatch.setattr(db_mod, "_CLAUDE_PROBE_DEADLINE", 0.0) + claude, *_ = db_mod.discover_model_services(WS, "token", probe_claude=True) + assert calls == [] + assert claude == {"opus": "system.ai.claude-opus-5-5"} + + class TestDiscoverModelServices: def test_buckets_families_by_name(self, monkeypatch): payload = {