From 2dae4ef8c31343e5a177595719da192abe1f97dd Mon Sep 17 00:00:00 2001 From: Rohit Agrawal Date: Sat, 29 Aug 2026 16:17:48 -0400 Subject: [PATCH 1/6] Manage Claude settings at OS scope --- README.md | 10 +- docs/os-managed-settings-design.md | 238 ++++++++++ src/ucode/agents/__init__.py | 12 +- src/ucode/agents/claude.py | 244 ++++++---- src/ucode/cli.py | 44 +- src/ucode/managed_files.py | 721 ++++++++++++++++++++++++++--- tests/conftest.py | 5 + tests/test_agent_claude.py | 135 +++--- tests/test_cli.py | 66 ++- tests/test_managed_files.py | 273 ++++++++++- tests/test_managed_resolve.py | 23 +- tests/test_managed_wizard.py | 8 +- tests/test_tracing.py | 5 + 13 files changed, 1477 insertions(+), 307 deletions(-) create mode 100644 docs/os-managed-settings-design.md diff --git a/README.md b/README.md index 3d61d639..7030ef97 100644 --- a/README.md +++ b/README.md @@ -243,8 +243,12 @@ ucode publish # publish it to the workspace ``` `ucode setup` 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 (for Claude -Code and Codex) whether the config writes the agent's own OS-level settings file or a ucode-only one. +Databricks-hosted models or an external Model Provider Service, the models to expose, and (for Codex) +whether the config writes the agent's own OS-level settings file or a ucode-only one. Interactive +Claude Code configuration installs its gateway configuration in the OS-managed settings scope so +enterprise settings cannot silently override ucode. Non-interactive and CI runs use the local file +without invoking `sudo`, and stop with an actionable error if an existing managed value conflicts. +Claude subscription relay is local-only because its loopback proxy exists only for that session. Claude Code is asked one model per family (opus/sonnet/haiku/fable), since it selects models by family alias; any family can be skipped. @@ -381,12 +385,14 @@ control the installation. |------|------| | `~/.codex/ucode.config.toml` (or legacy `~/.codex/config.toml`) | Codex | | `~/.claude/ucode-settings.json` | Claude Code settings generated by ucode | +| `/etc/claude-code/managed-settings.json` (Linux) or `/Library/Application Support/ClaudeCode/managed-settings.json` (macOS) | Claude Code OS-managed settings | | `~/.gemini/.env` | Gemini CLI | | `~/.config/opencode/opencode.json` | OpenCode | | `~/.copilot/.env` | GitHub Copilot CLI | | `~/.pi/agent/models.json` | Pi | | `~/.cursor/mcp.json` | Cursor Agent (MCP servers only) | | `~/.ucode/managed-state.json` | The managed config — authored by `ucode setup` (admins) and refreshed from the workspace on launch | +| `~/.ucode/managed-backups/` | Baseline backups for OS-managed files changed by ucode | Existing files are backed up before being overwritten. `ucode revert` restores backups. diff --git a/docs/os-managed-settings-design.md b/docs/os-managed-settings-design.md new file mode 100644 index 00000000..16a0c8f1 --- /dev/null +++ b/docs/os-managed-settings-design.md @@ -0,0 +1,238 @@ +# OS-Managed Agent Settings + +## Summary + +Claude Code and Codex give OS-managed settings higher precedence than user settings. Previously, +ucode wrote only its local configuration unless an administrator enabled +`use_as_global_settings`. An existing machine-managed file could therefore silently override the +gateway endpoint, authentication helper, provider headers, or model selected by ucode. + +The two stacked PRs make precedence handling deterministic: + +1. The Claude PR adds the shared managed-file lifecycle and applies it to Claude Code. +2. The Codex PR reuses that lifecycle for TOML, applies it to Codex, and removes + `use_as_global_settings`. + +After both PRs merge, interactive configuration reconciles the agent's OS-managed file by default. +Non-interactive and CI execution never elevates privileges and instead uses local settings when the +managed file is compatible. + +## Configuration Files + +| Agent | Local ucode configuration | OS-managed configuration | +| --- | --- | --- | +| Claude Code | `~/.claude/ucode-settings.json` | Linux: `/etc/claude-code/managed-settings.json`; macOS: `/Library/Application Support/ClaudeCode/managed-settings.json` | +| Codex | `~/.codex/ucode.config.toml` | `/etc/codex/managed_config.toml` | + +The local file is always written. The OS-managed file is additionally reconciled during interactive +configuration, except for Claude subscription relay. + +## Interactive Detection + +An invocation may modify OS-managed settings only when standard input is a TTY. Standard output does +not affect the decision, so piping logs does not disable an otherwise interactive configuration. +CI, pipes, cron jobs, and headless subprocesses normally have non-TTY standard input and therefore +remain local-only. + +This is a new shared ucode distinction. The previous implementation inferred interactivity from +command shape in some flows and did not guard managed-file writes consistently. + +## Behavior Matrix + +| Invocation | Managed file | Behavior | +| --- | --- | --- | +| Interactive | Absent | Create it from the ucode configuration after recording an absent baseline. | +| Interactive | Unrelated or partially populated | Preserve unrelated values and add or update all ucode-owned values. | +| Interactive | Conflicting | Back up the baseline, replace the conflicting ucode-owned values, and verify. | +| Interactive | Already identical | Continue without a backup, write, or `sudo` invocation. | +| Non-interactive | Absent | Use the local ucode file. Do not create the managed file. | +| Non-interactive | Ucode-owned values absent or equal | Use the local ucode file. Do not modify the managed file. | +| Non-interactive | Ucode-owned value conflicts | Stop before launching because the higher-precedence value would override ucode. | +| Any | Invalid, unreadable, or symlinked | Stop without modifying the file because precedence cannot be established safely. | + +`ucode configure`, first-time `ucode claude` or `ucode codex`, and later launches all use the same +agent-specific reconciliation path. A first-time launch from an interactive terminal can therefore +request administrator permission. A first-time non-interactive launch remains local-only. + +## Interactive Reconciliation + +For each agent, ucode: + +1. Strictly parses the existing managed JSON or TOML document. +2. Produces the desired document by applying the same overlay used for the local ucode file. +3. Preserves settings outside the paths owned by ucode. +4. Preserves enterprise Claude permission-deny entries while adding ucode-required entries. +5. Records the original baseline before the first change. +6. Requests administrator permission and performs an atomic privileged replacement. +7. Reads the installed file back and verifies its exact contents. +8. Records the last-applied snapshot, owned paths, and a launch fingerprint. + +An existing managed file is reconciled even when it does not currently conflict. This ensures every +ucode-required value exists at the highest-precedence scope and avoids separate behavior for absent, +partial, and conflicting files. + +## Privileged Write Transaction + +The shared writer handles ordinary MDM-installed and root-owned files rather than treating them as +errors: + +- refuses symlink destinations; +- creates a root-owned destination directory when it is absent; +- stages the new file in the destination directory for a same-filesystem atomic rename; +- clones an existing file before replacing its contents, preserving ownership, mode, ACLs, and + extended attributes; +- creates new files as root-owned and world-readable; +- detects, clears, and restores macOS `schg`, `uchg`, `sappnd`, and `uappnd` flags; +- detects, clears, and restores Linux immutable and append-only attributes; +- verifies the resulting contents after replacement; +- retries once only when device management restored the exact pre-write contents; +- preserves a concurrently changed policy instead of overwriting it. + +The privilege boundary is interactive. Non-interactive paths do not invoke either normal `sudo` or +`sudo -n`. + +Root access cannot sustainably override an actively enforced policy. If an MDM process immediately +restores the original file twice, ucode stops with an error instead of repeatedly fighting the +management agent. A different concurrent update is also preserved and reported. + +## Backup Model + +Backups live under `~/.ucode/managed-backups/` with directory mode `0700` and file mode `0600`. +The manifest records, per agent: + +- whether the managed file originally existed; +- its absolute path; +- the baseline snapshot and SHA-256 digest; +- the last ucode-applied snapshot and SHA-256 digest; +- the setting paths owned by ucode. + +The baseline is created before the first managed change and is not replaced by subsequent +configurations. Later configurations update only the last-applied snapshot. Snapshot filenames are +validated, digests are checked before use, and symlinked backup directories or manifests are +rejected. + +If ucode cannot complete a write or revert, the backup remains available for a later retry. + +## `ucode revert` + +`ucode revert` extends the existing local-file restoration with managed-file restoration: + +- If the current file exactly matches ucode's last-applied snapshot, restore the exact baseline. +- If ucode created the file, delete it. +- If an administrator or MDM changed the file later, perform a three-way revert. +- Restore original values only where the current value still matches ucode's last-applied value. +- Remove only list entries added by ucode while preserving externally added entries. +- Preserve externally changed values rather than replacing them with stale baseline values. +- Refuse an unsafe or unparsable merge and retain the backup. + +A revert that needs to change an OS-managed file must run interactively. A successful revert removes +that agent's backup record. The existing local configuration and ucode state cleanup still occur as +part of the command. + +## Cached Launches + +After a successful managed reconciliation or compatibility check, ucode stores: + +- the managed path; +- the verification scope; +- device and inode numbers; +- size; +- nanosecond modification and change times. + +A cached launch performs one `stat()` call and compares this fingerprint. When it matches, ucode +does not read, parse, back up, write, or invoke `sudo`. When it changes, the normal reconciliation or +compatibility path runs again. This catches later MDM replacement without adding meaningful latency +to unchanged launches. + +The verification scopes distinguish: + +- an interactively reconciled managed file; +- a managed file verified as compatible with local settings; +- a managed file verified as compatible with Claude relay. + +A compatibility fingerprint created non-interactively cannot suppress the next interactive +reconciliation. + +## Claude Subscription Relay + +Claude relay is intentionally local-only. It routes through a loopback refresh proxy whose address +and lifetime belong to one `ucode claude` session. Persisting that address in OS-managed settings +would break bare `claude` launches and future sessions. + +Relay therefore never creates or updates the managed file. It allows an absent file or unrelated +enterprise settings, but blocks these higher-precedence conflicts: + +- `apiKeyHelper`; +- `env.ANTHROPIC_BASE_URL`; +- `env.ANTHROPIC_CUSTOM_HEADERS`. + +If ucode wrote those values during an earlier standard configuration, the user runs `ucode revert` +interactively before switching to relay. External conflicting values require administrator action or +standard Databricks authentication. + +## Status and Messages + +`ucode status` reports, for Claude and Codex: + +- managed settings path; +- state: not configured, current, compatible local settings, compatible relay settings, drifted, + invalid, unreadable, missing, or unsupported; +- whether a managed baseline backup is available. + +Interactive updates announce the backup location, administrator-permission request, and verified +result. An identical file produces no elevation message. + +Representative blockers are: + +```text +Claude Code configuration cannot be applied non-interactively because OS-managed settings at + override ucode values: env.ANTHROPIC_BASE_URL. Run `ucode configure --agent claude` from an +interactive terminal or contact your administrator. +``` + +```text +Claude Code managed settings at were updated but immediately restored by device management. +Contact your administrator. +``` + +```text +Cannot safely update Codex managed settings at : . ucode did not modify the file. +Repair it or contact your administrator. +``` + +Write, verification, parse, symlink, and managed-conflict failures block the agent launch. Recovery +information always identifies the file and recommends either an interactive configure/revert or +administrator help. + +## Removal of `use_as_global_settings` + +The final behavior has no managed-config scope choice: + +- Claude and Codex reconcile OS-managed settings automatically during interactive configuration. +- Other agents continue using their existing local configuration because they do not have the same + supported self-refreshing managed-file path. +- The Codex PR removes `use_as_global_settings` from schemas, resolution, setup prompts, summaries, + documentation, and tests. + +The Claude PR temporarily leaves Codex's legacy interpretation in place so that the first PR is +independently safe. The stacked Codex PR removes the remaining field and transitional code. + +## PR Boundaries + +### PR 1: Claude Code + +- Add shared strict reads, fingerprints, compatibility checks, secure backups, atomic privileged + writes, immutable-flag handling, verification, status, and three-way revert helpers. +- Make interactive Claude configuration reconcile OS-managed JSON by default. +- Add non-interactive local fallback with conflict detection. +- Add Claude relay-specific safety checks. +- Add Claude managed status and revert output. +- Remove Claude from the legacy `use_as_global_settings` setup choice. + +### PR 2: Codex + +- Stack on the Claude PR and reuse the shared lifecycle with strict TOML parsing and serialization. +- Make interactive Codex configuration reconcile OS-managed TOML by default. +- Add non-interactive local fallback with conflict detection. +- Add Codex fingerprinted cached launches, status, and revert behavior. +- Remove all remaining `use_as_global_settings` code and documentation. diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py index d2480390..333e2700 100644 --- a/src/ucode/agents/__init__.py +++ b/src/ucode/agents/__init__.py @@ -69,15 +69,9 @@ DEFAULT_TOOL = "codex" BUNDLE_VERSION = 1 -# Agents that can mirror ucode's managed config into the tool's NATIVE default config file, so a -# bare `claude` / `codex` (not just `ucode `) picks up the gateway settings. This is gated by -# the admin's `use_as_global_settings` choice in `ucode setup`. Only agents whose gateway auth -# self-refreshes qualify: claude's `apiKeyHelper` and codex's `ucode auth-token` command both re-mint -# tokens on their own, so the native file keeps working indefinitely. The other agents bake a -# short-lived bearer token with no bare-launch refresher (opencode/pi/gemini) or expose no native -# config file at all (copilot is env-var only), so they're excluded — and `ucode setup` doesn't even -# ask them the machine-wide question. -GLOBAL_SETTINGS_AGENTS = frozenset({"claude", "codex"}) +# Codex still honors the legacy managed-config opt-in until its follow-up migration. Claude always +# reconciles its OS-managed settings and therefore no longer appears in the setup prompt. +GLOBAL_SETTINGS_AGENTS = frozenset({"codex"}) # ucode tool -> `databricks aitools` agent id. gemini/pi aren't supported. AITOOLS_AGENT_TOKENS = { diff --git a/src/ucode/agents/claude.py b/src/ucode/agents/claude.py index 222e780d..057d262a 100644 --- a/src/ucode/agents/claude.py +++ b/src/ucode/agents/claude.py @@ -38,7 +38,18 @@ AUTHORIZATION_HEADER, ) from ucode.launcher import exec_or_spawn -from ucode.managed_files import OS, current_os, write_managed_file +from ucode.managed_files import ( + OS, + current_os, + managed_file_conflicts, + managed_file_is_verified, + managed_file_status, + managed_writes_allowed, + mark_managed_file_verified, + read_managed_file, + reconcile_managed_file, + revert_managed_file, +) from ucode.smart_routing import v2 as smart_routing_v2 from ucode.smart_routing.claude_hooks import ( remove_smart_routing_hooks, @@ -48,7 +59,7 @@ from ucode.state import get_provider_service, mark_tool_managed, save_state from ucode.telemetry import agent_version, ucode_version from ucode.tracing import tracing_env -from ucode.ui import print_err, print_note, print_success, print_warning +from ucode.ui import print_note, print_success, print_warning GATEWAY_MODEL_DISCOVERY_ENV_VAR = "ENABLE_CLAUDE_CODE_GATEWAY_MODEL_DISCOVERY" CLAUDE_CONFIG_DIR = Path.home() / ".claude" @@ -175,19 +186,62 @@ def _managed_settings_path() -> Path | None: return None -def _managed_relayed_conflicts() -> tuple[Path, list[str]] | None: - """Enterprise managed-settings keys that would break relayed auth, if any. - The managed scope always wins (per key) over the --settings file and - subscription OAuth, so a managed value here overrides what ucode writes: - 'apiKeyHelper' shadows the subscription login, 'env.ANTHROPIC_BASE_URL' - clobbers our loopback proxy URL, and 'env.ANTHROPIC_CUSTOM_HEADERS' drops - the Databricks-Model-Provider-Service routing headers — each sends traffic - somewhere the relayed token swap can't reach or route correctly. - Returns (path, conflicting-key-labels) or None when there's no conflict.""" +def _parse_managed_settings(text: str) -> dict: + try: + settings = json.loads(text) + except json.JSONDecodeError as exc: + raise RuntimeError( + f"invalid JSON at line {exc.lineno}, column {exc.colno}: {exc.msg}" + ) from exc + if not isinstance(settings, dict): + raise RuntimeError("the top-level JSON value must be an object") + return settings + + +def _dump_managed_settings(settings: dict) -> str: + return json.dumps(settings, indent=2) + "\n" + + +def managed_settings_are_current(state: dict) -> bool: path = _managed_settings_path() - if path is None or not path.is_file(): - return None - settings = read_json_safe(path) + if path is None: + return True + if state.get("claude_relayed"): + required_scope = "relay-compatible" + elif managed_writes_allowed(): + required_scope = "managed" + else: + required_scope = None + return managed_file_is_verified(state, "claude", path, required_scope=required_scope) + + +def managed_settings_status(state: dict) -> tuple[Path | None, str, str]: + path = _managed_settings_path() + status, backup = managed_file_status(state, "claude", path, parser=_parse_managed_settings) + return path, status, backup + + +def revert_managed_settings() -> str: + return revert_managed_file( + "claude", + display="Claude Code", + parser=_parse_managed_settings, + dumper=_dump_managed_settings, + ) + + +def _managed_relayed_conflicts(path: Path) -> list[str]: + """Return managed settings that would override Claude subscription relay auth.""" + text = read_managed_file(path) + if text is None: + return [] + try: + settings = _parse_managed_settings(text) + except RuntimeError as exc: + raise RuntimeError( + f"Cannot safely inspect Claude Code managed settings at {path}: {exc}. Repair the " + "file or contact your administrator." + ) from exc conflicts: list[str] = [] if settings.get("apiKeyHelper"): conflicts.append("apiKeyHelper") @@ -197,45 +251,7 @@ def _managed_relayed_conflicts() -> tuple[Path, list[str]] | None: conflicts.append("env.ANTHROPIC_BASE_URL") if env.get("ANTHROPIC_CUSTOM_HEADERS"): conflicts.append("env.ANTHROPIC_CUSTOM_HEADERS") - return (path, conflicts) if conflicts else None - - -def _managed_pinned_model() -> tuple[Path, str] | None: - """Model that enterprise managed settings force Claude Code to launch with, - if any. Only `ANTHROPIC_MODEL` sets the launch model — the - `ANTHROPIC_DEFAULT_*` family aliases just remap what each tier resolves to - when selected, so they don't change the default. Doesn't break relayed auth, - but silently overrides Claude Code's model, so we surface it. Returns - (path, model_id) or None when the file is absent or `ANTHROPIC_MODEL` is unset.""" - path = _managed_settings_path() - if path is None or not path.is_file(): - return None - settings = read_json_safe(path) - env = settings.get("env") - if not isinstance(env, dict) or not env.get("ANTHROPIC_MODEL"): - return None - return (path, str(env["ANTHROPIC_MODEL"])) - - -def managed_settings_model_overrides() -> Path | None: - """Path to enterprise managed settings when they pin a model ucode selects with, else None. - - The enterprise scope outranks the ``--settings`` file ucode passes, so a model set there wins - over the one an admin published in the workspace's managed config — and unlike the user and - project scopes it can't be excluded with ``--setting-sources``. Callers surface this as a warning - so a developer whose models don't match their admin's config knows where to look. - - Only the keys ucode actually writes count. The ``_NAME`` companions in - :data:`CLAUDE_MANAGED_MODEL_ENV_KEYS` are picker labels that select nothing, so an enterprise - value there can't override anything and warning about it would be noise.""" - path = _managed_settings_path() - if path is None or not path.is_file(): - return None - env = read_json_safe(path).get("env") - if not isinstance(env, dict): - return None - selecting_keys = (key for key in CLAUDE_MANAGED_MODEL_ENV_KEYS if not key.endswith("_NAME")) - return path if any(env.get(key) for key in selecting_keys) else None + return conflicts def relayed_proxy_base_url(state: dict) -> str: @@ -577,6 +593,16 @@ def write_tool_config( routing_enabled = smart_routing_enabled(state) and provider is None if routing_enabled: managed_keys = managed_keys + [["hooks", event] for event in CLAUDE_ROUTING_HOOK_EVENTS] + managed_file_keys = list(managed_keys) + for path in ( + [["env", key] for key in CLAUDE_MANAGED_MODEL_ENV_KEYS] + + [["env", key] for key in CLAUDE_REMOVED_ENV_KEYS] + + [["env", key] for key in CLAUDE_TRACING_ENV_KEYS] + + [["hooks", "Stop"]] + + [["hooks", event] for event in CLAUDE_ROUTING_HOOK_EVENTS] + ): + if path not in managed_file_keys: + managed_file_keys.append(path) def _compose(base: dict) -> dict: # deepcopy the overlay per file so merging into one base can't alias nested dicts into @@ -612,8 +638,7 @@ def _compose(base: dict) -> dict: write_json_file(CLAUDE_SETTINGS_PATH, _compose(read_json_safe(CLAUDE_SETTINGS_PATH))) - if state.get("write_managed_config"): - _write_managed_settings(_compose, relayed) + _reconcile_managed_settings(state, _compose, managed_file_keys, relayed) if web_search_model: web_search_entry = _web_search_mcp_entry( @@ -641,34 +666,90 @@ def _compose(base: dict) -> dict: return state -def _write_managed_settings(compose: Callable[[dict], dict], relayed: bool) -> None: - """Write ucode's config into Claude Code's OS managed-settings.json so a bare `claude` works. +def _reconcile_managed_settings( + state: dict, + compose: Callable[[dict], dict], + owned_paths: list[list[str]], + relayed: bool, +) -> None: + """Reconcile Claude Code's OS-managed settings so a bare ``claude`` uses the gateway. - Runs only under use_as_global_settings. The managed file is root-owned and the highest-precedence - scope, so it applies whether or not `ucode` launches `claude`. The same compose (merge overlay + - prune stale keys) that produced the private file is applied to the existing managed file, so any - real IT-authored keys already there survive. The write goes through the sudo path in - `managed_files` (drift-suppressed, so no password prompt when unchanged). + The managed file is root-owned and the highest-precedence scope, so every normal Claude + configuration mirrors ucode's settings there. The same compose operation that produced the + private file is applied to the existing managed file, preserving unrelated IT-authored keys. Relayed launches are skipped: they depend on a per-session loopback refresh proxy that only runs during `ucode claude`, so a bare `claude` could not reach the gateway anyway. """ - if relayed: - print_warning( - "Claude subscription-relay launches use a per-session proxy a bare `claude` can't " - "reach, so ucode did not write the managed settings file. Launch with `ucode claude` " - "to use the relay." - ) - return path = _managed_settings_path() if path is None: print_warning( "Machine-wide Claude settings aren't supported on this platform; skipped the managed " - "settings write." + "settings." ) return - desired = json.dumps(compose(read_json_safe(path)), indent=2) - write_managed_file(path, desired, display="Claude Code") + if path.is_symlink(): + raise RuntimeError( + f"Refusing to use Claude Code managed settings through symlink {path}. Replace it " + "with a regular file or contact your administrator." + ) + if relayed: + conflicts = _managed_relayed_conflicts(path) + if conflicts: + raise RuntimeError( + "Claude subscription relay cannot start because enterprise managed settings " + f"define {', '.join(conflicts)} at {path}. Ask your administrator to remove " + "those entries or use standard Databricks authentication. If ucode previously " + "created them, run `ucode revert` from an interactive terminal first." + ) + mark_managed_file_verified(state, "claude", path, scope="relay-compatible") + return + + current_text = read_managed_file(path) + try: + existing = _parse_managed_settings(current_text) if current_text is not None else {} + except RuntimeError as exc: + raise RuntimeError( + f"Cannot safely update Claude Code managed settings at {path}: {exc}. " + "ucode did not modify the file. Repair it or contact your administrator." + ) from exc + managed_before = copy.deepcopy(existing) + desired_settings = compose(existing) + _preserve_permission_denies(managed_before, desired_settings) + if not managed_writes_allowed(): + conflicts = managed_file_conflicts(managed_before, desired_settings, owned_paths) + if conflicts: + raise RuntimeError( + "Claude Code configuration cannot be applied non-interactively because " + f"OS-managed settings at {path} override ucode values: {', '.join(conflicts)}. " + "Run `ucode configure --agent claude` from an interactive terminal or contact " + "your administrator." + ) + mark_managed_file_verified(state, "claude", path, scope="local-compatible") + return + reconcile_managed_file( + path, + _dump_managed_settings(desired_settings), + tool="claude", + display="Claude Code", + owned_paths=owned_paths, + ) + mark_managed_file_verified(state, "claude", path) + + +def _preserve_permission_denies(existing: dict, desired: dict) -> None: + existing_permissions = existing.get("permissions") + desired_permissions = desired.get("permissions") + if not isinstance(existing_permissions, dict) or not isinstance(desired_permissions, dict): + return + existing_denies = existing_permissions.get("deny") + desired_denies = desired_permissions.get("deny") + if not isinstance(existing_denies, list) or not isinstance(desired_denies, list): + return + desired_permissions["deny"] = [ + *existing_denies, + *(rule for rule in desired_denies if rule not in existing_denies), + ] def _is_tracing_stop_hook(hook: object) -> bool: @@ -1097,25 +1178,6 @@ def _launch_relayed(state: dict, binary: str, tool_args: list[str]) -> None: """Relayed launch: sign into the Claude subscription, start the loopback refresh proxy, then run Claude Code alongside it (the proxy must outlive the exec, so we spawn-and-wait rather than replacing the process).""" - conflict = _managed_relayed_conflicts() - if conflict is not None: - managed_path, keys = conflict - print_err( - "Enterprise managed settings are present, which Claude Code always " - "applies over relayed (Claude Max/Enterprise) auth. Remove " - f"{', '.join(keys)} from {managed_path} file before running relayed auth." - ) - raise SystemExit(1) - - pinned_model = _managed_pinned_model() - if pinned_model is not None: - managed_path, model_id = pinned_model - print_warning( - f"Default model ANTHROPIC_MODEL: {model_id} is set in your " - f"enterprise-managed settings ({managed_path}) and may override ucode " - "settings. Remove this entry if you encounter issues." - ) - _ensure_subscription_login() workspace = state["workspace"] port = state.get("relayed_proxy_port") diff --git a/src/ucode/cli.py b/src/ucode/cli.py index c352e51b..8e4046ca 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -77,7 +77,6 @@ managed_provider_service, managed_supplies_models, managed_unservable_models, - managed_use_as_global_settings, recommended_agent, resolve_state, ) @@ -1046,6 +1045,13 @@ def status() -> int: ", ".join(tool_mcp_servers) if tool_mcp_servers else "none saved by ucode", ) print_kv("Config file", str(config_path) if config_path.exists() else "missing") + if tool == "claude": + managed_path, managed_status, backup_status = claude_agent.managed_settings_status( + state + ) + print_kv("OS-managed settings", managed_status) + print_kv("Managed settings file", str(managed_path) if managed_path else "unsupported") + print_kv("Managed settings backup", backup_status) console.print() print_heading("Skills") @@ -1101,6 +1107,7 @@ def revert() -> int: state = load_state() managed_configs = state.get("managed_configs") or {} mcp_results = revert_mcp_configs(state) + claude_managed_result = claude_agent.revert_managed_settings() results: dict[str, bool] = { tool: restore_file( @@ -1122,6 +1129,7 @@ def revert() -> int: print_kv(f"{spec['display']} config", "restored" if results[tool] else "unchanged") if legacy_codex_stripped: print_kv("Codex shared config", "ucode entries removed") + print_kv("Claude Code OS-managed settings", claude_managed_result) print_kv("Pi settings", "restored" if pi_settings_restored else "unchanged") for client, spec in MCP_CLIENTS.items(): print_kv( @@ -1844,7 +1852,10 @@ def _can_launch_from_cached_config( return False if tool == "claude": - return claude_agent.CLAUDE_SETTINGS_PATH.exists() + return ( + claude_agent.CLAUDE_SETTINGS_PATH.exists() + and claude_agent.managed_settings_are_current(state) + ) return codex_agent.has_ucode_config() @@ -1942,18 +1953,6 @@ def _launch_tool( f"Your workspace's managed config lists no {TOOL_SPECS[tool]['display']}-servable " f"models ({', '.join(unservable)}); using your discovered models instead." ) - # The enterprise scope outranks the --settings file ucode writes, so a model pinned - # there quietly beats the admin's — point at the file rather than let the mismatch - # look like a ucode bug. Suppressed under use_as_global_settings: there ucode itself - # authored that managed-settings file, so its model keys are the admin's config, not an - # external override. - if tool == "claude" and not managed_use_as_global_settings(managed, "claude"): - overrides = claude_agent.managed_settings_model_overrides() - if overrides is not None: - print_warning( - f"Default models are set in your enterprise managed settings at " - f"{overrides}, which may override your admin's managed config." - ) elif managed_agent_config_enabled(): print_note("No managed coding agent config found; using your own settings") if managed is not None: @@ -2060,23 +2059,6 @@ def _launch_tool( # the id can't ride `resolved_model` — it is threaded separately as `custom_model`. if model and tool != "claude": resolved_model = model - # Claude Code's enterprise managed-settings scope (e.g. a dbexec install) - # outranks the --settings file ucode writes AND can't be excluded with --setting-sources, - # so a model pinned there silently wins over `--model`. Warn so a launch that ignores the - # requested model looks like the misconfiguration it is, not a ucode bug. - # Suppressed when ucode authored the managed-settings file itself (use_as_global_settings) - # — the pinned model is then ucode's own, deliberately applied, not a surprise override. - managed_owns_claude = managed is not None and managed_use_as_global_settings( - managed, "claude" - ) - if model and tool == "claude" and not managed_owns_claude: - enterprise = claude_agent.managed_settings_model_overrides() - if enterprise is not None: - print_warning( - f"Your enterprise managed settings at {enterprise} pin the Claude model, " - f"which overrides `--model {model}` — Claude Code will launch on the pinned " - "model instead. Edit or remove that file to use --model." - ) state = configure_tool( tool, state, diff --git a/src/ucode/managed_files.py b/src/ucode/managed_files.py index 40e58803..aa2c9157 100644 --- a/src/ucode/managed_files.py +++ b/src/ucode/managed_files.py @@ -1,35 +1,37 @@ -"""Write agent config into OS-level *managed settings* files. +"""Safely manage root-owned, highest-precedence agent settings files. -These files are root-owned and the highest-precedence config scope for their agent — a bare -``claude`` / ``codex`` (launched directly, without ucode) reads them, so writing here is what makes -the gateway config apply outside ``ucode ``: - -- Claude Code: ``/etc/claude-code/managed-settings.json`` (Linux), - ``/Library/Application Support/ClaudeCode/managed-settings.json`` (macOS) -- Codex: ``/etc/codex/managed_config.toml`` (Linux + macOS) - -The write is guarded by a **drift check**: it reads the world-readable file WITHOUT sudo and does -nothing when it already matches, so the common no-op launch never prompts for a password; only a -real change shells out to ``sudo`` (temp file → ``sudo cp``), clearing and restoring the immutable -flag (``chattr``/``chflags``) that a fleet golden image may have set. Writing needs root, so the -first write (or one after the config changes) prompts for the developer's sudo password. +Interactive updates preserve unrelated policy, retain a private baseline for ``ucode revert``, and +verify the privileged atomic replacement. Non-interactive runs only check whether existing managed +values are compatible with ucode's local settings. """ from __future__ import annotations +import hashlib +import json import os import shlex import subprocess import sys import tempfile +from collections.abc import Callable +from copy import deepcopy from enum import Enum from pathlib import Path +from typing import Any, cast -from ucode.config_io import is_dry_run -from ucode.ui import console, print_err, print_warning +from ucode.config_io import APP_DIR, is_dry_run +from ucode.ui import console, print_err, print_note, print_success, print_warning # Absolute path so a stripped PATH (desktop/GUI launchers) still finds it. _SUDO = "/usr/bin/sudo" +MANAGED_BACKUP_DIR = APP_DIR / "managed-backups" +MANAGED_BACKUP_MANIFEST_PATH = MANAGED_BACKUP_DIR / "manifest.json" +MANAGED_FINGERPRINT_VERSION = 1 +_MISSING = object() + +ManagedParser = Callable[[str], dict] +ManagedDumper = Callable[[dict], str] class OS(Enum): @@ -70,6 +72,522 @@ def _read_existing(path: Path) -> str: return "" +def read_managed_file(path: Path) -> str | None: + """Read a managed file strictly, returning ``None`` only when it is absent.""" + try: + return path.read_text(encoding="utf-8") + except FileNotFoundError: + return None + except OSError as exc: + raise RuntimeError(f"Cannot read managed settings at {path}: {exc}") from exc + + +def managed_file_fingerprint(path: Path) -> dict[str, int | bool]: + """Return metadata sufficient to detect normal MDM replacement or in-place edits.""" + try: + stat = path.stat() + except FileNotFoundError: + return {"exists": False} + except OSError as exc: + raise RuntimeError(f"Cannot inspect managed settings at {path}: {exc}") from exc + return { + "exists": True, + "device": stat.st_dev, + "inode": stat.st_ino, + "size": stat.st_size, + "mtime_ns": stat.st_mtime_ns, + "ctime_ns": stat.st_ctime_ns, + } + + +def managed_file_is_verified( + state: dict, tool: str, path: Path, *, required_scope: str | None = None +) -> bool: + """Fast cached-launch guard: one stat and no content parsing when unchanged.""" + records = state.get("managed_file_fingerprints") + if not isinstance(records, dict): + return False + record = records.get(tool) + if not isinstance(record, dict): + return False + if record.get("version") != MANAGED_FINGERPRINT_VERSION or record.get("path") != str(path): + return False + if required_scope is not None and record.get("scope") != required_scope: + return False + try: + return record.get("fingerprint") == managed_file_fingerprint(path) + except RuntimeError: + return False + + +def mark_managed_file_verified( + state: dict, tool: str, path: Path, *, scope: str = "managed" +) -> None: + records = dict(state.get("managed_file_fingerprints") or {}) + records[tool] = { + "version": MANAGED_FINGERPRINT_VERSION, + "path": str(path), + "scope": scope, + "fingerprint": managed_file_fingerprint(path), + } + state["managed_file_fingerprints"] = records + + +def managed_writes_allowed() -> bool: + """Managed writes are interactive setup work; scripts and CI use local settings.""" + return sys.stdin.isatty() + + +def managed_file_conflicts( + existing: dict, desired: dict, owned_paths: list[list[str]] +) -> list[str]: + """Return managed leaves that would override ucode's local settings.""" + conflicts: list[str] = [] + for path in owned_paths: + existing_value = _path_value(existing, path) + if existing_value is _MISSING: + continue + if existing_value != _path_value(desired, path): + conflicts.append(".".join(path)) + return conflicts + + +def managed_file_status( + state: dict, + tool: str, + path: Path | None, + *, + parser: ManagedParser | None = None, +) -> tuple[str, str]: + """Return a read-only status and backup label for ``ucode status``.""" + if path is None: + return "unsupported", "none" + try: + fingerprint = managed_file_fingerprint(path) + except RuntimeError: + return "unreadable", _backup_label(tool) + records = state.get("managed_file_fingerprints") + record = records.get(tool) if isinstance(records, dict) else None + if not fingerprint.get("exists"): + status = "missing" if isinstance(record, dict) else "not configured" + elif not isinstance(record, dict): + status = "not configured" + elif managed_file_is_verified(state, tool, path): + scope = record.get("scope") + if scope == "local-compatible": + status = "compatible (local settings)" + elif scope == "relay-compatible": + status = "compatible (relay settings)" + else: + status = "current" + else: + status = "drifted" + if parser is not None and status in {"current", "drifted", "not configured"}: + try: + text = read_managed_file(path) + except RuntimeError: + status = "unreadable" + else: + try: + if text is not None: + parser(text) + except RuntimeError: + status = "invalid" + return status, _backup_label(tool) + + +def reconcile_managed_file( + path: Path, + desired_text: str, + *, + tool: str, + display: str, + owned_paths: list[list[str]], +) -> str: + """Back up, atomically write, and verify one OS-managed settings file. + + The first pre-ucode contents are retained until ``ucode revert``. Subsequent writes update only + the last-applied snapshot used for drift-safe three-way restoration. + """ + if not managed_files_supported(): + print_warning( + f"{display}: OS-managed settings aren't supported on this platform; skipped {path}." + ) + return "unsupported" + if not managed_writes_allowed() and not is_dry_run(): + raise RuntimeError( + f"Refusing to update {display} managed settings at {path} non-interactively. " + "Run the command from an interactive terminal." + ) + if path.is_symlink(): + raise RuntimeError( + f"Refusing to update {display} managed settings through symlink {path}. " + "Replace it with a regular file or contact your administrator." + ) + current_text = read_managed_file(path) + if current_text == desired_text: + return "unchanged" + if is_dry_run(): + console.print(f"\n[bold]\\[dry run] {path} (via sudo)[/bold]\n{desired_text}") + return "written" + + created = current_text is None + backup_created = _ensure_backup(tool, path, current_text) + if backup_created: + print_note(f"{display}: original managed settings backed up under {MANAGED_BACKUP_DIR}.") + print_note(f"{display}: administrator permission is required to update {path}.") + if read_managed_file(path) != current_text: + raise RuntimeError( + f"{display} managed settings changed while ucode was preparing the update. " + "ucode preserved the newer file; run the command again." + ) + for attempt in range(2): + try: + _sudo_replace(path, desired_text) + except PermissionError as exc: + raise RuntimeError( + f"{display} cannot start because ucode could not update {path}: {exc}. " + "Run the ucode command from an interactive terminal and approve the administrator " + "prompt, or contact your administrator." + ) from exc + except subprocess.CalledProcessError as exc: + raise RuntimeError(_sudo_failure_message(path, display, exc)) from exc + + written_text = read_managed_file(path) + if written_text == desired_text: + break + if attempt == 0 and written_text == current_text: + print_warning( + f"{display} managed settings were restored during the update; retrying once." + ) + continue + if written_text == current_text: + raise RuntimeError( + f"{display} managed settings at {path} were updated but immediately restored by " + "device management. Contact your administrator." + ) + raise RuntimeError( + f"{display} managed settings changed concurrently at {path}. ucode will not overwrite " + "the newer policy; run the command again or contact your administrator." + ) + _record_last_applied(tool, path, desired_text, owned_paths) + print_success(f"{display} managed settings {'created' if created else 'updated'} and verified") + return "created" if created else "written" + + +def revert_managed_file( + tool: str, + *, + display: str, + parser: ManagedParser, + dumper: ManagedDumper, +) -> str: + """Restore one managed file from its baseline while preserving later external edits.""" + manifest = _load_manifest() + entry = _manifest_files(manifest).get(tool) + if not isinstance(entry, dict): + return "unchanged" + path = Path(str(entry.get("path") or "")) + if not path.is_absolute(): + raise RuntimeError(f"Invalid managed-settings backup path for {display}.") + if path.is_symlink(): + raise RuntimeError( + f"Refusing to restore {display} managed settings through symlink {path}." + ) + current_text = read_managed_file(path) + original_text = _original_text(entry) + last_text = _snapshot_text(entry, "last_applied_file") + + if current_text == last_text: + desired_text = original_text + elif current_text is None or last_text is None: + desired_text = current_text + else: + try: + current_doc = parser(current_text) + original_doc = parser(original_text) if original_text is not None else {} + last_doc = parser(last_text) + except Exception as exc: # noqa: BLE001 + raise RuntimeError( + f"Cannot safely revert {display} managed settings at {path}: {exc}" + ) from exc + owned_paths = entry.get("owned_paths") + paths = owned_paths if isinstance(owned_paths, list) else [] + reverted = _three_way_revert(current_doc, original_doc, last_doc, paths) + desired_text = dumper(reverted) + + if desired_text != current_text: + if not managed_writes_allowed(): + raise RuntimeError( + f"Cannot restore {display} managed settings non-interactively. Run `ucode revert` " + "from an interactive terminal." + ) + try: + if desired_text is None: + _sudo_remove(path) + else: + _sudo_replace(path, desired_text) + except (PermissionError, subprocess.CalledProcessError) as exc: + raise RuntimeError( + f"Could not restore {display} managed settings at {path}. The backup was retained " + f"under {MANAGED_BACKUP_DIR}. Resolve the permission issue and run `ucode revert` " + "again." + ) from exc + if read_managed_file(path) != desired_text: + raise RuntimeError( + f"Could not verify restored {display} managed settings at {path}. The backup was " + f"retained under {MANAGED_BACKUP_DIR}." + ) + + _delete_backup(tool, manifest, entry) + if original_text is None and desired_text is None: + return "removed" + if current_text != last_text: + return "ucode entries removed; external changes preserved" + return "restored" + + +def _sha256(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _manifest_files(manifest: dict) -> dict: + files = manifest.get("files") + if not isinstance(files, dict): + files = {} + manifest["files"] = files + return files + + +def _load_manifest() -> dict: + try: + if MANAGED_BACKUP_MANIFEST_PATH.is_symlink(): + raise RuntimeError( + f"Refusing to read symlinked managed-settings backup manifest at " + f"{MANAGED_BACKUP_MANIFEST_PATH}." + ) + if not MANAGED_BACKUP_MANIFEST_PATH.exists(): + return {"version": 1, "files": {}} + manifest = json.loads(MANAGED_BACKUP_MANIFEST_PATH.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise RuntimeError( + f"Cannot read managed-settings backup manifest at {MANAGED_BACKUP_MANIFEST_PATH}: {exc}" + ) from exc + if not isinstance(manifest, dict) or manifest.get("version") != 1: + raise RuntimeError( + f"Unsupported managed-settings backup manifest at {MANAGED_BACKUP_MANIFEST_PATH}." + ) + return manifest + + +def _write_private_file(path: Path, text: str) -> None: + if MANAGED_BACKUP_DIR.is_symlink(): + raise RuntimeError(f"Refusing to use symlinked backup directory {MANAGED_BACKUP_DIR}.") + MANAGED_BACKUP_DIR.mkdir(parents=True, exist_ok=True) + os.chmod(MANAGED_BACKUP_DIR, 0o700) + with tempfile.NamedTemporaryFile( + mode="w", dir=MANAGED_BACKUP_DIR, delete=False, encoding="utf-8" + ) as tmp: + tmp.write(text) + tmp_path = Path(tmp.name) + try: + os.chmod(tmp_path, 0o600) + os.replace(tmp_path, path) + finally: + if tmp_path.exists(): + tmp_path.unlink() + + +def _write_manifest(manifest: dict) -> None: + _write_private_file(MANAGED_BACKUP_MANIFEST_PATH, json.dumps(manifest, indent=2) + "\n") + + +def _backup_filename(tool: str, path: Path) -> str: + suffix = path.suffix or ".txt" + return f"{tool}-managed-settings.backup{suffix}" + + +def _last_applied_filename(tool: str, path: Path) -> str: + suffix = path.suffix or ".txt" + return f"{tool}-managed-settings.last-applied{suffix}" + + +def _ensure_backup(tool: str, path: Path, current_text: str | None) -> bool: + manifest = _load_manifest() + files = _manifest_files(manifest) + existing = files.get(tool) + if isinstance(existing, dict): + if existing.get("path") != str(path): + raise RuntimeError( + f"The saved {tool} managed-settings backup targets {existing.get('path')}, not " + f"{path}. Run `ucode revert` before configuring this path." + ) + if existing.get("original_existed"): + _original_text(existing) + return False + + entry: dict[str, Any] = { + "path": str(path), + "original_existed": current_text is not None, + "owned_paths": [], + } + if current_text is not None: + backup_file = _backup_filename(tool, path) + _write_private_file(MANAGED_BACKUP_DIR / backup_file, current_text) + entry["backup_file"] = backup_file + entry["original_sha256"] = _sha256(current_text) + files[tool] = entry + _write_manifest(manifest) + return True + + +def _record_last_applied( + tool: str, path: Path, desired_text: str, owned_paths: list[list[str]] +) -> None: + manifest = _load_manifest() + entry = _manifest_files(manifest).get(tool) + if not isinstance(entry, dict): + raise RuntimeError(f"Missing managed-settings backup metadata for {tool}.") + last_file = _last_applied_filename(tool, path) + _write_private_file(MANAGED_BACKUP_DIR / last_file, desired_text) + entry["last_applied_file"] = last_file + entry["last_applied_sha256"] = _sha256(desired_text) + known_paths = entry.get("owned_paths") if isinstance(entry.get("owned_paths"), list) else [] + for owned_path in owned_paths: + if owned_path not in known_paths: + known_paths.append(list(owned_path)) + entry["owned_paths"] = known_paths + _write_manifest(manifest) + + +def _snapshot_text(entry: dict, key: str) -> str | None: + filename = entry.get(key) + if not isinstance(filename, str): + return None + path = _snapshot_path(filename) + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + raise RuntimeError(f"Cannot read managed-settings snapshot at {path}: {exc}") from exc + hash_key = "original_sha256" if key == "backup_file" else "last_applied_sha256" + expected_hash = entry.get(hash_key) + if not isinstance(expected_hash, str) or _sha256(text) != expected_hash: + raise RuntimeError(f"Managed-settings snapshot failed integrity verification at {path}.") + return text + + +def _original_text(entry: dict) -> str | None: + if not entry.get("original_existed"): + return None + return _snapshot_text(entry, "backup_file") + + +def _backup_label(tool: str) -> str: + try: + entry = _manifest_files(_load_manifest()).get(tool) + except RuntimeError: + return "invalid" + return "available" if isinstance(entry, dict) else "none" + + +def _delete_backup(tool: str, manifest: dict, entry: dict) -> None: + for key in ("backup_file", "last_applied_file"): + filename = entry.get(key) + if isinstance(filename, str): + try: + _snapshot_path(filename).unlink(missing_ok=True) + except OSError as exc: + raise RuntimeError(f"Could not remove managed-settings backup: {exc}") from exc + _manifest_files(manifest).pop(tool, None) + _write_manifest(manifest) + + +def _snapshot_path(filename: str) -> Path: + if Path(filename).name != filename: + raise RuntimeError(f"Invalid managed-settings snapshot filename: {filename}") + return MANAGED_BACKUP_DIR / filename + + +def _path_value(doc: dict, path: list[str]) -> object: + node: object = doc + for key in path: + if not isinstance(node, dict) or key not in node: + return _MISSING + node = cast(dict, node)[key] + return node + + +def _set_path_value(doc: dict, path: list[str], value: object) -> None: + node = doc + for key in path[:-1]: + child = node.get(key) + if not isinstance(child, dict): + child = {} + node[key] = child + node = child + node[path[-1]] = deepcopy(value) + + +def _delete_path_value(doc: dict, path: list[str]) -> None: + parents: list[tuple[dict, str]] = [] + node = doc + for key in path[:-1]: + child = node.get(key) + if not isinstance(child, dict): + return + parents.append((node, key)) + node = child + node.pop(path[-1], None) + for parent, key in reversed(parents): + child = parent.get(key) + if isinstance(child, dict) and not child: + parent.pop(key, None) + + +def _owned_path(value: object) -> list[str] | None: + if not isinstance(value, list) or not value or not all(isinstance(part, str) for part in value): + return None + return cast(list[str], value) + + +def _three_way_revert(current: dict, original: dict, last: dict, paths: list) -> dict: + reverted = deepcopy(current) + for raw_path in paths: + path = _owned_path(raw_path) + if path is None: + continue + current_value = _path_value(reverted, path) + original_value = _path_value(original, path) + last_value = _path_value(last, path) + if current_value == last_value: + if original_value is _MISSING: + _delete_path_value(reverted, path) + else: + _set_path_value(reverted, path, original_value) + continue + if not isinstance(current_value, list) or not isinstance(last_value, list): + continue + original_list = original_value if isinstance(original_value, list) else [] + additions = [item for item in last_value if item not in original_list] + cleaned = [item for item in current_value if item not in additions] + if cleaned: + _set_path_value(reverted, path, cleaned) + else: + _delete_path_value(reverted, path) + return reverted + + +def _sudo_remove(path: Path) -> None: + original_flags = _clear_immutable(path) + try: + subprocess.run( + _sudo_command("rm", "-f", str(path)), capture_output=True, text=True, check=True + ) + finally: + if original_flags and path.exists(): + _restore_immutable(path, original_flags) + + def write_managed_file(path: Path, desired_text: str, *, display: str) -> str: """Write ``desired_text`` to a root-owned managed file, only when it differs (drift check). @@ -89,6 +607,12 @@ def write_managed_file(path: Path, desired_text: str, *, display: str) -> str: if is_dry_run(): console.print(f"\n[bold]\\[dry run] {path} (via sudo)[/bold]\n{desired_text}") return "written" + if not managed_writes_allowed(): + print_warning( + f"{display}: skipped the OS-managed settings update at {path} because the command " + "is non-interactive." + ) + return "skipped" try: _sudo_replace(path, desired_text) except PermissionError as exc: @@ -104,69 +628,129 @@ def write_managed_file(path: Path, desired_text: str, *, display: str) -> str: def _sudo_replace(path: Path, desired_text: str) -> None: - """Replace ``path`` with ``desired_text`` via sudo (temp file → ``sudo cp``), handling immutability. - - Writes the payload to a user-owned temp file first (no sudo), then copies it into place with - ``sudo`` and makes it world-readable so the file it lays down is readable by the agent binary - regardless of who launched it. - """ - subprocess.run([_SUDO, "mkdir", "-p", str(path.parent)], check=True) + """Atomically replace ``path`` via sudo while preserving metadata and file flags.""" + if not managed_writes_allowed(): + raise RuntimeError("Refusing to invoke sudo for managed settings non-interactively.") + try: + parent_existed = path.parent.exists() + except OSError: + parent_existed = True + subprocess.run(_sudo_command("mkdir", "-p", str(path.parent)), check=True) + if not parent_existed: + subprocess.run(_sudo_command("chown", "0:0", str(path.parent)), check=True) + subprocess.run(_sudo_command("chmod", "755", str(path.parent)), check=True) with tempfile.NamedTemporaryFile( mode="w", suffix=path.suffix or ".tmp", delete=False, encoding="utf-8" ) as tmp: tmp.write(desired_text) tmp_path = tmp.name + staging_path: str | None = None + original_flags: tuple[str, ...] = () try: - restore_immutable = _clear_immutable(path) - try: - # capture_output so the CalledProcessError on failure (e.g. still-immutable dest) carries - # cp's stderr for an actionable message. + result = subprocess.run( + _sudo_command("mktemp", str(path.parent / f".{path.name}.ucode.XXXXXX")), + capture_output=True, + text=True, + check=True, + ) + staging_path = result.stdout.strip() + if not staging_path or Path(staging_path).parent != path.parent: + raise RuntimeError(f"sudo mktemp returned an invalid staging path for {path}.") + + path_exists = path.exists() + if path_exists: + original_flags = _clear_immutable(path) + preserve_args = ["-p"] if current_os() is OS.MACOS else ["--preserve=all"] subprocess.run( - [_SUDO, "cp", tmp_path, str(path)], capture_output=True, text=True, check=True + _sudo_command("cp", *preserve_args, str(path), staging_path), + capture_output=True, + text=True, + check=True, ) - subprocess.run([_SUDO, "chmod", "a+rx", str(path.parent)], check=True) - subprocess.run([_SUDO, "chmod", "a+r", str(path)], check=True) - finally: - if restore_immutable: - _restore_immutable(path) + _clear_immutable(Path(staging_path)) + + subprocess.run( + _sudo_command("cp", tmp_path, staging_path), + capture_output=True, + text=True, + check=True, + ) + if not path_exists: + subprocess.run(_sudo_command("chown", "0:0", staging_path), check=True) + subprocess.run(_sudo_command("chmod", "644", staging_path), check=True) + + subprocess.run( + _sudo_command("mv", "-f", staging_path, str(path)), + capture_output=True, + text=True, + check=True, + ) + staging_path = None + if original_flags: + _restore_immutable(path, original_flags) + original_flags = () finally: + if original_flags and path.exists(): + _restore_immutable(path, original_flags) os.unlink(tmp_path) + if staging_path: + subprocess.run( + _sudo_command("rm", "-f", staging_path), + capture_output=True, + text=True, + check=False, + ) -def _clear_immutable(path: Path) -> bool: - """Clear an immutable flag a fleet golden image may have set. Returns whether to restore it. - - macOS: preserve JAMF's system-immutable ``schg`` across the update — inspect, unlock only when - set, and report that it must be restored. Linux: best-effort ``chattr -i`` (not every filesystem - supports it), never restored. - """ +def _clear_immutable(path: Path) -> tuple[str, ...]: + """Clear immutable/append-only flags and return the flags that must be restored.""" try: - # `path.exists()` stats the file; under a root-locked parent dir (e.g. a 750 /etc/codex we - # haven't opened yet) that raises PermissionError. There's nothing to unlock we can see, and - # the subsequent `sudo cp` (as root) overwrites regardless, so treat it as "nothing to clear". if not path.exists(): - return False + return () except OSError: - return False + return () if current_os() is OS.MACOS: result = subprocess.run( ["/usr/bin/stat", "-f", "%Sf", str(path)], capture_output=True, text=True, check=False ) - if result.returncode == 0 and "schg" in result.stdout.strip().split(","): + if result.returncode != 0: + return () + supported = {"schg", "uchg", "sappnd", "uappnd"} + flags = tuple(flag for flag in result.stdout.strip().split(",") if flag in supported) + if flags: subprocess.run( - [_SUDO, "chflags", "noschg", str(path)], capture_output=True, text=True, check=True + _sudo_command("chflags", ",".join(f"no{flag}" for flag in flags), str(path)), + capture_output=True, + text=True, + check=True, ) - return True - return False - subprocess.run([_SUDO, "chattr", "-i", str(path)], capture_output=True, text=True, check=False) - return False - - -def _restore_immutable(path: Path) -> None: - """Re-set macOS's ``schg`` flag after a write. Best-effort so it never masks the write result.""" + return flags result = subprocess.run( - [_SUDO, "chflags", "schg", str(path)], capture_output=True, text=True, check=False + _sudo_command("lsattr", "-d", str(path)), capture_output=True, text=True, check=False ) + if result.returncode != 0 or not result.stdout.strip(): + return () + attributes = result.stdout.split()[0] + flags = tuple(flag for flag in ("i", "a") if flag in attributes) + if flags: + subprocess.run( + _sudo_command("chattr", f"-{''.join(flags)}", str(path)), + capture_output=True, + text=True, + check=True, + ) + return flags + + +def _restore_immutable(path: Path, flags: tuple[str, ...]) -> None: + """Restore immutable/append-only flags after replacing a managed file.""" + if not flags: + return + if current_os() is OS.MACOS: + command = _sudo_command("chflags", ",".join(flags), str(path)) + else: + command = _sudo_command("chattr", f"+{''.join(flags)}", str(path)) + result = subprocess.run(command, capture_output=True, text=True, check=False) if result.returncode != 0: print_warning(f"Could not restore the immutable flag on {path}.") @@ -176,7 +760,7 @@ def _report_sudo_failure(path: Path, display: str, exc: subprocess.CalledProcess cause — cp fails with EPERM even under root — so point at the OS-specific clear command.""" stderr = (exc.stderr or "").strip() if isinstance(exc.stderr, str) else "" cmd = exc.cmd or [] - cp_failed = len(cmd) >= 2 and cmd[1] == "cp" + cp_failed = "cp" in cmd[1:3] if cp_failed and "Operation not permitted" in stderr: quoted = shlex.quote(str(path)) clear_cmd = f"sudo {'chflags noschg' if current_os() is OS.MACOS else 'chattr -i'} {quoted}" @@ -186,3 +770,26 @@ def _report_sudo_failure(path: Path, display: str, exc: subprocess.CalledProcess ) else: print_err(f"{display}: failed to write managed settings at {path}: {stderr or exc}") + + +def _sudo_failure_message(path: Path, display: str, exc: subprocess.CalledProcessError) -> str: + stderr = (exc.stderr or "").strip() if isinstance(exc.stderr, str) else "" + cmd = exc.cmd or [] + cp_failed = "cp" in cmd[1:3] + if cp_failed and "Operation not permitted" in stderr: + return ( + f"{display} managed settings at {path} are immutable and could not be updated. " + "Contact your administrator." + ) + return ( + f"{display} cannot start because ucode could not update {path}: {stderr or exc}. " + "Run the ucode command from an interactive terminal and approve the administrator prompt, " + "or contact your administrator." + ) + + +def _sudo_command(*args: str) -> list[str]: + """Build a sudo command only for an explicitly interactive managed-file operation.""" + if not managed_writes_allowed(): + raise RuntimeError("Refusing to invoke sudo for managed settings non-interactively.") + return [_SUDO, *args] diff --git a/tests/conftest.py b/tests/conftest.py index 4ad5f4d1..34f8a7de 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -33,6 +33,11 @@ def _isolate_ucode_state(tmp_path, monkeypatch): state_dir.mkdir() monkeypatch.setattr(state_mod, "STATE_PATH", state_dir / "state.json") monkeypatch.setattr(config_io_mod, "APP_DIR", state_dir) + backup_dir = state_dir / "managed-backups" + monkeypatch.setattr(managed_files_mod, "MANAGED_BACKUP_DIR", backup_dir) + monkeypatch.setattr( + managed_files_mod, "MANAGED_BACKUP_MANIFEST_PATH", backup_dir / "manifest.json" + ) def reject_privileged_write(path, _desired_text): pytest.fail( diff --git a/tests/test_agent_claude.py b/tests/test_agent_claude.py index dc99ab21..2f46f699 100644 --- a/tests/test_agent_claude.py +++ b/tests/test_agent_claude.py @@ -15,6 +15,11 @@ WS = "https://example.databricks.com" +@pytest.fixture(autouse=True) +def _avoid_real_managed_settings(monkeypatch): + monkeypatch.setattr(claude, "_managed_settings_path", lambda: None) + + class TestClaudeSpec: def test_binary(self): assert claude.SPEC["binary"] == "claude" @@ -488,6 +493,7 @@ def _patch(self, monkeypatch, existing, written): ) monkeypatch.setattr(claude, "save_state", lambda state: None) monkeypatch.setattr(claude, "_register_web_search_mcp", lambda *a, **kw: True) + monkeypatch.setattr(claude, "managed_writes_allowed", lambda: True) def test_strips_stale_disable_experimental_betas(self, monkeypatch): existing = {"env": {"CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS": "1"}} @@ -505,7 +511,7 @@ def test_strips_stale_disable_experimental_betas(self, monkeypatch): class TestWriteToolConfigManagedSettings: - """use_as_global_settings: also write Claude Code's OS managed-settings.json (via sudo, mocked).""" + """Every normal configuration also writes Claude Code's OS-managed settings.""" def _patch(self, monkeypatch, private_writes, managed_writes, existing_by_path=None): existing_by_path = existing_by_path or {} @@ -523,20 +529,29 @@ def _patch(self, monkeypatch, private_writes, managed_writes, existing_by_path=N ) monkeypatch.setattr(claude, "save_state", lambda state: None) monkeypatch.setattr(claude, "_register_web_search_mcp", lambda *a, **kw: True) + monkeypatch.setattr(claude, "managed_writes_allowed", lambda: True) # Deterministic managed path, and a mocked sudo writer so NO real sudo/`/etc` write happens. monkeypatch.setattr(claude, "_managed_settings_path", lambda: FAKE_MANAGED_PATH) + monkeypatch.setattr( + claude, + "read_managed_file", + lambda path: ( + json.dumps(existing_by_path[str(path)]) if str(path) in existing_by_path else None + ), + ) + monkeypatch.setattr(claude, "mark_managed_file_verified", lambda *a, **kw: None) - def fake_write_managed(path, text, *, display): + def fake_write_managed(path, text, **kwargs): managed_writes.append((str(path), text)) return "written" - monkeypatch.setattr(claude, "write_managed_file", fake_write_managed) + monkeypatch.setattr(claude, "reconcile_managed_file", fake_write_managed) - def test_writes_managed_file_when_flagged(self, monkeypatch): + def test_writes_managed_file_by_default(self, monkeypatch): private_writes: list = [] managed_writes: list = [] self._patch(monkeypatch, private_writes, managed_writes) - state = {"workspace": WS, "codex_models": [], "write_managed_config": True} + state = {"workspace": WS, "codex_models": []} claude.write_tool_config(state, "databricks-claude-sonnet-4") # Private file still written; managed file written too. assert str(claude.CLAUDE_SETTINGS_PATH) in [p for p, _ in private_writes] @@ -548,7 +563,7 @@ def test_managed_file_preserves_other_keys(self, monkeypatch): # An IT-authored key already in the managed file must survive the merge. existing = {str(FAKE_MANAGED_PATH): {"env": {"MY_OWN": "keep"}}} self._patch(monkeypatch, private_writes, managed_writes, existing) - state = {"workspace": WS, "codex_models": [], "write_managed_config": True} + state = {"workspace": WS, "codex_models": []} claude.write_tool_config(state, "databricks-claude-sonnet-4") _, text = managed_writes[0] written = json.loads(text) @@ -556,13 +571,15 @@ def test_managed_file_preserves_other_keys(self, monkeypatch): assert written["env"]["ANTHROPIC_BASE_URL"] assert written["apiKeyHelper"] - def test_no_managed_write_by_default(self, monkeypatch): + def test_managed_file_preserves_enterprise_permission_denies(self, monkeypatch): private_writes: list = [] managed_writes: list = [] - self._patch(monkeypatch, private_writes, managed_writes) - state = {"workspace": WS, "codex_models": []} + existing = {str(FAKE_MANAGED_PATH): {"permissions": {"deny": ["Bash(rm:*)"]}}} + self._patch(monkeypatch, private_writes, managed_writes, existing) + state = {"workspace": WS, "codex_models": ["databricks-gpt-5"]} claude.write_tool_config(state, "databricks-claude-sonnet-4") - assert managed_writes == [] + _, text = managed_writes[0] + assert json.loads(text)["permissions"]["deny"] == ["Bash(rm:*)", "WebSearch"] def test_relayed_skips_managed_write(self, monkeypatch): private_writes: list = [] @@ -571,10 +588,63 @@ def test_relayed_skips_managed_write(self, monkeypatch): self._patch(monkeypatch, private_writes, managed_writes) monkeypatch.setattr(claude, "print_warning", lambda msg: warns.append(msg)) monkeypatch.setattr(claude, "relayed_proxy_base_url", lambda state: "http://127.0.0.1:9999") - state = {"workspace": WS, "codex_models": [], "write_managed_config": True} + monkeypatch.setattr(claude, "_managed_relayed_conflicts", lambda path: []) + state = {"workspace": WS, "codex_models": []} claude.write_tool_config(state, "databricks-claude-sonnet-4", relayed=True) assert managed_writes == [] - assert any("bare `claude`" in w for w in warns) + assert warns == [] + + def test_relayed_fails_on_conflicting_managed_auth(self, monkeypatch): + private_writes: list = [] + managed_writes: list = [] + existing = {str(FAKE_MANAGED_PATH): {"apiKeyHelper": "enterprise-helper"}} + self._patch(monkeypatch, private_writes, managed_writes, existing) + monkeypatch.setattr(claude, "relayed_proxy_base_url", lambda state: "http://127.0.0.1:9999") + state = {"workspace": WS, "codex_models": []} + + with pytest.raises(RuntimeError, match="run `ucode revert`"): + claude.write_tool_config(state, "databricks-claude-sonnet-4", relayed=True) + + assert managed_writes == [] + + def test_relayed_rejects_invalid_managed_json(self, monkeypatch): + private_writes: list = [] + managed_writes: list = [] + self._patch(monkeypatch, private_writes, managed_writes) + monkeypatch.setattr(claude, "read_managed_file", lambda path: "{") + monkeypatch.setattr(claude, "relayed_proxy_base_url", lambda state: "http://127.0.0.1:9999") + state = {"workspace": WS, "codex_models": []} + + with pytest.raises(RuntimeError, match="Cannot safely inspect"): + claude.write_tool_config(state, "databricks-claude-sonnet-4", relayed=True) + + assert managed_writes == [] + + def test_noninteractive_uses_local_settings_when_managed_file_is_compatible(self, monkeypatch): + private_writes: list = [] + managed_writes: list = [] + self._patch(monkeypatch, private_writes, managed_writes) + monkeypatch.setattr(claude, "managed_writes_allowed", lambda: False) + state = {"workspace": WS, "codex_models": []} + + claude.write_tool_config(state, "databricks-claude-sonnet-4") + + assert managed_writes == [] + + def test_noninteractive_fails_when_managed_file_conflicts(self, monkeypatch): + private_writes: list = [] + managed_writes: list = [] + existing = { + str(FAKE_MANAGED_PATH): {"env": {"ANTHROPIC_BASE_URL": "https://other.example.com"}} + } + self._patch(monkeypatch, private_writes, managed_writes, existing) + monkeypatch.setattr(claude, "managed_writes_allowed", lambda: False) + state = {"workspace": WS, "codex_models": []} + + with pytest.raises(RuntimeError, match="cannot be applied non-interactively"): + claude.write_tool_config(state, "databricks-claude-sonnet-4") + + assert managed_writes == [] class TestRegisterWebSearchMcp: @@ -1227,44 +1297,3 @@ def test_disable_removes_only_ucode_hooks(self, tmp_path, monkeypatch): assert state.get(claude.SMART_ROUTING_STATE_KEY) is None assert list(doc["hooks"]) == ["PreToolUse"] assert doc["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "user-policy" - - -class TestManagedSettingsModelOverrides: - """Enterprise managed settings outrank ucode's --settings, so a model pinned there beats the - one an admin published — worth pointing a developer at the file.""" - - @staticmethod - def _write(monkeypatch, tmp_path, payload): - path = tmp_path / "managed-settings.json" - path.write_text(json.dumps(payload), encoding="utf-8") - monkeypatch.setattr(claude, "_managed_settings_path", lambda: path) - return path - - @pytest.mark.parametrize( - "key", - ["ANTHROPIC_MODEL", "ANTHROPIC_DEFAULT_OPUS_MODEL", "ANTHROPIC_DEFAULT_HAIKU_MODEL"], - ) - def test_reports_the_path_when_a_model_is_pinned(self, monkeypatch, tmp_path, key): - path = self._write(monkeypatch, tmp_path, {"env": {key: "system.ai.claude-opus-5"}}) - assert claude.managed_settings_model_overrides() == path - - def test_none_for_name_companions_that_select_nothing(self, monkeypatch, tmp_path): - # The `_NAME` keys are picker labels, so an enterprise value there overrides no model. - self._write(monkeypatch, tmp_path, {"env": {"ANTHROPIC_DEFAULT_OPUS_MODEL_NAME": "Opus 5"}}) - assert claude.managed_settings_model_overrides() is None - - def test_none_when_no_model_keys_are_set(self, monkeypatch, tmp_path): - self._write(monkeypatch, tmp_path, {"env": {"SOMETHING_ELSE": "1"}}) - assert claude.managed_settings_model_overrides() is None - - def test_none_when_env_block_is_absent(self, monkeypatch, tmp_path): - self._write(monkeypatch, tmp_path, {"permissions": {}}) - assert claude.managed_settings_model_overrides() is None - - def test_none_on_platforms_without_managed_settings(self, monkeypatch): - monkeypatch.setattr(claude, "_managed_settings_path", lambda: None) - assert claude.managed_settings_model_overrides() is None - - def test_none_when_the_file_does_not_exist(self, monkeypatch, tmp_path): - monkeypatch.setattr(claude, "_managed_settings_path", lambda: tmp_path / "missing.json") - assert claude.managed_settings_model_overrides() is None diff --git a/tests/test_cli.py b/tests/test_cli.py index 98a8f99b..1897e963 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -45,6 +45,7 @@ def no_state_writes(): patch("ucode.agents.__init__.save_state"), patch("ucode.agents.codex.save_state"), patch("ucode.agents.claude.save_state"), + patch("ucode.agents.claude._managed_settings_path", return_value=None), patch("ucode.agents.gemini.save_state"), patch("ucode.agents.opencode.save_state"), ): @@ -535,46 +536,6 @@ def test_provider_sets_transient_claude_launch_marker(self): assert result.exit_code == 0, result.output assert mock_launch.call_args.args[1]["_claude_launch_provider"] == "main.default.anthropic" - def test_warns_when_enterprise_settings_pin_the_model(self): - # Claude Code's enterprise managed-settings scope outranks the --settings file ucode writes, - # so --model is silently ignored; warn instead of launching on the "wrong" model unexplained. - from pathlib import Path - - with ( - patch("ucode.cli.ensure_bootstrap_dependencies"), - patch("ucode.cli.load_state", return_value=MINIMAL_STATE), - patch("ucode.cli.ensure_provider_state", return_value=MINIMAL_STATE), - patch("ucode.cli.configure_shared_state", return_value=MINIMAL_STATE), - patch("ucode.cli.resolve_launch_model", return_value=(MINIMAL_STATE, "system.ai.opus")), - patch("ucode.cli.configure_tool", return_value=MINIMAL_STATE), - patch("ucode.cli._fetch_managed_config", return_value=(None, False)), - patch( - "ucode.cli.claude_agent.managed_settings_model_overrides", - return_value=Path("/etc/claude-code/managed-settings.json"), - ), - patch("ucode.cli.launch_agent"), - ): - result = runner.invoke(app, ["claude", "--model", "main.aarushi.claude-opus-5"]) - assert result.exit_code == 0, result.output - assert "enterprise managed settings" in _strip_ansi(result.output) - assert "overrides `--model main.aarushi.claude-opus-5`" in _strip_ansi(result.output) - - def test_no_enterprise_warning_when_no_managed_settings(self): - with ( - patch("ucode.cli.ensure_bootstrap_dependencies"), - patch("ucode.cli.load_state", return_value=MINIMAL_STATE), - patch("ucode.cli.ensure_provider_state", return_value=MINIMAL_STATE), - patch("ucode.cli.configure_shared_state", return_value=MINIMAL_STATE), - patch("ucode.cli.resolve_launch_model", return_value=(MINIMAL_STATE, "system.ai.opus")), - patch("ucode.cli.configure_tool", return_value=MINIMAL_STATE), - patch("ucode.cli._fetch_managed_config", return_value=(None, False)), - patch("ucode.cli.claude_agent.managed_settings_model_overrides", return_value=None), - patch("ucode.cli.launch_agent"), - ): - result = runner.invoke(app, ["claude", "--model", "main.aarushi.claude-opus-5"]) - assert result.exit_code == 0, result.output - assert "enterprise managed settings" not in _strip_ansi(result.output) - class TestMcpSubcommands: def test_web_search_subcommand_help(self): @@ -1283,6 +1244,31 @@ def test_accepts_configured_codex_launch(self): is True ) + def test_accepts_claude_only_when_managed_settings_are_verified(self, tmp_path): + import ucode.cli as cli_mod + + settings_path = tmp_path / "ucode-settings.json" + settings_path.write_text("{}", encoding="utf-8") + with ( + patch("ucode.cli.managed_agent_config_enabled", return_value=False), + patch("ucode.cli.claude_agent.CLAUDE_SETTINGS_PATH", settings_path), + patch("ucode.cli.claude_agent.managed_settings_are_current", return_value=True), + ): + assert ( + cli_mod._can_launch_from_cached_config("claude", MINIMAL_STATE, **self._kwargs()) + is True + ) + + with ( + patch("ucode.cli.managed_agent_config_enabled", return_value=False), + patch("ucode.cli.claude_agent.CLAUDE_SETTINGS_PATH", settings_path), + patch("ucode.cli.claude_agent.managed_settings_are_current", return_value=False), + ): + assert ( + cli_mod._can_launch_from_cached_config("claude", MINIMAL_STATE, **self._kwargs()) + is False + ) + @pytest.mark.parametrize( "override", [ diff --git a/tests/test_managed_files.py b/tests/test_managed_files.py index dbb14f7b..40f39348 100644 --- a/tests/test_managed_files.py +++ b/tests/test_managed_files.py @@ -1,11 +1,8 @@ -"""Tests for managed_files.py — the isaac-style sudo writer for OS managed settings files. - -Every test mocks the actual privileged step (`_sudo_replace`), so NO real `sudo` / `/etc` write -ever runs. The behavior that matters here is the drift check: an unchanged file must not shell out. -""" +"""Tests for managed settings without real ``sudo`` or ``/etc`` writes.""" from __future__ import annotations +import json import subprocess import pytest @@ -25,6 +22,7 @@ def _reset_dry_run(): def _supported(monkeypatch): # Pin platform support on so tests are deterministic on any host. monkeypatch.setattr(managed_files, "managed_files_supported", lambda: True) + monkeypatch.setattr(managed_files.sys.stdin, "isatty", lambda: True) def _capture_sudo(monkeypatch): @@ -35,6 +33,14 @@ def _capture_sudo(monkeypatch): return calls +@pytest.fixture +def backup_dir(tmp_path, monkeypatch): + path = tmp_path / "managed-backups" + monkeypatch.setattr(managed_files, "MANAGED_BACKUP_DIR", path) + monkeypatch.setattr(managed_files, "MANAGED_BACKUP_MANIFEST_PATH", path / "manifest.json") + return path + + class TestWriteManagedFile: def test_unchanged_content_does_not_sudo(self, tmp_path, monkeypatch): path = tmp_path / "managed.json" @@ -92,7 +98,7 @@ def boom(path, text): class TestClearImmutableStatDenied: - def test_stat_denied_path_returns_false_without_raising(self, monkeypatch): + def test_stat_denied_path_returns_no_flags_without_raising(self, monkeypatch): # Regression: `_clear_immutable` ran an unguarded path.exists() inside the sudo write; under a # root-locked /etc/codex that raised PermissionError and aborted the write ("without root"). class _StatDenied: @@ -103,4 +109,257 @@ def exists(self): monkeypatch.setattr( managed_files.subprocess, "run", lambda *a, **k: pytest.fail("should not shell out") ) - assert managed_files._clear_immutable(_StatDenied()) is False + assert managed_files._clear_immutable(_StatDenied()) == () + + +class TestImmutableFlags: + def test_macos_flags_are_cleared_and_restored(self, tmp_path, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("{}", encoding="utf-8") + calls: list[list[str]] = [] + + def run(command, **kwargs): + calls.append(command) + stdout = "schg,uchg\n" if command[0] == "/usr/bin/stat" else "" + return subprocess.CompletedProcess(command, 0, stdout=stdout, stderr="") + + monkeypatch.setattr(managed_files, "current_os", lambda: managed_files.OS.MACOS) + monkeypatch.setattr(managed_files.subprocess, "run", run) + + flags = managed_files._clear_immutable(path) + managed_files._restore_immutable(path, flags) + + assert flags == ("schg", "uchg") + assert ["/usr/bin/sudo", "chflags", "noschg,nouchg", str(path)] in calls + assert ["/usr/bin/sudo", "chflags", "schg,uchg", str(path)] in calls + + def test_linux_flags_are_cleared_and_restored(self, tmp_path, monkeypatch): + path = tmp_path / "managed.toml" + path.write_text("", encoding="utf-8") + calls: list[list[str]] = [] + + def run(command, **kwargs): + calls.append(command) + stdout = "----ia------- managed.toml\n" if command[1] == "lsattr" else "" + return subprocess.CompletedProcess(command, 0, stdout=stdout, stderr="") + + monkeypatch.setattr(managed_files, "current_os", lambda: managed_files.OS.LINUX) + monkeypatch.setattr(managed_files.subprocess, "run", run) + + flags = managed_files._clear_immutable(path) + managed_files._restore_immutable(path, flags) + + assert flags == ("i", "a") + assert ["/usr/bin/sudo", "chattr", "-ia", str(path)] in calls + assert ["/usr/bin/sudo", "chattr", "+ia", str(path)] in calls + + +class TestManagedFileLifecycle: + def test_reconcile_refuses_symlink_target(self, tmp_path, backup_dir, monkeypatch): + target = tmp_path / "real.json" + target.write_text("{}", encoding="utf-8") + path = tmp_path / "managed.json" + path.symlink_to(target) + monkeypatch.setattr( + managed_files, "_sudo_replace", lambda *args: pytest.fail("must not write") + ) + + with pytest.raises(RuntimeError, match="Refusing to update"): + managed_files.reconcile_managed_file( + path, + '{"ucode": true}\n', + tool="claude", + display="Claude Code", + owned_paths=[["ucode"]], + ) + + def test_reconcile_backs_up_before_write(self, tmp_path, backup_dir, monkeypatch): + path = tmp_path / "managed.json" + path.write_text('{"enterprise": true}\n', encoding="utf-8") + + def replace(target, text): + assert (backup_dir / "claude-managed-settings.backup.json").exists() + target.write_text(text, encoding="utf-8") + + monkeypatch.setattr(managed_files, "_sudo_replace", replace) + result = managed_files.reconcile_managed_file( + path, + '{"enterprise": true, "ucode": true}\n', + tool="claude", + display="Claude Code", + owned_paths=[["ucode"]], + ) + + assert result == "written" + assert (backup_dir / "claude-managed-settings.backup.json").read_text() == ( + '{"enterprise": true}\n' + ) + manifest = json.loads((backup_dir / "manifest.json").read_text()) + assert manifest["files"]["claude"]["original_existed"] is True + + def test_unchanged_file_never_creates_backup(self, tmp_path, backup_dir, monkeypatch): + path = tmp_path / "managed.json" + path.write_text("same", encoding="utf-8") + monkeypatch.setattr( + managed_files, "_sudo_replace", lambda *args: pytest.fail("must not write") + ) + + result = managed_files.reconcile_managed_file( + path, + "same", + tool="claude", + display="Claude Code", + owned_paths=[["env"]], + ) + + assert result == "unchanged" + assert not backup_dir.exists() + + def test_verified_check_uses_fingerprint(self, tmp_path): + path = tmp_path / "managed.json" + path.write_text("current", encoding="utf-8") + state: dict = {} + managed_files.mark_managed_file_verified(state, "claude", path) + + assert managed_files.managed_file_is_verified(state, "claude", path) is True + path.write_text("changed-content", encoding="utf-8") + assert managed_files.managed_file_is_verified(state, "claude", path) is False + + def test_revert_restores_exact_original(self, tmp_path, backup_dir, monkeypatch): + path = tmp_path / "managed.json" + path.write_text('{"enterprise": true}\n', encoding="utf-8") + monkeypatch.setattr( + managed_files, + "_sudo_replace", + lambda target, text: target.write_text(text, encoding="utf-8"), + ) + managed_files.reconcile_managed_file( + path, + '{"enterprise": true, "ucode": true}\n', + tool="claude", + display="Claude Code", + owned_paths=[["ucode"]], + ) + + result = managed_files.revert_managed_file( + "claude", + display="Claude Code", + parser=json.loads, + dumper=lambda doc: json.dumps(doc) + "\n", + ) + + assert result == "restored" + assert path.read_text() == '{"enterprise": true}\n' + assert json.loads((backup_dir / "manifest.json").read_text())["files"] == {} + + def test_revert_removes_file_created_by_ucode(self, tmp_path, backup_dir, monkeypatch): + path = tmp_path / "managed.json" + monkeypatch.setattr( + managed_files, + "_sudo_replace", + lambda target, text: target.write_text(text, encoding="utf-8"), + ) + monkeypatch.setattr(managed_files, "_sudo_remove", lambda target: target.unlink()) + managed_files.reconcile_managed_file( + path, + '{"ucode": true}\n', + tool="claude", + display="Claude Code", + owned_paths=[["ucode"]], + ) + + result = managed_files.revert_managed_file( + "claude", + display="Claude Code", + parser=json.loads, + dumper=lambda doc: json.dumps(doc) + "\n", + ) + + assert result == "removed" + assert not path.exists() + + def test_revert_preserves_external_changes(self, tmp_path, backup_dir, monkeypatch): + path = tmp_path / "managed.json" + path.write_text('{"enterprise": "original"}\n', encoding="utf-8") + monkeypatch.setattr( + managed_files, + "_sudo_replace", + lambda target, text: target.write_text(text, encoding="utf-8"), + ) + managed_files.reconcile_managed_file( + path, + '{"enterprise": "original", "ucode": "gateway"}\n', + tool="claude", + display="Claude Code", + owned_paths=[["ucode"]], + ) + path.write_text( + '{"enterprise": "new-policy", "ucode": "gateway", "new": true}\n', + encoding="utf-8", + ) + + result = managed_files.revert_managed_file( + "claude", + display="Claude Code", + parser=json.loads, + dumper=lambda doc: json.dumps(doc, sort_keys=True) + "\n", + ) + + assert result == "ucode entries removed; external changes preserved" + assert json.loads(path.read_text()) == {"enterprise": "new-policy", "new": True} + + def test_reconcile_retries_exact_mdm_restore_once(self, tmp_path, backup_dir, monkeypatch): + path = tmp_path / "managed.json" + path.write_text('{"enterprise": true}\n', encoding="utf-8") + calls = 0 + + def restore_original(target, text): + nonlocal calls + calls += 1 + target.write_text('{"enterprise": true}\n', encoding="utf-8") + + monkeypatch.setattr(managed_files, "_sudo_replace", restore_original) + + with pytest.raises(RuntimeError, match="immediately restored by device management"): + managed_files.reconcile_managed_file( + path, + '{"enterprise": true, "ucode": true}\n', + tool="claude", + display="Claude Code", + owned_paths=[["ucode"]], + ) + + assert calls == 2 + + def test_reconcile_preserves_concurrent_policy_change(self, tmp_path, backup_dir, monkeypatch): + path = tmp_path / "managed.json" + path.write_text('{"enterprise": "old"}\n', encoding="utf-8") + + def external_update(target, text): + target.write_text('{"enterprise": "new"}\n', encoding="utf-8") + + monkeypatch.setattr(managed_files, "_sudo_replace", external_update) + + with pytest.raises(RuntimeError, match="changed concurrently"): + managed_files.reconcile_managed_file( + path, + '{"enterprise": "old", "ucode": true}\n', + tool="claude", + display="Claude Code", + owned_paths=[["ucode"]], + ) + + assert json.loads(path.read_text()) == {"enterprise": "new"} + + +def test_managed_writes_disabled_without_tty(monkeypatch): + monkeypatch.setattr(managed_files.sys.stdin, "isatty", lambda: False) + + assert managed_files.managed_writes_allowed() is False + + +def test_sudo_command_refuses_noninteractive_execution(monkeypatch): + monkeypatch.setattr(managed_files, "managed_writes_allowed", lambda: False) + + with pytest.raises(RuntimeError, match="Refusing to invoke sudo"): + managed_files._sudo_command("cp", "a", "b") diff --git a/tests/test_managed_resolve.py b/tests/test_managed_resolve.py index 6a621984..3c912678 100644 --- a/tests/test_managed_resolve.py +++ b/tests/test_managed_resolve.py @@ -185,15 +185,13 @@ def test_layers_provider_without_dropping_other_tools(self): class TestGlobalSettings: - def test_only_claude_and_codex_support_global_settings(self): - # This set gates both the write path AND the `ucode setup` machine-wide prompt. Adding an - # agent whose token can't self-refresh here would re-introduce a config that breaks in ~1h. + def test_only_codex_keeps_the_legacy_global_settings_flag(self): from ucode.agents import GLOBAL_SETTINGS_AGENTS - assert GLOBAL_SETTINGS_AGENTS == frozenset({"claude", "codex"}) + assert GLOBAL_SETTINGS_AGENTS == frozenset({"codex"}) - def test_flag_true_for_opted_in_supported_agent(self): - assert managed_use_as_global_settings(MANAGED, "claude") is True + def test_claude_no_longer_honors_the_legacy_flag(self): + assert managed_use_as_global_settings(MANAGED, "claude") is False def test_flag_false_when_not_opted_in(self): # codex is enabled but never marked machine-wide. @@ -204,17 +202,15 @@ def test_flag_ignored_for_unsupported_agent(self): managed = {"enabled_agents": {"gemini": {"use_as_global_settings": True}}} assert managed_use_as_global_settings(managed, "gemini") is False - def test_resolve_sets_transient_write_managed_config(self): + def test_resolve_does_not_set_transient_flag_for_claude(self): resolved = resolve_state(MANAGED, _state(), "claude") - assert resolved["write_managed_config"] is True + assert "write_managed_config" not in resolved def test_resolve_omits_flag_when_not_opted_in(self): resolved = resolve_state(MANAGED, _state(), "codex") assert "write_managed_config" not in resolved - def test_write_managed_config_is_not_persisted(self): - # It lives only for the config-write; save_state (via _without_managed_overlay) drops it so - # a later non-managed launch never writes the managed settings file. + def test_removed_claude_flag_is_not_persisted(self): resolved = resolve_state(MANAGED, _state(), "claude") assert "write_managed_config" not in _without_managed_overlay(resolved) @@ -237,12 +233,13 @@ def real_state_file(self, tmp_path, monkeypatch): monkeypatch.setattr(claude, "CLAUDE_BACKUP_PATH", tmp_path / "backup.json") managed_settings_path = tmp_path / "managed-settings.json" monkeypatch.setattr(claude, "_managed_settings_path", lambda: managed_settings_path) + monkeypatch.setattr(claude, "managed_writes_allowed", lambda: True) - def write_managed_file(path, desired_text, *, display): + def reconcile_managed_file(path, desired_text, **kwargs): path.write_text(desired_text, encoding="utf-8") return "written" - monkeypatch.setattr(claude, "write_managed_file", write_managed_file) + monkeypatch.setattr(claude, "reconcile_managed_file", reconcile_managed_file) # Seed a developer whose own opus choice differs from the manifest's. state_mod.save_state( { diff --git a/tests/test_managed_wizard.py b/tests/test_managed_wizard.py index 8d3513dd..d14d8f20 100644 --- a/tests/test_managed_wizard.py +++ b/tests/test_managed_wizard.py @@ -1842,12 +1842,12 @@ def test_single_model_agent_needs_no_extra_line(self, capsys): assert "models:" not in out def test_scope_label_only_for_global_capable_agents(self, capsys): - # claude/codex can use global settings, so they carry the scope; gemini can't, so it doesn't. + # Codex retains the legacy scope choice; Claude now always installs managed settings. manifest = { - "default_agent": "claude", + "default_agent": "codex", "enabled_agents": { - "claude": { - "model_config": {"default_model": "system.ai.claude-opus-4-8"}, + "codex": { + "model_config": {"default_model": "system.ai.gpt-5"}, "use_as_global_settings": True, }, "gemini": {"model_config": {"default_model": "system.ai.gemini-3-flash"}}, diff --git a/tests/test_tracing.py b/tests/test_tracing.py index 27e58913..50cca72c 100644 --- a/tests/test_tracing.py +++ b/tests/test_tracing.py @@ -17,6 +17,11 @@ SHARED_EXPERIMENT_ID = "111" +@pytest.fixture(autouse=True) +def _avoid_real_managed_settings(monkeypatch): + monkeypatch.setattr(claude, "_managed_settings_path", lambda: None) + + def _enabled_state(profile: str | None = None) -> dict: return { "workspace": WS, From e254e50abaf09aeb7d8b87f09eade8d28db222ef Mon Sep 17 00:00:00 2001 From: Rohit Agrawal Date: Sat, 29 Aug 2026 17:05:52 -0400 Subject: [PATCH 2/6] Simplify interactive configure prompts --- src/ucode/agents/__init__.py | 27 +++---- src/ucode/cli.py | 37 +++++----- src/ucode/managed_files.py | 46 ++++++++++-- tests/test_agents_init.py | 133 ++++++----------------------------- tests/test_cli.py | 102 ++++++++++----------------- tests/test_managed_files.py | 25 +++++++ 6 files changed, 157 insertions(+), 213 deletions(-) diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py index 333e2700..c12b0748 100644 --- a/src/ucode/agents/__init__.py +++ b/src/ucode/agents/__init__.py @@ -24,6 +24,7 @@ map_claude_family_models, resolve_provider_service, ) +from ucode.managed_files import managed_write_batch from ucode.state import get_provider_service, load_state, save_state from ucode.telemetry import agent_version from ucode.ui import ( @@ -68,6 +69,7 @@ DEFAULT_TOOL = "codex" BUNDLE_VERSION = 1 +_MANAGED_SETTINGS_TOOLS = {"claude"} # Codex still honors the legacy managed-config opt-in until its follow-up migration. Claude always # reconciles its OS-managed settings and therefore no longer appears in the setup prompt. @@ -140,16 +142,6 @@ def _required_update_message(tool: str) -> str | None: return checker() -def _confirm_update_installed_tool_binary(tool: str) -> bool: - spec = TOOL_SPECS[tool] - update = _MODULES[tool].is_update_available() - - if not update: - return False - current, latest = update - return prompt_yes_no(f"(Optional) Update {spec['display']} from {current} to {latest}?") - - def _too_new_downgrade(tool: str) -> tuple[str, str] | None: """Return (installed_version, downgrade_target) when the installed tool is too new to work, or None. Agents opt in by defining `too_new_downgrade`.""" @@ -207,9 +199,6 @@ def install_tool_binary( print_warning(required_update) if not _update_installed_tool_binary(tool): raise RuntimeError(_minimum_version_error(tool) or required_update) - elif prompt_optional_updates and _confirm_update_installed_tool_binary(tool): - _update_installed_tool_binary(tool) - version_error = _minimum_version_error(tool) if version_error: raise RuntimeError(version_error) @@ -425,7 +414,8 @@ def configure_single_tool(tool: str, state: dict) -> dict: raise RuntimeError( f"{TOOL_SPECS[tool]['display']} is not available on this workspace.{detail}" ) - state = _configure_one(tool, state, provider) + with managed_write_batch(_managed_settings_displays([tool])): + state = _configure_one(tool, state, provider) available_tools = list(set((state.get("available_tools") or []) + [tool])) state["available_tools"] = available_tools save_state(state) @@ -455,8 +445,9 @@ def configure_selected_tools(state: dict, tools: list[str]) -> dict: replacing it, so a previously-configured tool the user didn't pick this run is preserved. """ - for tool in tools: - state = _configure_one(tool, state, get_provider_service(state, tool)) + with managed_write_batch(_managed_settings_displays(tools)): + for tool in tools: + state = _configure_one(tool, state, get_provider_service(state, tool)) existing = state.get("available_tools") or [] state["available_tools"] = sorted(set(existing) | set(tools)) @@ -465,6 +456,10 @@ def configure_selected_tools(state: dict, tools: list[str]) -> dict: return state +def _managed_settings_displays(tools: list[str]) -> list[str]: + return [TOOL_SPECS[tool]["display"] for tool in tools if tool in _MANAGED_SETTINGS_TOOLS] + + def configure_all_tools(state: dict) -> dict: """Discover available tools on the workspace and configure all of them. diff --git a/src/ucode/cli.py b/src/ucode/cli.py index 8e4046ca..66b987f4 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -133,7 +133,6 @@ prompt_for_tools, prompt_for_workspace, prompt_yes_no, - prompt_yes_no_default, set_verbosity, spinner, status_badge, @@ -950,16 +949,6 @@ def configure_workspace_command( for tool_name in picked: state = _maybe_select_provider_service(tool_name, state) - # Last question in the interactive flow: opt out of AI Tools. When a flag - # already decided it, configure_shared_state persisted that; skip the prompt. - # The default is the resolved prior choice, so Enter won't undo a past opt-out. - if databricks_ai_tools_enabled is None and offer_provider: - state["databricks_ai_tools_enabled"] = prompt_yes_no_default( - "Install Databricks AI Tools for your coding agents? " - "This adds Databricks skills and plugins.", - default=state.get("databricks_ai_tools_enabled", True), - ) - state = configure_selected_tools(state, picked) summary_lines = [f"[bold]Workspace:[/bold] [cyan]{state['workspace']}[/cyan]"] @@ -1200,6 +1189,17 @@ def _configure_agents_for_mcp( return configured +def _configure_optional_setup() -> None: + if not prompt_yes_no("Configure MCP servers, skills, and plugins?"): + return + + state = load_state() + state["databricks_ai_tools_enabled"] = True + save_state(state) + install_databricks_ai_tools_for_agents(state.get("available_tools") or [], state) + configure_mcp_command() + + @mcp_app.command("add") def mcp_add( location: Annotated[ @@ -2701,6 +2701,7 @@ def configure( # Set True only in the fully-interactive branch below; gates the optional # MCP setup prompt so flag-driven / scripted runs are never interrupted. fully_interactive = False + combined_optional_setup = False if agent is not None: tool = normalize_tool(agent) install_tool_binary( @@ -2773,6 +2774,12 @@ def configure( else: # Tool binaries are installed after the user picks which agents # they want, in configure_workspace_command. + combined_optional_setup = ( + not flag_driven_workspace and enable_databricks_ai_tools is None + ) + if combined_optional_setup: + # Defer AI Tools until the combined optional setup prompt below. + skip_kwargs["databricks_ai_tools_enabled"] = False if workspace_entries is None: configure_workspace_command( prompt_optional_updates=prompt_optional_updates, @@ -2820,11 +2827,9 @@ def configure( "interactive picker." ) configure_mcp_command(services=services) - # Offer MCP setup as the natural next step of interactive configuration, - # so users discover it without needing to know `configure mcp` exists. - # Skipped in dry-run and non-interactive/flag-driven runs (which stay - # scriptable), and when --dry-run is set. - if fully_interactive and not dry_run and prompt_yes_no("Configure MCP servers now?"): + if combined_optional_setup and not dry_run: + _configure_optional_setup() + elif fully_interactive and not dry_run and prompt_yes_no("Configure MCP servers now?"): configure_mcp_command() except typer.Exit: # `typer.Exit` subclasses RuntimeError, so it has to be re-raised ahead of the handler diff --git a/src/ucode/managed_files.py b/src/ucode/managed_files.py index aa2c9157..1d78e5e1 100644 --- a/src/ucode/managed_files.py +++ b/src/ucode/managed_files.py @@ -14,7 +14,8 @@ import subprocess import sys import tempfile -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextlib import contextmanager from copy import deepcopy from enum import Enum from pathlib import Path @@ -29,6 +30,8 @@ MANAGED_BACKUP_MANIFEST_PATH = MANAGED_BACKUP_DIR / "manifest.json" MANAGED_FINGERPRINT_VERSION = 1 _MISSING = object() +_managed_write_batch: tuple[str, ...] = () +_managed_write_notice_shown = False ManagedParser = Callable[[str], dict] ManagedDumper = Callable[[dict], str] @@ -138,6 +141,38 @@ def managed_writes_allowed() -> bool: return sys.stdin.isatty() +@contextmanager +def managed_write_batch(displays: list[str]) -> Iterator[None]: + """Group setup messaging for agents configured in one command.""" + global _managed_write_batch, _managed_write_notice_shown + + previous_batch = _managed_write_batch + previous_notice = _managed_write_notice_shown + _managed_write_batch = tuple(dict.fromkeys(displays)) + _managed_write_notice_shown = False + try: + yield + if _managed_write_notice_shown: + print_success(f"Settings configured for {' and '.join(_managed_write_batch)}") + finally: + _managed_write_batch = previous_batch + _managed_write_notice_shown = previous_notice + + +def _print_managed_write_permission(display: str) -> None: + global _managed_write_notice_shown + + if not _managed_write_batch: + print_note(f"Enter password to configure settings for {display}.") + return + if _managed_write_notice_shown: + return + + displays = " and ".join(_managed_write_batch) + print_note(f"Enter password to configure settings for {displays}.") + _managed_write_notice_shown = True + + def managed_file_conflicts( existing: dict, desired: dict, owned_paths: list[list[str]] ) -> list[str]: @@ -232,10 +267,8 @@ def reconcile_managed_file( return "written" created = current_text is None - backup_created = _ensure_backup(tool, path, current_text) - if backup_created: - print_note(f"{display}: original managed settings backed up under {MANAGED_BACKUP_DIR}.") - print_note(f"{display}: administrator permission is required to update {path}.") + _ensure_backup(tool, path, current_text) + _print_managed_write_permission(display) if read_managed_file(path) != current_text: raise RuntimeError( f"{display} managed settings changed while ucode was preparing the update. " @@ -271,7 +304,8 @@ def reconcile_managed_file( "the newer policy; run the command again or contact your administrator." ) _record_last_applied(tool, path, desired_text, owned_paths) - print_success(f"{display} managed settings {'created' if created else 'updated'} and verified") + if not _managed_write_batch: + print_success(f"Settings configured for {display}") return "created" if created else "written" diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py index de40affb..a2519504 100644 --- a/tests/test_agents_init.py +++ b/tests/test_agents_init.py @@ -3,6 +3,7 @@ from __future__ import annotations import subprocess +from contextlib import contextmanager import pytest @@ -360,7 +361,7 @@ def fake_run(*args, **kwargs): assert install_tool_binary("opencode", strict=False) is False - def test_updates_existing_binary_when_requested(self, monkeypatch, capsys): + def test_existing_binary_does_not_prompt_for_optional_update(self, monkeypatch, capsys): calls: list[list[str]] = [] def fake_which(binary: str) -> str | None: @@ -372,106 +373,19 @@ def fake_run(args, **kwargs): monkeypatch.setattr("ucode.agents.shutil.which", fake_which) monkeypatch.setattr("ucode.agents.subprocess.run", fake_run) - monkeypatch.setattr("ucode.agents._confirm_update_installed_tool_binary", lambda _: True) - - assert install_tool_binary("opencode", strict=False, update_existing=True) is True - assert calls == [["npm", "install", "-g", "opencode-ai"]] - output = capsys.readouterr().out - assert "Updating OpenCode..." in output - assert "OpenCode is up to date" in output - - def test_skips_existing_binary_update_when_latest_is_not_newer(self, monkeypatch, capsys): - calls: list[list[str]] = [] - prompt_calls: list[str] = [] - - def fake_which(binary: str) -> str | None: - return f"/usr/bin/{binary}" - - def fake_run(args, **kwargs): - calls.append(args) - return subprocess.CompletedProcess(args, 0) - - monkeypatch.setattr("ucode.agents.shutil.which", fake_which) - monkeypatch.setattr("ucode.agents.subprocess.run", fake_run) - monkeypatch.setattr("ucode.agents.opencode.is_update_available", lambda: None) monkeypatch.setattr( - "ucode.agents.prompt_yes_no", lambda prompt: prompt_calls.append(prompt) or True + "ucode.agents.opencode.is_update_available", + lambda: (_ for _ in ()).throw(AssertionError("should not check for optional updates")), ) - - assert install_tool_binary("opencode", strict=False, update_existing=True) is True - assert calls == [] - assert prompt_calls == [] - assert "Updating OpenCode..." not in capsys.readouterr().out - - def test_prompts_and_updates_existing_binary_when_newer_version_exists( - self, monkeypatch, capsys - ): - calls: list[list[str]] = [] - prompt_calls: list[str] = [] - - def fake_which(binary: str) -> str | None: - return f"/usr/bin/{binary}" - - def fake_run(args, **kwargs): - calls.append(args) - return subprocess.CompletedProcess(args, 0) - - monkeypatch.setattr("ucode.agents.shutil.which", fake_which) - monkeypatch.setattr("ucode.agents.subprocess.run", fake_run) - monkeypatch.setattr("ucode.agents.opencode.is_update_available", lambda: ("1.2.3", "1.2.4")) monkeypatch.setattr( - "ucode.agents.prompt_yes_no", lambda prompt: prompt_calls.append(prompt) or True + "ucode.agents.prompt_yes_no", + lambda prompt: (_ for _ in ()).throw(AssertionError("should not prompt")), ) - assert install_tool_binary("opencode", strict=False, update_existing=True) is True - assert prompt_calls == ["(Optional) Update OpenCode from 1.2.3 to 1.2.4?"] - assert calls == [["npm", "install", "-g", "opencode-ai"]] - assert "Updating OpenCode..." in capsys.readouterr().out - - def test_skips_existing_binary_update_when_user_declines(self, monkeypatch, capsys): - calls: list[list[str]] = [] - - def fake_which(binary: str) -> str | None: - return f"/usr/bin/{binary}" - - def fake_run(args, **kwargs): - calls.append(args) - return subprocess.CompletedProcess(args, 0) - - monkeypatch.setattr("ucode.agents.shutil.which", fake_which) - monkeypatch.setattr("ucode.agents.subprocess.run", fake_run) - monkeypatch.setattr("ucode.agents._confirm_update_installed_tool_binary", lambda _: False) - assert install_tool_binary("opencode", strict=False, update_existing=True) is True assert calls == [] assert "Updating OpenCode..." not in capsys.readouterr().out - def test_optional_update_prompt_suppressed_when_disabled(self, monkeypatch): - """prompt_optional_updates=False must skip the optional update check - entirely — the confirm prompt should never be reached.""" - - def fake_which(binary: str) -> str | None: - return f"/usr/bin/{binary}" - - monkeypatch.setattr("ucode.agents.shutil.which", fake_which) - monkeypatch.setattr("ucode.agents._minimum_version_error", lambda _: None) - monkeypatch.setattr("ucode.agents._required_update_message", lambda _: None) - - def boom(_tool: str) -> bool: - raise AssertionError("optional update prompt should not be reached") - - monkeypatch.setattr("ucode.agents._confirm_update_installed_tool_binary", boom) - - assert ( - install_tool_binary( - "opencode", - strict=False, - update_existing=True, - prompt_optional_updates=False, - ) - is True - ) - def test_required_update_runs_even_when_optional_prompt_disabled(self, monkeypatch): """A required (minimum-version) update is forced regardless of the prompt_optional_updates preference.""" @@ -516,11 +430,6 @@ def fake_run(args, **kwargs): monkeypatch.setattr("ucode.agents.shutil.which", fake_which) monkeypatch.setattr("ucode.agents.subprocess.run", fake_run) monkeypatch.setattr("ucode.agents.gemini.too_new_downgrade", lambda: ("0.45.0", "0.44.1")) - # The optional-update path must never be reached for a too-new tool. - monkeypatch.setattr( - "ucode.agents._confirm_update_installed_tool_binary", - lambda _: (_ for _ in ()).throw(AssertionError("should not reach optional update")), - ) monkeypatch.setattr( "ucode.agents.prompt_yes_no", lambda prompt: prompt_calls.append(prompt) or True ) @@ -581,19 +490,6 @@ def fake_run(args, **kwargs): assert calls == [] assert "newer than the latest version known to work" in capsys.readouterr().out - def test_update_failure_keeps_existing_binary_available(self, monkeypatch): - def fake_which(binary: str) -> str | None: - return f"/usr/bin/{binary}" - - def fake_run(*args, **kwargs): - raise subprocess.CalledProcessError(1, args[0]) - - monkeypatch.setattr("ucode.agents.shutil.which", fake_which) - monkeypatch.setattr("ucode.agents.subprocess.run", fake_run) - monkeypatch.setattr("ucode.agents._confirm_update_installed_tool_binary", lambda _: True) - - assert install_tool_binary("opencode", strict=True, update_existing=True) is True - def test_ensure_tool_binary_available_raises_when_missing(self, monkeypatch): monkeypatch.setattr("ucode.agents.shutil.which", lambda _: None) @@ -602,6 +498,23 @@ def test_ensure_tool_binary_available_raises_when_missing(self, monkeypatch): class TestConfigureSelectedTools: + def test_groups_managed_permission_notice(self, monkeypatch): + batches: list[list[str]] = [] + + @contextmanager + def capture_batch(displays): + batches.append(displays) + yield + + monkeypatch.setattr(agents_mod, "managed_write_batch", capture_batch) + monkeypatch.setattr(agents_mod, "_configure_one", lambda tool, state, provider: state) + monkeypatch.setattr(agents_mod, "save_state", lambda state: None) + monkeypatch.setattr(agents_mod, "install_databricks_ai_tools_for_agents", lambda *_: None) + + configure_selected_tools({}, ["codex", "claude"]) + + assert batches == [["Claude Code"]] + def test_merges_with_existing_available_tools(self, monkeypatch): """Configuring a new tool should not drop previously-configured tools from state['available_tools'].""" diff --git a/tests/test_cli.py b/tests/test_cli.py index 1897e963..6a73ccc6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1345,43 +1345,58 @@ def test_no_flag_calls_configure_all(self): patch("ucode.cli.install_databricks_cli"), patch("ucode.cli.install_tool_binary"), patch("ucode.cli.configure_workspace_command") as mock_cfg, - # Fully-interactive configure ends by offering the MCP step; decline it. - patch("ucode.cli.prompt_yes_no", return_value=False) as mock_mcp_prompt, - patch("ucode.cli.configure_mcp_command") as mock_mcp, + patch("ucode.cli._configure_optional_setup") as mock_optional_setup, ): - # No flag: the AI Tools prompt happens later, inside - # configure_workspace_command, so nothing is forwarded here. result = runner.invoke(app, ["configure"]) assert result.exit_code == 0, result.output - mock_cfg.assert_called_once_with(prompt_optional_updates=True) - mock_mcp_prompt.assert_called_once() - mock_mcp.assert_not_called() + mock_cfg.assert_called_once_with( + prompt_optional_updates=True, databricks_ai_tools_enabled=False + ) + mock_optional_setup.assert_called_once_with() - def test_interactive_accepting_mcp_prompt_runs_mcp_config(self): + def test_optional_setup_installs_ai_tools_and_configures_mcp(self): + import ucode.cli as cli_mod + + state = {"available_tools": ["claude", "codex"]} with ( - patch("ucode.cli.install_databricks_cli"), - patch("ucode.cli.install_tool_binary"), - patch("ucode.cli.configure_workspace_command"), - patch("ucode.cli.prompt_yes_no", return_value=True), + patch("ucode.cli.prompt_yes_no", return_value=True) as mock_prompt, + patch("ucode.cli.load_state", return_value=state), + patch("ucode.cli.save_state") as mock_save, + patch("ucode.cli.install_databricks_ai_tools_for_agents") as mock_install, patch("ucode.cli.configure_mcp_command") as mock_mcp, ): - result = runner.invoke(app, ["configure"]) - assert result.exit_code == 0, result.output + cli_mod._configure_optional_setup() + + mock_prompt.assert_called_once_with("Configure MCP servers, skills, and plugins?") + assert state["databricks_ai_tools_enabled"] is True + mock_save.assert_called_once_with(state) + mock_install.assert_called_once_with(["claude", "codex"], state) mock_mcp.assert_called_once_with() + def test_optional_setup_decline_does_nothing(self): + import ucode.cli as cli_mod + + with ( + patch("ucode.cli.prompt_yes_no", return_value=False), + patch("ucode.cli.load_state") as mock_load, + patch("ucode.cli.configure_mcp_command") as mock_mcp, + ): + cli_mod._configure_optional_setup() + + mock_load.assert_not_called() + mock_mcp.assert_not_called() + def test_agents_flag_skips_mcp_prompt(self): # Flag-driven (non-interactive) runs must stay scriptable: no MCP prompt. with ( patch("ucode.cli.install_databricks_cli"), patch("ucode.cli.install_tool_binary"), patch("ucode.cli.configure_workspace_command"), - patch("ucode.cli.prompt_yes_no") as mock_prompt, - patch("ucode.cli.configure_mcp_command") as mock_mcp, + patch("ucode.cli._configure_optional_setup") as mock_optional_setup, ): result = runner.invoke(app, ["configure", "--agents", "claude,codex"]) assert result.exit_code == 0, result.output - mock_prompt.assert_not_called() - mock_mcp.assert_not_called() + mock_optional_setup.assert_not_called() def test_agents_flag_calls_configure_with_tools(self): with ( @@ -1534,7 +1549,9 @@ def test_skip_upgrade_flag_disables_optional_update_prompt(self): ): result = runner.invoke(app, ["configure", "--skip-upgrade"]) assert result.exit_code == 0, result.output - mock_cfg.assert_called_once_with(prompt_optional_updates=False) + mock_cfg.assert_called_once_with( + prompt_optional_updates=False, databricks_ai_tools_enabled=False + ) def test_disable_databricks_ai_tools_forwards_false_and_skips_prompt(self): # An explicit flag suppresses the interactive prompt and forwards the choice. @@ -1570,51 +1587,6 @@ def test_enable_databricks_ai_tools_with_agents_forwards_true(self): databricks_ai_tools_enabled=True, ) - def _stub_interactive_configure(self, monkeypatch, shared_state): - """Wire configure_workspace_command's interactive path; return captured info.""" - import ucode.cli as cli_mod - - monkeypatch.setattr(cli_mod, "configure_shared_state", lambda *a, **k: shared_state) - monkeypatch.setattr(cli_mod, "check_gateway_endpoint", lambda s, t: t == "claude") - monkeypatch.setattr(cli_mod, "install_tool_binary", lambda *a, **k: True) - monkeypatch.setattr(cli_mod, "_maybe_select_provider_service", lambda tool, s: s) - monkeypatch.setattr(cli_mod, "validate_all_tools", lambda s: None) - monkeypatch.setattr(cli_mod, "validate_tool", lambda t: (True, None)) - monkeypatch.setattr( - cli_mod, "_prompt_for_configuration", lambda tool=None: ("https://w", None) - ) - monkeypatch.setattr(cli_mod, "prompt_for_tools", lambda options: ["claude"]) - captured = {} - monkeypatch.setattr( - cli_mod, - "configure_selected_tools", - lambda s, tools: captured.update(state=dict(s)) or s, - ) - prompt_calls = [] - monkeypatch.setattr( - cli_mod, - "prompt_yes_no_default", - lambda msg, *, default: prompt_calls.append(default) or default, - ) - return cli_mod, captured, prompt_calls - - def test_interactive_prompt_default_yes_when_no_prior_optout(self, monkeypatch): - # No prior opt-out -> prompt defaults to yes; state carries True into install. - state = {**MINIMAL_STATE, "available_tools": [], "databricks_ai_tools_enabled": True} - cli_mod, captured, prompt_calls = self._stub_interactive_configure(monkeypatch, state) - cli_mod.configure_workspace_command() - assert prompt_calls == [True] # default derived from resolved prior choice - assert captured["state"]["databricks_ai_tools_enabled"] is True - - def test_interactive_prompt_defaults_to_prior_optout(self, monkeypatch): - # configure_shared_state resolved a prior --disable to False; the prompt must - # default to no so Enter doesn't silently re-enable it. - state = {**MINIMAL_STATE, "available_tools": [], "databricks_ai_tools_enabled": False} - cli_mod, captured, prompt_calls = self._stub_interactive_configure(monkeypatch, state) - cli_mod.configure_workspace_command() - assert prompt_calls == [False] - assert captured["state"]["databricks_ai_tools_enabled"] is False - def test_skip_upgrade_flag_with_agent_skips_optional_update(self): with ( patch("ucode.cli.install_databricks_cli"), diff --git a/tests/test_managed_files.py b/tests/test_managed_files.py index 40f39348..ba4b6b5e 100644 --- a/tests/test_managed_files.py +++ b/tests/test_managed_files.py @@ -197,6 +197,31 @@ def replace(target, text): manifest = json.loads((backup_dir / "manifest.json").read_text()) assert manifest["files"]["claude"]["original_existed"] is True + def test_batch_messages_name_all_agents_once(self, tmp_path, backup_dir, monkeypatch): + notes: list[str] = [] + successes: list[str] = [] + + monkeypatch.setattr(managed_files, "print_note", notes.append) + monkeypatch.setattr(managed_files, "print_success", successes.append) + monkeypatch.setattr( + managed_files, + "_sudo_replace", + lambda target, text: target.write_text(text, encoding="utf-8"), + ) + + with managed_files.managed_write_batch(["Codex", "Claude Code"]): + for tool in ("codex", "claude"): + managed_files.reconcile_managed_file( + tmp_path / f"{tool}.json", + '{"ucode": true}\n', + tool=tool, + display=tool.title(), + owned_paths=[["ucode"]], + ) + + assert notes == ["Enter password to configure settings for Codex and Claude Code."] + assert successes == ["Settings configured for Codex and Claude Code"] + def test_unchanged_file_never_creates_backup(self, tmp_path, backup_dir, monkeypatch): path = tmp_path / "managed.json" path.write_text("same", encoding="utf-8") From d41774335cb520f566f48b4012c7f0990edf5d4b Mon Sep 17 00:00:00 2001 From: Rohit Agrawal Date: Sat, 29 Aug 2026 17:08:19 -0400 Subject: [PATCH 3/6] Remove obsolete configure prompt fixture --- tests/test_cli.py | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 6a73ccc6..cd47de21 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -52,15 +52,6 @@ def no_state_writes(): yield -@pytest.fixture(autouse=True) -def no_blocking_ai_tools_prompt(): - """The interactive configure flow prompts for AI Tools; default it to yes so - tests that drive that path don't block reading stdin. Tests that assert on the - prompt override this with their own patch.""" - with patch("ucode.cli.prompt_yes_no_default", lambda msg, *, default: default): - yield - - MINIMAL_STATE = { "workspace": "https://example.databricks.com", "base_urls": { @@ -1559,14 +1550,12 @@ def test_disable_databricks_ai_tools_forwards_false_and_skips_prompt(self): patch("ucode.cli.install_databricks_cli"), patch("ucode.cli.install_tool_binary"), patch("ucode.cli.configure_workspace_command") as mock_cfg, - patch("ucode.cli.prompt_yes_no_default") as mock_prompt, # Fully-interactive configure ends by offering the MCP step; decline it. patch("ucode.cli.prompt_yes_no", return_value=False), patch("ucode.cli.configure_mcp_command"), ): result = runner.invoke(app, ["configure", "--disable-databricks-ai-tools"]) assert result.exit_code == 0, result.output - mock_prompt.assert_not_called() mock_cfg.assert_called_once_with( prompt_optional_updates=True, databricks_ai_tools_enabled=False ) From bd460975a9cdbfadb11d99aa6fea3fd53f11e8ce Mon Sep 17 00:00:00 2001 From: Rohit Agrawal Date: Sat, 29 Aug 2026 17:10:32 -0400 Subject: [PATCH 4/6] Defer optional tools to final setup step --- src/ucode/agents/__init__.py | 7 ++++-- src/ucode/cli.py | 42 +++++++++++++++++++++--------------- tests/test_agents_init.py | 7 ++++++ tests/test_cli.py | 25 +++++++++------------ 4 files changed, 47 insertions(+), 34 deletions(-) diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py index c12b0748..d50106e4 100644 --- a/src/ucode/agents/__init__.py +++ b/src/ucode/agents/__init__.py @@ -437,7 +437,9 @@ def _configure_one(tool: str, state: dict, provider: str | None) -> dict: return configure_tool(tool, state, model) -def configure_selected_tools(state: dict, tools: list[str]) -> dict: +def configure_selected_tools( + state: dict, tools: list[str], *, install_ai_tools: bool = True +) -> dict: """Configure the given tools. Caller is responsible for ensuring each tool is available on the workspace. @@ -452,7 +454,8 @@ def configure_selected_tools(state: dict, tools: list[str]) -> dict: existing = state.get("available_tools") or [] state["available_tools"] = sorted(set(existing) | set(tools)) save_state(state) - install_databricks_ai_tools_for_agents(tools, state) + if install_ai_tools: + install_databricks_ai_tools_for_agents(tools, state) return state diff --git a/src/ucode/cli.py b/src/ucode/cli.py index 66b987f4..74ddaa57 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -838,6 +838,7 @@ def configure_workspace_command( skip_unavailable: bool = False, fable_enabled: bool | None = None, databricks_ai_tools_enabled: bool | None = None, + offer_optional_setup: bool = False, ) -> int: if tool is not None and selected_tools is not None: raise RuntimeError("Use either --agent or --agents, not both.") @@ -949,7 +950,10 @@ def configure_workspace_command( for tool_name in picked: state = _maybe_select_provider_service(tool_name, state) - state = configure_selected_tools(state, picked) + if offer_optional_setup: + state = configure_selected_tools(state, picked, install_ai_tools=False) + else: + state = configure_selected_tools(state, picked) summary_lines = [f"[bold]Workspace:[/bold] [cyan]{state['workspace']}[/cyan]"] for tool_name in picked: @@ -969,11 +973,13 @@ def configure_workspace_command( if skip_validate: print_note("Skipping agent validation (--skip-validate).") - return 0 - # Limit validation to just-configured tools so we don't re-validate - # previously-configured tools the user didn't touch this run. - validate_state = {**state, "available_tools": picked} - validate_all_tools(validate_state) + else: + # Limit validation to just-configured tools so we don't re-validate + # previously-configured tools the user didn't touch this run. + validate_state = {**state, "available_tools": picked} + validate_all_tools(validate_state) + if offer_optional_setup and not is_dry_run(): + _configure_optional_setup(state, picked) return 0 @@ -1189,14 +1195,14 @@ def _configure_agents_for_mcp( return configured -def _configure_optional_setup() -> None: - if not prompt_yes_no("Configure MCP servers, skills, and plugins?"): +def _configure_optional_setup(state: dict, tools: list[str]) -> None: + enabled = prompt_yes_no("Configure MCP servers, skills, and plugins?") + state["databricks_ai_tools_enabled"] = enabled + save_state(state) + if not enabled: return - state = load_state() - state["databricks_ai_tools_enabled"] = True - save_state(state) - install_databricks_ai_tools_for_agents(state.get("available_tools") or [], state) + install_databricks_ai_tools_for_agents(tools, state) configure_mcp_command() @@ -2778,8 +2784,7 @@ def configure( not flag_driven_workspace and enable_databricks_ai_tools is None ) if combined_optional_setup: - # Defer AI Tools until the combined optional setup prompt below. - skip_kwargs["databricks_ai_tools_enabled"] = False + skip_kwargs["offer_optional_setup"] = True if workspace_entries is None: configure_workspace_command( prompt_optional_updates=prompt_optional_updates, @@ -2827,9 +2832,12 @@ def configure( "interactive picker." ) configure_mcp_command(services=services) - if combined_optional_setup and not dry_run: - _configure_optional_setup() - elif fully_interactive and not dry_run and prompt_yes_no("Configure MCP servers now?"): + if ( + fully_interactive + and not combined_optional_setup + and not dry_run + and prompt_yes_no("Configure MCP servers now?") + ): configure_mcp_command() except typer.Exit: # `typer.Exit` subclasses RuntimeError, so it has to be re-raised ahead of the handler diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py index a2519504..eb061735 100644 --- a/tests/test_agents_init.py +++ b/tests/test_agents_init.py @@ -123,6 +123,13 @@ def test_configure_selected_tools_triggers_install(self, monkeypatch): agents_mod.configure_selected_tools({"profile": "myprof"}, ["codex"]) assert captured == {"agents": ["codex"], "profile": "myprof"} + def test_configure_selected_tools_can_defer_install(self, monkeypatch): + captured = self._stub_configure(monkeypatch) + agents_mod.configure_selected_tools( + {"profile": "myprof"}, ["codex"], install_ai_tools=False + ) + assert captured == {} + class TestNormalizeTool: @pytest.mark.parametrize( diff --git a/tests/test_cli.py b/tests/test_cli.py index cd47de21..e92f80d4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1336,14 +1336,10 @@ def test_no_flag_calls_configure_all(self): patch("ucode.cli.install_databricks_cli"), patch("ucode.cli.install_tool_binary"), patch("ucode.cli.configure_workspace_command") as mock_cfg, - patch("ucode.cli._configure_optional_setup") as mock_optional_setup, ): result = runner.invoke(app, ["configure"]) assert result.exit_code == 0, result.output - mock_cfg.assert_called_once_with( - prompt_optional_updates=True, databricks_ai_tools_enabled=False - ) - mock_optional_setup.assert_called_once_with() + mock_cfg.assert_called_once_with(prompt_optional_updates=True, offer_optional_setup=True) def test_optional_setup_installs_ai_tools_and_configures_mcp(self): import ucode.cli as cli_mod @@ -1351,12 +1347,11 @@ def test_optional_setup_installs_ai_tools_and_configures_mcp(self): state = {"available_tools": ["claude", "codex"]} with ( patch("ucode.cli.prompt_yes_no", return_value=True) as mock_prompt, - patch("ucode.cli.load_state", return_value=state), patch("ucode.cli.save_state") as mock_save, patch("ucode.cli.install_databricks_ai_tools_for_agents") as mock_install, patch("ucode.cli.configure_mcp_command") as mock_mcp, ): - cli_mod._configure_optional_setup() + cli_mod._configure_optional_setup(state, ["claude", "codex"]) mock_prompt.assert_called_once_with("Configure MCP servers, skills, and plugins?") assert state["databricks_ai_tools_enabled"] is True @@ -1369,12 +1364,16 @@ def test_optional_setup_decline_does_nothing(self): with ( patch("ucode.cli.prompt_yes_no", return_value=False), - patch("ucode.cli.load_state") as mock_load, + patch("ucode.cli.save_state") as mock_save, + patch("ucode.cli.install_databricks_ai_tools_for_agents") as mock_install, patch("ucode.cli.configure_mcp_command") as mock_mcp, ): - cli_mod._configure_optional_setup() + state = {} + cli_mod._configure_optional_setup(state, ["claude"]) - mock_load.assert_not_called() + assert state["databricks_ai_tools_enabled"] is False + mock_save.assert_called_once_with(state) + mock_install.assert_not_called() mock_mcp.assert_not_called() def test_agents_flag_skips_mcp_prompt(self): @@ -1383,11 +1382,9 @@ def test_agents_flag_skips_mcp_prompt(self): patch("ucode.cli.install_databricks_cli"), patch("ucode.cli.install_tool_binary"), patch("ucode.cli.configure_workspace_command"), - patch("ucode.cli._configure_optional_setup") as mock_optional_setup, ): result = runner.invoke(app, ["configure", "--agents", "claude,codex"]) assert result.exit_code == 0, result.output - mock_optional_setup.assert_not_called() def test_agents_flag_calls_configure_with_tools(self): with ( @@ -1540,9 +1537,7 @@ def test_skip_upgrade_flag_disables_optional_update_prompt(self): ): result = runner.invoke(app, ["configure", "--skip-upgrade"]) assert result.exit_code == 0, result.output - mock_cfg.assert_called_once_with( - prompt_optional_updates=False, databricks_ai_tools_enabled=False - ) + mock_cfg.assert_called_once_with(prompt_optional_updates=False, offer_optional_setup=True) def test_disable_databricks_ai_tools_forwards_false_and_skips_prompt(self): # An explicit flag suppresses the interactive prompt and forwards the choice. From 422b1af7ff3f90279d976460475a7f24a8430170 Mon Sep 17 00:00:00 2001 From: Rohit Agrawal Date: Sat, 29 Aug 2026 17:16:27 -0400 Subject: [PATCH 5/6] Remove optional agent update checks --- src/ucode/agent_updates.py | 31 ---------------------- src/ucode/agents/claude.py | 5 ---- src/ucode/agents/codex.py | 5 ---- src/ucode/agents/copilot.py | 5 ---- src/ucode/agents/gemini.py | 20 -------------- src/ucode/agents/opencode.py | 5 ---- src/ucode/agents/pi.py | 5 ---- tests/test_agent_gemini.py | 18 ------------- tests/test_agent_updates.py | 51 +----------------------------------- tests/test_agents_init.py | 9 ------- 10 files changed, 1 insertion(+), 153 deletions(-) diff --git a/src/ucode/agent_updates.py b/src/ucode/agent_updates.py index 175c19dd..fe77184d 100644 --- a/src/ucode/agent_updates.py +++ b/src/ucode/agent_updates.py @@ -69,34 +69,3 @@ def latest_version_below(package: str, ceiling: tuple[int, int, int]) -> str | N # npm returns versions in ascending order, so the last entry is newest. return pool[-1] - -def available_npm_package_update(package: str) -> tuple[str, str] | None: - if not shutil.which("npm"): - return None - try: - result = subprocess.run( - ["npm", "outdated", "-g", "--json", package], - capture_output=True, - text=True, - timeout=10, - check=False, - ) - except (FileNotFoundError, subprocess.TimeoutExpired, OSError): - return None - - # npm outdated exits 1 when it finds outdated packages. - if result.returncode not in (0, 1) or not result.stdout.strip(): - return None - try: - outdated = json.loads(result.stdout) - except json.JSONDecodeError: - return None - - package_update = outdated.get(package) - if not isinstance(package_update, dict): - return None - current = package_update.get("current") - latest = package_update.get("latest") - if not isinstance(current, str) or not isinstance(latest, str): - return None - return current, latest diff --git a/src/ucode/agents/claude.py b/src/ucode/agents/claude.py index 057d262a..6645a2c2 100644 --- a/src/ucode/agents/claude.py +++ b/src/ucode/agents/claude.py @@ -15,7 +15,6 @@ from pathlib import Path from typing import cast -from ucode.agent_updates import available_npm_package_update from ucode.anthropic_model_discovery_proxy import ( start_proxy as start_anthropic_model_discovery_proxy, ) @@ -105,10 +104,6 @@ ) -def is_update_available() -> tuple[str, str] | None: - return available_npm_package_update(SPEC["package"]) - - def _resolve_web_search_model(state: dict) -> str | None: """Pick the model the web_search MCP server should call. Prefers an explicit override in state, otherwise the first endpoint discovered as diff --git a/src/ucode/agents/codex.py b/src/ucode/agents/codex.py index 40568a6e..db256425 100644 --- a/src/ucode/agents/codex.py +++ b/src/ucode/agents/codex.py @@ -11,7 +11,6 @@ import tomlkit -from ucode.agent_updates import available_npm_package_update from ucode.config_io import ( APP_DIR, ToolSpec, @@ -76,10 +75,6 @@ _GPT_RE = re.compile(r"(?:databricks-)?gpt-(\d+)(?:[.-](\d+))?(?:[.-](\d+))?(-.+|[a-z].*)?") -def is_update_available() -> tuple[str, str] | None: - return available_npm_package_update(SPEC["package"]) - - def _parse_version(value: str) -> tuple[int, int, int] | None: match = re.search(r"(\d+)\.(\d+)\.(\d+)", value) if not match: diff --git a/src/ucode/agents/copilot.py b/src/ucode/agents/copilot.py index 19a52b8e..3e876fbc 100644 --- a/src/ucode/agents/copilot.py +++ b/src/ucode/agents/copilot.py @@ -20,7 +20,6 @@ import threading from pathlib import Path -from ucode.agent_updates import available_npm_package_update from ucode.config_io import ( APP_DIR, ToolSpec, @@ -66,10 +65,6 @@ ] -def is_update_available() -> tuple[str, str] | None: - return available_npm_package_update(SPEC["package"]) - - def default_model(state: dict) -> str | None: """Prefer Claude sonnet, then opus/haiku, then codex. diff --git a/src/ucode/agents/gemini.py b/src/ucode/agents/gemini.py index 5e54571a..a45850c9 100644 --- a/src/ucode/agents/gemini.py +++ b/src/ucode/agents/gemini.py @@ -75,26 +75,6 @@ def latest_working_version() -> str | None: return latest_version_below(SPEC["package"], MAX_GEMINI_VERSION) -def is_update_available() -> tuple[str, str] | None: - """Offer an update only toward a known-working version. - - The npm `latest` tag points at the broken >= 0.45 line, so the generic - "outdated" check would steer clients onto the regression. Instead we - compare the installed build against the latest working release and only - surface an upgrade when it is genuinely newer (and still safe). - """ - installed = _parse_version(agent_version(SPEC["binary"])) - if installed is None: - return None - target = latest_working_version() - if target is None: - return None - target_base = _parse_version(target) - if target_base is None or target_base <= installed: - return None - return f"{installed[0]}.{installed[1]}.{installed[2]}", target - - def too_new_version() -> str | None: """Return the installed version string when it exceeds the safe ceiling. diff --git a/src/ucode/agents/opencode.py b/src/ucode/agents/opencode.py index 19adff71..b7803d66 100644 --- a/src/ucode/agents/opencode.py +++ b/src/ucode/agents/opencode.py @@ -7,7 +7,6 @@ import subprocess import threading -from ucode.agent_updates import available_npm_package_update from ucode.config_io import ( APP_DIR, ToolSpec, @@ -45,10 +44,6 @@ ] -def is_update_available() -> tuple[str, str] | None: - return available_npm_package_update(SPEC["package"]) - - def _resolve_model_selector(model: str, opencode_models: dict[str, list[str]]) -> str: """Return an OpenCode model selector in provider/model form when possible.""" if model.startswith(("databricks-anthropic/", "databricks-google/", "databricks-oss/")): diff --git a/src/ucode/agents/pi.py b/src/ucode/agents/pi.py index b6069114..a673a548 100644 --- a/src/ucode/agents/pi.py +++ b/src/ucode/agents/pi.py @@ -32,7 +32,6 @@ import subprocess import threading -from ucode.agent_updates import available_npm_package_update from ucode.config_io import ( APP_DIR, ToolSpec, @@ -79,10 +78,6 @@ LEGACY_PROVIDER_NAMES = ("databricks-anthropic", "databricks-codex", "databricks-oss") -def is_update_available() -> tuple[str, str] | None: - return available_npm_package_update(SPEC["package"]) - - def _resolve_model_selector( model: str, claude_models: dict[str, str], diff --git a/tests/test_agent_gemini.py b/tests/test_agent_gemini.py index 5cd70fa3..3cdadf43 100644 --- a/tests/test_agent_gemini.py +++ b/tests/test_agent_gemini.py @@ -156,24 +156,6 @@ def test_too_new_downgrade_none_when_no_target(self, monkeypatch): monkeypatch.setattr(gemini, "latest_version_below", lambda pkg, ceiling: None) assert gemini.too_new_downgrade() is None - def test_update_only_offered_toward_working_version(self, monkeypatch): - # Installed 0.40.0, latest working 0.44.1 -> offer the upgrade. - monkeypatch.setattr(gemini, "agent_version", lambda binary: "0.40.0") - monkeypatch.setattr(gemini, "latest_version_below", lambda pkg, ceiling: "0.44.1") - assert gemini.is_update_available() == ("0.40.0", "0.44.1") - - def test_no_update_when_already_at_working_version(self, monkeypatch): - monkeypatch.setattr(gemini, "agent_version", lambda binary: "0.44.1") - monkeypatch.setattr(gemini, "latest_version_below", lambda pkg, ceiling: "0.44.1") - assert gemini.is_update_available() is None - - def test_no_update_offered_toward_broken_version(self, monkeypatch): - # Even when a newer 0.45 exists, the target stays below the ceiling. - monkeypatch.setattr(gemini, "agent_version", lambda binary: "0.44.1") - monkeypatch.setattr(gemini, "latest_version_below", lambda pkg, ceiling: "0.44.1") - assert gemini.is_update_available() is None - - class TestGeminiValidateCmd: def test_starts_with_binary(self): cmd = gemini.validate_cmd("gemini") diff --git a/tests/test_agent_updates.py b/tests/test_agent_updates.py index 297eac18..230e1a01 100644 --- a/tests/test_agent_updates.py +++ b/tests/test_agent_updates.py @@ -5,56 +5,7 @@ import json import subprocess -from ucode.agent_updates import ( - available_npm_package_update, - latest_version_below, - published_versions, -) - - -def test_returns_none_when_npm_missing(monkeypatch): - monkeypatch.setattr("ucode.agent_updates.shutil.which", lambda _: None) - - assert available_npm_package_update("opencode-ai") is None - - -def test_returns_none_when_package_is_current(monkeypatch): - monkeypatch.setattr("ucode.agent_updates.shutil.which", lambda _: "/usr/bin/npm") - monkeypatch.setattr( - "ucode.agent_updates.subprocess.run", - lambda *args, **kwargs: subprocess.CompletedProcess(args[0], 0, stdout="{}", stderr=""), - ) - - assert available_npm_package_update("opencode-ai") is None - - -def test_returns_current_and_latest_when_outdated(monkeypatch): - monkeypatch.setattr("ucode.agent_updates.shutil.which", lambda _: "/usr/bin/npm") - - def fake_run(*args, **kwargs): - return subprocess.CompletedProcess( - args[0], - 1, - stdout='{"opencode-ai":{"current":"1.2.3","wanted":"1.2.4","latest":"1.2.4"}}', - stderr="", - ) - - monkeypatch.setattr("ucode.agent_updates.subprocess.run", fake_run) - - assert available_npm_package_update("opencode-ai") == ("1.2.3", "1.2.4") - - -def test_returns_none_for_malformed_output(monkeypatch): - monkeypatch.setattr("ucode.agent_updates.shutil.which", lambda _: "/usr/bin/npm") - monkeypatch.setattr( - "ucode.agent_updates.subprocess.run", - lambda *args, **kwargs: subprocess.CompletedProcess( - args[0], 1, stdout="not json", stderr="" - ), - ) - - assert available_npm_package_update("opencode-ai") is None - +from ucode.agent_updates import latest_version_below, published_versions _GEMINI_VERSIONS = [ "0.43.0", diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py index eb061735..f808c4f5 100644 --- a/tests/test_agents_init.py +++ b/tests/test_agents_init.py @@ -57,11 +57,6 @@ def test_each_spec_has_required_keys(self): def test_default_tool_is_codex(self): assert DEFAULT_TOOL == "codex" - def test_each_agent_exposes_update_check(self): - for tool, module in agents_mod._MODULES.items(): - assert callable(module.is_update_available), f"{tool} missing is_update_available" - - class TestInstallAiToolsForAgents: def _capture(self, monkeypatch): captured = {} @@ -380,10 +375,6 @@ def fake_run(args, **kwargs): monkeypatch.setattr("ucode.agents.shutil.which", fake_which) monkeypatch.setattr("ucode.agents.subprocess.run", fake_run) - monkeypatch.setattr( - "ucode.agents.opencode.is_update_available", - lambda: (_ for _ in ()).throw(AssertionError("should not check for optional updates")), - ) monkeypatch.setattr( "ucode.agents.prompt_yes_no", lambda prompt: (_ for _ in ()).throw(AssertionError("should not prompt")), From 4fbab0a82f58750a996e099562b43688812dd856 Mon Sep 17 00:00:00 2001 From: Rohit Agrawal Date: Sat, 29 Aug 2026 17:17:14 -0400 Subject: [PATCH 6/6] Format update-check cleanup --- src/ucode/agent_updates.py | 1 - tests/test_agent_gemini.py | 1 + tests/test_agents_init.py | 1 + 3 files changed, 2 insertions(+), 1 deletion(-) diff --git a/src/ucode/agent_updates.py b/src/ucode/agent_updates.py index fe77184d..2ca32f9e 100644 --- a/src/ucode/agent_updates.py +++ b/src/ucode/agent_updates.py @@ -68,4 +68,3 @@ def latest_version_below(package: str, ceiling: tuple[int, int, int]) -> str | N pool = stable or at_max # npm returns versions in ascending order, so the last entry is newest. return pool[-1] - diff --git a/tests/test_agent_gemini.py b/tests/test_agent_gemini.py index 3cdadf43..1f3f67c5 100644 --- a/tests/test_agent_gemini.py +++ b/tests/test_agent_gemini.py @@ -156,6 +156,7 @@ def test_too_new_downgrade_none_when_no_target(self, monkeypatch): monkeypatch.setattr(gemini, "latest_version_below", lambda pkg, ceiling: None) assert gemini.too_new_downgrade() is None + class TestGeminiValidateCmd: def test_starts_with_binary(self): cmd = gemini.validate_cmd("gemini") diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py index f808c4f5..7525ae46 100644 --- a/tests/test_agents_init.py +++ b/tests/test_agents_init.py @@ -57,6 +57,7 @@ def test_each_spec_has_required_keys(self): def test_default_tool_is_codex(self): assert DEFAULT_TOOL == "codex" + class TestInstallAiToolsForAgents: def _capture(self, monkeypatch): captured = {}