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_agent_config_enabled, managed_launch_state
+from ucode.managed_resolve import managed_default_model, managed_provider_service
 from ucode.mcp import (
     MCP_CLIENTS,
     SKILLS_MCP_KIND,
@@ -1170,10 +1172,53 @@ 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)
         routing_agent = _ROUTING_AGENTS.get(tool)
+        # 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
+        # stale state from before a tool added a new bundle). Under a provider
+        # this heavy discovery is skipped (only a web-search model is fetched).
+        state = configure_shared_state(
+            state["workspace"],
+            profile=state.get("profile"),
+            tools=[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.
+        managed = None
+        if managed_agent_config_enabled():
+            # The spinner covers the read; the outcome is printed after it so the line survives
+            # (a spinner erases itself, leaving nothing to explain which settings won).
+            with spinner("Checking for a managed coding agent config..."):
+                state, managed = managed_launch_state(state, tool, skip_preflight=skip_preflight)
+            if managed is not None:
+                print_success("Applied your workspace's managed coding agent config")
+            else:
+                print_note("No managed coding agent config found; using your own settings")
+        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
+        # Checked after the managed config settles `provider`: an admin-set provider must trip this
+        # guard too, or routing would be persisted as on while a provider is active.
         if routing_agent is not None and enable_smart_routing_flag and provider:
             raise RuntimeError(
                 f"{TOOL_SPECS[tool]['display']} smart routing cannot be enabled with "
@@ -1182,26 +1227,18 @@ def _launch_tool(
         # 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:
+                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)
-        # 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
-        # stale state from before a tool added a new bundle). Under a provider
-        # this heavy discovery is skipped (only a web-search model is fetched).
-        state = configure_shared_state(
-            state["workspace"],
-            profile=state.get("profile"),
-            tools=[tool],
-            skip_model_discovery=bool(provider),
-            skip_preflight=skip_preflight,
-        )
         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
@@ -1214,7 +1251,12 @@ def _launch_tool(
             # the workspace has no matching Databricks models.
             resolved_model = None
         else:
-            state, resolved_model = resolve_launch_model(tool, state, None)
+            # A managed default_model is the model the admin wants sessions to start on, so it goes
+            # in as the explicit model rather than being applied afterwards: for codex the proto has
+            # no model list at all, so passing it here is the only way a launch succeeds when the
+            # workspace's own discovery turned up nothing.
+            managed_model = managed_default_model(managed, tool) if managed is not None else None
+            state, resolved_model = resolve_launch_model(tool, state, managed_model)
             if routing_agent is not None and routing_agent.smart_routing_enabled(state):
                 display = TOOL_SPECS[tool]["display"]
                 with spinner(f"Selecting a {display} model with smart routing..."):
@@ -1231,6 +1273,14 @@ def _launch_tool(
                     print_warning(
                         f"Smart routing was unavailable ({routing_error}); using {resolved_model}."
                     )
+            # The admin's model outranks a smart-routing pick too. Claude only launches on it when
+            # pinned as ANTHROPIC_MODEL (route_root_model); other agents take `resolved_model`,
+            # which already holds it from resolve_launch_model above.
+            if managed_model:
+                if tool == "claude":
+                    route_root_model = managed_model
+                else:
+                    resolved_model = managed_model
         state = configure_tool(
             tool,
             state,
@@ -1241,6 +1291,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,10 +21,16 @@
 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"
 
+# Opt-in switch while the feature is in bug bash: unset means launches ignore managed configs
+# entirely and behave exactly as they did before.
+MANAGED_CONFIG_ENV_VAR = "ENABLE_MANAGED_AGENT_CONFIG"
+
 # 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."
@@ -237,12 +245,15 @@ def get_managed_config(workspace: str, token: str) -> tuple[dict | None, str | N
 
     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.
+    - ``(None, None)`` — the workspace definitively has no managed config (not an error);
+    - ``(None, reason)`` — the read didn't settle the question; ``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)``.
+    The distinction matters to callers that cache: only ``(None, None)`` is authoritative enough to
+    clear a previously stored config. "No config defined" arrives two ways depending on the backend
+    — an empty listing (HTTP 200 with no configs) or a NOT_FOUND — and both collapse to
+    ``(None, None)``. Anything else, including a PERMISSION_DENIED, leaves the question unanswered
+    and is surfaced as a failure: an admin may have published a config the developer can't read,
+    which they need to know about rather than silently launch without.
 
     v0 stores at most one config per workspace, so the first entry is the workspace's config.
     """
@@ -258,19 +269,34 @@ def get_managed_config(workspace: str, token: str) -> tuple[dict | None, str | N
 
 
 def _is_not_found(reason: str) -> bool:
-    """True when a read failure reason indicates the config simply doesn't exist yet.
+    """True when a read failure reason means the workspace definitively has no managed config.
 
     ``_http_get_json`` formats failures as ``HTTP <code> <text>[: <body>]``; 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 _is_permission_denied(reason: str) -> bool:
+    """True when the read was refused rather than answering whether a config exists.
+
+    The read is meant to be available to any workspace user, so a refusal means the workspace's
+    managed config isn't readable by this developer — worth telling them about, since an admin may
+    have published a config that silently isn't reaching them. It settles nothing about whether one
+    exists, so a cached config is left in place rather than cleared."""
+    lowered = reason.lower()
+    return "http 403" in lowered or "permission_denied" 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.
+
+    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 +333,118 @@ 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:
+        # A refused read leaves the cached config alone: it says nothing about whether the admin's
+        # config still exists, unlike a successful "no config" answer below.
+        return _persisted_fallback(workspace, reason, refused=_is_permission_denied(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, *, refused: bool = False) -> dict | None:
+    """Return the last persisted config for ``workspace`` after a failed fetch.
+
+    Warns only when there is a config to fall back on, because then the launch proceeds on an admin
+    policy that may be out of date. With nothing persisted there is no managed config in play at
+    all, so staying quiet keeps someone with (say) an expired session from being told about a
+    feature they don't use — including when the read was ``refused``, since a refusal is no evidence
+    that a config exists.
+    """
+    # 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 not persisted:
+        return None
+    summary = _summarize_read_failure(reason)
+    if refused:
+        print_warning(
+            f"Your workspace's managed config is not readable by you ({summary}); using the last "
+            "one saved for this workspace. Ask an admin to grant access."
+        )
+    else:
+        print_warning(
+            f"Could not read your workspace's managed config ({summary}); "
+            "using the last one saved for this workspace."
+        )
+    return persisted
+
+
+def _summarize_read_failure(reason: str) -> str:
+    """Condense a read failure into one short line fit for a terminal warning.
+
+    ``_http_get_json`` appends the raw response body, which for a gateway error is a multi-line JSON
+    blob (error_code, message, request_id, trace ids). Surface just the status and the API's own
+    message; the full text is still available under ``UCODE_DEBUG=1``.
+    """
+    status, _, body = reason.partition(": ")
+    body = body.strip()
+    if body.startswith("{"):
+        try:
+            parsed = json.loads(body)
+        except json.JSONDecodeError:
+            parsed = None
+        if isinstance(parsed, dict):
+            message = _str(parsed.get("message")) or _str(parsed.get("error_code"))
+            if message:
+                return f"{status.strip()}: {message}"
+        return status.strip()
+    condensed = " ".join(reason.split())
+    return condensed if len(condensed) <= 160 else condensed[:157] + "..."
+
+
+def managed_agent_config_enabled() -> bool:
+    """True when managed coding-agent configs are switched on for this run.
+
+    Opt-in while the feature is being bug-bashed: without the env var set, launches behave exactly
+    as they did before and never read the workspace's config."""
+    return os.environ.get(MANAGED_CONFIG_ENV_VAR, "").strip().lower() in ("1", "true", "yes")
+
+
+def managed_launch_state(
+    state: dict, tool: str, *, skip_preflight: bool = False
+) -> 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.
+
+    ``skip_preflight`` mirrors the launch flag: managed/headless launchers pass it to avoid
+    per-launch network calls, so the config is read from the last persisted copy instead of being
+    re-fetched (which means it can be arbitrarily stale until a normal launch refreshes it).
+    """
+    if not managed_agent_config_enabled():
+        return state, None
+    if skip_preflight:
+        managed = load_managed_state(state.get("workspace")) or None
+    else:
+        managed = refresh_managed_config(state)
+    if managed is None:
+        return state, None
+    return resolve_state(managed, state, tool), managed