diff --git a/src/ucode/databricks.py b/src/ucode/databricks.py index 1d32f31..d9cf166 100644 --- a/src/ucode/databricks.py +++ b/src/ucode/databricks.py @@ -1374,6 +1374,31 @@ def discover_model_services( return claude_models, codex_models, gemini_models, oss_models, None +# --- Managed coding-agent config (admin-authored, developer-read) ----------- + +# The workspace-admin authors a CodingAgentConfig via the AI Gateway; developers read it +# (non-admin) through the List endpoint and apply it locally. +_CODING_AGENT_CONFIGS_API_PATH = "/api/ai-gateway/v2/coding-agent-configs" + + +def fetch_managed_coding_agent_configs(workspace: str, token: str) -> tuple[list[dict], str | None]: + """List the workspace's managed CodingAgentConfig(s) via the AI Gateway.""" + hostname = workspace_hostname(workspace) + url = f"https://{hostname}{_CODING_AGENT_CONFIGS_API_PATH}" + payload, reason = _http_get_json(url, token, timeout=30) + if reason is not None: + return [], reason + if isinstance(payload, dict): + configs = payload.get("coding_agent_configs") or [] + elif isinstance(payload, list): + configs = payload + else: + return [], "coding-agent-configs listing returned an unexpected response shape" + if not isinstance(configs, list): + return [], "coding-agent-configs listing returned an unexpected response shape" + return [c for c in configs if isinstance(c, dict)], None + + # --- MCP services (parallel to model services) ----------------------------- diff --git a/src/ucode/managed_config.py b/src/ucode/managed_config.py new file mode 100644 index 0000000..ec39805 --- /dev/null +++ b/src/ucode/managed_config.py @@ -0,0 +1,317 @@ +"""Admin-authored managed coding-agent config: fetch, normalize, and local persistence. + +An org admin authors a ``CodingAgentConfig`` through the Databricks AI Gateway; developers read it +(non-admin) and ``ucode`` applies it locally. This module owns the developer-read half: + +- fetching the raw manifest (via :func:`ucode.databricks.fetch_managed_coding_agent_configs`), +- normalizing the proto-JSON into a stable internal dict keyed by ucode's own tool names, and +- persisting it to ``~/.ucode/managed-state.json`` (0600) so launches can reconcile against it. + +Reconciliation against the local ``state.json`` and applying the manifest to agents live in later +changes; this module deliberately stops at "read + normalize + persist". +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import cast + +import ucode.config_io as config_io +from ucode.databricks import fetch_managed_coding_agent_configs + +MANAGED_STATE_PATH = config_io.APP_DIR / "managed-state.json" + +# Shown to a developer when their workspace has no admin-defined managed config yet — the normal +# case, not an error. Kept here so the CLI (which surfaces it) uses one consistent message. +NO_MANAGED_CONFIG_MESSAGE = "No coding-agent config has been set up by your workspace admin yet." + +# CodingAgent proto enum -> ucode tool name. Anything unrecognized (e.g. a newer agent this ucode +# build doesn't know) is dropped during normalization rather than guessed at. +_AGENT_ENUM_TO_TOOL: dict[str, str] = { + "CODING_AGENT_CLAUDE_CODE": "claude", + "CODING_AGENT_CODEX": "codex", + "CODING_AGENT_GEMINI": "gemini", + "CODING_AGENT_COPILOT": "copilot", + "CODING_AGENT_PI": "pi", + "CODING_AGENT_OPENCODE": "opencode", +} + +# McpServerType proto enum -> ucode's short type tag. Mirrors the selection prefixes in ``mcp.py``; +# the actual name->URL resolution happens there when the manifest is applied (a later change). +_MCP_TYPE_ENUM_TO_TAG: dict[str, str] = { + "MCP_SERVER_TYPE_UC_SERVICE": "mcp-service", + "MCP_SERVER_TYPE_EXTERNAL": "external", + "MCP_SERVER_TYPE_GENIE": "genie-space", + "MCP_SERVER_TYPE_VECTOR_SEARCH": "vector-search", + "MCP_SERVER_TYPE_UC_FUNCTIONS": "uc-functions", + "MCP_SERVER_TYPE_DATABRICKS_APP": "app", + "MCP_SERVER_TYPE_DATABRICKS_SQL": "sql", +} + + +def _as_dict(value: object) -> dict[str, object]: + """Return ``value`` as a ``dict[str, object]`` when it is a dict, else an empty dict. + + Centralizes the isinstance-narrowing so downstream ``.get`` calls type-check (a bare + ``isinstance(x, dict)`` narrows to ``dict[Never, Never]``, which rejects string keys).""" + return cast("dict[str, object]", value) if isinstance(value, dict) else {} + + +def _str(value: object) -> str | None: + """Return a non-empty stripped string, or None.""" + if isinstance(value, str): + stripped = value.strip() + return stripped or None + return None + + +def _str_list(value: object) -> list[str]: + if not isinstance(value, list): + return [] + out: list[str] = [] + for item in value: + s = _str(item) + if s: + out.append(s) + return out + + +def _normalize_model_config(model_config: object) -> dict | None: + """Normalize an ``AgentModelConfig`` oneof into ``{model_provider_service?, default_model?, + models}``. + + The proto is a oneof over per-agent variants (claude/codex/opencode/pi/gemini/copilot). We + don't care which variant tag it is here — the enclosing agent already tells us — so we read the + common fields. Claude's ``models`` is a dict of family slots; the rest are a flat list. Returns + None when there's no usable model config. + """ + mc = _as_dict(model_config) + if not mc: + return None + # Unwrap the oneof: take whichever single variant sub-dict is present. + variant = next((_as_dict(v) for v in mc.values() if isinstance(v, dict)), None) + if not variant: + return None + result: dict = {} + mps = _str(variant.get("model_provider_service")) + if mps: + result["model_provider_service"] = mps + default_model = _str(variant.get("default_model")) + if default_model: + result["default_model"] = default_model + models = variant.get("models") + if isinstance(models, dict): + # Claude family slots (default_opus_model, default_sonnet_model, ...). + slots = {k: _str(v) for k, v in _as_dict(models).items() if _str(v)} + if slots: + result["models"] = slots + else: + model_list = _str_list(models) + if model_list: + result["models"] = model_list + return result or None + + +def _normalize_enabled_agent(entry: object) -> tuple[str, dict] | None: + """Normalize one ``EnabledAgent`` into ``(tool, agent_config)``, or None if unusable. + + Drops entries whose agent enum is unset/unknown to this ucode build. + """ + entry_dict = _as_dict(entry) + if not entry_dict: + return None + tool = _AGENT_ENUM_TO_TOOL.get(_str(entry_dict.get("agent")) or "") + if tool is None: + return None + config_in = _as_dict(entry_dict.get("config")) + agent_config: dict = {} + if isinstance(config_in.get("use_as_global_settings"), bool): + agent_config["use_as_global_settings"] = config_in["use_as_global_settings"] + headers = config_in.get("custom_headers") + if isinstance(headers, dict): + clean = { + k: v for k, v in _as_dict(headers).items() if isinstance(k, str) and isinstance(v, str) + } + if clean: + agent_config["custom_headers"] = clean + tracing_table = _tracing_table(config_in.get("tracing_config")) + if tracing_table: + agent_config["tracing_table"] = tracing_table + model_config = _normalize_model_config(config_in.get("model_config")) + if model_config is not None: + agent_config["model_config"] = model_config + return tool, agent_config + + +def _tracing_table(tracing: object) -> str | None: + """Extract ``TracingConfig.table`` (a UC table FQN), or None.""" + return _str(_as_dict(tracing).get("table")) + + +def _normalize_mcp_servers(value: object) -> list[dict]: + if not isinstance(value, list): + return [] + out: list[dict] = [] + for entry in value: + entry_dict = _as_dict(entry) + name = _str(entry_dict.get("name")) + tag = _MCP_TYPE_ENUM_TO_TAG.get(_str(entry_dict.get("type")) or "") + if name and tag: + out.append({"name": name, "type": tag}) + return out + + +def _normalize_budget_policy(value: object) -> dict | None: + bp = _as_dict(value) + if not bp: + return None + policy: dict = {} + display_name = _str(bp.get("display_name")) + if display_name: + policy["display_name"] = display_name + budget_id = _str(bp.get("budget_id")) + if budget_id: + policy["budget_id"] = budget_id + tiers: list[dict] = [] + raw_tiers = bp.get("tiers") + for tier in raw_tiers if isinstance(raw_tiers, list) else []: + tier_dict = _as_dict(tier) + pct = tier_dict.get("spending_percentage") + if not isinstance(pct, (int, float)) or isinstance(pct, bool): + continue + tier_out: dict = {"spending_percentage": float(pct)} + agent = _AGENT_ENUM_TO_TOOL.get(_str(tier_dict.get("default_agent")) or "") + if agent: + tier_out["default_agent"] = agent + model = _str(tier_dict.get("default_model")) + if model: + tier_out["default_model"] = model + tiers.append(tier_out) + if tiers: + policy["tiers"] = tiers + return policy or None + + +def normalize_managed_config(raw: dict) -> dict: + """Normalize a raw ``CodingAgentConfig`` proto-JSON dict into ucode's internal shape. + + The internal shape uses ucode's own tool names and short MCP type tags so downstream reconcile + and apply code never touches proto enum spellings. Unknown agents / MCP types are dropped. + """ + raw = _as_dict(raw) + result: dict = {} + name = _str(raw.get("name")) + if name: + result["name"] = name + default_agent = _AGENT_ENUM_TO_TOOL.get(_str(raw.get("default_agent")) or "") + if default_agent: + result["default_agent"] = default_agent + enabled_agents: dict[str, dict] = {} + raw_agents = raw.get("enabled_agents") + for entry in raw_agents if isinstance(raw_agents, list) else []: + normalized = _normalize_enabled_agent(entry) + if normalized is not None: + tool, agent_config = normalized + enabled_agents[tool] = agent_config + if enabled_agents: + result["enabled_agents"] = enabled_agents + mcp_servers = _normalize_mcp_servers(raw.get("mcp_servers")) + if mcp_servers: + result["mcp_servers"] = mcp_servers + skill_names = _str_list(_as_dict(raw.get("skills")).get("names")) + if skill_names: + result["skills"] = {"names": skill_names} + tracing_table = _tracing_table(raw.get("tracing")) + if tracing_table: + result["tracing_table"] = tracing_table + budget_policy = _normalize_budget_policy(raw.get("budget_policy")) + if budget_policy is not None: + result["budget_policy"] = budget_policy + return result + + +def get_managed_config(workspace: str, token: str) -> tuple[dict | None, str | None]: + """Fetch and normalize the workspace's managed config. + + Returns ``(config, reason)``: + - ``(config, None)`` — the normalized manifest for the workspace's single config; + - ``(None, None)`` — no managed config is defined for the workspace (not an error); + - ``(None, reason)`` — the read failed; ``reason`` says why. + + "No config defined" arrives two ways depending on the backend: an empty listing (HTTP 200 + with no configs) or a NOT_FOUND (HTTP 404). Both are the normal, non-error case for a workspace + whose admin hasn't set one up, so both collapse to ``(None, None)``. + + v0 stores at most one config per workspace, so the first entry is the workspace's config. + """ + configs, reason = fetch_managed_coding_agent_configs(workspace, token) + if reason is not None: + # A NOT_FOUND means the admin hasn't defined a config for this workspace — not a failure. + if _is_not_found(reason): + return None, None + return None, reason + if not configs: + return None, None + return normalize_managed_config(configs[0]), None + + +def _is_not_found(reason: str) -> bool: + """True when a read failure reason indicates the config simply doesn't exist yet. + + ``_http_get_json`` formats failures as ``HTTP [: ]``; a NOT_FOUND surfaces + as an ``HTTP 404`` there (and the API's error body carries ``NOT_FOUND``).""" + lowered = reason.lower() + return "http 404" in lowered or "not_found" in lowered + + +def save_managed_state(workspace: str, config: dict) -> None: + """Persist the normalized managed config to ``~/.ucode/managed-state.json`` at mode 0600. + + The file is org-authored, not developer-editable — 0600 keeps it readable/writable only by the + user (a light guard; hard enforcement / sudo ownership is a separate concern). No-op in dry-run. + """ + if config_io.is_dry_run(): + return + payload = {"workspace": workspace, "config": config} + config_io.ensure_parent_dir(MANAGED_STATE_PATH) + try: + MANAGED_STATE_PATH.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") + except OSError as exc: + raise RuntimeError(f"Failed to write managed state file: {MANAGED_STATE_PATH}") from exc + _restrict_permissions(MANAGED_STATE_PATH) + + +def _restrict_permissions(path: Path) -> None: + """Best-effort chmod 0600. No-op where unsupported (e.g. Windows), where the effective + read-only guarantee is left to a later change.""" + try: + os.chmod(path, 0o600) + except (OSError, NotImplementedError): + pass + + +def load_managed_state(workspace: str | None) -> dict | None: + """Load the persisted managed config for ``workspace``, or None if absent/mismatched. + + Returns the normalized config dict (the ``config`` field), only when the stored file is for the + same workspace — so a stale file from another workspace is ignored rather than misapplied. + """ + if not workspace: + return None + data = config_io.read_json_safe(MANAGED_STATE_PATH) + if data.get("workspace") != workspace: + return None + config = data.get("config") + return config if isinstance(config, dict) else None + + +def delete_managed_state() -> None: + """Remove the managed-state file, if any. No-op in dry-run.""" + if config_io.is_dry_run(): + return + try: + MANAGED_STATE_PATH.unlink(missing_ok=True) + except OSError as exc: + raise RuntimeError(f"Failed to remove managed state file: {MANAGED_STATE_PATH}") from exc diff --git a/tests/test_e2e.py b/tests/test_e2e.py index 025ced4..001ac69 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -10,10 +10,13 @@ from __future__ import annotations +import atexit import json import os import shutil import subprocess +import tempfile +from pathlib import Path from urllib import error as urllib_error from urllib import request as urllib_request @@ -64,6 +67,17 @@ def _run_agent( ) +def _codex_home_outside_tmp() -> Path: + """Create a fresh CODEX_HOME under the user's home dir, registered for cleanup at exit. + + pytest's ``tmp_path`` lives under ``/tmp``; codex (>=0.134) refuses to create its helper + binaries when ``CODEX_HOME`` is under a temporary dir, so launching codex from ``tmp_path`` + fails before doing anything. Rooting CODEX_HOME under ``$HOME`` sidesteps that guard.""" + home = Path(tempfile.mkdtemp(prefix=".ucode-e2e-codex-", dir=Path.home())) + atexit.register(shutil.rmtree, home, ignore_errors=True) + return home + + def _run_gemini_gateway_smoke(workspace: str, model: str, token: str) -> str: """Call the Gemini gateway directly with a text-only prompt. @@ -438,7 +452,7 @@ def test_launch_codex_per_model(self, tmp_path, monkeypatch, e2e_state, e2e_work models = self._codex_models(e2e_state) monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path) - config_dir = tmp_path / "codex_home" / ".codex" + config_dir = _codex_home_outside_tmp() / ".codex" config_dir.mkdir(parents=True) config_path = config_dir / "ucode.config.toml" backup_path = tmp_path / "codex-config.backup.toml" @@ -536,13 +550,21 @@ class TestModelProviderLaunch: @staticmethod def _first_service(tool: str, workspace: str, token: str) -> str: - names, reason = list_tool_provider_services(tool, workspace, token) + services, reason = list_model_provider_services(workspace, token) if is_model_provider_feature_unavailable(reason): pytest.skip("Model Provider Service feature not enabled on this workspace") if reason is not None: pytest.skip(f"could not list provider services: {reason}") + # Relayed (subscription-relay) services can only be invoked through the credential-swap + # launch path, so the plain provider launch these tests exercise gets a 400. Skip them and + # pick a normal service instead. + names = [ + s["name"] for s in services if service_usable_for_tool(tool, s) and not s.get("relayed") + ] if not names: - pytest.skip(f"no {tool} model provider services available on this workspace") + pytest.skip( + f"no non-relayed {tool} model provider services available on this workspace" + ) return names[0] @staticmethod @@ -554,10 +576,17 @@ def test_launch_claude_through_provider( self, tmp_path, monkeypatch, e2e_state, e2e_workspace, e2e_token ): import ucode.config_io as config_io_mod - from ucode.agents import claude + from ucode.agents import claude, resolve_provider_models _require_binary("claude") provider = self._first_service("claude", e2e_workspace, e2e_token) + state = {**e2e_state, "workspace": e2e_workspace} + # Resolve the provider's models exactly as the launch path does: an Anthropic service + # returns None (canonical names route via the header), while a Bedrock service returns the + # per-family provider-side ids to pin — without which the gateway 403s ("not in the allowed + # models list") because Claude Code's canonical name isn't a Bedrock-routable model. + provider_models, error, _relayed = resolve_provider_models("claude", state, provider) + assert error is None, f"provider={provider} could not resolve models: {error}" config_dir = tmp_path / "claude_config" config_dir.mkdir() @@ -567,10 +596,8 @@ def test_launch_claude_through_provider( with pytest.MonkeyPatch().context() as mp: mp.setattr("ucode.state.save_state", lambda s: None) - # No model pinned — the provider header (written into the settings - # env block) routes the agent's own canonical model name. claude.write_tool_config( - {**e2e_state, "workspace": e2e_workspace}, None, provider=provider + state, None, provider=provider, provider_models=provider_models ) env = { @@ -597,7 +624,7 @@ def test_launch_codex_through_provider( provider = self._first_service("codex", e2e_workspace, e2e_token) monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path) - config_dir = tmp_path / "codex_home" / ".codex" + config_dir = _codex_home_outside_tmp() / ".codex" config_dir.mkdir(parents=True) monkeypatch.setattr(codex, "CODEX_CONFIG_PATH", config_dir / "ucode.config.toml") monkeypatch.setattr(codex, "CODEX_BACKUP_PATH", tmp_path / "codex-config.backup.toml") diff --git a/tests/test_managed_config.py b/tests/test_managed_config.py new file mode 100644 index 0000000..6229a44 --- /dev/null +++ b/tests/test_managed_config.py @@ -0,0 +1,249 @@ +"""Tests for managed_config.py — fetch/normalize/persist of the admin-authored managed config.""" + +from __future__ import annotations + +import os +import stat + +import pytest + +import ucode.databricks as db_mod +import ucode.managed_config as mc_mod +from ucode.managed_config import ( + get_managed_config, + load_managed_state, + normalize_managed_config, + save_managed_state, +) + +# A representative raw CodingAgentConfig proto-JSON manifest (mirrors what the API returns). +RAW_MANIFEST = { + "name": "coding-agent-configs/abc-123", + "workspace_id": 1653573648247579, + "default_agent": "CODING_AGENT_CLAUDE_CODE", + "enabled_agents": [ + { + "agent": "CODING_AGENT_CLAUDE_CODE", + "config": { + "use_as_global_settings": True, + "custom_headers": {"x-databricks-workspace": "eng-ml-inference"}, + "tracing_config": {"table": "main.default.ucode_traces"}, + "model_config": { + "claude": { + "default_model": "system.ai.claude-opus-4-8", + "models": { + "default_opus_model": "system.ai.claude-opus-4-8", + "default_sonnet_model": "system.ai.claude-sonnet-4-6", + "default_haiku_model": "system.ai.claude-haiku-4-5", + }, + } + }, + }, + }, + { + "agent": "CODING_AGENT_OPENCODE", + "config": { + "model_config": { + "opencode": { + "default_model": "system.ai.claude-opus-4-8", + "models": ["system.ai.claude-opus-4-8", "system.ai.kimi-k2-7-code"], + } + } + }, + }, + ], + "mcp_servers": [ + {"name": "system.ai.github", "type": "MCP_SERVER_TYPE_UC_SERVICE"}, + {"name": "some-space-id", "type": "MCP_SERVER_TYPE_GENIE"}, + ], + "skills": {"names": ["system.ai.pdf-extraction"]}, + "tracing": {"table": "main.default.ucode_traces"}, + "budget_policy": { + "display_name": "paved-path", + "budget_id": "c6563b45-df9a-4b19-afb2-d42dc2b52576", + "tiers": [ + { + "spending_percentage": 0.8, + "default_agent": "CODING_AGENT_CLAUDE_CODE", + "default_model": "system.ai.claude-sonnet-4-6", + }, + { + "spending_percentage": 1.0, + "default_agent": "CODING_AGENT_OPENCODE", + "default_model": "system.ai.kimi-k2-7-code", + }, + ], + }, +} + + +class TestNormalize: + def test_full_manifest_maps_enums_to_tool_names(self): + cfg = normalize_managed_config(RAW_MANIFEST) + assert cfg["name"] == "coding-agent-configs/abc-123" + assert cfg["default_agent"] == "claude" + assert set(cfg["enabled_agents"]) == {"claude", "opencode"} + + def test_claude_agent_config_fields(self): + claude = normalize_managed_config(RAW_MANIFEST)["enabled_agents"]["claude"] + assert claude["use_as_global_settings"] is True + assert claude["custom_headers"] == {"x-databricks-workspace": "eng-ml-inference"} + assert claude["tracing_table"] == "main.default.ucode_traces" + assert claude["model_config"]["default_model"] == "system.ai.claude-opus-4-8" + assert claude["model_config"]["models"]["default_opus_model"] == "system.ai.claude-opus-4-8" + + def test_opencode_model_list_is_flat(self): + opencode = normalize_managed_config(RAW_MANIFEST)["enabled_agents"]["opencode"] + assert opencode["model_config"]["models"] == [ + "system.ai.claude-opus-4-8", + "system.ai.kimi-k2-7-code", + ] + + def test_mcp_servers_map_type_enums_to_tags(self): + mcp = normalize_managed_config(RAW_MANIFEST)["mcp_servers"] + assert mcp == [ + {"name": "system.ai.github", "type": "mcp-service"}, + {"name": "some-space-id", "type": "genie-space"}, + ] + + def test_skills_and_tracing_and_budget(self): + cfg = normalize_managed_config(RAW_MANIFEST) + assert cfg["skills"] == {"names": ["system.ai.pdf-extraction"]} + assert cfg["tracing_table"] == "main.default.ucode_traces" + assert cfg["budget_policy"]["budget_id"] == "c6563b45-df9a-4b19-afb2-d42dc2b52576" + assert cfg["budget_policy"]["tiers"][1]["default_agent"] == "opencode" + + @pytest.mark.parametrize("agent_enum", ["CODING_AGENT_FUTURE", "CODING_AGENT_UNSPECIFIED"]) + def test_unrecognized_agent_enum_dropped(self, agent_enum): + raw = {"enabled_agents": [{"agent": agent_enum, "config": {}}]} + assert "enabled_agents" not in normalize_managed_config(raw) + + def test_unknown_mcp_type_dropped(self): + raw = {"mcp_servers": [{"name": "x", "type": "MCP_SERVER_TYPE_UNSPECIFIED"}]} + assert "mcp_servers" not in normalize_managed_config(raw) + + def test_empty_manifest_yields_empty_dict(self): + assert normalize_managed_config({}) == {} + + +class TestGetManagedConfig: + def test_returns_normalized_first_config(self, monkeypatch): + monkeypatch.setattr( + mc_mod, "fetch_managed_coding_agent_configs", lambda ws, tok: ([RAW_MANIFEST], None) + ) + cfg, reason = get_managed_config("https://ws", "tok") + assert reason is None + assert cfg["default_agent"] == "claude" + + def test_no_config_is_not_an_error(self, monkeypatch): + monkeypatch.setattr( + mc_mod, "fetch_managed_coding_agent_configs", lambda ws, tok: ([], None) + ) + cfg, reason = get_managed_config("https://ws", "tok") + assert cfg is None + assert reason is None + + def test_fetch_failure_surfaces_reason(self, monkeypatch): + monkeypatch.setattr( + mc_mod, + "fetch_managed_coding_agent_configs", + lambda ws, tok: ([], "HTTP 500 Server Error"), + ) + cfg, reason = get_managed_config("https://ws", "tok") + assert cfg is None + assert reason == "HTTP 500 Server Error" + + @pytest.mark.parametrize( + "not_found_reason", + [ + "HTTP 404 Not Found", + 'HTTP 404 Not Found: {"error_code":"NOT_FOUND","message":"..."}', + 'HTTP 400 Bad Request: {"error_code":"NOT_FOUND"}', + ], + ) + def test_not_found_is_treated_as_no_config(self, monkeypatch, not_found_reason): + # A NOT_FOUND from the read means the admin hasn't defined a config — the normal + # no-config case, not an error, so it collapses to (None, None). + monkeypatch.setattr( + mc_mod, + "fetch_managed_coding_agent_configs", + lambda ws, tok: ([], not_found_reason), + ) + cfg, reason = get_managed_config("https://ws", "tok") + assert cfg is None + assert reason is None + + +class TestPersistence: + @pytest.fixture(autouse=True) + def _managed_path(self, tmp_path, monkeypatch): + path = tmp_path / ".ucode" / "managed-state.json" + monkeypatch.setattr(mc_mod, "MANAGED_STATE_PATH", path) + return path + + def test_save_then_load_round_trips(self, _managed_path): + cfg = normalize_managed_config(RAW_MANIFEST) + save_managed_state("https://ws.example.com", cfg) + loaded = load_managed_state("https://ws.example.com") + assert loaded == cfg + + def test_saved_file_is_0600(self, _managed_path): + save_managed_state("https://ws.example.com", {"default_agent": "claude"}) + mode = stat.S_IMODE(os.stat(_managed_path).st_mode) + # Owner-only read/write; no group/other bits. + assert mode == 0o600 + + def test_load_ignores_other_workspace(self, _managed_path): + save_managed_state("https://ws-a.example.com", {"default_agent": "claude"}) + assert load_managed_state("https://ws-b.example.com") is None + + def test_load_missing_returns_none(self, _managed_path): + assert load_managed_state("https://ws.example.com") is None + + def test_load_none_workspace_returns_none(self, _managed_path): + assert load_managed_state(None) is None + + def test_delete_removes_file(self, _managed_path): + save_managed_state("https://ws.example.com", {"default_agent": "claude"}) + assert _managed_path.exists() + mc_mod.delete_managed_state() + assert not _managed_path.exists() + + def test_delete_missing_is_noop(self, _managed_path): + mc_mod.delete_managed_state() # should not raise + + +class TestFetchClient: + """fetch_managed_coding_agent_configs lives in databricks.py; test its response parsing.""" + + def test_extracts_configs_list(self, monkeypatch): + payload = {"coding_agent_configs": [RAW_MANIFEST]} + monkeypatch.setattr( + db_mod, + "_http_get_json", + lambda url, token, timeout=10: (payload, None), + ) + configs, reason = db_mod.fetch_managed_coding_agent_configs("https://ws", "tok") + assert reason is None + assert len(configs) == 1 + assert configs[0]["default_agent"] == "CODING_AGENT_CLAUDE_CODE" + + def test_empty_list_when_no_configs(self, monkeypatch): + monkeypatch.setattr( + db_mod, + "_http_get_json", + lambda url, token, timeout=10: ({}, None), + ) + configs, reason = db_mod.fetch_managed_coding_agent_configs("https://ws", "tok") + assert configs == [] + assert reason is None + + def test_http_failure_surfaces_reason(self, monkeypatch): + monkeypatch.setattr( + db_mod, + "_http_get_json", + lambda url, token, timeout=10: (None, "HTTP 403 Forbidden"), + ) + configs, reason = db_mod.fetch_managed_coding_agent_configs("https://ws", "tok") + assert configs == [] + assert reason == "HTTP 403 Forbidden"