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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions src/ucode/agents/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
map_claude_family_models,
resolve_provider_service,
)
from ucode.managed_files import restore_managed_file
from ucode.state import get_provider_service, load_state, save_state
from ucode.telemetry import agent_version
from ucode.ui import (
Expand Down Expand Up @@ -89,6 +90,32 @@
}


def _os_managed_file_path(tool: str):
if tool == "claude":
return claude._managed_settings_path()
if tool == "codex":
return codex._managed_config_path()
return None


def reconcile_global_settings(managed: dict | None) -> None:
"""Undo ucode's OS-managed-file write for each global-settings agent the current config no longer
marks global, so switching workspaces stops a prior one's settings applying to a bare agent.

Reconciles both agents, not just the launched one; an agent still marked global is left for its
writer to (re)write. Idempotent, drift-suppressed, and nonfatal. ``managed_use_as_global_settings``
is imported locally to avoid a cycle (managed_resolve imports ``GLOBAL_SETTINGS_AGENTS`` from here).
"""
from ucode.managed_resolve import managed_use_as_global_settings

for tool in GLOBAL_SETTINGS_AGENTS:
if managed_use_as_global_settings(managed or {}, tool):
continue
path = _os_managed_file_path(tool)
if path is not None:
restore_managed_file(path, display=TOOL_SPECS[tool]["display"])


def install_databricks_ai_tools_for_agents(tools: list[str], state: dict) -> None:
"""Install Databricks AI Tools for the coding agents that support them
(gemini/pi have no ``aitools`` support and are dropped)."""
Expand Down
5 changes: 5 additions & 0 deletions src/ucode/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
install_tool_binary,
normalize_tool,
provider_permission_error,
reconcile_global_settings,
resolve_launch_model,
resolve_provider_models,
validate_all_tools,
Expand Down Expand Up @@ -282,6 +283,7 @@ def _resolve_workspace_then_maybe_reject(
managed, coding_agent_config_feature_disabled = refresh_managed_config(
{"workspace": workspace, "profile": profile}
)
reconcile_global_settings(managed)
if not managed and not coding_agent_config_feature_disabled:
_maybe_offer_admin_setup(workspace, profile)
if not managed:
Expand Down Expand Up @@ -1910,6 +1912,8 @@ def _launch_tool(
f"which overrides `--model {model}` — Claude Code will launch on the pinned "
"model instead. Edit or remove that file to use --model."
)
if managed_agent_config_enabled():
reconcile_global_settings(managed)
state = configure_tool(
tool,
state,
Expand Down Expand Up @@ -2091,6 +2095,7 @@ def _launch_managed_default(
else:
with spinner("Loading..."):
managed, coding_agent_config_feature_disabled = refresh_managed_config(state)
reconcile_global_settings(managed)
if not managed and not coding_agent_config_feature_disabled:
_print_no_managed_config_guidance(current, state.get("profile"))
if not managed:
Expand Down
158 changes: 158 additions & 0 deletions src/ucode/managed_files.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,17 @@

from __future__ import annotations

import json
import os
import shlex
import subprocess
import sys
import tempfile
from collections.abc import Callable
from enum import Enum
from pathlib import Path

from ucode import config_io
from ucode.config_io import is_dry_run
from ucode.ui import console, print_err, print_warning

Expand Down Expand Up @@ -89,6 +92,7 @@ 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"
snapshot = _snapshot_original(path)
try:
_sudo_replace(path, desired_text)
except PermissionError as exc:
Expand All @@ -100,6 +104,7 @@ def write_managed_file(path: Path, desired_text: str, *, display: str) -> str:
except subprocess.CalledProcessError as exc:
_report_sudo_failure(path, display, exc)
return "skipped"
_record_original(path, snapshot)
return "written"


Expand Down Expand Up @@ -186,3 +191,156 @@ 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 _ledger_path() -> Path:
"""Machine-global record of each OS-managed file's pre-ucode state (OS files are machine-wide,
unlike per-workspace ``state.json``), so reconcile can restore the original instead of pruning keys."""
return config_io.APP_DIR / "managed-os-backups.json"


def _read_ledger() -> dict:
path = _ledger_path()
try:
if not path.exists():
return {}
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {}
return data if isinstance(data, dict) else {}


def _write_ledger(ledger: dict) -> None:
try:
config_io.APP_DIR.mkdir(parents=True, exist_ok=True)
_ledger_path().write_text(json.dumps(ledger, indent=2), encoding="utf-8")
except OSError:
pass


def _snapshot_original(path: Path) -> dict:
"""Read ``path``'s pre-ucode state into a ledger entry.

``contents`` is the exact text for a readable file, ``None`` for one absent or unreadable (kept
distinct from a genuinely empty ``""``). When the unprivileged stat or read fails (a root-locked
parent, or a root-only file under a readable dir) it falls back to a privileged probe via
:func:`_sudo_snapshot`, because guessing here would either leave a ucode-created file uncleaned or
delete a pre-existing one. Runs only on the write path, so the fallback adds no extra sudo prompt.
"""
try:
existed = path.exists()
except OSError:
return _sudo_snapshot(path)
if not existed:
return {"existed": False, "contents": None}
try:
return {"existed": True, "contents": path.read_text(encoding="utf-8")}
except OSError:
return _sudo_snapshot(path)


def _sudo_snapshot(path: Path) -> dict:
"""Determine ``path``'s pre-ucode state with sudo, for a root-locked parent or root-only file.

A file sudo confirms absent is recorded ``existed=False`` so cleanup removes the file ucode is
about to create; one readable only as root is captured verbatim; one unreadable even as root stays
``existed=True/contents=None`` (left untouched on cleanup). A sudo failure is harmless: the write
that follows fails the same way, so this snapshot is never recorded.
"""
present = subprocess.run([_SUDO, "test", "-e", str(path)], capture_output=True, check=False)
if present.returncode != 0:
return {"existed": False, "contents": None}
read = subprocess.run([_SUDO, "cat", str(path)], capture_output=True, text=True, check=False)
if read.returncode == 0:
return {"existed": True, "contents": read.stdout}
return {"existed": True, "contents": None}


def _record_original(path: Path, snapshot: dict) -> None:
"""Persist ``snapshot`` to the ledger under ``path``, once (call only after a successful write, so
a failed sudo leaves no false ownership record). No-op if already recorded."""
key = str(path)
ledger = _read_ledger()
if key in ledger:
return
ledger[key] = snapshot
_write_ledger(ledger)


def _forget_original(key: str) -> None:
if is_dry_run():
return
ledger = _read_ledger()
if key in ledger:
del ledger[key]
_write_ledger(ledger)


def restore_managed_file(path: Path, *, display: str) -> str:
"""Undo ucode's write to ``path``, restoring the pre-ucode original or removing a ucode-created
file. Returns ``"restored"``, ``"removed"``, ``"unchanged"``, or ``"skipped"``.

No ledger entry means ucode never wrote it, so nothing happens. Drift-suppressed (an already-matching
or already-absent file makes no sudo call) and never raises: sudo failures warn and return ``"skipped"``.
"""
if not managed_files_supported():
return "unchanged"
key = str(path)
entry = _read_ledger().get(key)
if not isinstance(entry, dict):
return "unchanged"
original = entry.get("contents")
if entry.get("existed"):
if not isinstance(original, str):
_forget_original(key)
return "unchanged"
if _read_existing(path) == original:
_forget_original(key)
return "unchanged"
if is_dry_run():
console.print(f"\n[bold]\\[dry run] restore {path} (via sudo)[/bold]\n{original}")
return "restored"
if not _sudo_managed_op(lambda: _sudo_replace(path, original), path, display):
return "skipped"
_forget_original(key)
return "restored"
if not _path_present(path):
_forget_original(key)
return "unchanged"
if is_dry_run():
console.print(f"\n[bold]\\[dry run] remove {path} (via sudo)[/bold]")
return "removed"
if not _sudo_managed_op(lambda: _sudo_remove(path), path, display):
return "skipped"
_forget_original(key)
return "removed"


def _path_present(path: Path) -> bool:
"""Whether ``path`` exists; a root-locked parent we can't stat is assumed present (let sudo try)."""
try:
return path.exists()
except OSError:
return True


def _sudo_managed_op(op: Callable[[], None], path: Path, display: str) -> bool:
"""Run a privileged managed-file op, turning its failures into a warning. Returns success."""
try:
op()
except PermissionError as exc:
print_err(
f"{display}: cannot update {path} without root ({exc}). Re-run with `sudo ucode ...` to "
"reconcile the machine-wide config."
)
return False
except subprocess.CalledProcessError as exc:
_report_sudo_failure(path, display, exc)
return False
return True


def _sudo_remove(path: Path) -> None:
"""Remove a root-owned managed file via sudo, clearing any immutable flag first."""
_clear_immutable(path)
subprocess.run([_SUDO, "rm", "-f", str(path)], capture_output=True, text=True, check=True)
50 changes: 50 additions & 0 deletions tests/test_agent_claude.py
Original file line number Diff line number Diff line change
Expand Up @@ -565,6 +565,56 @@ def test_relayed_skips_managed_write(self, monkeypatch):
assert any("bare `claude`" in w for w in warns)


class TestClaudeManagedSettingsLifecycle:
"""Workspace A (global settings) -> B (none): the real capture+restore path, sudo mocked."""

def _patch(self, tmp_path, monkeypatch):
import ucode.managed_files as mf

managed_path = tmp_path / "etc-claude-code" / "managed-settings.json"
monkeypatch.setattr(claude, "_managed_settings_path", lambda: managed_path)
monkeypatch.setattr(claude, "write_json_file", lambda path, payload: None)
monkeypatch.setattr(claude, "save_state", lambda state: None)
monkeypatch.setattr(claude, "backup_existing_file", lambda *a, **kw: True)
monkeypatch.setattr(claude, "_register_web_search_mcp", lambda *a, **kw: True)
monkeypatch.setattr(mf, "managed_files_supported", lambda: True)

def _fake_replace(path, text):
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(text, encoding="utf-8")

monkeypatch.setattr(mf, "_sudo_replace", _fake_replace)
monkeypatch.setattr(mf, "_sudo_remove", lambda path: Path(path).unlink())
return managed_path

def test_switch_away_restores_it_authored_file(self, tmp_path, monkeypatch):
import ucode.managed_files as mf

managed_path = self._patch(tmp_path, monkeypatch)
managed_path.parent.mkdir(parents=True, exist_ok=True)
managed_path.write_text('{"env": {"IT_KEY": "keep"}}', encoding="utf-8")

state = {"workspace": WS, "codex_models": [], "write_managed_config": True}
claude.write_tool_config(state, "databricks-claude-sonnet-4")
written = json.loads(managed_path.read_text(encoding="utf-8"))
assert written["env"]["IT_KEY"] == "keep" # IT key preserved through the merge
assert written["env"]["ANTHROPIC_BASE_URL"] # ucode keys applied

assert mf.restore_managed_file(managed_path, display="Claude Code") == "restored"
assert json.loads(managed_path.read_text(encoding="utf-8")) == {"env": {"IT_KEY": "keep"}}

def test_switch_away_removes_ucode_created_file(self, tmp_path, monkeypatch):
import ucode.managed_files as mf

managed_path = self._patch(tmp_path, monkeypatch) # no file on disk before ucode
state = {"workspace": WS, "codex_models": [], "write_managed_config": True}
claude.write_tool_config(state, "databricks-claude-sonnet-4")
assert managed_path.exists()
assert mf.restore_managed_file(managed_path, display="Claude Code") == "removed"
assert not managed_path.exists()


class TestRegisterWebSearchMcp:
def test_clears_existing_then_adds(self, monkeypatch):
import ucode.mcp as mcp_mod
Expand Down
51 changes: 51 additions & 0 deletions tests/test_agent_codex.py
Original file line number Diff line number Diff line change
Expand Up @@ -718,3 +718,54 @@ def test_no_managed_write_by_default(self, tmp_path, monkeypatch):
state = {"workspace": WS, "codex_models": ["gpt-5"]}
codex.write_tool_config(state)
assert not managed_path.exists()


class TestCodexManagedConfigLifecycle:
"""Workspace A (global settings) -> B (none): the real capture+restore path, sudo mocked."""

def _patch(self, tmp_path, monkeypatch):
import ucode.managed_files as mf

config_path = tmp_path / ".codex" / "ucode.config.toml"
managed_path = tmp_path / "etc-codex" / "managed_config.toml"
monkeypatch.setattr(codex, "CODEX_CONFIG_PATH", config_path)
monkeypatch.setattr(codex, "CODEX_BACKUP_PATH", tmp_path / "codex-ucode-config.backup.toml")
monkeypatch.setattr(codex, "agent_version", lambda binary: "0.134.0")
monkeypatch.setattr(codex, "save_state", lambda state: None)
monkeypatch.setattr(codex, "_managed_config_path", lambda: managed_path)
monkeypatch.setattr(mf, "managed_files_supported", lambda: True)

def _fake_replace(path, text):
p = Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(text, encoding="utf-8")

monkeypatch.setattr(mf, "_sudo_replace", _fake_replace)
monkeypatch.setattr(mf, "_sudo_remove", lambda path: Path(path).unlink())
return managed_path

def test_switch_away_restores_it_authored_file(self, tmp_path, monkeypatch):
import ucode.managed_files as mf

managed_path = self._patch(tmp_path, monkeypatch)
managed_path.parent.mkdir(parents=True, exist_ok=True)
managed_path.write_text('approval_policy = "on-request"\n', encoding="utf-8")

state = {"workspace": WS, "codex_models": ["gpt-5"], "write_managed_config": True}
codex.write_tool_config(state)
doc = read_toml_safe(managed_path)
assert doc["approval_policy"] == "on-request" # IT key preserved through the merge
assert doc["model"] == "gpt-5" # ucode keys applied

assert mf.restore_managed_file(managed_path, display="Codex") == "restored"
assert managed_path.read_text(encoding="utf-8") == 'approval_policy = "on-request"\n'

def test_switch_away_removes_ucode_created_file(self, tmp_path, monkeypatch):
import ucode.managed_files as mf

managed_path = self._patch(tmp_path, monkeypatch) # no file on disk before ucode
state = {"workspace": WS, "codex_models": ["gpt-5"], "write_managed_config": True}
codex.write_tool_config(state)
assert managed_path.exists()
assert mf.restore_managed_file(managed_path, display="Codex") == "removed"
assert not managed_path.exists()
Loading
Loading