[AIGTWY-4573] Configure branching: managed-config fetch and cache foundation

src/ucode/managed_config.py

@@ -345,11 +345,14 @@ class FetchedManagedConfig(NamedTuple):
 
 
 class ManagedConfigResult(NamedTuple):
-    """The launch-path refresh outcome: the ``manifest`` to apply (None when absent or dropped) and
-    ``feature_disabled``, True when the coding-agent-configs feature is off server-side."""
+    """The launch-path refresh outcome: the ``manifest`` to apply (None when absent or dropped),
+    ``feature_disabled`` True when the coding-agent-configs feature is off server-side, and
+    ``definitively_absent`` True when the absence is definitive (NOT_FOUND or feature disabled)
+    rather than due to a transient fetch failure."""
 
     manifest: dict | None
     feature_disabled: bool
+    definitively_absent: bool
 
 
 def _as_dict(value: object) -> dict[str, object]:
@@ -688,27 +691,27 @@ def refresh_managed_config(state: dict) -> ManagedConfigResult:
     """
     workspace = state.get("workspace")
     if not workspace:
-        return ManagedConfigResult(None, False)
+        return ManagedConfigResult(None, False, False)
     try:
         token = get_databricks_token(workspace, state.get("profile"))
     except RuntimeError as exc:
-        return ManagedConfigResult(_persisted_fallback(workspace, str(exc)), False)
+        return ManagedConfigResult(_persisted_fallback(workspace, str(exc)), False, False)
     raw, reason = get_managed_config(workspace, token)
     if reason is not None:
         if _is_feature_disabled(reason):
             save_managed_state(workspace, {})
-            return ManagedConfigResult(None, True)
+            return ManagedConfigResult(None, True, True)
         fallback = _persisted_fallback(workspace, reason, refused=_is_permission_denied(reason))
-        return ManagedConfigResult(fallback, False)
+        return ManagedConfigResult(fallback, False, False)
     if raw is None:
         # Record that this workspace has no config, rather than leaving an earlier one on disk:
         # the file doubles as the fallback above, so a removed policy would otherwise come back
         # into force after the next transient outage.
         save_managed_state(workspace, {})
-        return ManagedConfigResult(None, False)
+        return ManagedConfigResult(None, False, True)
     # Persist the raw config verbatim; hand callers the normalized manifest they expect.
     save_managed_state(workspace, raw)
-    return ManagedConfigResult(normalize_managed_config(raw), False)
+    return ManagedConfigResult(normalize_managed_config(raw), False, False)
 
 
 def _is_feature_disabled(reason: str) -> bool:

src/ucode/managed_files.py

@@ -31,6 +31,10 @@
 _MISSING = object()
 _managed_write_batch: tuple[str, ...] = ()
 _managed_write_notice_shown = False
+# When set, reconcile_managed_file skips the OS-managed (sudo) write entirely. The launch path uses
+# this so a plain launch never re-writes /etc — that write is owned by `ug configure` and by the
+# launch-time apply that runs only when the managed configuration actually changed.
+_managed_writes_suppressed = False
 
 ManagedParser = Callable[[str], dict]
 ManagedDumper = Callable[[dict], str]
@@ -154,6 +158,24 @@ def managed_write_batch(displays: list[str]) -> Iterator[None]:
         _managed_write_notice_shown = previous_notice
 
 
+@contextmanager
+def suppressed_managed_writes() -> Iterator[None]:
+    """Within this context, :func:`reconcile_managed_file` skips the OS-managed (sudo) write.
+
+    Used by the launch path so a launch that isn't re-applying the managed configuration never
+    touches the root-owned /etc file (and never prompts for a password); the user-level config is
+    still written by the caller.
+    """
+    global _managed_writes_suppressed
+
+    prev = _managed_writes_suppressed
+    _managed_writes_suppressed = True
+    try:
+        yield
+    finally:
+        _managed_writes_suppressed = prev
+
+
 def _print_managed_write_permission(display: str) -> None:
     global _managed_write_notice_shown
 
@@ -239,6 +261,8 @@ def reconcile_managed_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 _managed_writes_suppressed:
+        return "unchanged"
     if not managed_files_supported():
         print_warning(
             f"{display}: OS-managed settings aren't supported on this platform; skipped {path}."

tests/test_managed_config.py

@@ -437,13 +437,17 @@ def test_persists_raw_and_returns_the_normalized_manifest(self, monkeypatch):
             mc_mod, "save_managed_state", lambda ws, cfg, **kwargs: saved.append((ws, cfg))
         )
         # The raw config is persisted verbatim; the caller gets the normalized manifest.
-        assert refresh_managed_config(_state()) == (normalize_managed_config(RAW_MANIFEST), False)
+        assert refresh_managed_config(_state()) == (
+            normalize_managed_config(RAW_MANIFEST),
+            False,
+            False,
+        )
         assert saved == [(WORKSPACE, RAW_MANIFEST)]
 
     def test_no_managed_config_returns_none(self, monkeypatch):
         monkeypatch.setattr(mc_mod, "get_managed_config", lambda ws, tok: (None, None))
         monkeypatch.setattr(mc_mod, "save_managed_state", lambda ws, cfg, **kwargs: None)
-        result, _ = refresh_managed_config(_state())
+        result, _, _ = refresh_managed_config(_state())
         assert result is None
 
     def test_read_failure_falls_back_to_the_persisted_config(self, monkeypatch):
@@ -452,7 +456,7 @@ def test_read_failure_falls_back_to_the_persisted_config(self, monkeypatch):
         monkeypatch.setattr(mc_mod, "get_managed_config", lambda ws, tok: (None, "HTTP 500"))
         monkeypatch.setattr(mc_mod, "load_managed_state", lambda ws: MANAGED)
         monkeypatch.setattr(mc_mod, "print_warning", lambda msg: warnings.append(msg))
-        assert refresh_managed_config(_state()) == (MANAGED, False)
+        assert refresh_managed_config(_state()) == (MANAGED, False, False)
         assert "HTTP 500" in warnings[0]
         assert "last one saved" in warnings[0]
 
@@ -464,7 +468,7 @@ def test_read_failure_without_persisted_config_is_silent(self, monkeypatch):
         monkeypatch.setattr(
             mc_mod, "print_warning", lambda msg: pytest.fail(f"should not warn: {msg}")
         )
-        result, _ = refresh_managed_config(_state())
+        result, _, _ = refresh_managed_config(_state())
         assert result is None
 
     def test_auth_failure_falls_back_to_the_persisted_config(self, monkeypatch):
@@ -476,7 +480,7 @@ def boom(ws, profile):
         monkeypatch.setattr(mc_mod, "get_databricks_token", boom)
         monkeypatch.setattr(mc_mod, "load_managed_state", lambda ws: MANAGED)
         monkeypatch.setattr(mc_mod, "print_warning", lambda msg: warnings.append(msg))
-        assert refresh_managed_config(_state()) == (MANAGED, False)
+        assert refresh_managed_config(_state()) == (MANAGED, False, False)
         assert "no token" in warnings[0]
 
     def test_auth_failure_without_persisted_config_is_silent(self, monkeypatch):
@@ -488,7 +492,7 @@ def boom(ws, profile):
         monkeypatch.setattr(
             mc_mod, "print_warning", lambda msg: pytest.fail(f"should not warn: {msg}")
         )
-        result, _ = refresh_managed_config(_state())
+        result, _, _ = refresh_managed_config(_state())
         assert result is None
 
     def test_permission_denied_without_cache_is_silent(self, monkeypatch):
@@ -500,7 +504,7 @@ def test_permission_denied_without_cache_is_silent(self, monkeypatch):
         monkeypatch.setattr(
             mc_mod, "print_warning", lambda msg: pytest.fail(f"should not warn: {msg}")
         )
-        result, _ = refresh_managed_config(_state())
+        result, _, _ = refresh_managed_config(_state())
         assert result is None
 
     def test_permission_denied_warns_and_keeps_the_cached_config(self, monkeypatch):
@@ -516,7 +520,7 @@ def test_permission_denied_warns_and_keeps_the_cached_config(self, monkeypatch):
             "save_managed_state",
             lambda ws, cfg, **kwargs: pytest.fail("must not clear the cache"),
         )
-        assert refresh_managed_config(_state()) == (MANAGED, False)
+        assert refresh_managed_config(_state()) == (MANAGED, False, False)
         assert "not readable by you" in warnings[0]
 
     def test_no_config_on_the_server_does_not_use_a_stale_persisted_file(self, monkeypatch):
@@ -527,7 +531,7 @@ def test_no_config_on_the_server_does_not_use_a_stale_persisted_file(self, monke
         monkeypatch.setattr(
             mc_mod, "load_managed_state", lambda ws: pytest.fail("must not fall back")
         )
-        result, _ = refresh_managed_config(_state())
+        result, _, _ = refresh_managed_config(_state())
         assert result is None
 
     def test_no_config_on_the_server_clears_the_persisted_one(self, monkeypatch):
@@ -539,7 +543,7 @@ def test_no_config_on_the_server_clears_the_persisted_one(self, monkeypatch):
             mc_mod, "save_managed_state", lambda ws, cfg, **kwargs: saved.append((ws, cfg))
         )
         monkeypatch.setattr(mc_mod, "load_managed_state", lambda ws: None)
-        result, _ = refresh_managed_config(_state())
+        result, _, _ = refresh_managed_config(_state())
         assert result is None
         assert saved == [(WORKSPACE, {})]
 
@@ -551,14 +555,14 @@ def test_empty_persisted_config_is_not_treated_as_a_fallback(self, monkeypatch):
         monkeypatch.setattr(
             mc_mod, "print_warning", lambda msg: pytest.fail(f"should not warn: {msg}")
         )
-        result, _ = refresh_managed_config(_state())
+        result, _, _ = refresh_managed_config(_state())
         assert result is None
 
     def test_no_workspace_is_a_noop(self, monkeypatch):
         monkeypatch.setattr(
             mc_mod, "get_managed_config", lambda ws, tok: pytest.fail("should not fetch")
         )
-        result, _ = refresh_managed_config({})
+        result, _, _ = refresh_managed_config({})
         assert result is None
 
     def test_feature_disabled_sets_flag_when_there_is_no_fallback(self, monkeypatch):
@@ -569,9 +573,10 @@ def test_feature_disabled_sets_flag_when_there_is_no_fallback(self, monkeypatch)
         monkeypatch.setattr(mc_mod, "load_managed_state", lambda ws: None)
         monkeypatch.setattr(mc_mod, "print_warning", lambda msg: None)
         state = _state()
-        result, flag = refresh_managed_config(state)
+        result, flag, definitively_absent = refresh_managed_config(state)
         assert result is None
         assert flag is True
+        assert definitively_absent is True
 
     def test_feature_disabled_ignores_a_cached_config_and_sets_the_flag(self, monkeypatch):
         # FEATURE_DISABLED is authoritative, so a config cached while the feature was enabled no
@@ -590,27 +595,30 @@ def test_feature_disabled_ignores_a_cached_config_and_sets_the_flag(self, monkey
             lambda msg: pytest.fail("feature-disabled must not warn about falling back to a cache"),
         )
         state = _state()
-        result, flag = refresh_managed_config(state)
+        result, flag, definitively_absent = refresh_managed_config(state)
         assert result is None
         assert flag is True
+        assert definitively_absent is True
         assert saved == [(WORKSPACE, {})]
 
     def test_transient_failure_does_not_set_the_flag(self, monkeypatch):
         monkeypatch.setattr(mc_mod, "get_managed_config", lambda ws, tok: (None, "HTTP 500"))
         monkeypatch.setattr(mc_mod, "load_managed_state", lambda ws: None)
         monkeypatch.setattr(mc_mod, "print_warning", lambda msg: None)
         state = _state()
-        result, flag = refresh_managed_config(state)
+        result, flag, definitively_absent = refresh_managed_config(state)
         assert result is None
         assert flag is False
+        assert definitively_absent is False
 
     def test_successful_no_config_clears_the_flag(self, monkeypatch):
         monkeypatch.setattr(mc_mod, "get_managed_config", lambda ws, tok: (None, None))
         monkeypatch.setattr(mc_mod, "save_managed_state", lambda ws, cfg, **kwargs: None)
         state = _state()
-        result, flag = refresh_managed_config(state)
+        result, flag, definitively_absent = refresh_managed_config(state)
         assert result is None
         assert flag is False
+        assert definitively_absent is True
 
 
 class TestRefreshAlwaysFetches:
@@ -640,9 +648,10 @@ def test_fetches_even_with_a_persisted_config(self, monkeypatch):
         # A previously-persisted config no longer short-circuits: every launch re-reads the workspace.
         save_managed_state(WORKSPACE, RAW_MANIFEST)
         calls = self._counting_fetch(monkeypatch)
-        result, flag = refresh_managed_config(_state())
+        result, flag, definitively_absent = refresh_managed_config(_state())
         assert result == normalize_managed_config(RAW_MANIFEST)
         assert flag is False
+        assert definitively_absent is False
         assert calls["n"] == 1
 
     def test_persists_the_fetched_config_raw_without_a_timestamp(self, monkeypatch):
@@ -790,3 +799,41 @@ def test_unparseable_decimals_become_none(self, monkeypatch):
         )
         rec, _ = mc_mod.get_model_recommendation("https://w", "tok")
         assert rec is not None and rec["current_spend"] is None
+
+
+class TestDefinitivelyAbsent:
+    """Test the definitively_absent field distinguishes definitive absence from transient failures."""
+
+    @pytest.fixture(autouse=True)
+    def _stub_token(self, monkeypatch):
+        monkeypatch.setattr(mc_mod, "get_databricks_token", lambda ws, profile: "tok")
+
+    def test_transient_failure_with_no_cache_is_not_definitive(self, monkeypatch):
+        # A transient fetch failure with no cache falls back to None, but it's not definitive.
+        monkeypatch.setattr(mc_mod, "get_managed_config", lambda ws, tok: (None, "HTTP 500"))
+        monkeypatch.setattr(mc_mod, "load_managed_state", lambda ws: None)
+        monkeypatch.setattr(mc_mod, "print_warning", lambda msg: None)
+        result, flag, definitively_absent = refresh_managed_config(_state())
+        assert result is None
+        assert flag is False
+        assert definitively_absent is False
+
+    def test_not_found_is_definitive(self, monkeypatch):
+        # A successful NOT_FOUND from the API is definitive.
+        monkeypatch.setattr(mc_mod, "get_managed_config", lambda ws, tok: (None, None))
+        monkeypatch.setattr(mc_mod, "save_managed_state", lambda ws, cfg, **kwargs: None)
+        result, flag, definitively_absent = refresh_managed_config(_state())
+        assert result is None
+        assert flag is False
+        assert definitively_absent is True
+
+    def test_feature_disabled_is_definitive(self, monkeypatch):
+        # FEATURE_DISABLED is authoritative and definitive.
+        reason = 'HTTP 400 Bad Request: {"error_code":"FEATURE_DISABLED"}'
+        monkeypatch.setattr(mc_mod, "get_managed_config", lambda ws, tok: (None, reason))
+        monkeypatch.setattr(mc_mod, "load_managed_state", lambda ws: None)
+        monkeypatch.setattr(mc_mod, "print_warning", lambda msg: None)
+        result, flag, definitively_absent = refresh_managed_config(_state())
+        assert result is None
+        assert flag is True
+        assert definitively_absent is True