diff --git a/src/ucode/agents/claude.py b/src/ucode/agents/claude.py index 5fad5225..9fbd4226 100644 --- a/src/ucode/agents/claude.py +++ b/src/ucode/agents/claude.py @@ -30,7 +30,7 @@ get_databricks_token, ) 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, remove_managed_file, write_managed_file from ucode.smart_routing.claude_hooks import ( remove_smart_routing_hooks, sync_smart_routing_hooks, @@ -559,7 +559,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) + _write_managed_settings(_compose, relayed, state["workspace"]) if web_search_model: _register_web_search_mcp(state["workspace"], web_search_model, state.get("profile")) @@ -576,14 +576,13 @@ def _compose(base: dict) -> dict: return state -def _write_managed_settings(compose: Callable[[dict], dict], relayed: bool) -> None: +def _write_managed_settings(compose: Callable[[dict], dict], relayed: bool, workspace: str) -> None: """Write ucode's config into Claude Code's OS managed-settings.json so a bare `claude` works. 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). + scope, so it applies whether or not `ucode` launches `claude`. The file is wholly ucode-owned; a + pre-existing unowned file is rejected rather than merged. The write goes through the sudo path + in `managed_files` (drift-suppressed, so no password prompt when unchanged). 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. @@ -602,8 +601,16 @@ def _write_managed_settings(compose: Callable[[dict], dict], relayed: bool) -> N "settings write." ) return - desired = json.dumps(compose(read_json_safe(path)), indent=2) - write_managed_file(path, desired, display="Claude Code") + desired = json.dumps(compose({}), indent=2) + write_managed_file(path, desired, display="Claude Code", workspace=workspace) + + +def clear_managed_settings() -> str: + """Remove Claude Code's machine-wide settings when ucode owns them.""" + path = _managed_settings_path() + if path is None: + return "unchanged" + return remove_managed_file(path, display="Claude Code") def _is_tracing_stop_hook(hook: object) -> bool: diff --git a/src/ucode/agents/codex.py b/src/ucode/agents/codex.py index 2b7415e2..01d3d424 100644 --- a/src/ucode/agents/codex.py +++ b/src/ucode/agents/codex.py @@ -26,7 +26,7 @@ get_databricks_token, ) 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, remove_managed_file, write_managed_file from ucode.smart_routing.codex_hooks import ( remove_smart_routing_hooks, sync_smart_routing_hooks, @@ -399,9 +399,9 @@ def _write_managed_config( use_pat: bool, provider: str | None, ) -> None: - """Merge the modern overlay into Codex's OS managed_config.toml, preserving any other keys there. + """Write the modern overlay into Codex's strictly ucode-owned OS managed config. - Written via the sudo path in `managed_files` (drift-suppressed). + A pre-existing unowned file is rejected by `managed_files`; ucode never merges into it. """ path = _managed_config_path() if path is None: @@ -413,12 +413,20 @@ def _write_managed_config( overlay = render_overlay( workspace, model, databricks_profile, use_pat=use_pat, provider=provider ) - doc = read_toml_safe(path) + doc: dict = {} deep_merge_dict(doc, overlay) if provider: # deep_merge can't drop keys; clear a `model` a prior non-provider run pinned. doc.pop("model", None) - write_managed_file(path, tomlkit.dumps(doc), display="Codex") + write_managed_file(path, tomlkit.dumps(doc), display="Codex", workspace=workspace) + + +def clear_managed_config() -> str: + """Remove Codex's machine-wide config when ucode owns it.""" + path = _managed_config_path() + if path is None: + return "unchanged" + return remove_managed_file(path, display="Codex") def default_model(state: dict) -> str | None: diff --git a/src/ucode/cli.py b/src/ucode/cli.py index 0cbbe14a..694d77a4 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -1445,6 +1445,23 @@ def _fetch_managed_config(state: dict) -> dict | None: return refresh_managed_config(state) +def _reconcile_global_managed_settings(managed: dict | None, tool: str) -> None: + """Remove a prior workspace's global file when the target does not want one.""" + if not managed_agent_config_enabled() or managed_use_as_global_settings(managed or {}, tool): + return + if tool == "claude": + result = claude_agent.clear_managed_settings() + elif tool == "codex": + result = codex_agent.clear_managed_config() + else: + return + if result == "skipped": + raise RuntimeError( + f"Could not clear stale machine-wide settings for {TOOL_SPECS[tool]['display']}. " + "Resolve the warning above before launching this workspace." + ) + + def _note_recommended_agent(recommendation: dict | None, tool: str) -> None: """Say when the budget tier points at a different agent than the one being launched. @@ -1651,6 +1668,7 @@ def _launch_tool( # control-plane round trip and any fallback warning it printed. if managed is None: managed = _fetch_managed_config(state) + _reconcile_global_managed_settings(managed, tool) # Checked before discovery, which can take tens of seconds, so a blocked launch fails fast. _reject_disabled_agent(managed, tool) # Discovery exists to find models and isn't needed for managed config that already names them. diff --git a/src/ucode/managed_files.py b/src/ucode/managed_files.py index 40e58803..c27951f8 100644 --- a/src/ucode/managed_files.py +++ b/src/ucode/managed_files.py @@ -1,4 +1,4 @@ -"""Write agent config into OS-level *managed settings* files. +"""Own and write agent config in OS-level *managed 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 @@ -8,15 +8,17 @@ ``/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. +ucode uses strict file ownership: it creates an adjacent ``.ucode-owner.json`` marker when it first +creates a managed file, and only updates or removes files carrying a valid marker whose content hash +still matches. A pre-existing unowned file is never modified. The marker persists the owning +workspace across sessions, allowing workspace A's file to become workspace B's and then be removed +when workspace C has no global managed config, without retaining a baseline copy. """ from __future__ import annotations +import hashlib +import json import os import shlex import subprocess @@ -30,6 +32,7 @@ # Absolute path so a stripped PATH (desktop/GUI launchers) still finds it. _SUDO = "/usr/bin/sudo" +_OWNER_VERSION = 1 class OS(Enum): @@ -70,8 +73,63 @@ def _read_existing(path: Path) -> str: return "" -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). +def _path_exists(path: Path) -> bool | None: + """Whether ``path`` exists, or None when permissions prevent determining it.""" + try: + return path.exists() + except OSError: + return None + + +def ownership_path(path: Path) -> Path: + """The durable ownership marker adjacent to ``path``.""" + return path.with_name(f"{path.name}.ucode-owner.json") + + +def _content_hash(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _load_owner(path: Path) -> dict | None: + marker = ownership_path(path) + marker_exists = _path_exists(marker) + if marker_exists is False: + return None + if marker_exists is None: + return {} + try: + value = json.loads(marker.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + if not isinstance(value, dict): + return {} + if ( + value.get("version") != _OWNER_VERSION + or value.get("owner") != "ucode" + or not isinstance(value.get("workspace"), str) + or not isinstance(value.get("sha256"), str) + ): + return {} + return value + + +def _owner_text(workspace: str, desired_text: str) -> str: + return ( + json.dumps( + { + "version": _OWNER_VERSION, + "owner": "ucode", + "workspace": workspace, + "sha256": _content_hash(desired_text), + }, + indent=2, + ) + + "\n" + ) + + +def write_managed_file(path: Path, desired_text: str, *, display: str, workspace: str) -> str: + """Write ``desired_text`` only when ucode strictly owns ``path``. Returns ``"written"``, ``"unchanged"``, or ``"skipped"``. Never raises: a permission or immutable failure is surfaced as an actionable message and reported as ``"skipped"`` so the launch still @@ -83,26 +141,96 @@ def write_managed_file(path: Path, desired_text: str, *, display: str) -> str: f"skipped {path}." ) return "skipped" - # Drift check first — reading is unprivileged, so an unchanged file never triggers a sudo prompt. - if _read_existing(path) == desired_text: + existing = _read_existing(path) + owner = _load_owner(path) + target_exists = _path_exists(path) + if owner is None and target_exists is not False: + print_warning( + f"{display}: {path} already exists and is not owned by ucode; " + "leaving it unchanged. Its machine-wide settings may override this ucode launch." + ) + return "skipped" + if owner == {}: + print_warning(f"{display}: ownership metadata for {path} is invalid; leaving it unchanged.") + return "skipped" + if owner is not None and target_exists is None: + print_warning(f"{display}: cannot verify ownership of {path}; leaving it unchanged.") + return "skipped" + if owner is not None and target_exists and owner["sha256"] != _content_hash(existing): + print_warning(f"{display}: {path} changed after ucode wrote it; leaving it unchanged.") + return "skipped" + marker = ownership_path(path) + marker_text = _owner_text(workspace, desired_text) + if existing == desired_text and owner is not None and owner["workspace"] == workspace: return "unchanged" if is_dry_run(): - console.print(f"\n[bold]\\[dry run] {path} (via sudo)[/bold]\n{desired_text}") + console.print(f"\n[bold]\\[dry run] {path} + {marker} (via sudo)[/bold]\n{desired_text}") return "written" + existed = target_exists is True + marker_existed = owner is not None + existing_marker = _read_existing(marker) if marker_existed else "" try: _sudo_replace(path, desired_text) + _sudo_replace(marker, marker_text) except PermissionError as exc: + _rollback_owned_write(path, existing, existed, marker, existing_marker, marker_existed) print_err( f"{display}: cannot write {path} without root ({exc}). Re-run with `sudo ucode ...` to " "apply the config machine-wide." ) return "skipped" except subprocess.CalledProcessError as exc: + _rollback_owned_write(path, existing, existed, marker, existing_marker, marker_existed) _report_sudo_failure(path, display, exc) return "skipped" return "written" +def remove_managed_file(path: Path, *, display: str) -> str: + """Remove ``path`` only when its ownership marker and content still match. + + Returns ``"removed"`, ``"unchanged"``, or ``"skipped"``. An unmarked file is not ours and is + therefore unchanged; drift is skipped so ucode never deletes a file another actor took over. + """ + marker = ownership_path(path) + owner = _load_owner(path) + if owner is None: + return "unchanged" + if not managed_files_supported(): + print_warning( + f"{display}: machine-wide managed settings aren't supported on this platform; " + f"could not remove {path}." + ) + return "skipped" + if owner == {}: + print_warning(f"{display}: ownership metadata for {path} is invalid; leaving it unchanged.") + return "skipped" + target_exists = _path_exists(path) + if target_exists is None: + print_warning(f"{display}: cannot verify ownership of {path}; leaving it unchanged.") + return "skipped" + if target_exists and owner["sha256"] != _content_hash(_read_existing(path)): + print_warning(f"{display}: {path} changed after ucode wrote it; leaving it unchanged.") + return "skipped" + if is_dry_run(): + console.print(f"\n[bold]\\[dry run] remove {path} + {marker} (via sudo)[/bold]") + return "removed" + try: + if target_exists: + _sudo_remove(path) + _sudo_remove(marker) + except PermissionError as exc: + print_err( + f"{display}: cannot remove {path} without root ({exc}). Re-run with `sudo ucode ...` " + "to clear the stale workspace-managed settings." + ) + return "skipped" + except subprocess.CalledProcessError as exc: + _report_sudo_failure(path, display, exc) + return "skipped" + return "removed" + + def _sudo_replace(path: Path, desired_text: str) -> None: """Replace ``path`` with ``desired_text`` via sudo (temp file → ``sudo cp``), handling immutability. @@ -133,6 +261,37 @@ def _sudo_replace(path: Path, desired_text: str) -> None: os.unlink(tmp_path) +def _sudo_remove(path: Path) -> None: + """Remove a ucode-owned managed file or marker through sudo.""" + _clear_immutable(path) + subprocess.run([_SUDO, "rm", "-f", str(path)], capture_output=True, text=True, check=True) + + +def _rollback_owned_write( + path: Path, + existing: str, + existed: bool, + marker: Path, + existing_marker: str, + marker_existed: bool, +) -> None: + """Best-effort rollback when updating the managed file and marker fails partway through.""" + try: + if existed: + _sudo_replace(path, existing) + else: + _sudo_remove(path) + if marker_existed: + _sudo_replace(marker, existing_marker) + else: + _sudo_remove(marker) + except (PermissionError, subprocess.CalledProcessError): + print_warning( + f"Could not fully roll back a failed managed-settings update at {path}; " + "inspect the file and its ucode ownership marker before retrying." + ) + + def _clear_immutable(path: Path) -> bool: """Clear an immutable flag a fleet golden image may have set. Returns whether to restore it. diff --git a/tests/test_agent_claude.py b/tests/test_agent_claude.py index b45b6303..2536838b 100644 --- a/tests/test_agent_claude.py +++ b/tests/test_agent_claude.py @@ -494,7 +494,7 @@ def _patch(self, monkeypatch, private_writes, managed_writes, existing_by_path=N # 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) - def fake_write_managed(path, text, *, display): + def fake_write_managed(path, text, *, display, workspace): managed_writes.append((str(path), text)) return "written" @@ -510,17 +510,18 @@ def test_writes_managed_file_when_flagged(self, monkeypatch): assert str(claude.CLAUDE_SETTINGS_PATH) in [p for p, _ in private_writes] assert [p for p, _ in managed_writes] == [str(FAKE_MANAGED_PATH)] - def test_managed_file_preserves_other_keys(self, monkeypatch): + def test_managed_file_does_not_merge_unowned_keys(self, monkeypatch): private_writes: list = [] managed_writes: list = [] - # An IT-authored key already in the managed file must survive the merge. + # Strict ownership means the desired document contains only ucode's settings. The real + # managed-file writer refuses this write because the existing file has no owner marker. 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} claude.write_tool_config(state, "databricks-claude-sonnet-4") _, text = managed_writes[0] written = json.loads(text) - assert written["env"]["MY_OWN"] == "keep" + assert "MY_OWN" not in written["env"] assert written["env"]["ANTHROPIC_BASE_URL"] assert written["apiKeyHelper"] diff --git a/tests/test_agent_codex.py b/tests/test_agent_codex.py index 54fe1aa5..652e932e 100644 --- a/tests/test_agent_codex.py +++ b/tests/test_agent_codex.py @@ -681,7 +681,7 @@ def _patch(self, tmp_path, monkeypatch): # can read the TOML back and NO real sudo/`/etc` write ever happens. monkeypatch.setattr(codex, "_managed_config_path", lambda: managed_path) - def fake_write_managed(path, text, *, display): + def fake_write_managed(path, text, *, display, workspace): Path(path).parent.mkdir(parents=True, exist_ok=True) Path(path).write_text(text, encoding="utf-8") return "written" @@ -699,7 +699,7 @@ def test_writes_managed_config_when_flagged(self, tmp_path, monkeypatch): assert doc["model"] == "gpt-5" assert "ucode-databricks" in doc["model_providers"] - def test_managed_config_preserves_other_keys(self, tmp_path, monkeypatch): + def test_managed_config_does_not_merge_unowned_keys(self, tmp_path, monkeypatch): _, managed_path = self._patch(tmp_path, monkeypatch) managed_path.parent.mkdir(parents=True, exist_ok=True) managed_path.write_text( @@ -709,8 +709,9 @@ def test_managed_config_preserves_other_keys(self, tmp_path, monkeypatch): codex.write_tool_config(state) doc = read_toml_safe(managed_path) - # ucode pins its own model, but other keys already in the managed file survive. - assert doc["approval_policy"] == "on-request" + # Strict ownership means the desired document contains only ucode's settings. The real + # managed-file writer refuses this write because the existing file has no owner marker. + assert "approval_policy" not in doc assert doc["model"] == "gpt-5" def test_no_managed_write_by_default(self, tmp_path, monkeypatch): diff --git a/tests/test_cli.py b/tests/test_cli.py index 2069cd2e..40317711 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2481,6 +2481,61 @@ def test_skip_managed_config_makes_the_fetch_a_no_op(self, monkeypatch): assert self._fetch({"workspace": "https://w"}) is None +class TestReconcileGlobalManagedSettings: + @staticmethod + def _reconcile(managed, tool): + import ucode.cli as cli_mod + + cli_mod._reconcile_global_managed_settings(managed, tool) + + def test_no_config_removes_owned_codex_file(self, monkeypatch): + monkeypatch.setenv("ENABLE_MANAGED_AGENT_CONFIG", "1") + calls: list[str] = [] + monkeypatch.setattr( + "ucode.cli.codex_agent.clear_managed_config", lambda: calls.append("codex") or "removed" + ) + + self._reconcile(None, "codex") + assert calls == ["codex"] + + def test_non_global_config_removes_owned_claude_file(self, monkeypatch): + monkeypatch.setenv("ENABLE_MANAGED_AGENT_CONFIG", "1") + calls: list[str] = [] + monkeypatch.setattr( + "ucode.cli.claude_agent.clear_managed_settings", + lambda: calls.append("claude") or "removed", + ) + + self._reconcile({"enabled_agents": {"claude": {}}}, "claude") + assert calls == ["claude"] + + def test_global_config_keeps_file_for_agent_writer(self, monkeypatch): + monkeypatch.setenv("ENABLE_MANAGED_AGENT_CONFIG", "1") + monkeypatch.setattr( + "ucode.cli.codex_agent.clear_managed_config", + lambda: pytest.fail("workspace B's writer should replace workspace A's owned file"), + ) + managed = {"enabled_agents": {"codex": {"use_as_global_settings": True}}} + + self._reconcile(managed, "codex") + + def test_cleanup_conflict_blocks_launch(self, monkeypatch): + monkeypatch.setenv("ENABLE_MANAGED_AGENT_CONFIG", "1") + monkeypatch.setattr("ucode.cli.codex_agent.clear_managed_config", lambda: "skipped") + + with pytest.raises(RuntimeError, match="Could not clear stale machine-wide settings"): + self._reconcile(None, "codex") + + def test_feature_off_does_not_touch_global_files(self, monkeypatch): + monkeypatch.delenv("ENABLE_MANAGED_AGENT_CONFIG", raising=False) + monkeypatch.setattr( + "ucode.cli.codex_agent.clear_managed_config", + lambda: pytest.fail("feature-off launch must not mutate global files"), + ) + + self._reconcile(None, "codex") + + class TestManagedConfigDecidesDiscoveryFromFreshRead: def test_a_removed_model_list_no_longer_skips_discovery(self, monkeypatch): """The sweep decision must come from the fetched config, not the cached one. diff --git a/tests/test_managed_files.py b/tests/test_managed_files.py index dbb14f7b..3c8fb31b 100644 --- a/tests/test_managed_files.py +++ b/tests/test_managed_files.py @@ -7,6 +7,7 @@ from __future__ import annotations import subprocess +from pathlib import Path import pytest @@ -32,6 +33,7 @@ def _capture_sudo(monkeypatch): monkeypatch.setattr( managed_files, "_sudo_replace", lambda path, text: calls.append((str(path), text)) ) + monkeypatch.setattr(managed_files, "_sudo_remove", lambda path: None) return calls @@ -40,35 +42,61 @@ def test_unchanged_content_does_not_sudo(self, tmp_path, monkeypatch): path = tmp_path / "managed.json" path.write_text("same", encoding="utf-8") calls = _capture_sudo(monkeypatch) - assert managed_files.write_managed_file(path, "same", display="X") == "unchanged" - # The whole point: an unchanged file never prompts for a password. + assert ( + managed_files.write_managed_file( + path, "same", display="X", workspace="https://workspace-a" + ) + == "skipped" + ) + # A matching file without an ownership marker is still not ours. assert calls == [] def test_changed_content_sudo_writes(self, tmp_path, monkeypatch): path = tmp_path / "managed.json" path.write_text("old", encoding="utf-8") calls = _capture_sudo(monkeypatch) - assert managed_files.write_managed_file(path, "new", display="X") == "written" - assert calls == [(str(path), "new")] + assert ( + managed_files.write_managed_file( + path, "new", display="X", workspace="https://workspace-a" + ) + == "skipped" + ) + assert calls == [] def test_absent_file_sudo_writes(self, tmp_path, monkeypatch): path = tmp_path / "managed.json" calls = _capture_sudo(monkeypatch) - assert managed_files.write_managed_file(path, "new", display="X") == "written" - assert calls == [(str(path), "new")] + assert ( + managed_files.write_managed_file( + path, "new", display="X", workspace="https://workspace-a" + ) + == "written" + ) + assert calls[0] == (str(path), "new") + assert calls[1][0] == str(managed_files.ownership_path(path)) def test_dry_run_does_not_sudo(self, tmp_path, monkeypatch): path = tmp_path / "managed.json" calls = _capture_sudo(monkeypatch) config_io.set_dry_run(True) - assert managed_files.write_managed_file(path, "new", display="X") == "written" + assert ( + managed_files.write_managed_file( + path, "new", display="X", workspace="https://workspace-a" + ) + == "written" + ) assert calls == [] def test_unsupported_platform_skips(self, tmp_path, monkeypatch): monkeypatch.setattr(managed_files, "managed_files_supported", lambda: False) calls = _capture_sudo(monkeypatch) path = tmp_path / "managed.json" - assert managed_files.write_managed_file(path, "new", display="X") == "skipped" + assert ( + managed_files.write_managed_file( + path, "new", display="X", workspace="https://workspace-a" + ) + == "skipped" + ) assert calls == [] def test_permission_error_is_skipped_not_raised(self, tmp_path, monkeypatch): @@ -79,7 +107,12 @@ def boom(path, text): monkeypatch.setattr(managed_files, "_sudo_replace", boom) # Never raises — the launch proceeds; the private ucode config still works. - assert managed_files.write_managed_file(path, "new", display="X") == "skipped" + assert ( + managed_files.write_managed_file( + path, "new", display="X", workspace="https://workspace-a" + ) + == "skipped" + ) def test_sudo_failure_is_skipped_not_raised(self, tmp_path, monkeypatch): path = tmp_path / "managed.json" @@ -88,7 +121,104 @@ def boom(path, text): raise subprocess.CalledProcessError(1, ["/usr/bin/sudo", "cp"], stderr="denied") monkeypatch.setattr(managed_files, "_sudo_replace", boom) - assert managed_files.write_managed_file(path, "new", display="X") == "skipped" + assert ( + managed_files.write_managed_file( + path, "new", display="X", workspace="https://workspace-a" + ) + == "skipped" + ) + + +class TestStrictOwnershipLifecycle: + @staticmethod + def _direct_sudo(monkeypatch): + def replace(path, text): + Path(path).parent.mkdir(parents=True, exist_ok=True) + Path(path).write_text(text, encoding="utf-8") + + def remove(path): + Path(path).unlink(missing_ok=True) + + monkeypatch.setattr(managed_files, "_sudo_replace", replace) + monkeypatch.setattr(managed_files, "_sudo_remove", remove) + + def test_workspace_a_to_b_to_c(self, tmp_path, monkeypatch): + self._direct_sudo(monkeypatch) + path = tmp_path / "managed.toml" + + assert ( + managed_files.write_managed_file(path, "workspace-a", display="X", workspace="A") + == "written" + ) + assert path.read_text(encoding="utf-8") == "workspace-a" + + assert ( + managed_files.write_managed_file(path, "workspace-b", display="X", workspace="B") + == "written" + ) + assert path.read_text(encoding="utf-8") == "workspace-b" + owner = managed_files._load_owner(path) + assert owner is not None and owner["workspace"] == "B" + + assert managed_files.remove_managed_file(path, display="X") == "removed" + assert not path.exists() + assert not managed_files.ownership_path(path).exists() + + def test_unchanged_owned_file_does_not_sudo(self, tmp_path, monkeypatch): + self._direct_sudo(monkeypatch) + path = tmp_path / "managed.toml" + managed_files.write_managed_file(path, "workspace-a", display="X", workspace="A") + calls = _capture_sudo(monkeypatch) + + assert ( + managed_files.write_managed_file(path, "workspace-a", display="X", workspace="A") + == "unchanged" + ) + assert calls == [] + + def test_does_not_overwrite_unowned_file(self, tmp_path, monkeypatch): + self._direct_sudo(monkeypatch) + path = tmp_path / "managed.toml" + path.write_text("enterprise", encoding="utf-8") + + assert ( + managed_files.write_managed_file(path, "workspace-a", display="X", workspace="A") + == "skipped" + ) + assert path.read_text(encoding="utf-8") == "enterprise" + assert not managed_files.ownership_path(path).exists() + + def test_does_not_remove_owned_file_after_external_change(self, tmp_path, monkeypatch): + self._direct_sudo(monkeypatch) + path = tmp_path / "managed.toml" + managed_files.write_managed_file(path, "workspace-a", display="X", workspace="A") + path.write_text("changed externally", encoding="utf-8") + + assert managed_files.remove_managed_file(path, display="X") == "skipped" + assert path.read_text(encoding="utf-8") == "changed externally" + assert managed_files.ownership_path(path).exists() + + def test_marker_write_failure_rolls_back_new_file(self, tmp_path, monkeypatch): + path = tmp_path / "managed.toml" + marker = managed_files.ownership_path(path) + + def replace(target, text): + Path(target).parent.mkdir(parents=True, exist_ok=True) + Path(target).write_text(text, encoding="utf-8") + if Path(target) == marker: + raise PermissionError("marker denied") + + monkeypatch.setattr(managed_files, "_sudo_replace", replace) + monkeypatch.setattr( + managed_files, "_sudo_remove", lambda target: Path(target).unlink(missing_ok=True) + ) + + assert ( + managed_files.write_managed_file(path, "workspace-a", display="X", workspace="A") + == "skipped" + ) + assert not path.exists() + assert not marker.exists() class TestClearImmutableStatDenied: