[AIGTWY-4901] Record the settings key paths ug writes

AGENTS.md

@@ -49,6 +49,14 @@ The two columns show what `ug` does without and with a [managed config](https://
 
 This is the intended behavior; we are still working to make the code match it in every case. In particular, `ug` today also clears some fields that this list leaves alone: the `env.ANTHROPIC_DEFAULT_*_MODEL_NAME` companions, `env.CLAUDE_CODE_ENABLE_GATEWAY_MODEL_DISCOVERY`, `env.CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS`, and Codex `model_reasoning_effort`. Codex `[otel]` is never written to a file.
 
+### How `ug` tracks what it wrote (provenance)
+
+`ug` shares these files with other writers (an admin, Isaac, MDM, or the developer), so when it cleans up a field it previously set it must not delete a value someone else owns at the same key. For that, `ug` keeps a provenance record at `~/.ucode/managed-backups/provenance.json`: for each agent and destination file, the key paths `ug` wrote and the exact value it wrote at each.
+
+- `ug` removes or restores a field only while the file still holds the value `ug` last wrote there. If anyone else has since changed that value, `ug` gives up the claim and leaves the field alone.
+- The record is refreshed after each confirmed write by `ug configure` (including the auto-configure a launch runs first), and `ug revert` clears it.
+- It is not a mirror of this whole table. It holds only the keys a cleanup path actually consults (the fields `ug` conditionally removes), not every field `ug` writes. Values other writers set are never recorded.
+
 <details>
 <summary>Claude Code</summary>
 

src/ucode/cli.py

@@ -125,6 +125,7 @@
     revert_mcp_configs,
 )
 from ucode.os_compatibility import subprocess_cross_os
+from ucode.provenance import clear_provenance
 from ucode.skills_download import (
     configure_location_skills_download_command,
     configure_selected_skills_download_command,
@@ -1279,6 +1280,7 @@ def revert() -> int:
     # place; restoring the per-profile file above does not undo that.
     legacy_codex_stripped = revert_legacy_shared_config()
     clear_state()
+    clear_provenance()
 
     print_heading("Revert")
     print_kv("Workspace", state.get("workspace") or "none")

src/ucode/provenance.py

@@ -0,0 +1,202 @@
+"""Per-file record of the settings key paths ug wrote, and the exact values it wrote there.
+
+Later reconciles use the record to tell ug's own keys apart from a user's, so ug only ever
+restores or deletes values it can prove it wrote. ``ug revert`` clears the store.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+import os
+from collections.abc import Iterable, Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+
+from ucode import managed_files
+from ucode.config_io import is_dry_run
+from ucode.os_compatibility.file_lock_cross_os import (
+    acquire_exclusive_file_lock,
+    release_file_lock,
+)
+
+KeyPath = tuple[str, ...]
+
+
+def load_provenance(agent: str, path: Path) -> dict[KeyPath, Any] | None:
+    """Return the key paths ug recorded for this file, or None when there is no record yet."""
+    store = _load_store()
+    if store is None:
+        return None
+    return _parse_entries(store["files"].get(_file_key(agent, path)))
+
+
+def save_provenance(agent: str, path: Path, owned: dict[KeyPath, Any]) -> None:
+    """Replace the recorded key paths and values ug wrote to this file."""
+    if is_dry_run():
+        return
+    with _store_lock():
+        store = _load_store() or {"files": {}}
+        store["files"][_file_key(agent, path)] = [
+            {"path": list(key_path), "value": value} for key_path, value in sorted(owned.items())
+        ]
+        managed_files._write_private_file(_store_path(), json.dumps(store, indent=2) + "\n")
+
+
+def clear_provenance() -> None:
+    """Delete the whole store, as part of ``ug revert``; a missing store is fine."""
+    if is_dry_run() or not _store_path().exists():
+        return
+    try:
+        with _store_lock():
+            _store_path().unlink(missing_ok=True)
+    except OSError as exc:
+        raise RuntimeError(f"Could not remove the provenance store: {exc}") from exc
+
+
+def owned_after_write(
+    owned: dict[KeyPath, Any],
+    generated: dict[KeyPath, Any],
+    final: dict,
+    before: dict | None = None,
+) -> dict[KeyPath, Any]:
+    """Keep generated paths that landed in ``final`` and prior owned paths still intact there.
+
+    With ``before`` (the file before this write), a generated value the file already held is
+    adopted only if that held value was still ug's own, so an equal value someone else wrote, or an
+    edit that happens to match, stays theirs.
+    """
+    result: dict[KeyPath, Any] = {
+        key_path: copy.deepcopy(value)
+        for key_path, value in generated.items()
+        if _is_intact(final, key_path, value)
+        and (
+            before is None
+            or not _is_intact(before, key_path, value)
+            or (key_path in owned and _is_intact(before, key_path, owned[key_path]))
+        )
+    }
+    for key_path, value in owned.items():
+        if key_path not in result and _is_intact(final, key_path, value):
+            result[key_path] = copy.deepcopy(value)
+    return result
+
+
+def values_at(key_paths: list[list[str]], doc: dict | None) -> dict[KeyPath, Any]:
+    """The values ``doc`` holds at ``key_paths``, skipping paths it lacks; None means no document."""
+    if doc is None:
+        return {}
+    owned: dict[KeyPath, Any] = {}
+    for key_path in map(tuple, key_paths):
+        value = _value_at(doc, key_path)
+        if value is not managed_files._MISSING:
+            owned[key_path] = copy.deepcopy(value)
+    return owned
+
+
+def retire(
+    doc: dict, owned: dict[KeyPath, Any], paths: Iterable[KeyPath], baseline: dict | None
+) -> None:
+    """Restore each owned, unedited path to its baseline value, or delete it when none exists."""
+    for key_path in paths:
+        if key_path in owned and _is_intact(doc, key_path, owned[key_path]):
+            _restore(doc, key_path, baseline)
+
+
+def retire_group(
+    doc: dict, owned: dict[KeyPath, Any], paths: Iterable[KeyPath], baseline: dict | None
+) -> None:
+    """Like :func:`retire`, but all or nothing.
+
+    A member that ug owns but was edited or deleted, or a present member that ug does not own and
+    that differs from the baseline, vetoes the group; the veto drops ug's ownership of every member,
+    so a later run never retires a remnant of someone else's group. Unowned members still at their
+    baseline value stay.
+    """
+    paths = list(paths)
+    present = [
+        key_path for key_path in paths if _value_at(doc, key_path) is not managed_files._MISSING
+    ]
+
+    def vetoes(key_path: KeyPath) -> bool:
+        if key_path in owned:
+            return not _is_intact(doc, key_path, owned[key_path])
+        return key_path in present and (
+            baseline is None or not _is_intact(baseline, key_path, _value_at(doc, key_path))
+        )
+
+    if any(vetoes(key_path) for key_path in paths):
+        for key_path in paths:
+            owned.pop(key_path, None)
+        return
+    for key_path in present:
+        _restore(doc, key_path, baseline)
+
+
+def _store_path() -> Path:
+    return managed_files.MANAGED_BACKUP_DIR / "provenance.json"
+
+
+@contextmanager
+def _store_lock() -> Iterator[None]:
+    """Serialize the store's read-modify-write across concurrent ug processes with an OS file lock.
+
+    Keep the lock file in place: unlinking it could let waiters lock different inodes.
+    """
+    backup_dir = managed_files.MANAGED_BACKUP_DIR
+    backup_dir.mkdir(parents=True, exist_ok=True)
+    os.chmod(backup_dir, 0o700)
+    with (backup_dir / "provenance.lock").open("a+b") as lock_file:
+        acquire_exclusive_file_lock(lock_file)
+        try:
+            yield
+        finally:
+            release_file_lock(lock_file)
+
+
+def _file_key(agent: str, path: Path) -> str:
+    return f"{agent}:{path}"
+
+
+def _load_store() -> dict | None:
+    try:
+        store = json.loads(_store_path().read_text(encoding="utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError):
+        return None
+    if not isinstance(store, dict) or not isinstance(store.get("files"), dict):
+        return None
+    return store
+
+
+def _parse_entries(entries: object) -> dict[KeyPath, Any] | None:
+    if not isinstance(entries, list):
+        return None
+    owned: dict[KeyPath, Any] = {}
+    for entry in entries:
+        if not isinstance(entry, dict) or "value" not in entry:
+            continue
+        path = entry.get("path")
+        if isinstance(path, list) and all(isinstance(part, str) for part in path):
+            owned[tuple(path)] = entry["value"]
+    return owned
+
+
+def _value_at(doc: dict, key_path: KeyPath) -> Any:
+    return managed_files._path_value(doc, list(key_path))
+
+
+def _is_intact(doc: dict, key_path: KeyPath, value: Any) -> bool:
+    """Whether ``doc`` still holds exactly ``value`` at ``key_path``, comparing types strictly."""
+    current = _value_at(doc, key_path)
+    if current is managed_files._MISSING:
+        return False
+    return managed_files.is_semantically_equal(current, value)
+
+
+def _restore(doc: dict, key_path: KeyPath, baseline: dict | None) -> None:
+    value = _value_at(baseline, key_path) if baseline is not None else managed_files._MISSING
+    if value is managed_files._MISSING:
+        managed_files._delete_path_value(doc, list(key_path))
+    else:
+        managed_files._set_path_value(doc, list(key_path), value)

tests/conftest.py

@@ -37,6 +37,7 @@ def _isolate_ucode_state(tmp_path, monkeypatch):
     import ucode.state as state_mod
     from ucode.agents import codex as codex_mod
 
+    config_io_mod.set_dry_run(False)
     state_dir = tmp_path / ".ucode"
     state_dir.mkdir()
     monkeypatch.setattr(state_mod, "STATE_PATH", state_dir / "state.json")

tests/test_cli.py

@@ -2825,6 +2825,40 @@ def test_reverts_mcp_configs_before_clearing_state(self):
         assert cleared == [True]
         assert "Claude Code MCP config: restored" in result.output
 
+    def test_clears_provenance_store(self):
+        from ucode import managed_files, provenance
+
+        provenance.save_provenance("claude", Path("/tmp/settings.json"), {("env", "A"): "1"})
+        store_path = managed_files.MANAGED_BACKUP_DIR / "provenance.json"
+        assert store_path.exists()
+
+        with (
+            patch("ucode.cli.load_state", return_value=MINIMAL_STATE),
+            patch("ucode.cli.restore_file", return_value=False),
+            patch("ucode.cli.revert_mcp_configs", return_value={}),
+            patch("ucode.cli.clear_state"),
+        ):
+            result = runner.invoke(app, ["revert"])
+
+        assert result.exit_code == 0, result.output
+        assert not store_path.exists()
+
+    def test_clears_state_even_when_clearing_provenance_fails(self):
+        cleared: list[bool] = []
+
+        with (
+            patch("ucode.cli.load_state", return_value=MINIMAL_STATE),
+            patch("ucode.cli.restore_file", return_value=False),
+            patch("ucode.cli.revert_mcp_configs", return_value={}),
+            patch("ucode.cli.clear_state", side_effect=lambda: cleared.append(True)),
+            patch("ucode.cli.clear_provenance", side_effect=RuntimeError("boom")),
+        ):
+            result = runner.invoke(app, ["revert"])
+
+        assert result.exit_code == 1
+        assert "boom" in _strip_ansi(result.output)
+        assert cleared == [True]
+
 
 class TestDoctorCommand:
     def test_invokes_doctor(self):