Add ucode setup: interactive managed-config authoring

.gitignore

@@ -5,3 +5,4 @@ build/
 dist/
 .venv/
 .DS_Store
+.isaac/

README.md

@@ -163,6 +163,40 @@ ucode configure skills --location main.default,ml.prod --mcp
 Each run prints the registered server, its URL, the configured agents, and its tools, and reminds
 you to run `ucode <agent>` (existing agent sessions need a restart before the MCP tools load).
 
+### Managed config for a workspace (admins)
+
+```bash
+ucode setup
+```
+
+Author the coding config your developers pick up automatically, instead of asking each of them to
+run `ucode configure` by hand. Restricted to workspace admins.
+
+The flow walks through the agents to enable and which one bare `ucode` launches, then per agent:
+Databricks-hosted models or an external Model Provider Service, the models to expose, and whether
+the config applies machine-wide or per user. Claude Code is asked one model per family
+(opus/sonnet/haiku/fable), since Claude Code selects models by family alias; any family can be
+skipped. It then offers tracing, managed MCP servers, skills, and a spend-based budget policy that
+switches the default agent and model as the workspace burns through a budget.
+
+The result is written to `~/.ucode/managed-settings.json`, which `ucode apply` publishes to the
+workspace. Your own agent configs are left alone, with one exception: answering yes to tracing, MCP
+servers, or skills runs the matching `ucode configure` step, which does configure this machine.
+
+```bash
+# Review the manifest and the exact payload `ucode apply` would publish.
+ucode setup show
+
+# Walk the flow without writing anything.
+ucode setup --dry-run
+
+# Skip the prompts and load a hand-written config instead (validated before saving).
+ucode setup --from-file ./managed-settings.json
+```
+
+Publishing replaces the workspace's config outright — there is no partial update yet, so anything
+skipped in a re-run is dropped.
+
 ---
 
 ## Other Commands
@@ -185,6 +219,9 @@ you to run `ucode <agent>` (existing agent sessions need a restart before the MC
 | `ucode configure skills --location main.default [--path <dir>]` | Download a schema's skills to disk (under `<dir>`, or your home dir) and register a schema-less skills MCP connection |
 | `ucode configure skills --location main.default --skill my-skill` | Download only the named skill(s) from a schema (comma-separated for several) |
 | `ucode configure skills --location main.default --mcp` | Expose a schema's skills as MCP tools (override-only) instead of downloading |
+| `ucode setup` | Author the workspace's managed coding config (workspace admins only) |
+| `ucode setup show` | Print the authored config and the payload `ucode apply` would publish |
+| `ucode setup --from-file <file>` | Load a hand-written managed config instead of running the prompts |
 
 ## Managed Local Files
 
@@ -199,6 +236,7 @@ you to run `ucode <agent>` (existing agent sessions need a restart before the MC
 | `~/.copilot/.env` | GitHub Copilot CLI |
 | `~/.pi/agent/models.json` | Pi |
 | `~/.cursor/mcp.json` | Cursor Agent (MCP servers only) |
+| `~/.ucode/managed-settings.json` | The managed config authored by `ucode setup` (admins) |
 
 Existing files are backed up before being overwritten. `ucode revert` restores backups.
 

src/ucode/cli.py

@@ -57,6 +57,7 @@
 )
 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.managed_wizard import setup_command, show_command
 from ucode.mcp import (
     MCP_CLIENTS,
     SKILLS_MCP_KIND,
@@ -862,6 +863,10 @@ def revert() -> int:
 app.add_typer(configure_app, name="configure", help="Configure workspace and tool settings.")
 mcp_app = typer.Typer(add_completion=False, no_args_is_help=True)
 app.add_typer(mcp_app, name="mcp", help="MCP servers exposed by ucode.")
+setup_app = typer.Typer(add_completion=False, no_args_is_help=False)
+app.add_typer(
+    setup_app, name="setup", help="Author the workspace's managed coding config (admins only)."
+)
 
 
 @mcp_app.command("web-search")
@@ -1915,6 +1920,53 @@ def configure_tracing(
         raise typer.Exit(130) from None
 
 
+@setup_app.callback(invoke_without_command=True)
+def setup(
+    ctx: typer.Context,
+    from_file: Annotated[
+        str | None,
+        typer.Option(
+            "--from-file",
+            help="Skip the interactive flow and load a hand-written managed config (JSON, in "
+            "ucode's manifest shape) instead. Validated before it is saved.",
+        ),
+    ] = None,
+    dry_run: Annotated[
+        bool,
+        typer.Option("--dry-run", help="Walk the flow without writing any files."),
+    ] = False,
+) -> None:
+    """Author the managed coding config for your workspace (workspace admins only)."""
+    if ctx.invoked_subcommand is not None:
+        return
+    set_dry_run(dry_run)
+    # `typer.Exit` subclasses RuntimeError, so it must be raised outside the try — inside, the
+    # `except RuntimeError` below would swallow it and report the exit code as an error message.
+    try:
+        install_databricks_cli()
+        code = setup_command(from_file=from_file)
+    except RuntimeError as exc:
+        print_err(str(exc))
+        raise typer.Exit(1) from None
+    except KeyboardInterrupt:
+        print_err("Interrupted.")
+        raise typer.Exit(130) from None
+    if code:
+        raise typer.Exit(code)
+
+
+@setup_app.command("show")
+def setup_show_cmd() -> None:
+    """Print the authored managed config and the payload `ucode apply` would publish."""
+    try:
+        code = show_command()
+    except RuntimeError as exc:
+        print_err(str(exc))
+        raise typer.Exit(1) from None
+    if code:
+        raise typer.Exit(code)
+
+
 @app.command("status")
 def status_cmd() -> None:
     """Show current workspace, tool configs, and saved model selections."""

src/ucode/databricks.py

@@ -331,6 +331,76 @@ def _http_get_bytes(url: str, token: str, *, timeout: int = 10) -> tuple[bytes |
         return None, f"network error: {exc.reason}"
 
 
+# Workspace group whose members are workspace admins. `ucode setup` / `ucode apply` are restricted
+# to this group because the coding-agent-config CRUD API enforces the same check server-side.
+WORKSPACE_ADMIN_GROUP = "admins"
+
+
+def is_workspace_admin(workspace: str, token: str) -> bool | None:
+    """Whether the caller is a workspace admin, via their SCIM `Me` group membership.
+
+    Returns True/False, or None when the check itself could not be made (SCIM unreachable or a
+    malformed response). Callers should treat None as "unknown" and proceed optimistically rather
+    than blocking: the API enforces the same check server-side, so a false negative here would
+    needlessly stop a legitimate admin, while a false positive just surfaces the server's
+    PERMISSION_DENIED later.
+    """
+    hostname = workspace_hostname(workspace)
+    payload, _ = _http_get_json(f"https://{hostname}/api/2.0/preview/scim/v2/Me", token)
+    if not isinstance(payload, dict):
+        return None
+    groups = payload.get("groups")
+    if not isinstance(groups, list):
+        # A well-formed `Me` for a user in no groups omits `groups` entirely, so this is a
+        # definitive "not an admin" rather than a failed check.
+        return False
+    return any(
+        isinstance(group, dict) and group.get("display") == WORKSPACE_ADMIN_GROUP
+        for group in groups
+    )
+
+
+# Workspace-scoped budget listing. Account-level budget APIs need account auth, which ucode does not
+# have; this endpoint resolves the workspace server-side from the caller's token.
+_WORKSPACE_BUDGETS_API_PATH = "/api/ai-gateway/v2/workspace-metrics/budgets"
+
+
+def list_workspace_budgets(workspace: str, token: str) -> tuple[list[dict], str | None]:
+    """List the AI Gateway budgets that apply to this workspace.
+
+    Returns ``(budgets, reason)`` where each budget is ``{"id": ..., "display_name": ...}``.
+    ``reason`` is None on success, otherwise it explains why the list is empty. ucode never creates
+    budgets — an admin picks an existing one to attach a spend-routing policy to.
+    """
+    hostname = workspace_hostname(workspace)
+    url = f"https://{hostname}{_WORKSPACE_BUDGETS_API_PATH}"
+    payload, reason = _http_get_json(url, token, timeout=30)
+    if reason is not None:
+        return [], reason
+    if not isinstance(payload, dict):
+        return [], "workspace budget listing returned an unexpected response shape"
+    raw = payload.get("workspace_ai_gateway_budgets")
+    if not isinstance(raw, list):
+        return [], "workspace budget listing returned no budgets"
+    budgets: list[dict] = []
+    for entry in raw:
+        if not isinstance(entry, dict):
+            continue
+        budget_id = entry.get("budget_configuration_id")
+        if not isinstance(budget_id, str) or not budget_id:
+            continue
+        display_name = entry.get("display_name")
+        budgets.append(
+            {
+                "id": budget_id,
+                "display_name": display_name if isinstance(display_name, str) else "",
+            }
+        )
+    if not budgets:
+        return [], "workspace budget listing returned no budgets"
+    return budgets, None
+
+
 def get_current_user_name(workspace: str, token: str) -> str | None:
     """Return the current user's login (email) via SCIM `Me`, or None on failure.
 
@@ -843,12 +913,20 @@ def run_databricks_login(workspace: str, profile: str | None = None) -> None:
     print_success("Databricks authentication complete")
 
 
-def ensure_databricks_auth(workspace: str, profile: str | None = None) -> None:
-    """Check auth and login only if needed (used by launch path)."""
+def ensure_databricks_auth(
+    workspace: str, profile: str | None = None, *, quiet: bool = False
+) -> None:
+    """Check auth and login only if needed (used by launch path).
+
+    ``quiet`` suppresses the "already available" line for a caller that only needs a token before
+    some later step re-authenticates and reports it — otherwise the same success prints twice. A
+    login that actually runs is never silent.
+    """
     with spinner("Checking Databricks auth..."):
         auth_is_valid = has_valid_databricks_auth(workspace, profile)
     if auth_is_valid:
-        print_success(f"Databricks auth already available for {workspace}")
+        if not quiet:
+            print_success(f"Databricks auth already available for {workspace}")
         return
     run_databricks_login(workspace, profile)
 
@@ -1272,12 +1350,42 @@ def _get_model_services_page(
     return payload, reason
 
 
+# Successful model-service listings for this process, keyed by workspace. The listing is a paginated
+# walk of the whole metastore catalog, and several callers want different views of the same result
+# (`discover_model_services` buckets it per family, `discover_claude_models_unbucketed` keeps the raw
+# Claude ids), so a single `ucode setup` run would otherwise page it twice. Cached per process, not
+# persisted: a long-lived process is not a thing here, and a new model appearing mid-command is not
+# worth a second walk. Failures are never cached, so a transient error still retries.
+_MODEL_SERVICES_CACHE: dict[str, list[str]] = {}
+
+# Same idea for the Model Provider Service listing (a different endpoint). It is workspace-wide and
+# filtered per agent afterwards, so `ucode setup` would otherwise re-list it once per MPS-capable
+# agent.
+_MODEL_PROVIDER_SERVICES_CACHE: dict[str, list[dict]] = {}
+
+
+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()
+
+
+def has_cached_model_provider_services(workspace: str) -> bool:
+    """True when :func:`list_model_provider_services` will answer from cache.
+
+    Lets a caller skip a progress spinner it doesn't need: the cold listing takes over a second, so
+    it deserves one, but repeating it per agent on an instant cache hit is just noise.
+    """
+    return workspace in _MODEL_PROVIDER_SERVICES_CACHE
+
+
 def list_model_services(
     workspace: str,
     token: str,
     *,
     page_size: int = _MODEL_SERVICES_PAGE_SIZE,
     max_pages: int = 100,
+    use_cache: bool = True,
 ) -> tuple[list[str], str | None]:
     """List all `system.ai.*` model ids via the UC model-services API.
 
@@ -1286,7 +1394,15 @@ def list_model_services(
     de-duplicated, sorted list of ``system.ai.<model-name>`` ids. Returns
     (ids, reason); reason is None on success, otherwise it describes why the
     list is empty (HTTP/network error or no services).
+
+    A successful result is memoized per workspace for the life of the process; pass
+    ``use_cache=False`` to force a fresh walk.
     """
+    if use_cache:
+        cached = _MODEL_SERVICES_CACHE.get(workspace)
+        if cached is not None:
+            return list(cached), None
+
     hostname = workspace_hostname(workspace)
     ids: list[str] = []
     page_token: str | None = None
@@ -1319,10 +1435,26 @@ def list_model_services(
 
     deduped = sorted(set(ids))
     if deduped:
+        if use_cache:
+            _MODEL_SERVICES_CACHE[workspace] = list(deduped)
         return deduped, None
     return [], last_reason or "model-services listing returned no models"
 
 
+def discover_claude_models_unbucketed(workspace: str, token: str) -> tuple[list[str], str | None]:
+    """Every `system.ai.claude-*` id on the workspace, unbucketed.
+
+    `discover_model_services` keeps only the newest id per family because the launch path pins one
+    model per Claude family alias. An admin authoring a managed config needs the alternatives too
+    (see `managed_setup.claude_family_candidates`), so this returns the full set without disturbing
+    that shape.
+    """
+    ids, reason = list_model_services(workspace, token)
+    if not ids:
+        return [], reason
+    return [m for m in ids if "claude-" in m.lower()], None
+
+
 def discover_model_services(
     workspace: str, token: str
 ) -> tuple[dict[str, str], list[str], list[str], list[str], str | None]:
@@ -1495,7 +1627,9 @@ def _provider_type_tag(provider_type: str | None) -> str:
     return tag.lower()
 
 
-def list_model_provider_services(workspace: str, token: str) -> tuple[list[dict], str | None]:
+def list_model_provider_services(
+    workspace: str, token: str, *, use_cache: bool = True
+) -> tuple[list[dict], str | None]:
     """List Unity Catalog Model Provider Services on the workspace.
 
     Returns ``(services, reason)`` where each service is
@@ -1505,7 +1639,20 @@ def list_model_provider_services(workspace: str, token: str) -> tuple[list[dict]
     Bedrock model names). ``relayed`` is True for a credential-less Anthropic
     service (Claude Max/Team/Enterprise subscription relay). A non-None
     ``reason`` means the listing call itself failed.
+
+    The listing is workspace-wide — it is filtered per agent afterwards by
+    :func:`service_usable_for_tool` — so a successful result is memoized per workspace for the life
+    of the process, like the model-services listing. Without that, `ucode setup` re-lists it once
+    per MPS-capable agent. Pass ``use_cache=False`` to force a fresh call.
     """
+    if use_cache:
+        cached = _MODEL_PROVIDER_SERVICES_CACHE.get(workspace)
+        if cached is not None:
+            # A fresh list of fresh dicts each time: callers treat the result as theirs (the wizard
+            # filters it per agent), so handing out the cached objects would let one caller's edit
+            # reach the next.
+            return [dict(service) for service in cached], None
+
     hostname = workspace_hostname(workspace)
     url = f"https://{hostname}/api/2.1/unity-catalog/model-provider-services"
     payload, reason = _http_get_json(url, token, timeout=30)
@@ -1542,6 +1689,8 @@ def list_model_provider_services(workspace: str, token: str) -> tuple[list[dict]
             }
         )
     services.sort(key=lambda s: s["name"])
+    if use_cache:
+        _MODEL_PROVIDER_SERVICES_CACHE[workspace] = [dict(service) for service in services]
     return services, None