Resolve managed coding-agent config over local state at launch

src/ucode/cli.py

@@ -55,6 +55,8 @@
     resolve_pat_token,
     run_databricks_login,
 )
+from ucode.managed_config import managed_launch_state
+from ucode.managed_resolve import managed_provider_service
 from ucode.mcp import (
     MCP_CLIENTS,
     SKILLS_MCP_KIND,
@@ -1170,6 +1172,9 @@ def _launch_tool(
         if needs_auto_configure:
             _auto_configure_tool(tool)
         state = ensure_provider_state(tool)
+        # Remembered before the fallback below collapses the two cases: a managed config may not
+        # silently override a provider the user typed on the command line (it errors instead).
+        explicit_provider = provider
         # An explicit --provider overrides the persisted choice; otherwise fall
         # back to whatever `ucode configure` saved for this tool.
         provider = provider or get_provider_service(state, tool)
@@ -1179,17 +1184,6 @@ def _launch_tool(
                 f"{TOOL_SPECS[tool]['display']} smart routing cannot be enabled with "
                 "--provider. Launch without a Model Provider Service and try again."
             )
-        # Validate the provider service before launching — it must exist, be a
-        # provider type this tool can route to (e.g. claude can't use an OpenAI
-        # or Foundry service), and, for Bedrock, expose Claude models to pin.
-        # Surfaces a clear error up front instead of a cryptic gateway failure
-        # mid-session. For a Bedrock service this also returns the model ids.
-        provider_models = None
-        relayed = False
-        if provider:
-            provider_models, error, relayed = resolve_provider_models(tool, state, provider)
-            if error:
-                raise RuntimeError(error)
         # Re-fetch model lists on every launch so newly-added Databricks
         # endpoints show up without a manual `ucode configure` (and so that
         # tools like pi which read multiple model bundles never run on
@@ -1202,6 +1196,38 @@ def _launch_tool(
             skip_model_discovery=bool(provider),
             skip_preflight=skip_preflight,
         )
+        # An admin-published managed config wins over the developer's own settings. Resolved before
+        # the provider and model are settled below, so each is decided once against the values that
+        # will actually be written — the two state files are never merged on disk.
+        state, managed = managed_launch_state(state, tool)
+        if managed is not None:
+            managed_provider = managed_provider_service(managed, tool)
+            if explicit_provider and managed_provider and managed_provider != explicit_provider:
+                # An explicit --provider that disagrees with the admin's is a hard error rather
+                # than a silent override: the user asked for something the managed config forbids,
+                # and quietly routing them elsewhere would hide it.
+                raise RuntimeError(
+                    f"You cannot launch {TOOL_SPECS[tool]['display']} with provider "
+                    f"{explicit_provider} because your admin has specified managed provider "
+                    f"{managed_provider}."
+                )
+            if managed_provider:
+                provider = managed_provider
+        # Validate the provider service before launching — it must exist, be a
+        # provider type this tool can route to (e.g. claude can't use an OpenAI
+        # or Foundry service), and, for Bedrock, expose Claude models to pin.
+        provider_models = None
+        relayed = False
+        if provider:
+            provider_models, error, relayed = resolve_provider_models(tool, state, provider)
+            if error:
+                if managed is not None and provider == managed_provider_service(managed, tool):
+                    # Clear error if the admin has Unity Catalog grants the developer doesn't.
+                    raise RuntimeError(
+                        f"Your admin's managed config specifies provider {provider} for "
+                        f"{TOOL_SPECS[tool]['display']}, which can't be used: {error}"
+                    )
+                raise RuntimeError(error)
         if routing_agent is not None and enable_smart_routing_flag:
             state = routing_agent.enable_smart_routing(state)
         # The router's per-launch pick for the root session. Codex pins it as the
@@ -1241,6 +1267,8 @@ def _launch_tool(
             route_root_model=route_root_model,
         )
         print_section(f"ucode with {TOOL_SPECS[tool]['display']}")
+        if managed is not None:
+            print_kv("Config", "workspace-managed")
         if provider:
             print_kv("Provider", provider)
         elif route_root_model:

src/ucode/managed_config.py

@@ -4,11 +4,13 @@
 (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.
+- normalizing the proto-JSON into a stable internal dict keyed by ucode's own tool names,
+- persisting it to ``~/.ucode/managed-state.json`` (0600), and
+- re-reading it on each launch, falling back to the persisted copy when the read fails.
 
-Reconciliation against the local ``state.json`` and applying the manifest to agents live in later
-changes; this module deliberately stops at "read + normalize + persist".
+:func:`managed_launch_state` is the launch path's entry point: it refreshes the manifest and hands
+back the state to configure the agent with. Deciding *which* value wins for a given key is
+:mod:`ucode.managed_resolve`'s job, kept separate so that logic stays pure and I/O-free.
 """
 
 from __future__ import annotations
@@ -19,7 +21,9 @@
 from typing import cast
 
 import ucode.config_io as config_io
-from ucode.databricks import fetch_managed_coding_agent_configs
+from ucode.databricks import fetch_managed_coding_agent_configs, get_databricks_token
+from ucode.managed_resolve import resolve_state
+from ucode.ui import print_warning
 
 MANAGED_STATE_PATH = config_io.APP_DIR / "managed-state.json"
 
@@ -271,6 +275,10 @@ def save_managed_state(workspace: str, config: dict) -> None:
 
     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.
+
+    An empty ``config`` records "this workspace has no managed config", which matters because the
+    file doubles as the fallback when a later read fails: without it, removing a config server-side
+    would leave the old one on disk to be reapplied after a transient outage.
     """
     if config_io.is_dry_run():
         return
@@ -307,11 +315,67 @@ def load_managed_state(workspace: str | None) -> dict | None:
     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
+def refresh_managed_config(state: dict) -> dict | None:
+    """Fetch the workspace's managed config and persist it, returning the normalized manifest.
+
+    Runs on every launch so a developer picks up an admin's edits without re-running
+    ``ucode configure``. Returns None when the workspace has no managed config — the normal case for
+    a workspace whose admin hasn't published one.
+
+    A failed fetch never blocks the launch: an unreachable control plane shouldn't stop someone from
+    coding. Instead it falls back to the last config persisted for this workspace, so the admin's
+    most recent known policy still applies; only when there is no persisted config either does the
+    launch fall through to the developer's own settings.
+    """
+    workspace = state.get("workspace")
+    if not workspace:
+        return None
     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
+        token = get_databricks_token(workspace, state.get("profile"))
+    except RuntimeError as exc:
+        return _persisted_fallback(workspace, str(exc))
+    managed, reason = get_managed_config(workspace, token)
+    if reason is not None:
+        return _persisted_fallback(workspace, reason)
+    if managed is None:
+        # Record that this workspace has no config, rather than leaving an earlier one on disk:
+        # the file doubles as the fallback above, so a removed policy would otherwise come back
+        # into force after the next transient outage.
+        save_managed_state(workspace, {})
+        return None
+    save_managed_state(workspace, managed)
+    return managed
+
+
+def _persisted_fallback(workspace: str, reason: str) -> dict | None:
+    """Return the last persisted config for ``workspace`` after a failed fetch, warning either way.
+
+    Distinguishes the two outcomes in the warning: continuing on a possibly-stale admin config is
+    materially different from continuing on the developer's own settings.
+    """
+    # An empty persisted config means the last successful read found none, so there is no admin
+    # policy to fall back to — treat it the same as having no file at all.
+    persisted = load_managed_state(workspace)
+    if persisted:
+        print_warning(
+            f"Could not read your workspace's managed config ({reason}); "
+            "using the last one saved for this workspace."
+        )
+        return persisted
+    print_warning(
+        f"Could not read your workspace's managed config ({reason}); using your local settings."
+    )
+    return None
+
+
+def managed_launch_state(state: dict, tool: str) -> tuple[dict, dict | None]:
+    """Return ``(state, managed)`` for launching ``tool`` under any managed config.
+
+    The returned state has the manifest's models and provider layered over the developer's own —
+    managed wins per key — so the settings file written from it reflects the admin's choices. When
+    the workspace has no managed config the state is handed back untouched.
+    """
+    managed = refresh_managed_config(state)
+    if managed is None:
+        return state, None
+    return resolve_state(managed, state, tool), managed

src/ucode/managed_resolve.py

@@ -0,0 +1,115 @@
+"""Resolve the effective agent settings from the managed config plus local ucode state.
+
+The admin-authored manifest (``~/.ucode/managed-state.json``, written by
+:mod:`ucode.managed_config`) and the developer's own ucode state (``~/.ucode/state.json``) stay
+separate files — they are never merged on disk. Instead this module resolves them *per key* at
+config-write time: whatever the manifest specifies wins, and anything it leaves unset falls back to
+the developer's ucode state. The resolved view is what gets rendered into the agent config files
+(e.g. ``~/.claude/ucode-settings.json``), so managed settings take precedence for every ``ucode``
+command without either file being rewritten.
+
+Only settings the developer set *through* ucode participate in the fallback. Settings they wrote by
+hand outside ucode (``~/.claude/settings.json``, etc.) are not read here — Claude Code merges
+those scopes itself at launch, underneath the file ucode passes via ``--settings``.
+
+Everything here is pure: no I/O, no mutation of the inputs. Fetching and persisting the manifest,
+and handing the resolved state to the agent config writers, live in :mod:`ucode.managed_config`.
+"""
+
+from __future__ import annotations
+
+from typing import cast
+
+from ucode.state import MANAGED_OVERLAY_KEY
+
+# Proto model-config slot -> the family key `claude.py`'s render_overlay reads. The manifest keeps
+# the proto spelling (`default_opus_model`), while ucode state and render_overlay both key claude
+# models by bare family (`opus`), so the two have to be bridged before the settings file is written.
+_CLAUDE_FAMILY_SLOTS = {
+    "default_opus_model": "opus",
+    "default_sonnet_model": "sonnet",
+    "default_haiku_model": "haiku",
+    "default_fable_model": "fable",
+}
+
+
+def _as_dict(value: object) -> dict[str, object]:
+    """Return ``value`` as a ``dict[str, object]`` when it is a dict, else an empty dict."""
+    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 _agent_entry(managed: dict, tool: str) -> dict[str, object]:
+    """Return the manifest's config for ``tool``, or an empty dict when it isn't enabled."""
+    enabled = _as_dict(_as_dict(managed).get("enabled_agents"))
+    return _as_dict(enabled.get(tool))
+
+
+def _agent_model_config(managed: dict, tool: str) -> dict[str, object]:
+    """Return the manifest's normalized ``model_config`` for ``tool``, if any."""
+    return _as_dict(_agent_entry(managed, tool).get("model_config"))
+
+
+def effective_agent_models(managed: dict, state: dict, tool: str) -> dict | list | None:
+    """Resolve ``tool``'s model list/slots from the manifest, falling back to ucode state.
+
+    Claude keys its models by family (``opus``/``sonnet``/``haiku``/``fable``) and resolves per
+    family, so a family the manifest omits keeps the developer's value. Every other agent stores a
+    flat list, which has no per-key identity — there the manifest's list replaces the local one
+    outright, or the local one stands when the manifest specifies none.
+    """
+    manifest_models = _agent_model_config(managed, tool).get("models")
+    if tool == "claude":
+        local = dict(_as_dict(state.get("claude_models")))
+        for slot, family in _CLAUDE_FAMILY_SLOTS.items():
+            model = _str(_as_dict(manifest_models).get(slot))
+            if model:
+                local[family] = model
+        return local or None
+    if isinstance(manifest_models, list):
+        models = [m for m in (_str(item) for item in manifest_models) if m]
+        if models:
+            return models
+    local_list = state.get(f"{tool}_models")
+    return local_list if local_list else None
+
+
+def managed_provider_service(managed: dict, tool: str) -> str | None:
+    """Return only the provider the managed config specifies for ``tool``, ignoring local state."""
+    return _str(_agent_model_config(managed, tool).get("model_provider_service"))
+
+
+def resolve_state(managed: dict, state: dict, tool: str) -> dict:
+    """Return a copy of ``state`` with ``tool``'s managed values layered on top.
+
+    ``write_tool_config`` reads its models and provider out of the state dict it is handed, so
+    handing it this resolved copy is what makes managed settings win. Each key the managed config
+    displaces is recorded under :data:`~ucode.state.MANAGED_OVERLAY_KEY` with the developer's own
+    value (None when they had none), which ``save_state`` swaps back before writing — so the admin's
+    settings reach the generated agent config files without ``state.json`` losing what the developer
+    configured. The two files are never merged on disk.
+    """
+    resolved = dict(state)
+    overlay: dict[str, object] = {}
+    models = effective_agent_models(managed, state, tool)
+    models_key = f"{tool}_models"
+    if models is not None and models != state.get(models_key):
+        overlay[models_key] = state.get(models_key)
+        resolved[models_key] = models
+    provider = managed_provider_service(managed, tool)
+    if provider:
+        providers = dict(_as_dict(state.get("provider_services")))
+        if providers.get(tool) != provider:
+            overlay["provider_services"] = state.get("provider_services")
+            providers[tool] = provider
+            resolved["provider_services"] = providers
+    if overlay:
+        resolved[MANAGED_OVERLAY_KEY] = overlay
+    return resolved

src/ucode/state.py

@@ -13,6 +13,10 @@
 
 STATE_PATH = APP_DIR / "state.json"
 STATE_VERSION = 3
+# Transient key holding the developer's own values for whatever a managed config layered over them.
+# Present only in memory: the layered values render the agent settings files, while `save_state`
+# restores what's under it so `state.json` keeps recording the developer's own configuration.
+MANAGED_OVERLAY_KEY = "_managed_overlay"
 AUTH_COMMAND_TIMEOUT_MS = 5000
 AUTH_REFRESH_INTERVAL_MS = 900_000
 
@@ -42,21 +46,47 @@ def load_state() -> dict:
 
 
 def save_state(state: dict) -> None:
-    """Save workspace state back into the per-workspace structure."""
+    """Save workspace state back into the per-workspace structure.
+
+    Values a managed config layered over the developer's own are stripped first (see
+    ``MANAGED_OVERLAY_KEY``), so an admin-published config takes effect through the generated agent
+    settings files without overwriting what the developer configured for themselves. Read
+    non-destructively: a launch can save more than once from the same dict (e.g. the relayed proxy
+    rewriting its port), and every one of those writes must restore the developer's values.
+    """
     if is_dry_run():
         return
     full = load_full_state()
     workspace = state.get("workspace") or full.get("current_workspace")
     if workspace:
         full["current_workspace"] = workspace
-        full["workspaces"][workspace] = hydrate_state(state)
+        full["workspaces"][workspace] = hydrate_state(_without_managed_overlay(state))
     try:
         APP_DIR.mkdir(parents=True, exist_ok=True)
         STATE_PATH.write_text(json.dumps(full, indent=2), encoding="utf-8")
     except OSError as exc:
         raise RuntimeError(f"Failed to write state file: {STATE_PATH}") from exc
 
 
+def _without_managed_overlay(state: dict) -> dict:
+    """Return ``state`` with managed-config values swapped back for the developer's own.
+
+    Returns a new dict and leaves ``state`` untouched, so the caller keeps the layered values it
+    needs for rendering and repeated saves stay idempotent.
+    """
+    overlay = state.get(MANAGED_OVERLAY_KEY)
+    if not isinstance(overlay, dict):
+        return state
+    persisted = {key: value for key, value in state.items() if key != MANAGED_OVERLAY_KEY}
+    for key, value in overlay.items():
+        # A key the developer never set is dropped rather than persisted as None.
+        if value is None:
+            persisted.pop(key, None)
+        else:
+            persisted[key] = value
+    return persisted
+
+
 def set_current_workspace(workspace: str | None) -> None:
     """Set ``current_workspace`` without touching the per-workspace blocks.