fix(config-remote): review round 9b: guard reads keep the caller's read base, dashboard model saves mask inherited route keys, MCP ack checks every toggle field

1. _write_config_state (atomic_config_write: /verbose, /footer, ...) runs its no-omission guard
   read under keeping_read_base, like save_config, so a poll between the caller's read and its
   save no longer becomes the base its stale document is diffed against.
2. Dashboard / profile model assignments (_apply_main_model_assignment: POST /api/model/set,
   profile builder, custom endpoints) share persist_model_selection's semantics under Remote
   Config: apply_model_selection(mask_cleared=True) stores null for a cleared key the effective
   block holds, and remote_model_selection_guard refuses the whole assignment when any changed
   model.* key is locked (a bulk save would otherwise omit it and land a half route).
   persist_model_selection now builds its keyed write from the same _merge_selection_updates.
3. profiles.configure MCP acknowledgement compares enabled AND the legacy disabled field per
   server, so a lock that kept disabled: true reports mcp_servers unapplied.

Regressions (each fails with the production change reverted):
test_poll_between_read_and_atomic_config_write_sends_only_the_callers_edit,
test_dashboard_model_switch_masks_inherited_endpoint_credentials,
test_dashboard_model_switch_refuses_whole_assignment_on_a_locked_route_key,
test_profile_editor_reports_mcp_unapplied_when_legacy_disabled_cleanup_is_locked.

hermes_cli/config.py

@@ -1972,7 +1972,10 @@ def _write_config_state(
     """Shared comment-preserving config writer; omission policy is selected by the public wrapper."""
     _refuse_failed_read(config_path, data)
     if not allow_omissions:
-        existing = require_readable_config_before_write(config_path)
+        # A guard read, not the caller's read: *data* came from the caller's own earlier read, so a
+        # Remote Config save must still be diffed against THAT read, not a poll that landed since.
+        from hermes_cli.config_backend import keeping_read_base
+        existing = keeping_read_base(lambda: require_readable_config_before_write(config_path))
         omitted = _omitted_config_paths(existing, data)
         if omitted:
             shown = ", ".join(omitted[:12])

hermes_cli/model_switch.py

@@ -1829,22 +1829,52 @@ def selection_route_changed(model_cfg: dict, result: ModelSwitchResult) -> bool:
     return normalize_route_base_url(model_cfg.get("base_url")) != normalize_route_base_url(result.base_url)
 
 
-def apply_model_selection(model_cfg: Any, result: ModelSwitchResult) -> dict:
+def apply_model_selection(model_cfg: Any, result: ModelSwitchResult, *, mask_cleared: bool = False) -> dict:
     """Apply the canonical shape to an in-memory ``model:`` dict (``None`` = key removed) for
-    callers that save a whole config document they are already mutating."""
-    return _merge_selection_updates(model_cfg, model_selection_config_updates(result, model_cfg))
+    callers that save a whole config document they are already mutating.
+
+    ``mask_cleared``: a key the switch clears that *model_cfg* holds is set to ``None`` instead of
+    removed. For a document saved to Remote Config, where *model_cfg* is the EFFECTIVE block and
+    a removal only drops the profile override, so an upper level's old-endpoint ``api_key`` /
+    ``key_env`` / ``context_length`` would still apply at the new endpoint (contract §5.4: a
+    stored null is "no value"). The same rule :func:`persist_model_selection` applies."""
+    return _merge_selection_updates(model_cfg, model_selection_config_updates(result, model_cfg),
+                                    mask_cleared=mask_cleared)
 
 
-def _merge_selection_updates(model_cfg: Any, updates: dict[str, Any]) -> dict:
+def _merge_selection_updates(model_cfg: Any, updates: dict[str, Any], *, mask_cleared: bool = False) -> dict:
     model_cfg = dict(model_cfg) if isinstance(model_cfg, dict) else {}
     for key, value in updates.items():
-        if value is None:
+        if value is None and not (mask_cleared and key in model_cfg):
             model_cfg.pop(key, None)
         else:
             model_cfg[key] = value
     return model_cfg
 
 
+def remote_model_selection_guard(before: Any, after: Any) -> None:
+    """Remote Config: refuse a whole-document model assignment when any ``model.*`` key it
+    changes is locked by an upper level. A bulk save omits a locked key instead of raising, which
+    would land the rest of the route (a new endpoint with the old provider, say) half-applied;
+    :func:`persist_model_selection` gets the same all-or-nothing refusal from its one keyed write.
+    No-op for the file backend."""
+    from hermes_cli.config_backend import ConfigLockedError, get_config_backend, supports_file_tooling
+    if supports_file_tooling():
+        return
+    from hermes_constants import get_hermes_home
+    before = before if isinstance(before, dict) else {}
+    after = after if isinstance(after, dict) else {}
+    backend, home = get_config_backend(), get_hermes_home()
+    for key in sorted(set(before) | set(after)):
+        if before.get(key, _MISSING) != after.get(key, _MISSING):
+            level = backend.locked(home, f"model.{key}")
+            if level:
+                raise ConfigLockedError(f"model.{key}", level)
+
+
+_MISSING = object()
+
+
 def persist_model_selection(result: ModelSwitchResult, config_path: Any = None) -> None:
     """Write a successful :func:`switch_model` result to ``config_path`` (default:
     ``HERMES_HOME/config.yaml`` — the context override or ``HERMES_HOME`` at call time).
@@ -1874,9 +1904,10 @@ def persist_model_selection(result: ModelSwitchResult, config_path: Any = None)
         # still holds is therefore masked with null ("no value"); an absent one is just unset.
         from hermes_cli.config_backend import Changes, write_changes
         held = planned_from if isinstance(planned_from, dict) else {}
+        masked = _merge_selection_updates(held, updates, mask_cleared=True)
         write_changes(path, Changes(
-            set={f"model.{k}": v for k, v in updates.items() if v is not None or k in held},
-            unset=tuple(f"model.{k}" for k, v in updates.items() if v is None and k not in held)))
+            set={f"model.{k}": masked[k] for k in updates if k in masked},
+            unset=tuple(f"model.{k}" for k in updates if k not in masked)))
         return
     with _CONFIG_LOCK:
         config = read_user_config_raw(path)

hermes_cli/web_server_config.py

@@ -517,12 +517,17 @@ def _apply_main_model_assignment(model_cfg: Any, result: "ModelSwitchResult", ap
     dropped so a stale secret cannot shadow it.
 
     Returns a new dict."""
-    from hermes_cli.model_switch import apply_model_selection
+    from hermes_cli.config_backend import supports_file_tooling
+    from hermes_cli.model_switch import apply_model_selection, remote_model_selection_guard
 
-    model_cfg = apply_model_selection(model_cfg, result)
+    before = model_cfg
+    # Remote Config: the caller saves the effective block as a whole document, so a cleared
+    # inherited key must be masked (null), and a locked route key refuses the whole assignment.
+    model_cfg = apply_model_selection(model_cfg, result, mask_cleared=not supports_file_tooling())
     if api_key.strip():
         model_cfg["api_key"] = api_key.strip()
         model_cfg.pop("api", None)
+    remote_model_selection_guard(before, model_cfg)
     return model_cfg
 
 

tests/plugins/config_backends/test_remote_backend.py

@@ -1701,3 +1701,68 @@ def test_model_switch_masks_inherited_endpoint_credentials(plane, monkeypatch):
     model = read_user_config_raw()["model"]
     assert model["base_url"] == "https://new.example/v1"
     assert all(model.get(k) is None for k in ("api_key", "key_env", "api_key_env", "context_length")), model
+
+
+def test_poll_between_read_and_atomic_config_write_sends_only_the_callers_edit(plane, monkeypatch):
+    """Same schedule through ``atomic_config_write`` (``/verbose``, ``/footer``): its no-omission
+    guard read must not become the base the caller's stale document is diffed against."""
+    from gateway import slash_commands as sc
+    plane.profile("default").update(values={"display": {"personality": "concise", "compact": False}}, version=1)
+    backend = get_config_backend()
+    orig = sc.atomic_config_write
+
+    def poll_before_write(path, doc):
+        p = plane.profile("default")
+        p["values"]["display"]["compact"] = True
+        p["version"] += 1
+        assert backend.poll_one(backend._state(plane.home))
+        return orig(path, doc)
+
+    monkeypatch.setattr(sc, "atomic_config_write", poll_before_write)
+    sc._write_raw_config_leaf(plane.home / "config.yaml", ("display", "personality"), "pirate")
+    bodies = [r["body"] for r in plane.patches()]
+    assert len(bodies) == 1 and bodies[0]["set"] == {"display.personality": "pirate"}, bodies
+    assert plane.profile("default")["values"]["display"] == {"personality": "pirate", "compact": True}
+
+
+def _dashboard_switch(monkeypatch, url):
+    import asyncio
+    from hermes_cli.model_switch import ModelSwitchResult
+    from hermes_cli.web_models import ModelAssignment
+    from hermes_cli.web_routers import models
+    result = ModelSwitchResult(success=True, new_model="new", target_provider="custom", provider_changed=True,
+                               base_url=url, api_mode="chat_completions", is_global=True)
+    monkeypatch.setattr(models, "_dashboard_code_skew_guard", lambda: None)
+    monkeypatch.setattr(models, "_prepare_main_assignment", lambda *a, **kw: (url, result))
+    return asyncio.run(models.set_model_assignment(ModelAssignment(
+        scope="main", provider="custom", model="new", base_url=url, confirm_expensive_model=True)))
+
+
+_OLD_ROUTE = {"default": "old", "provider": "custom:old", "base_url": "https://old.example/v1",
+              "api_key": "${OLD_KEY}", "key_env": "OLD_KEY", "api_key_env": "OLD_KEY", "context_length": 123456}
+
+
+def test_dashboard_model_switch_masks_inherited_endpoint_credentials(plane, monkeypatch):
+    """POST /api/model/set saves the whole document. Under Remote Config its model block is the
+    effective one, so the endpoint-bound keys it clears must be masked (null), not dropped from
+    the profile, or the old endpoint's key would reach the new endpoint."""
+    from hermes_cli.config import read_user_config_raw
+    url = "https://new.example/v1"
+    plane.upper = {"model": dict(_OLD_ROUTE), "providers": {"known-new": {"base_url": url}}}
+    assert _dashboard_switch(monkeypatch, url)["ok"] is True
+    model = read_user_config_raw()["model"]
+    assert model["base_url"] == url
+    assert all(model.get(k) is None for k in ("api_key", "key_env", "api_key_env", "context_length")), model
+
+
+def test_dashboard_model_switch_refuses_whole_assignment_on_a_locked_route_key(plane, monkeypatch):
+    """A lock on one route key refuses the dashboard assignment entirely (409/403), like
+    persist_model_selection: nothing is sent, so the old route stays whole."""
+    from fastapi import HTTPException
+    url = "https://new.example/v1"
+    plane.upper = {"model": dict(_OLD_ROUTE), "providers": {"known-new": {"base_url": url}}}
+    plane.upper_locks = [{"path": "model.base_url", "level": "tenant"}]
+    with pytest.raises(HTTPException) as exc:
+        _dashboard_switch(monkeypatch, url)
+    assert exc.value.status_code in (403, 409), exc.value
+    assert plane.patches() == []

tests/plugins/config_backends/test_remote_grouped_and_migration_reads.py

@@ -314,6 +314,53 @@ def test_migration_recursion_still_sees_its_private_copy(plane, monkeypatch):
     assert plane.patches() == []
 
 
+class _Transport:
+    def __init__(self):
+        self.frames, self.done = [], threading.Event()
+
+    def write(self, obj):
+        self.frames.append(obj)
+        self.done.set()
+        return True
+
+    def close(self):
+        pass
+
+
+def _configure(server, rid, **params):
+    import json
+    t = _Transport()
+    ret = server.dispatch({"jsonrpc": "2.0", "id": rid, "method": "profiles.configure",
+                           "params": {"name": "default", **params}}, t)
+    if ret is not None:
+        t.write(ret)
+    assert t.done.wait(10), "no terminal response"
+    return json.loads(json.dumps(t.frames[-1]))["result"]
+
+
+def test_profile_editor_reports_mcp_unapplied_when_legacy_disabled_cleanup_is_locked(plane, monkeypatch):
+    """Enabling an MCP server writes ``enabled: true`` AND drops the legacy ``disabled``. When a
+    lock keeps ``disabled: true``, the editor (which honours ``disabled``) still shows the server
+    off, so the edit must not be acknowledged; unlocked, it lands and the editor agrees."""
+    from hermes_cli.config import read_user_config_raw
+    from tui_gateway.methods_profiles import _mcp_entry_enabled
+    plane.upper = {"mcp_servers": {"example": {"command": "unused", "enabled": False}}}
+    plane.profile("default")["values"] = {"mcp_servers": {"example": {"disabled": True}}}
+    plane.upper_locks = [{"path": "mcp_servers.example.disabled", "level": "tenant"}]
+    server = _tui_server(monkeypatch, plane.home)
+
+    locked = _configure(server, 1, enabled_mcp_servers=["example"])
+    assert locked["applied"]["mcp_servers"] is False and locked["ok"] is False, locked
+    assert _mcp_entry_enabled(read_user_config_raw()["mcp_servers"]["example"]) is False
+
+    plane.upper_locks = []
+    backend = get_config_backend()
+    backend.poll_one(backend._state(plane.home))
+    unlocked = _configure(server, 2, enabled_mcp_servers=["example"])
+    assert unlocked["applied"]["mcp_servers"] is True and unlocked["ok"] is True, unlocked
+    assert _mcp_entry_enabled(read_user_config_raw()["mcp_servers"]["example"]) is True
+
+
 def test_profile_editor_reports_a_locked_skills_edit_as_not_applied(plane, monkeypatch):
     """``profiles.configure`` saves its sections through bulk ``save_config``, which omits a
     Remote Config lock instead of raising. The editor must report that section unapplied, not

tui_gateway/methods_profiles.py

@@ -749,8 +749,19 @@ def _configure_cfg_sections(profile_dir, params, applied) -> None:
             mcp_cfg = load_config() or {}
             applied["mcp_servers"] = _best_effort(lambda: _save_mcp_toggles(
                 mcp_cfg, params["enabled_mcp_servers"], launch_mcp, save_config)) \
-                and _landed(mcp_cfg, ("mcp_servers",), lambda v: {
-                    k: e.get("enabled") for k, e in v.items() if isinstance(e, dict)} if isinstance(v, dict) else v)
+                and _landed(mcp_cfg, ("mcp_servers",), _mcp_toggle_state)
+
+
+def _mcp_toggle_state(servers):
+    """Every toggle field ``_save_mcp_toggles`` writes, per server: ``enabled`` AND the legacy
+    ``disabled`` it removes. A lock that kept ``disabled: true`` leaves the server off for the
+    editor (and on for the runtime) although ``enabled`` landed, so it must not read as applied."""
+    if not isinstance(servers, dict):
+        return servers
+    return {k: (e.get("enabled"), e.get("disabled", _ABSENT)) for k, e in servers.items() if isinstance(e, dict)}
+
+
+_ABSENT = "<absent>"
 
 
 def _landed(cfg: dict, path: tuple, key=lambda v: v) -> bool: