Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 55 additions & 0 deletions examples/managed-config.stub.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
{
"spec_version": 1,
"update_time": "2026-09-11T00:00:00Z",
"retrieved_time": "2026-09-11T00:00:00Z",
"default_agent": "CODING_AGENT_CLAUDE_CODE",
"enabled_agents": [
{
"agent": "CODING_AGENT_CLAUDE_CODE",
"config": {
"models": {
"model_services": [
"system.ai.claude-opus-4-8",
"system.ai.claude-sonnet-4-6",
"system.ai.claude-haiku-4-5"
]
},
"default_models": {
"default_model": "system.ai.claude-opus-4-8",
"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"
},
"smart_routing": { "enabled": false },
"http_headers": {},
"tracing": { "enabled": true }
}
},
{
"agent": "CODING_AGENT_CODEX",
"config": {
"models": { "model_provider_service": "main.default.openai-mps" },
"default_models": { "default_model": "gpt-5.4" },
"tracing": { "enabled": false }
}
}
],
"mcp_servers": {
"names": ["system.ai.github"],
"tags": ["MCP_SERVER_TYPE_UC_SERVICE"]
},
"skills": {
"names": ["system.ai.pdf-extraction"],
"tags": []
},
"spend_tiers": {
"budget_id": "00000000-0000-0000-0000-000000000000",
"tiers": [
{
"spending_percentage": 0.9,
"recommended_agent": "CODING_AGENT_CODEX",
"recommended_model": "gpt-5.4"
}
]
}
}
146 changes: 119 additions & 27 deletions src/ucode/agents/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,18 @@
resolve_provider_service,
)
from ucode.managed_files import managed_write_batch
from ucode.state import get_provider_service, load_state, save_state
from ucode.managed_resolve import (
managed_provider_service,
managed_supplies_models,
resolve_state,
)
from ucode.state import (
_without_managed_overlay,
get_provider_service,
load_state,
save_state,
set_provider_service,
)
from ucode.telemetry import agent_version
from ucode.ui import (
console,
Expand Down Expand Up @@ -472,24 +483,40 @@ def launch(
_MODULES[tool].launch(state, tool_args, options=options)


def check_gateway_endpoint(state: dict, tool: str) -> bool:
"""V2-only: a tool is available iff we discovered models for it."""
def check_gateway_endpoint(state: dict, tool: str, managed: dict | None = None) -> bool:
"""V2-only: a tool is available iff we discovered models for it or the managed config supplies them.

FIX 2: An agent whose models come only from the managed config (no discovered models)
must count as available. Check both discovered models and managed-supplied models.
"""
# Check discovered models
if tool == "claude":
return bool(state.get("claude_models"))
if tool == "opencode":
return bool(state.get("opencode_models"))
if tool == "codex":
return bool(state.get("codex_models"))
if tool == "gemini":
return bool(state.get("gemini_models"))
if tool == "copilot":
return bool(state.get("claude_models")) or bool(state.get("codex_models"))
if tool == "pi":
return (
discovered = bool(state.get("claude_models"))
elif tool == "opencode":
discovered = bool(state.get("opencode_models"))
elif tool == "codex":
discovered = bool(state.get("codex_models"))
elif tool == "gemini":
discovered = bool(state.get("gemini_models"))
elif tool == "copilot":
discovered = bool(state.get("claude_models")) or bool(state.get("codex_models"))
elif tool == "pi":
discovered = (
bool(state.get("claude_models"))
or bool(state.get("codex_models"))
or bool(state.get("gemini_models"))
)
else:
return False

# If discovered models exist, the tool is available
if discovered:
return True

# If managed config supplies models, the tool is available
if managed and managed_supplies_models(managed, tool):
return True

return False


Expand All @@ -514,14 +541,41 @@ def _availability_failure_detail(tool: str, state: dict) -> str:
return " (" + "; ".join(parts) + ")"


def configure_single_tool(tool: str, state: dict) -> dict:
"""Check availability, configure, and persist state for one tool only."""
def resolve_managed_for_tool(managed: dict | None, state: dict, tool: str) -> dict:
"""State with the managed config applied for ``tool`` and a persisted provider cleared when a
managed static-list or discovery-location source displaces it, so availability and validation
see the effective config rather than the developer's own settings."""
if managed is None:
return state
resolved = resolve_state(managed, state, tool)
if managed_supplies_models(managed, tool) and not managed_provider_service(managed, tool):
resolved = set_provider_service(resolved, tool, None)
return resolved


def configure_single_tool(tool: str, state: dict, managed: dict | None = None) -> dict:
"""Check availability, configure, and persist state for one tool only.

If managed config is provided, it is applied to the state (its settings take
precedence) and the provider precedence rules are enforced (FIX 2).
"""
# Apply managed config before resolving provider, so admin settings win
if managed is not None:
state = resolve_state(managed, state, tool)

provider = get_provider_service(state, tool)

# When the managed config names its own model source without a provider, clear any persisted
# provider so the managed source drives the picker/catalog.
if managed is not None and managed_supplies_models(managed, tool):
if not managed_provider_service(managed, tool):
provider = None

# A Model Provider Service routes through the same gateway and pins no
# Databricks model, so the per-tool model availability check doesn't apply.
if not provider:
with spinner(f"Checking {TOOL_SPECS[tool]['display']} availability..."):
ok = check_gateway_endpoint(state, tool)
ok = check_gateway_endpoint(state, tool, managed=managed)
if not ok:
detail = _availability_failure_detail(tool, state)
raise RuntimeError(
Expand Down Expand Up @@ -559,19 +613,48 @@ def _configure_one(tool: str, state: dict, provider: str | None) -> dict:


def configure_selected_tools(
state: dict, tools: list[str], *, install_ai_tools: bool = True
state: dict, tools: list[str], *, install_ai_tools: bool = True, managed: dict | None = None
) -> dict:
"""Configure the given tools. Caller is responsible for ensuring each tool
is available on the workspace.

Merges newly-configured tools into state['available_tools'] rather than
replacing it, so a previously-configured tool the user didn't pick this
run is preserved.
run is preserved. If managed config is provided, it is applied to each tool
(its settings take precedence) and the provider precedence rules are enforced.
"""
# FIX A: Resolve each tool from a clean (overlay-free) state that accumulates
# configuration writes but never accumulates overlays. This ensures each tool's
# managed overlay is independent and doesn't leak into persisted state.
developer_state = state
with managed_write_batch(_managed_settings_displays(tools)):
for tool in tools:
state = _configure_one(tool, state, get_provider_service(state, tool))

# Apply managed config before resolving provider, so admin settings win.
# Resolve from the accumulated state (without overlay) to preserve each
# tool's configuration writes, but use fresh overlays per tool.
tool_state = developer_state
if managed is not None:
tool_state = resolve_state(managed, developer_state, tool)

provider = get_provider_service(tool_state, tool)

# When the managed config names its own model source without a provider, clear any
# persisted provider so the managed source drives the picker/catalog.
if managed is not None and managed_supplies_models(managed, tool):
if not managed_provider_service(managed, tool):
provider = None

# Configure this tool
tool_configured = _configure_one(tool, tool_state, provider)
# Strip the overlay from this tool's result: the managed values reached the config
# file through resolution above, and now the developer's original state is restored
# for persistence. Non-overlay modifications (auth, etc.) are kept.
if managed is not None:
developer_state = _without_managed_overlay(tool_configured)
else:
developer_state = tool_configured

state = developer_state
existing = state.get("available_tools") or []
state["available_tools"] = sorted(set(existing) | set(tools))
save_state(state)
Expand Down Expand Up @@ -623,20 +706,27 @@ def ensure_provider_state(tool: str) -> dict:
return state


def validate_tool(tool: str) -> tuple[bool, str]:
"""Invoke a tool with a simple prompt to verify it works. Returns (ok, error_msg)."""
def validate_tool(tool: str, state: dict | None = None) -> tuple[bool, str]:
"""Invoke a tool with a simple prompt to verify it works. Returns (ok, error_msg).

FIX 1: When ``state`` is provided (post-configure validation), use it as-is since it
holds the managed-resolved values that were written to the agent config. When ``state``
is None (other paths), load the persisted state.
"""
if state is None:
state = load_state()
spec = TOOL_SPECS[tool]
binary = spec["binary"]
module = _MODULES[tool]
# Some configs (e.g. claude relayed) can't be probed with a live message —
# the proxy + subscription login only exist at launch. Trust the written config.
if hasattr(module, "skip_validation") and module.skip_validation(load_state()):
if hasattr(module, "skip_validation") and module.skip_validation(state):
return True, ""
cmd = module.validate_cmd(binary)
env = None
if hasattr(module, "validate_env"):
try:
env = module.validate_env(load_state())
env = module.validate_env(state)
except RuntimeError:
env = None
try:
Expand Down Expand Up @@ -685,7 +775,7 @@ def provider_permission_error(tool: str, state: dict, err: str) -> str:
return err


def validate_all_tools(state: dict) -> None:
def validate_all_tools(state: dict, managed_config: dict | None = None) -> None:
from rich.panel import Panel # local to avoid bumping module-level deps

from ucode.agents.pi import PI_SETTINGS_BACKUP_PATH, PI_SETTINGS_PATH
Expand All @@ -710,7 +800,9 @@ def validate_all_tools(state: dict) -> None:
if tool not in available_tools:
continue
with spinner(f"Validating {spec['display']}..."):
ok, err = validate_tool(tool)
ok, err = validate_tool(
tool, state=resolve_managed_for_tool(managed_config, state, tool)
)
results.append((tool, ok))
if ok:
print_success(f"{spec['display']} is working")
Expand Down
Loading
Loading