feat: scoped threshold-config merge for non-admin saves

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NfPpSpccTWBfZg1FTveyaU

hbd/server/config_access.py

@@ -139,3 +139,30 @@ def merge_hosts_scoped(existing: Any, payload: Any, username: str,
             raise ScopedMergeError(f"host {name!r}: only the owner may delete a host")
 
     return result
+
+
+def merge_threshold_configs_scoped(existing: Any, payload: Any,
+                                   username: str) -> Dict[str, Any]:
+    """Return a new threshold_configs section with the user's own configs
+    replaced by *payload*.
+
+    Global and foreign-owned configs are preserved and may not appear in the
+    payload ('default' included).  Own configs missing from the payload are
+    deleted.  Every payload entry gets its owner forced to *username*.
+    """
+    existing = existing or {}
+    payload = payload or {}
+    result: Dict[str, Any] = {
+        n: c for n, c in existing.items()
+        if not (isinstance(c, dict) and c.get("owner") == username)
+    }
+    for name, entry in payload.items():
+        if name == "default":
+            raise ScopedMergeError("threshold config 'default' is global and admin-managed")
+        old = existing.get(name)
+        if old is not None and (not isinstance(old, dict) or old.get("owner") != username):
+            raise ScopedMergeError(f"threshold config {name!r}: not owned by you")
+        new_cfg = dict(entry) if isinstance(entry, dict) else {}
+        new_cfg["owner"] = username
+        result[name] = new_cfg
+    return result

tests/test_config_access.py

@@ -190,3 +190,58 @@ def test_merge_hosts_rejects_unknown_fields():
 def test_merge_hosts_owner_clearing_list_removes_key():
     result = _merge_hosts(_full_payload(mine={"owner": "alice", "watch": True}))
     assert "notification_channels" not in result["mine"]
+
+
+# ---------------------------------------------------------------------------
+# merge_threshold_configs_scoped
+# ---------------------------------------------------------------------------
+
+MERGE_EXISTING_TCS = {
+    "default": {"thresholds": {"cpu": {"load": {"warning": 2}}}},
+    "alice_tc": {"owner": "alice", "thresholds": {"cpu": {"load": {"warning": 3}}}},
+    "bob_tc": {"owner": "bob", "thresholds": {}},
+    "global_tc": {"thresholds": {}},
+}
+
+
+def _merge_tcs(payload):
+    return ca.merge_threshold_configs_scoped(dict(MERGE_EXISTING_TCS), payload, "alice")
+
+
+def test_merge_tcs_new_config_owner_forced():
+    result = _merge_tcs({"alice_tc": {"thresholds": {}},
+                         "new_tc": {"thresholds": {}, "owner": "bob"}})
+    assert result["new_tc"]["owner"] == "alice"
+
+
+def test_merge_tcs_edit_own():
+    result = _merge_tcs({"alice_tc": {"thresholds": {"mem": {"used": {"warning": 90}}}}})
+    assert result["alice_tc"]["thresholds"] == {"mem": {"used": {"warning": 90}}}
+    assert result["alice_tc"]["owner"] == "alice"
+
+
+def test_merge_tcs_delete_own_when_missing():
+    result = _merge_tcs({})
+    assert "alice_tc" not in result
+
+
+def test_merge_tcs_preserves_global_and_foreign():
+    result = _merge_tcs({"alice_tc": {"thresholds": {}}})
+    assert result["default"] == MERGE_EXISTING_TCS["default"]
+    assert result["bob_tc"] == MERGE_EXISTING_TCS["bob_tc"]
+    assert result["global_tc"] == MERGE_EXISTING_TCS["global_tc"]
+
+
+def test_merge_tcs_rejects_default():
+    with pytest.raises(ca.ScopedMergeError, match="default"):
+        _merge_tcs({"alice_tc": {"thresholds": {}}, "default": {"thresholds": {}}})
+
+
+def test_merge_tcs_rejects_global_name_collision():
+    with pytest.raises(ca.ScopedMergeError, match="global_tc"):
+        _merge_tcs({"alice_tc": {"thresholds": {}}, "global_tc": {"thresholds": {}}})
+
+
+def test_merge_tcs_rejects_foreign_owned():
+    with pytest.raises(ca.ScopedMergeError, match="bob_tc"):
+        _merge_tcs({"alice_tc": {"thresholds": {}}, "bob_tc": {"thresholds": {}}})