diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py index 40813892..91608e46 100644 --- a/src/ucode/agents/__init__.py +++ b/src/ucode/agents/__init__.py @@ -85,16 +85,22 @@ } -def install_databricks_ai_tools_for_agents(tools: list[str], state: dict) -> None: +def install_databricks_ai_tools_for_agents( + tools: list[str], state: dict, *, force_refresh: bool = False +) -> None: """Install Databricks AI Tools for supported agents. Gemini and Pi have no ``aitools`` support and are dropped. + + This runs only during ``ug configure``. ``force_refresh`` reads the managed config fresh; a + caller that already refreshed this launch (the main configure path) leaves it False so the gate + reuses that read instead of adding another control-plane round trip. """ if not state.get("databricks_ai_tools_enabled"): return # An admin's managed config governs the workspace, so ucode does not # self-install AI Tools under one (may become a managed-config option later). - if refresh_managed_config(state).manifest is not None: + if refresh_managed_config(state, force_refresh=force_refresh).manifest is not None: return agents = [AITOOLS_AGENT_TOKENS[tool] for tool in tools if tool in AITOOLS_AGENT_TOKENS] if not agents: diff --git a/src/ucode/cli.py b/src/ucode/cli.py index 933df438..592c2e74 100644 --- a/src/ucode/cli.py +++ b/src/ucode/cli.py @@ -780,7 +780,9 @@ def configure_workspace_command( ) state = states[0] state = configure_single_tool(tool, state) - install_databricks_ai_tools_for_agents([tool], state) + # No managed refresh precedes this branch, so read fresh here: `ug configure` never decides + # from a stale cache. + install_databricks_ai_tools_for_agents([tool], state, force_refresh=True) spec = TOOL_SPECS[tool] console.print( Panel( @@ -807,8 +809,9 @@ def configure_workspace_command( save_state(state) # A published managed config means the admin dictates the setup: apply it to every enabled agent - # now rather than prompting the developer to pick. - managed, _ = refresh_managed_config(state) + # now rather than prompting the developer to pick. Configure always reads fresh so it never + # applies a config the admin has since changed. + managed, _ = refresh_managed_config(state, force_refresh=True) if managed is not None: _announce_managed_config(managed) for tool_name in managed_enabled_tools(managed): diff --git a/src/ucode/config_io.py b/src/ucode/config_io.py index 3444abd9..4e941993 100644 --- a/src/ucode/config_io.py +++ b/src/ucode/config_io.py @@ -170,7 +170,7 @@ def read_json_safe(path: Path) -> dict: if not path.exists(): return {} data = json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): + except (OSError, UnicodeError, json.JSONDecodeError): return {} return data if isinstance(data, dict) else {} diff --git a/src/ucode/managed_config.py b/src/ucode/managed_config.py index 368a7dcd..5b793f37 100644 --- a/src/ucode/managed_config.py +++ b/src/ucode/managed_config.py @@ -7,8 +7,9 @@ - fetching the raw manifest (via :func:`ucode.databricks.fetch_managed_coding_agent_configs`), - normalizing the proto-JSON into a stable internal dict keyed by ucode's own tool names, - persisting it via :func:`save_managed_state` / :func:`load_managed_state` — the launch path pulls - the published copy into this file, and -- re-reading it on each launch, falling back to the persisted copy when the read fails. + the published copy into this file, stamped with a ``retrieved_at`` and its outcome, and +- re-reading it on each launch (reusing a read younger than :data:`MANAGED_CONFIG_TTL` rather than + re-fetching), falling back to the persisted copy when the read fails. There is deliberately one file: the workspace is the source of truth, so the pulled copy lives in ``managed-config.json`` and a launch re-reads it from there. @@ -26,7 +27,7 @@ import os import re from dataclasses import dataclass -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from pathlib import Path from typing import NamedTuple, cast @@ -64,6 +65,17 @@ MAX_SPEC_VERSION = 1 +# Launch-path cache TTL. `ug` / `ug ` reuse a managed-config read younger than this instead of +# hitting the control plane every launch; `ug configure` forces a fresh read (force_refresh=True). +MANAGED_CONFIG_TTL = timedelta(minutes=5) + +# The last authoritative read's outcome, persisted alongside the config so a cached wrapper can be +# replayed without a GET. "none" and "feature_disabled" both persist an empty config, so the outcome +# is what tells them apart. +_OUTCOME_PUBLISHED = "published" +_OUTCOME_NONE = "none" +_OUTCOME_FEATURE_DISABLED = "feature_disabled" + # A per-family default-model key in the `default_models` map, e.g. `default_opus_model`. Matches the # server's `default_.+_model` validation so a new Claude family needs no ucode change. The bare # `default_model` overall default does not match (no family segment) and is read on its own. @@ -604,7 +616,12 @@ def _is_unsupported_spec(reason: str) -> bool: return "spec_version" in reason.lower() -def save_managed_state(workspace: str, config: dict) -> None: +def _utcnow() -> datetime: + """Current UTC time. A seam so the launch-path cache TTL can be exercised deterministically.""" + return datetime.now(UTC) + + +def save_managed_state(workspace: str, config: dict, *, outcome: str | None = None) -> None: """Persist the raw managed config to ``~/.ucode/managed-config.json`` at mode 0600. ``config`` is stored verbatim as the gateway returned it (byte-identical to the GET), so the file @@ -615,8 +632,15 @@ def save_managed_state(workspace: str, config: dict) -> None: An empty ``config`` records "this workspace has no managed config", which matters because the file doubles as the fallback when a later read fails: without it, removing a config server-side would leave the old one on disk to be reapplied after a transient outage. + + ``outcome``, when set, stamps the read time and its result (published / none / feature_disabled) + so a later launch can reuse this read within :data:`MANAGED_CONFIG_TTL` without a GET. Only an + authoritative read passes it; a failed refresh persists nothing and so never advances the stamp. """ payload: dict = {"workspace": workspace, "config": config} + if outcome is not None: + payload["retrieved_at"] = _utcnow().isoformat() + payload["outcome"] = outcome if config_io.is_dry_run(): # Print rather than write, matching how the agent config writers behave under --dry-run. console.print( @@ -681,15 +705,44 @@ def managed_state_workspace() -> str | None: return workspace if isinstance(workspace, str) and workspace else None -def refresh_managed_config(state: dict) -> ManagedConfigResult: - """Fetch the workspace's managed config fresh and persist it as a :class:`ManagedConfigResult`. +def _cached_result_if_fresh(workspace: str) -> ManagedConfigResult | None: + """The persisted read for ``workspace`` replayed as a result, if still within the TTL. + + Returns None (forcing a fresh fetch) when the wrapper is for another workspace, predates this + cache format (no ``outcome`` / ``retrieved_at``), or its stamp is missing, unparseable, in the + future, or at least :data:`MANAGED_CONFIG_TTL` old. + """ + data = config_io.read_json_safe(MANAGED_CONFIG_PATH) + if data.get("workspace") != workspace: + return None + # Reuses the RFC-3339 parser the update-time watermark uses; None (missing/unparseable) is stale. + retrieved_at = _parse_update_time(_str(data.get("retrieved_at"))) + if retrieved_at is None: + return None + age = _utcnow() - retrieved_at + if age < timedelta(0) or age >= MANAGED_CONFIG_TTL: + return None + outcome = data.get("outcome") + if outcome == _OUTCOME_FEATURE_DISABLED: + return ManagedConfigResult(None, True) + if outcome == _OUTCOME_NONE: + return ManagedConfigResult(None, False) + if outcome == _OUTCOME_PUBLISHED and isinstance(data.get("config"), dict): + return ManagedConfigResult(normalize_managed_config(data["config"]), False) + return None + + +def refresh_managed_config(state: dict, *, force_refresh: bool = False) -> ManagedConfigResult: + """Fetch the workspace's managed config and persist it as a :class:`ManagedConfigResult`. Runs on every launch so a developer picks up an admin's edits without re-running - ``ucode configure``. It always hits the control plane; whether the fetched config is *newer* than - what was last applied — and so whether the launch re-applies the settings — is the caller's - decision, via :func:`managed_config_is_newer` against the persisted applied watermark. The - manifest is None when the workspace has no managed config — the normal case for a workspace whose - admin hasn't published one. + ``ucode configure``. A launch reuses the last read when it is younger than + :data:`MANAGED_CONFIG_TTL`, so back-to-back launches don't each hit the control plane; + ``force_refresh`` (used by ``ug configure``) skips the cache and always reads fresh. Whether the + fetched config is *newer* than what was last applied, and so whether the launch re-applies the + settings, is the caller's decision, via :func:`managed_config_is_newer` against the persisted + applied watermark. The manifest is None when the workspace has no managed config, the normal + case for a workspace whose admin hasn't published one. A failed fetch never blocks the launch: an unreachable control plane shouldn't stop someone from coding. Instead it falls back to the last config persisted for this workspace, so the admin's @@ -707,6 +760,10 @@ def refresh_managed_config(state: dict) -> ManagedConfigResult: workspace = state.get("workspace") if not workspace: return ManagedConfigResult(None, False) + if not force_refresh: + cached = _cached_result_if_fresh(workspace) + if cached is not None: + return cached try: token = get_databricks_token(workspace, state.get("profile")) except RuntimeError as exc: @@ -714,7 +771,7 @@ def refresh_managed_config(state: dict) -> ManagedConfigResult: raw, reason = get_managed_config(workspace, token) if reason is not None: if _is_feature_disabled(reason): - save_managed_state(workspace, {}) + save_managed_state(workspace, {}, outcome=_OUTCOME_FEATURE_DISABLED) return ManagedConfigResult(None, True) fallback = _persisted_fallback(workspace, reason, refused=_is_permission_denied(reason)) return ManagedConfigResult(fallback, False) @@ -722,10 +779,10 @@ def refresh_managed_config(state: dict) -> ManagedConfigResult: # 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, {}) + save_managed_state(workspace, {}, outcome=_OUTCOME_NONE) return ManagedConfigResult(None, False) # Persist the raw config verbatim; hand callers the normalized manifest they expect. - save_managed_state(workspace, raw) + save_managed_state(workspace, raw, outcome=_OUTCOME_PUBLISHED) return ManagedConfigResult(normalize_managed_config(raw), False) diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py index 8c85d429..6f3abbe7 100644 --- a/tests/test_agents_init.py +++ b/tests/test_agents_init.py @@ -85,7 +85,9 @@ def _capture(self, monkeypatch, *, managed=None): lambda agents, profile: captured.update(agents=agents, profile=profile), ) monkeypatch.setattr( - agents_mod, "refresh_managed_config", lambda state: ManagedConfigResult(managed, False) + agents_mod, + "refresh_managed_config", + lambda state, **_k: ManagedConfigResult(managed, False), ) return captured @@ -135,6 +137,23 @@ def test_skipped_under_empty_managed_config(self, monkeypatch): ) assert captured == {} # install_ai_tools never called + def test_forwards_force_refresh_to_managed_read(self, monkeypatch): + # The gate forwards force_refresh so `ug configure --agent` (no prior refresh) reads fresh, + # while the main configure path (already refreshed) reuses its read instead of re-fetching. + seen: list[bool] = [] + monkeypatch.setattr(agents_mod, "install_ai_tools", lambda agents, profile: None) + monkeypatch.setattr( + agents_mod, + "refresh_managed_config", + lambda state, *, force_refresh=False: ( + seen.append(force_refresh) or ManagedConfigResult(None, False) + ), + ) + state = {"profile": "p", "databricks_ai_tools_enabled": True} + install_databricks_ai_tools_for_agents(["claude"], state) + install_databricks_ai_tools_for_agents(["claude"], state, force_refresh=True) + assert seen == [False, True] + class TestConfigureWiresAiToolsInstall: """AI Tools install is a `ucode configure`-only step. `configure_selected_tools` @@ -151,7 +170,9 @@ def _stub_configure(self, monkeypatch): lambda agents, profile: captured.update(agents=agents, profile=profile), ) monkeypatch.setattr( - agents_mod, "refresh_managed_config", lambda state: ManagedConfigResult(None, False) + agents_mod, + "refresh_managed_config", + lambda state, **_k: ManagedConfigResult(None, False), ) return captured diff --git a/tests/test_cli.py b/tests/test_cli.py index bff007c9..d4c5ae38 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2651,7 +2651,7 @@ def _no_managed_config(self, monkeypatch): # `ug configure` now fetches the managed config, which shells out to the `databricks` CLI. # Default it to absent so these personal-flow tests never hit the CLI (it isn't on CI); # the managed-branch test overrides this. - monkeypatch.setattr(cli_mod, "refresh_managed_config", lambda state: (None, False)) + monkeypatch.setattr(cli_mod, "refresh_managed_config", lambda state, **_k: (None, False)) @pytest.mark.parametrize(("keys", "expected"), [(" \r", ["codex"]), ("\r", [])]) def test_interactive_picker_installs_only_checked_agents(self, monkeypatch, keys, expected): @@ -2763,7 +2763,7 @@ def test_managed_config_applies_all_enabled_and_skips_selection(self, monkeypatc monkeypatch.setattr( cli_mod, "refresh_managed_config", - lambda s: ({"enabled_agents": {"claude": {}, "codex": {}}}, False), + lambda s, **_k: ({"enabled_agents": {"claude": {}, "codex": {}}}, False), ) monkeypatch.setattr(cli_mod, "check_gateway_endpoint", lambda s, t: True) monkeypatch.setattr(cli_mod, "resolve_state", lambda managed, s, tool: s) @@ -2794,7 +2794,7 @@ def test_managed_config_registers_mcp_servers_after_configuring_agents(self, mon "enabled_agents": {"claude": {}, "codex": {}}, "mcp_servers": {"names": ["x.y.z"]}, } - monkeypatch.setattr(cli_mod, "refresh_managed_config", lambda s: (managed, False)) + monkeypatch.setattr(cli_mod, "refresh_managed_config", lambda s, **_k: (managed, False)) monkeypatch.setattr(cli_mod, "check_gateway_endpoint", lambda s, t: True) monkeypatch.setattr(cli_mod, "resolve_state", lambda m, s, tool: s) monkeypatch.setattr(cli_mod, "_print_managed_summary", lambda *a, **k: None) @@ -2826,7 +2826,7 @@ def test_managed_configure_accumulates_available_tools_for_all_agents(self, monk monkeypatch.setattr( cli_mod, "refresh_managed_config", - lambda s: ({"enabled_agents": {"claude": {}, "codex": {}}}, False), + lambda s, **_k: ({"enabled_agents": {"claude": {}, "codex": {}}}, False), ) monkeypatch.setattr(cli_mod, "check_gateway_endpoint", lambda s, t: True) # Mirror production: resolve_state hands each iteration a fresh copy of `state`. @@ -2890,7 +2890,7 @@ def test_unmanaged_workspace_reconciles_managed_mcp_servers(self, monkeypatch): state = {**MINIMAL_STATE, "available_tools": []} monkeypatch.setattr(cli_mod, "configure_shared_state", lambda *a, **k: state) - monkeypatch.setattr(cli_mod, "refresh_managed_config", lambda s: (None, False)) + monkeypatch.setattr(cli_mod, "refresh_managed_config", lambda s, **_k: (None, False)) 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( @@ -3590,7 +3590,7 @@ def test_falls_back_to_legacy_when_uc_empty(self, monkeypatch): class TestConfigureNoLongerValidates: @pytest.fixture(autouse=True) def _no_managed_config(self, monkeypatch): - monkeypatch.setattr(cli_mod, "refresh_managed_config", lambda state: (None, False)) + monkeypatch.setattr(cli_mod, "refresh_managed_config", lambda state, **_k: (None, False)) def test_configure_completes_without_probe(self, monkeypatch): import ucode.cli as cli_mod diff --git a/tests/test_config_io.py b/tests/test_config_io.py index 9ee1c58d..1cb325ee 100644 --- a/tests/test_config_io.py +++ b/tests/test_config_io.py @@ -232,6 +232,13 @@ def test_read_json_safe_non_dict(self, tmp_path): p.write_text("[1, 2, 3]", encoding="utf-8") assert read_json_safe(p) == {} + def test_read_json_safe_non_utf8(self, tmp_path): + # A file with non-UTF-8 bytes must read as absent rather than raising, so a corrupted + # cache file can't crash a launch that reads it. + p = tmp_path / "binary.json" + p.write_bytes(b"\xff\xfe\x00not utf-8") + assert read_json_safe(p) == {} + def test_read_toml_safe_missing_file(self, tmp_path): doc = read_toml_safe(tmp_path / "missing.toml") assert dict(doc) == {} diff --git a/tests/test_managed_config.py b/tests/test_managed_config.py index 02593889..cd4e26c1 100644 --- a/tests/test_managed_config.py +++ b/tests/test_managed_config.py @@ -5,6 +5,7 @@ import json import os import stat +from datetime import UTC, datetime, timedelta import pytest @@ -366,6 +367,19 @@ def test_workspace_is_stored_alongside_the_config(self, _managed_path): def test_workspace_is_none_when_absent(self, _managed_path): assert managed_state_workspace() is None + def test_outcome_stamps_retrieved_at_and_outcome(self, _managed_path, monkeypatch): + # A read's outcome carries a retrieved_at stamp so the launch path can reuse it within the + # TTL; without an outcome the wrapper stays the bare {workspace, config} shape. + monkeypatch.setattr(mc_mod, "_utcnow", lambda: NOW) + save_managed_state("https://ws.example.com", RAW_MANIFEST, outcome="published") + stored = json.loads(_managed_path.read_text(encoding="utf-8")) + assert stored["outcome"] == "published" + assert stored["retrieved_at"] == NOW.isoformat() + save_managed_state("https://ws.example.com", {"default_agent": "claude"}) + bare = json.loads(_managed_path.read_text(encoding="utf-8")) + assert "retrieved_at" not in bare + assert "outcome" not in bare + def test_dry_run_writes_nothing(self, _managed_path, monkeypatch): # Under --dry-run the config writers print instead of touching disk, so a launch that # dry-runs an admin's authored draft never overwrites it. @@ -639,17 +653,17 @@ def test_successful_no_config_clears_the_flag(self, monkeypatch): assert flag is False -class TestRefreshAlwaysFetches: - """The launch-time refresh always hits the control plane; the 30-minute TTL is gone. +NOW = datetime(2026, 9, 16, 12, 0, 0, tzinfo=UTC) + - Whether to re-apply the fetched config is decided separately by the caller via - ``managed_config_is_newer`` against the persisted applied watermark, so refresh never short- - circuits on a cached copy. - """ +class TestRefreshTTL: + """A launch reuses a read younger than ``MANAGED_CONFIG_TTL``; ``ug configure`` (force_refresh) + and a stale/absent/foreign cache re-read the control plane.""" @pytest.fixture(autouse=True) - def _stub_token(self, monkeypatch): + def _stub_token_and_clock(self, monkeypatch): monkeypatch.setattr(mc_mod, "get_databricks_token", lambda ws, profile: "tok") + monkeypatch.setattr(mc_mod, "_utcnow", lambda: NOW) @staticmethod def _counting_fetch(monkeypatch, result=(RAW_MANIFEST, None)): @@ -662,28 +676,159 @@ def fetch(ws, tok): monkeypatch.setattr(mc_mod, "get_managed_config", fetch) return calls - def test_fetches_even_with_a_persisted_config(self, monkeypatch): - # A previously-persisted config no longer short-circuits: every launch re-reads the workspace. + @staticmethod + def _write_cache(*, config, outcome, retrieved_at, workspace=WORKSPACE): + payload = { + "workspace": workspace, + "config": config, + "outcome": outcome, + "retrieved_at": retrieved_at.isoformat(), + } + mc_mod.MANAGED_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) + mc_mod.MANAGED_CONFIG_PATH.write_text(json.dumps(payload), encoding="utf-8") + + @staticmethod + def _no_fetch(monkeypatch): + monkeypatch.setattr( + mc_mod, + "get_managed_config", + lambda ws, tok: pytest.fail("fresh cache must not hit the control plane"), + ) + + def test_fresh_published_cache_short_circuits(self, monkeypatch): + self._write_cache( + config=RAW_MANIFEST, outcome="published", retrieved_at=NOW - timedelta(minutes=1) + ) + self._no_fetch(monkeypatch) + assert refresh_managed_config(_state()) == (normalize_managed_config(RAW_MANIFEST), False) + + def test_fresh_no_config_cache_short_circuits(self, monkeypatch): + self._write_cache(config={}, outcome="none", retrieved_at=NOW - timedelta(minutes=1)) + self._no_fetch(monkeypatch) + assert refresh_managed_config(_state()) == (None, False) + + def test_fresh_feature_disabled_cache_short_circuits(self, monkeypatch): + self._write_cache( + config={}, outcome="feature_disabled", retrieved_at=NOW - timedelta(minutes=1) + ) + self._no_fetch(monkeypatch) + assert refresh_managed_config(_state()) == (None, True) + + def test_stale_cache_refetches(self, monkeypatch): + # A read at or past the TTL is stale: re-read rather than reuse it. + self._write_cache( + config=RAW_MANIFEST, outcome="published", retrieved_at=NOW - timedelta(minutes=5) + ) + calls = self._counting_fetch(monkeypatch) + refresh_managed_config(_state()) + assert calls["n"] == 1 + + def test_future_dated_cache_refetches(self, monkeypatch): + # A future stamp (clock skew or a tampered file) is not trusted as fresh. + self._write_cache( + config=RAW_MANIFEST, outcome="published", retrieved_at=NOW + timedelta(minutes=1) + ) + calls = self._counting_fetch(monkeypatch) + refresh_managed_config(_state()) + assert calls["n"] == 1 + + def test_missing_timestamp_refetches(self, monkeypatch): + # An old-format file (no retrieved_at) is never treated as a fresh cache. save_managed_state(WORKSPACE, RAW_MANIFEST) calls = self._counting_fetch(monkeypatch) - result, flag = refresh_managed_config(_state()) - assert result == normalize_managed_config(RAW_MANIFEST) - assert flag is False + refresh_managed_config(_state()) assert calls["n"] == 1 - def test_persists_the_fetched_config_raw_without_a_timestamp(self, monkeypatch): - self._counting_fetch(monkeypatch) + def test_invalid_timestamp_refetches(self, monkeypatch): + mc_mod.MANAGED_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) + mc_mod.MANAGED_CONFIG_PATH.write_text( + json.dumps( + { + "workspace": WORKSPACE, + "config": RAW_MANIFEST, + "outcome": "published", + "retrieved_at": "not-a-timestamp", + } + ), + encoding="utf-8", + ) + calls = self._counting_fetch(monkeypatch) refresh_managed_config(_state()) - # The on-disk payload is the raw config verbatim and carries no retrieved_at field. - stored = json.loads(mc_mod.MANAGED_CONFIG_PATH.read_text(encoding="utf-8")) - assert "retrieved_at" not in stored - assert stored["config"] == RAW_MANIFEST + assert calls["n"] == 1 + + def test_other_workspace_cache_refetches(self, monkeypatch): + self._write_cache( + config=RAW_MANIFEST, + outcome="published", + retrieved_at=NOW - timedelta(minutes=1), + workspace="https://other.example.com", + ) + calls = self._counting_fetch(monkeypatch) + refresh_managed_config(_state()) + assert calls["n"] == 1 + + def test_force_refresh_bypasses_a_fresh_cache(self, monkeypatch): + # `ug configure` passes force_refresh=True so it never applies a since-changed config. + self._write_cache( + config=RAW_MANIFEST, outcome="published", retrieved_at=NOW - timedelta(minutes=1) + ) + calls = self._counting_fetch(monkeypatch) + refresh_managed_config(_state(), force_refresh=True) + assert calls["n"] == 1 def test_first_launch_with_no_cache_fetches(self, monkeypatch): calls = self._counting_fetch(monkeypatch) refresh_managed_config(_state()) assert calls["n"] == 1 + def test_non_utf8_cache_file_refetches(self, monkeypatch): + # A corrupted (non-UTF-8) file read on every launch must not crash: it reads as absent, so + # the launch falls through to a fresh fetch rather than raising UnicodeDecodeError. + mc_mod.MANAGED_CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True) + mc_mod.MANAGED_CONFIG_PATH.write_bytes(b"\xff\xfe not utf-8") + calls = self._counting_fetch(monkeypatch) + refresh_managed_config(_state()) + assert calls["n"] == 1 + + def test_successful_read_stamps_retrieved_at_and_outcome(self, monkeypatch): + self._counting_fetch(monkeypatch) + refresh_managed_config(_state()) + stored = json.loads(mc_mod.MANAGED_CONFIG_PATH.read_text(encoding="utf-8")) + assert stored["config"] == RAW_MANIFEST + assert stored["outcome"] == "published" + assert stored["retrieved_at"] == NOW.isoformat() + + def test_no_config_read_caches_none_outcome(self, monkeypatch): + self._counting_fetch(monkeypatch, result=(None, None)) + refresh_managed_config(_state()) + stored = json.loads(mc_mod.MANAGED_CONFIG_PATH.read_text(encoding="utf-8")) + assert stored["config"] == {} + assert stored["outcome"] == "none" + assert stored["retrieved_at"] == NOW.isoformat() + + def test_feature_disabled_read_caches_feature_disabled_outcome(self, monkeypatch): + reason = 'HTTP 400 Bad Request: {"error_code":"FEATURE_DISABLED"}' + self._counting_fetch(monkeypatch, result=(None, reason)) + monkeypatch.setattr(mc_mod, "print_warning", lambda msg: None) + refresh_managed_config(_state()) + stored = json.loads(mc_mod.MANAGED_CONFIG_PATH.read_text(encoding="utf-8")) + assert stored["config"] == {} + assert stored["outcome"] == "feature_disabled" + assert stored["retrieved_at"] == NOW.isoformat() + + def test_failed_read_does_not_advance_retrieved_at(self, monkeypatch): + # A transient failure falls back to the last good config and must leave the stamp untouched, + # so the stale read cannot masquerade as fresh on the next launch. + stamped = NOW - timedelta(minutes=10) + self._write_cache(config=RAW_MANIFEST, outcome="published", retrieved_at=stamped) + self._counting_fetch(monkeypatch, result=(None, "HTTP 500")) + monkeypatch.setattr(mc_mod, "print_warning", lambda msg: None) + result, flag = refresh_managed_config(_state()) + assert result == normalize_managed_config(RAW_MANIFEST) + assert flag is False + stored = json.loads(mc_mod.MANAGED_CONFIG_PATH.read_text(encoding="utf-8")) + assert stored["retrieved_at"] == stamped.isoformat() + class TestManagedUpdateTime: def test_reads_top_level_update_time(self):