feat: wire post-sync mutual exclusion invariant repair into apply_synced_settings

- Call enforce_mutual_exclusion(current) in apply_synced_settings() after
  applying SYNCABLE_KEYS and before save_settings()
- Import enforce_mutual_exclusion from muxplex.views inside the function
  to avoid circular import issues
- Update docstring to describe the mutual exclusion invariant repair step
- Add test_apply_synced_settings_enforces_mutual_exclusion to verify that
  sessions present in both hidden_sessions and a view's sessions are
  removed from hidden_sessions after sync (task-7)
This commit is contained in:
Brian Krabach
2026-04-15 11:33:49 -07:00
parent 569ba44848
commit a3ff22dc02
2 changed files with 47 additions and 0 deletions
+7
View File
@@ -166,11 +166,18 @@ def apply_synced_settings(incoming_settings: dict, incoming_timestamp: float) ->
Only applies keys that are in SYNCABLE_KEYS. Sets settings_updated_at Only applies keys that are in SYNCABLE_KEYS. Sets settings_updated_at
to the incoming timestamp (NOT time.time()) to prevent sync loops. to the incoming timestamp (NOT time.time()) to prevent sync loops.
After applying synced keys, enforces the mutual exclusion invariant:
any session key that appears in both hidden_sessions and a view's sessions
is removed from hidden_sessions (visibility wins over hiding).
""" """
from muxplex.views import enforce_mutual_exclusion
current = load_settings() current = load_settings()
for key in SYNCABLE_KEYS: for key in SYNCABLE_KEYS:
if key in incoming_settings: if key in incoming_settings:
current[key] = incoming_settings[key] current[key] = incoming_settings[key]
enforce_mutual_exclusion(current)
current["settings_updated_at"] = incoming_timestamp current["settings_updated_at"] = incoming_timestamp
save_settings(current) save_settings(current)
return current return current
+40
View File
@@ -1013,3 +1013,43 @@ def test_patch_settings_syncs_views():
assert settings["settings_updated_at"] > 0, ( assert settings["settings_updated_at"] > 0, (
"settings_updated_at must be > 0 after patching a syncable key (views)" "settings_updated_at must be > 0 after patching a syncable key (views)"
) )
# ============================================================
# Post-sync mutual exclusion invariant repair (task-7)
# ============================================================
def test_apply_synced_settings_enforces_mutual_exclusion(redirect_settings_path):
"""apply_synced_settings must call enforce_mutual_exclusion after applying synced keys.
Pre-populate settings with hidden_sessions=['abc:dev'] and views=[].
Apply incoming sync with views=[{'name': 'Work', 'sessions': ['abc:dev']}] and
hidden_sessions=['abc:dev'].
Assert 'abc:dev' NOT in result['hidden_sessions'] and IS in result['views'][0]['sessions'].
"""
import json
# Pre-populate settings with hidden_sessions containing 'abc:dev'
redirect_settings_path.write_text(
json.dumps({"hidden_sessions": ["abc:dev"], "views": []})
)
# Apply incoming sync that has 'abc:dev' in BOTH views and hidden_sessions
incoming = {
"views": [{"name": "Work", "sessions": ["abc:dev"]}],
"hidden_sessions": ["abc:dev"],
}
result = apply_synced_settings(incoming, 1712600000.0)
# enforce_mutual_exclusion should remove 'abc:dev' from hidden_sessions
# because it now appears in a view's sessions
assert "abc:dev" not in result["hidden_sessions"], (
f"'abc:dev' must be removed from hidden_sessions after mutual exclusion repair; "
f"got hidden_sessions={result['hidden_sessions']!r}"
)
# 'abc:dev' must still be in the view's sessions
assert "abc:dev" in result["views"][0]["sessions"], (
f"'abc:dev' must remain in views[0]['sessions'] after mutual exclusion repair; "
f"got views={result['views']!r}"
)