Skip to content
7 changes: 7 additions & 0 deletions src/ucode/agents/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
print_success,
print_warning,
prompt_yes_no,
prompt_yes_no_default,
spinner,
)

Expand Down Expand Up @@ -201,6 +202,12 @@ def install_tool_binary(

if not too_new and version_error:
print_warning(version_error)
# Native upgraders run in place, so confirm before mutating the install;
# EOF/piped runs take the default and upgrade (a required fix must not stall).
if tool in _NATIVE_UPGRADE_COMMANDS and not prompt_yes_no_default(
f"Upgrade {spec['display']} if available?", default=True
):
raise RuntimeError(version_error)
if not _update_installed_tool_binary(tool):
raise RuntimeError(version_error)
version_error = _minimum_version_error(tool)
Expand Down
40 changes: 10 additions & 30 deletions src/ucode/agents/claude.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,11 +89,8 @@
CLAUDE_USER_SETTINGS_PATH = CLAUDE_CONFIG_DIR / "settings.json"
CLAUDE_BACKUP_PATH = APP_DIR / "claude-ucode-settings.backup.json"
WEB_SEARCH_MCP_STATE_KEY = "claude_web_search_mcp"
MINIMUM_CLAUDE_VERSION = (2, 1, 248)
MINIMUM_CLAUDE_VERSION_TEXT = "2.1.248"
# managedMcpServers needs Claude Code 2.1.259+; older versions ignore it and fall back to user scope.
MANAGED_MCP_MIN_VERSION = (2, 1, 259)
MANAGED_MCP_MIN_VERSION_TEXT = "2.1.259"
MINIMUM_CLAUDE_VERSION = (2, 1, 259)
MINIMUM_CLAUDE_VERSION_TEXT = "2.1.259"
MANAGED_MCP_SETTINGS_KEY = "managedMcpServers"

SPEC: ToolSpec = {
Expand All @@ -116,25 +113,15 @@ def _parse_version(value: str) -> tuple[int, int, int] | None:
return int(major), int(minor), int(patch)


def _minimum_version_requirement_message(version: str) -> str:
feature = "Smart routing" if smart_routing_v2.smart_routing_enabled() else "Model discovery"
return (
f"{feature} requires Claude Code {MINIMUM_CLAUDE_VERSION_TEXT} or newer. "
f"Your current version is Claude Code {version}."
)


def minimum_version_error() -> str | None:
if (
os.environ.get(GATEWAY_MODEL_DISCOVERY_ENV_VAR) != "1"
and not smart_routing_v2.smart_routing_enabled()
):
return None
version = agent_version(SPEC["binary"])
parsed = _parse_version(version)
if parsed is None or parsed >= MINIMUM_CLAUDE_VERSION:
return None
return _minimum_version_requirement_message(version)
return (
f"ug requires Claude Code {MINIMUM_CLAUDE_VERSION_TEXT} or newer. "
f"Your current version is Claude Code {version}."
)


def _resolve_web_search_model(state: dict) -> str | None:
Expand Down Expand Up @@ -706,24 +693,17 @@ def remove_claude_mcp_server(name: str, scope: str) -> bool:
raise RuntimeError(f"Failed to remove MCP server '{name}' via claude CLI.") from exc


def _version_supports_managed_mcp() -> bool:
parsed = _parse_version(agent_version(SPEC["binary"]))
return parsed is not None and parsed >= MANAGED_MCP_MIN_VERSION


def managed_mcp_uses_managed_file(workspace: str, *, use_pat: bool) -> bool:
"""Whether Claude's managed MCP servers belong in the OS-managed file rather than user scope.

The OS-managed ``managedMcpServers`` key is additive (it never touches the developer's own
servers) but Claude Code reads it only from a real managed source, only since 2.1.259, and only
as a remote HTTP server it can drive OAuth against itself. So it fits only when the platform
supports the sudo reconcile, the run is interactive, the CLI is new enough, the developer is not
on PAT auth (which needs the stdio proxy), and the workspace publishes the ``claude-code`` OAuth
client. Every other case falls back to the user-scope registration."""
servers) but Claude Code reads it only from a real managed source, and only as a remote HTTP
server it can drive OAuth against itself. So it fits only when the platform supports the sudo
reconcile, the run is interactive, the developer is not on PAT auth (which needs the stdio
proxy), and the workspace publishes the ``claude-code`` OAuth client. Every other case falls back to the user-scope registration."""
return (
managed_files_supported()
and managed_writes_allowed()
and _version_supports_managed_mcp()
and not use_pat
and oauth_client_available(workspace, CLAUDE_CODE_OAUTH_CLIENT_ID)
)
Expand Down
30 changes: 9 additions & 21 deletions src/ucode/agents/codex.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,10 +102,11 @@
MODEL_SERVICE_PARENT_SCHEMA_HEADER,
],
]
MINIMUM_CODEX_VERSION = (0, 134, 0)
MINIMUM_CODEX_VERSION_TEXT = "0.134.0"
MINIMUM_ROUTING_CODEX_VERSION = (0, 145, 0)
MINIMUM_ROUTING_CODEX_VERSION_TEXT = "0.145.0"
MINIMUM_CODEX_VERSION = (0, 145, 0)
MINIMUM_CODEX_VERSION_TEXT = "0.145.0"
# Codex 0.134.0 introduced per-profile config files; older releases use the legacy layout.
LEGACY_LAYOUT_CODEX_VERSION = (0, 134, 0)
LEGACY_LAYOUT_CODEX_VERSION_TEXT = "0.134.0"
# Retained only to identify and remove state written by the legacy persisted opt-in.
SMART_ROUTING_STATE_KEY = smart_routing_v2.LEGACY_STATE_KEY
APP_SERVER_SMART_ROUTING_STARTING_MODEL = "gpt-5.6-luna"
Expand Down Expand Up @@ -142,17 +143,11 @@ def _parse_version(value: str) -> tuple[int, int, int] | None:


def minimum_version_error() -> str | None:
"""Return the active smart-routing version blocker, if any."""
if not smart_routing_v2.smart_routing_enabled():
return None
version = agent_version(SPEC["binary"])
parsed = _parse_version(version)
if parsed is None or parsed >= MINIMUM_ROUTING_CODEX_VERSION:
if parsed is None or parsed >= MINIMUM_CODEX_VERSION:
return None
return (
"Codex smart routing requires Codex "
f"{MINIMUM_ROUTING_CODEX_VERSION_TEXT} or newer; found {version}."
)
return f"ug requires Codex {MINIMUM_CODEX_VERSION_TEXT} or newer; found {version}."


def _use_legacy_layout() -> bool:
Expand All @@ -167,7 +162,7 @@ def _use_legacy_layout() -> bool:
parsed = _parse_version(agent_version(SPEC["binary"]))
if parsed is None:
return False
return parsed < MINIMUM_CODEX_VERSION
return parsed < LEGACY_LAYOUT_CODEX_VERSION


def has_ucode_config() -> bool:
Expand Down Expand Up @@ -943,7 +938,7 @@ def launch(
if _use_legacy_layout():
print_warning_err(
f"Codex {agent_version(binary)} is outdated. Upgrade Codex to "
f"{MINIMUM_CODEX_VERSION_TEXT} or newer, then run `codex --version` to verify "
f"{LEGACY_LAYOUT_CODEX_VERSION_TEXT} or newer, then run `codex --version` to verify "
"the active installation."
)
_run_codex(
Expand Down Expand Up @@ -1010,13 +1005,6 @@ def launch(
def _launch_smart_routing(state: dict, tool_args: list[str]) -> None:
"""Launch the Codex TUI through the smart-routing interposer."""
binary = SPEC["binary"]
version_text = agent_version(binary)
parsed_version = _parse_version(version_text)
if parsed_version is not None and parsed_version < MINIMUM_ROUTING_CODEX_VERSION:
raise RuntimeError(
"Codex smart routing requires Codex "
f"{MINIMUM_ROUTING_CODEX_VERSION_TEXT} or newer; found {version_text}."
)

configured_model = _smart_routing_config_model(state)
# Prefer the custom catalog if it exists.
Expand Down
129 changes: 129 additions & 0 deletions tests/integration/test_ug_version_floor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
"""Version-floor gate: ug enforces minimum agent CLI versions at launch.

One scenario per agent, driven against the real installed CLI with a
below-floor pin (`--claude-version 2.1.258 --codex-version 0.144.0`, or the
matching workflow_dispatch inputs). The gate only exists below the floor, and
the suite forbids skips, so each scenario returns early when the pinned CLI
already meets the floor.
"""

from __future__ import annotations

import re
import subprocess
import time

import pytest
from utils.terminal import TerminalProcess

pytestmark = pytest.mark.installation

_FLOORS = {"claude": (2, 1, 259), "codex": (0, 145, 0)}
_REQUIREMENTS = {
"claude": "ug requires Claude Code 2.1.259 or newer",
"codex": "ug requires Codex 0.145.0 or newer",
}
_UPGRADE_PROMPTS = {
"claude": "Upgrade Claude Code if available?",
"codex": "Upgrade Codex if available?",
}
_UPGRADE_NOTES = {"claude": "Upgrading Claude Code", "codex": "Upgrading Codex"}
_UPGRADE_FAILURES = {
"claude": "Could not update Claude Code",
"codex": "Could not update Codex",
}
# Shown by the auto-configure flow once the launch clears the floor gate.
_PAST_GATE_SIGNALS = ("Loading Databricks workspaces", "Workspace URL", "Select workspace:")


def _installed_version(session, binary: str) -> tuple[int, int, int]:
result = subprocess.run(
[binary, "--version"],
env=session.env,
capture_output=True,
text=True,
timeout=30,
check=False,
)
match = re.search(r"(\d+)\.(\d+)\.(\d+)", result.stdout + result.stderr)
assert match, f"`{binary} --version` reported no semver: {result!r}"
return int(match.group(1)), int(match.group(2)), int(match.group(3))


def _wait_for_exit(tui: TerminalProcess, timeout: int = 60) -> None:
deadline = time.monotonic() + timeout
while not tui.ended and time.monotonic() < deadline:
tui.read()
assert tui.ended, f"Process did not exit within {timeout}s:\n{tui.visible}"
tui.child.close(force=False)


def _past_gate(text: str) -> bool:
return any(signal in text for signal in _PAST_GATE_SIGNALS)


def _launch(session, agent: str, name: str) -> TerminalProcess:
tui = TerminalProcess(session, agent, [str(session.binary), agent], name)
tui.wait_for(lambda text: _UPGRADE_PROMPTS[agent] in text, "required-upgrade prompt")
return tui


def _assert_gate_holds_after_failed_upgrade(tui: TerminalProcess, agent: str) -> None:
"""The floor clears only when the accepted upgrade actually succeeds.

A working upgrader carries the launch past the gate to workspace
configuration; a failed one (e.g. registry unreachable) must leave the
launch blocked with the requirement restated.
"""
tui.wait_for(
lambda text: _past_gate(text) or _UPGRADE_FAILURES[agent] in text or tui.ended,
"upgrade outcome",
timeout=600,
)
transcript = "".join(tui.output)
if _UPGRADE_FAILURES[agent] in transcript:
_wait_for_exit(tui)
assert _REQUIREMENTS[agent] in transcript
assert tui.child.exitstatus != 0
else:
assert _past_gate(transcript), f"Launch stalled after upgrade:\n{tui.visible}"


def _check_version_floor(session, agent: str) -> None:
if _installed_version(session, agent) >= _FLOORS[agent]:
return # The gate only exists below the floor; skips are forbidden here.
with _launch(session, agent, "floor-decline") as tui:
tui.send("n\r", "decline the required upgrade")
_wait_for_exit(tui)
transcript = "".join(tui.output)
assert _REQUIREMENTS[agent] in transcript
assert _UPGRADE_NOTES[agent] not in transcript
assert tui.child.exitstatus != 0
with _launch(session, agent, "floor-accept") as tui:
tui.send("y\r", "accept the required upgrade")
tui.wait_for(
lambda text: _UPGRADE_NOTES[agent] in text, "native upgrade attempt", timeout=60
)
_assert_gate_holds_after_failed_upgrade(tui, agent)


@pytest.mark.claude
def test_claude_launch_below_version_floor(session):
"""Scenario: launch Claude Code pinned below the version floor.

Expected: ug requires an upgrade first. Declining blocks the launch with the
requirement restated; accepting runs the native upgrader and the floor
clears only when the upgrade succeeds.
"""
_check_version_floor(session, "claude")


@pytest.mark.codex
def test_codex_launch_below_version_floor(session):
"""Scenario: launch Codex pinned below the version floor.

Expected: ug requires an upgrade first. Declining blocks the launch with the
requirement restated; accepting runs the native upgrader and the floor
clears only when the upgrade succeeds.
"""
_check_version_floor(session, "codex")
43 changes: 4 additions & 39 deletions tests/test_agent_claude.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,52 +46,24 @@ def test_display(self):


class TestMinimumVersion:
@pytest.mark.parametrize("version", ["2.1.248", "2.1.250", "3.0.0"])
@pytest.mark.parametrize("version", ["2.1.259", "2.1.260", "3.0.0"])
def test_supported_version(self, monkeypatch, version):
monkeypatch.setenv(v2.ENABLE_SMART_ROUTING_ENV_VAR, "1")
monkeypatch.setattr(claude, "agent_version", lambda _binary: version)

assert claude.minimum_version_error() is None

def test_older_version_requires_update(self, monkeypatch):
monkeypatch.setenv(v2.ENABLE_SMART_ROUTING_ENV_VAR, "1")
monkeypatch.setattr(claude, "agent_version", lambda _binary: "2.1.247")
monkeypatch.setattr(claude, "agent_version", lambda _binary: "2.1.258")

assert claude.minimum_version_error() == (
"Smart routing requires Claude Code 2.1.248 or newer. "
"Your current version is Claude Code 2.1.247."
)

def test_older_version_requires_update_for_model_discovery(self, monkeypatch):
monkeypatch.setenv(claude.GATEWAY_MODEL_DISCOVERY_ENV_VAR, "1")
monkeypatch.setattr(claude, "agent_version", lambda _binary: "2.1.247")

expected = (
"Model discovery requires Claude Code 2.1.248 or newer. "
"Your current version is Claude Code 2.1.247."
"ug requires Claude Code 2.1.259 or newer. Your current version is Claude Code 2.1.258."
)
assert claude.minimum_version_error() == expected

def test_smart_routing_message_wins_when_both_features_are_enabled(self, monkeypatch):
monkeypatch.setenv(v2.ENABLE_SMART_ROUTING_ENV_VAR, "1")
monkeypatch.setenv(claude.GATEWAY_MODEL_DISCOVERY_ENV_VAR, "1")
monkeypatch.setattr(claude, "agent_version", lambda _binary: "2.1.247")

assert claude.minimum_version_error().startswith("Smart routing requires")

def test_unknown_version_does_not_block(self, monkeypatch):
monkeypatch.setenv(v2.ENABLE_SMART_ROUTING_ENV_VAR, "1")
monkeypatch.setattr(claude, "agent_version", lambda _binary: "unknown")

assert claude.minimum_version_error() is None

def test_older_version_is_not_validated_without_discovery_features(self, monkeypatch):
monkeypatch.delenv(v2.ENABLE_SMART_ROUTING_ENV_VAR, raising=False)
monkeypatch.delenv(claude.GATEWAY_MODEL_DISCOVERY_ENV_VAR, raising=False)
monkeypatch.setattr(claude, "agent_version", lambda _binary: "2.1.247")

assert claude.minimum_version_error() is None


class TestRenderOverlay:
def test_long_context_suffix_supports_major_only_claude_versions(self):
Expand Down Expand Up @@ -2348,22 +2320,15 @@ def test_reconfigure_does_not_back_up_generated_file(self, tmp_path, monkeypatch


class TestManagedMcpUsesManagedFile:
def _wire(
self, monkeypatch, *, supported=True, interactive=True, version="2.1.259", oauth=True
):
def _wire(self, monkeypatch, *, supported=True, interactive=True, oauth=True):
monkeypatch.setattr(claude, "managed_files_supported", lambda: supported)
monkeypatch.setattr(claude, "managed_writes_allowed", lambda: interactive)
monkeypatch.setattr(claude, "agent_version", lambda _binary: version)
monkeypatch.setattr(claude, "oauth_client_available", lambda ws, client_id: oauth)

def test_true_when_all_conditions_hold(self, monkeypatch):
self._wire(monkeypatch)
assert claude.managed_mcp_uses_managed_file(WS, use_pat=False) is True

def test_false_on_old_claude(self, monkeypatch):
self._wire(monkeypatch, version="2.1.258")
assert claude.managed_mcp_uses_managed_file(WS, use_pat=False) is False

def test_false_under_pat(self, monkeypatch):
self._wire(monkeypatch)
assert claude.managed_mcp_uses_managed_file(WS, use_pat=True) is False
Expand Down
16 changes: 10 additions & 6 deletions tests/test_agent_codex.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,16 +29,20 @@ def test_display(self):


class TestMinimumVersion:
def test_smart_routing_old_version_requires_update(self, monkeypatch):
monkeypatch.setenv(codex.smart_routing_v2.ENABLE_SMART_ROUTING_ENV_VAR, "1")
@pytest.mark.parametrize("version", ["0.145.0", "0.148.0", "1.0.0"])
def test_supported_version(self, monkeypatch, version):
monkeypatch.setattr(codex, "agent_version", lambda _binary: version)

assert codex.minimum_version_error() is None

def test_older_version_requires_update(self, monkeypatch):
monkeypatch.setattr(codex, "agent_version", lambda _binary: "0.144.0")

expected = "Codex smart routing requires Codex 0.145.0 or newer; found 0.144.0."
expected = "ug requires Codex 0.145.0 or newer; found 0.144.0."
assert codex.minimum_version_error() == expected

def test_old_version_is_not_blocked_without_smart_routing(self, monkeypatch):
monkeypatch.delenv(codex.smart_routing_v2.ENABLE_SMART_ROUTING_ENV_VAR, raising=False)
monkeypatch.setattr(codex, "agent_version", lambda _binary: "0.144.0")
def test_unknown_version_does_not_block(self, monkeypatch):
monkeypatch.setattr(codex, "agent_version", lambda _binary: "unknown")

assert codex.minimum_version_error() is None

Expand Down
Loading
Loading