feat(views): prune stale session keys with local-only bookkeeping (Phase 4)

New pruning.py sidecar module with load_pruning_state / save_pruning_state. The sidecar lives at ~/.config/muxplex/pruning.json and is NEVER in SYNCABLE_KEYS — bookkeeping is per-device, not federated.

prune_stale_keys(settings, live_keys, *, pruning_state, grace_seconds, now) drops keys from hidden_sessions and view.sessions after they have been missing past the configurable grace period (default 24h, syncable via stale_key_grace_hours).

Wired into _run_poll_cycle as step 14 with try/except so failures never abort the cycle.

The prune ACTION (settings change) syncs normally via existing LWW; the bookkeeping does not.

24 new tests pass; total muxplex/tests/ = 1263 passed.

README.md

@@ -247,6 +247,7 @@ All settings are stored in `~/.config/muxplex/settings.json`.
 | `sort_order` | `manual` | Session ordering: `manual`, `alphabetical`, `recent` |
 | `hidden_sessions` | `[]` | Sessions hidden from the dashboard |
 | `views` | `[]` | Named session views for grouping and filtering sessions |
+| `stale_key_grace_hours` | `24.0` | Hours before a session key absent from all live sessions is pruned from views/hidden_sessions (syncable; per-device bookkeeping is local-only) |
 | `window_size_largest` | `false` | Auto-set tmux `window-size largest` on connect |
 | `auto_open_created` | `true` | Auto-open newly created sessions |
 | `new_session_template` | `tmux new-session -d -s {name}` | Command template for creating sessions |

muxplex/main.py

@@ -70,7 +70,10 @@ from muxplex.settings import (
     load_federation_key,
     load_settings,
     patch_settings,
+    save_settings,
 )
+from muxplex.pruning import load_pruning_state, save_pruning_state
+from muxplex.views import prune_stale_keys
 from muxplex.identity import load_device_id
 from muxplex.ttyd import kill_orphan_ttyd, kill_ttyd, spawn_ttyd, TTYD_PORT
 
@@ -269,6 +272,48 @@ async def _run_poll_cycle() -> None:
             except Exception:
                 _log.exception("settings sync cycle error")
 
+    # 14. Prune stale session keys from views and hidden_sessions.
+    #
+    #     Each device only knows its own live sessions natively.  The live_keys
+    #     set below includes both the bare session name (legacy bare-name entries)
+    #     and the canonical device_id:name form.  Remote-device keys tracked in
+    #     views/hidden_sessions are also covered: if this device has not seen a
+    #     remote key for the full grace period the key is pruned locally; the
+    #     remote device handles pruning for its own keys independently.  The grace
+    #     period prevents thrash when sessions briefly disappear during restarts
+    #     or sync gaps.
+    #
+    #     Pruning bookkeeping (first-missed-at timestamps) is NEVER written to
+    #     settings.json and is NEVER sent to peers — it lives in pruning.json.
+    #     The prune action (removing dead keys from settings) IS a normal
+    #     settings write that syncs via the existing LWW mechanism.
+    try:
+        _prune_settings = load_settings()
+        _prune_state = load_pruning_state()
+        _grace_hours = float(_prune_settings.get("stale_key_grace_hours", 24.0))
+        _grace_seconds = _grace_hours * 3600.0
+
+        _local_device_id = load_device_id()
+        _live_keys: set[str] = set()
+        for _name in names:
+            # Include both the bare name (for legacy stored entries) and the
+            # canonical device_id:name form.
+            _live_keys.add(_name)
+            _live_keys.add(f"{_local_device_id}:{_name}")
+
+        _prune_settings, _prune_state, _prune_changed = prune_stale_keys(
+            _prune_settings,
+            _live_keys,
+            pruning_state=_prune_state,
+            grace_seconds=_grace_seconds,
+        )
+        save_pruning_state(_prune_state)
+        if _prune_changed:
+            # Stale keys were removed — persist (triggers LWW sync on next cycle).
+            save_settings(_prune_settings)
+    except Exception:
+        _log.exception("stale-key prune cycle error")
+
 
 # ---------------------------------------------------------------------------
 # Poll loop

muxplex/pruning.py

@@ -0,0 +1,56 @@
+"""
+Local sidecar bookkeeping for stale-key pruning.
+
+Stale-key pruning is a per-device concern: each device independently tracks
+which session keys it has failed to observe, and prunes them from its own
+settings once the grace period expires.  That bookkeeping must NEVER be synced
+to peers — it is stored in a local sidecar file outside settings.json.
+
+The prune ACTION (removing stale keys from view.sessions / hidden_sessions) IS
+a normal settings write and DOES sync via the existing LWW mechanism.
+
+See docs/plans/2026-05-17-hidden-state-redesign-design.md, Phase 4 and the
+section "Stale key pruning (separate concern, local-only state)".
+"""
+
+import json
+from pathlib import Path
+
+PRUNING_STATE_PATH = Path.home() / ".config" / "muxplex" / "pruning.json"
+
+
+def load_pruning_state() -> dict:
+    """Load local pruning bookkeeping from the sidecar file.
+
+    Returns an empty dict on absent file or corrupt JSON — never raises for
+    either condition.  Unexpected errors (e.g. PermissionError) propagate.
+
+    The returned dict has the shape::
+
+        {
+            "first_missed_at": {
+                "dev1:dead-session": 1747512345.0,
+                ...
+            }
+        }
+    """
+    try:
+        text = PRUNING_STATE_PATH.read_text(encoding="utf-8", errors="replace")
+        data = json.loads(text)
+        if not isinstance(data, dict):
+            return {}
+        return data
+    except FileNotFoundError:
+        return {}
+    except (json.JSONDecodeError, ValueError):
+        return {}
+
+
+def save_pruning_state(state: dict) -> None:
+    """Write pruning bookkeeping to the sidecar file.
+
+    Creates parent directories as needed.  Matches the direct-write style of
+    save_settings (indent=2, trailing newline).
+    """
+    PRUNING_STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
+    PRUNING_STATE_PATH.write_text(json.dumps(state, indent=2) + "\n")

muxplex/settings.py

@@ -57,6 +57,11 @@ DEFAULT_SETTINGS: dict = {
     "sidebarOpen": None,
     "settings_updated_at": 0.0,
     "_schema_version": SCHEMA_VERSION,
+    # Grace period (hours) before a session key missing from all live sessions
+    # is removed from views/hidden_sessions. Syncable so the operator can tune
+    # it federation-wide; the per-device first-missed-at bookkeeping that drives
+    # the actual prune is local-only (pruning.json, never synced).
+    "stale_key_grace_hours": 24.0,
 }
 
 SYNCABLE_KEYS: frozenset[str] = frozenset(
@@ -82,6 +87,10 @@ SYNCABLE_KEYS: frozenset[str] = frozenset(
         # Schema version — sent so peers can detect our version, but never
         # accepted from the wire (see apply_synced_settings).
         "_schema_version",
+        # Pruning grace period — synced so the operator can tune it
+        # federation-wide. The per-device bookkeeping (first-missed-at
+        # timestamps) is NOT synced; it lives in pruning.json locally.
+        "stale_key_grace_hours",
     }
 )
 

muxplex/tests/test_pruning.py

@@ -0,0 +1,177 @@
+"""
+Tests for muxplex/pruning.py — local sidecar bookkeeping for stale-key pruning.
+
+The pruning sidecar (pruning.json) is NEVER synced to peers.  These tests verify:
+- load_pruning_state() returns {} on absent file
+- load_pruning_state() returns {} on corrupt JSON (never crashes)
+- round-trip: save then load returns the same data
+- the sidecar path constant is the expected XDG-style location
+- the sidecar is distinct from settings.json (different constant, different path)
+"""
+
+import json
+from pathlib import Path
+
+import pytest
+
+import muxplex.pruning as pruning_mod
+from muxplex.pruning import load_pruning_state, save_pruning_state
+
+
+# ---------------------------------------------------------------------------
+# Autouse fixture: redirect PRUNING_STATE_PATH to tmp_path for all tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def redirect_pruning_state_path(tmp_path, monkeypatch):
+    """Redirect PRUNING_STATE_PATH to a temporary file for all tests."""
+    fake_path = tmp_path / "pruning.json"
+    monkeypatch.setattr(pruning_mod, "PRUNING_STATE_PATH", fake_path)
+    return fake_path
+
+
+# ---------------------------------------------------------------------------
+# Default path check (against the module constant, not the redirected one)
+# ---------------------------------------------------------------------------
+
+
+def test_pruning_state_path_is_expected_location():
+    """PRUNING_STATE_PATH must be ~/.config/muxplex/pruning.json."""
+    # The autouse fixture redirects PRUNING_STATE_PATH at test time, so we
+    # verify the expected path by construction rather than reading the (already
+    # patched) module constant.
+    expected = Path.home() / ".config" / "muxplex" / "pruning.json"
+    # Path structure: ~/.config/muxplex/pruning.json
+    assert expected.name == "pruning.json"
+    assert expected.parent.name == "muxplex"
+    assert expected.parent.parent.name == ".config"
+    assert expected.parent.parent.parent == Path.home()
+
+
+def test_pruning_state_path_differs_from_settings_path():
+    """PRUNING_STATE_PATH and SETTINGS_PATH must be different files."""
+    from muxplex.settings import SETTINGS_PATH
+
+    pruning_default = Path.home() / ".config" / "muxplex" / "pruning.json"
+    assert pruning_default != SETTINGS_PATH, (
+        "pruning.json and settings.json must be distinct files — "
+        "pruning bookkeeping must never be mixed with syncable settings"
+    )
+
+
+# ---------------------------------------------------------------------------
+# load_pruning_state — missing file
+# ---------------------------------------------------------------------------
+
+
+def test_load_pruning_state_returns_empty_when_file_absent():
+    """load_pruning_state() returns {} when the sidecar file does not exist."""
+    # The redirected path points to a non-existent file (fixture only creates the dir).
+    result = load_pruning_state()
+    assert result == {}, (
+        f"load_pruning_state() must return {{}} for absent file, got: {result!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# load_pruning_state — corrupt JSON
+# ---------------------------------------------------------------------------
+
+
+def test_load_pruning_state_returns_empty_on_corrupt_json(redirect_pruning_state_path):
+    """load_pruning_state() returns {} on corrupt JSON — never raises."""
+    redirect_pruning_state_path.write_text("NOT VALID JSON {{{{")
+
+    result = load_pruning_state()
+
+    assert result == {}, (
+        f"load_pruning_state() must return {{}} on corrupt JSON, got: {result!r}"
+    )
+
+
+def test_load_pruning_state_returns_empty_on_truncated_file(
+    redirect_pruning_state_path,
+):
+    """load_pruning_state() returns {} on a file with stray/truncated bytes."""
+    redirect_pruning_state_path.write_bytes(b"\xff\xfe truncated")
+
+    result = load_pruning_state()
+
+    assert result == {}, (
+        f"load_pruning_state() must return {{}} on stray bytes, got: {result!r}"
+    )
+
+
+def test_load_pruning_state_returns_empty_on_non_dict_json(
+    redirect_pruning_state_path,
+):
+    """load_pruning_state() returns {} when JSON parses to a non-dict (e.g. a list)."""
+    redirect_pruning_state_path.write_text(json.dumps([1, 2, 3]))
+
+    result = load_pruning_state()
+
+    assert result == {}, (
+        f"load_pruning_state() must return {{}} when JSON root is not a dict, "
+        f"got: {result!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Round-trip: save then load
+# ---------------------------------------------------------------------------
+
+
+def test_save_then_load_round_trip():
+    """save_pruning_state then load_pruning_state returns the same data."""
+    state = {
+        "first_missed_at": {
+            "dev1:dead-session": 1747512345.0,
+            "dev2:another-gone": 1747512000.0,
+        }
+    }
+
+    save_pruning_state(state)
+    loaded = load_pruning_state()
+
+    assert loaded == state, (
+        f"round-trip save/load must preserve data exactly; got: {loaded!r}"
+    )
+
+
+def test_save_creates_parent_directories(tmp_path, monkeypatch):
+    """save_pruning_state creates parent directories as needed."""
+    nested_path = tmp_path / "a" / "b" / "pruning.json"
+    monkeypatch.setattr(pruning_mod, "PRUNING_STATE_PATH", nested_path)
+
+    save_pruning_state({"first_missed_at": {}})
+
+    assert nested_path.exists(), "save_pruning_state must create parent directories"
+
+
+def test_save_writes_valid_json(redirect_pruning_state_path):
+    """save_pruning_state writes well-formed JSON (parseable by json.loads)."""
+    state = {"first_missed_at": {"dev1:x": 1234567890.0}}
+    save_pruning_state(state)
+
+    raw = redirect_pruning_state_path.read_text()
+    parsed = json.loads(raw)
+    assert parsed == state
+
+
+def test_save_empty_state_round_trips(redirect_pruning_state_path):
+    """An empty pruning state saves and loads cleanly."""
+    save_pruning_state({})
+    loaded = load_pruning_state()
+    assert loaded == {}
+
+
+def test_save_overwrites_previous_state(redirect_pruning_state_path):
+    """Subsequent saves overwrite the previous sidecar contents."""
+    save_pruning_state({"first_missed_at": {"dev1:old": 111.0}})
+    save_pruning_state({"first_missed_at": {"dev1:new": 222.0}})
+
+    loaded = load_pruning_state()
+    assert loaded == {"first_missed_at": {"dev1:new": 222.0}}, (
+        f"save must overwrite previous state; got: {loaded!r}"
+    )

muxplex/tests/test_settings.py

@@ -1137,3 +1137,37 @@ def test_peer_supports_v2_handles_legacy_payload():
     assert peer_supports_v2({}) is False  # legacy peer omits the field
     assert peer_supports_v2({"_schema_version": "garbage"}) is False
     assert peer_supports_v2({"_schema_version": None}) is False
+
+
+# ============================================================
+# stale_key_grace_hours (Phase 4)
+# See docs/plans/2026-05-17-hidden-state-redesign-design.md
+# ============================================================
+
+
+def test_stale_key_grace_hours_default_is_24():
+    """DEFAULT_SETTINGS must include stale_key_grace_hours defaulting to 24.0."""
+    assert "stale_key_grace_hours" in DEFAULT_SETTINGS, (
+        "DEFAULT_SETTINGS must include 'stale_key_grace_hours'"
+    )
+    assert DEFAULT_SETTINGS["stale_key_grace_hours"] == 24.0, (
+        f"stale_key_grace_hours default must be 24.0, "
+        f"got: {DEFAULT_SETTINGS['stale_key_grace_hours']!r}"
+    )
+
+
+def test_stale_key_grace_hours_is_in_syncable_keys():
+    """stale_key_grace_hours must be in SYNCABLE_KEYS so peers can share the grace period."""
+    assert "stale_key_grace_hours" in SYNCABLE_KEYS, (
+        "stale_key_grace_hours must be in SYNCABLE_KEYS"
+    )
+
+
+def test_stale_key_grace_hours_is_not_pruning_state():
+    """pruning_state bookkeeping must NOT be in SYNCABLE_KEYS under any circumstances."""
+    assert "pruning_state" not in SYNCABLE_KEYS, (
+        "pruning_state must never be in SYNCABLE_KEYS (it is local-only bookkeeping)"
+    )
+    assert "first_missed_at" not in SYNCABLE_KEYS, (
+        "first_missed_at must never be in SYNCABLE_KEYS (it is local-only bookkeeping)"
+    )

muxplex/tests/test_views.py

@@ -9,6 +9,7 @@ from muxplex.views import (
     hide,
     is_hidden,
     normalize_session_keys,
+    prune_stale_keys,
     remove_from_all_views,
     remove_membership,
     unhide,
@@ -569,3 +570,273 @@ def test_unhide_does_not_touch_views():
     unhide(settings, "dev1:a")
     assert settings["views"][0]["sessions"] == original_work
     assert settings["views"][1]["sessions"] == original_personal
+
+
+# ---------------------------------------------------------------------------
+# Stale-key pruning (Phase 4)
+# See docs/plans/2026-05-17-hidden-state-redesign-design.md, Phase 4 and
+# the section "Stale key pruning (separate concern, local-only state)".
+# ---------------------------------------------------------------------------
+
+_GRACE = 86400.0  # 24 hours in seconds — default grace period used in tests
+_T0 = 1_700_000_000.0  # arbitrary stable "now" for deterministic tests
+
+
+def _pruning_settings() -> dict:
+    """Settings with one view and a hidden_sessions list for pruning tests."""
+    return {
+        "hidden_sessions": ["dev1:hidden"],
+        "views": [
+            {
+                "name": "Work",
+                "sessions": ["dev1:work-a", "dev1:work-b"],
+            },
+            {
+                "name": "Personal",
+                "sessions": ["dev1:personal-a"],
+            },
+        ],
+    }
+
+
+# 1. Live key clears bookkeeping
+def test_prune_live_key_clears_first_missed_at():
+    """If a key is in live_keys, any existing first_missed_at entry is removed."""
+    settings = _pruning_settings()
+    pruning_state = {"first_missed_at": {"dev1:work-a": _T0 - 1000.0}}
+
+    _, ps, changed = prune_stale_keys(
+        settings,
+        live_keys={"dev1:work-a", "dev1:work-b", "dev1:hidden", "dev1:personal-a"},
+        pruning_state=pruning_state,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+
+    assert "dev1:work-a" not in ps["first_missed_at"]
+    assert changed is False
+
+
+# 2. Newly missing key records first_missed_at, settings unchanged
+def test_prune_newly_missing_key_records_timestamp():
+    """A key that is in settings but absent from live_keys gets a first_missed_at entry."""
+    settings = _pruning_settings()
+    pruning_state = {"first_missed_at": {}}
+
+    _, ps, changed = prune_stale_keys(
+        settings,
+        live_keys={"dev1:work-b", "dev1:hidden", "dev1:personal-a"},
+        # dev1:work-a is missing from live_keys
+        pruning_state=pruning_state,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+
+    assert ps["first_missed_at"]["dev1:work-a"] == _T0
+    assert changed is False
+    # key still in settings — not pruned yet
+    assert "dev1:work-a" in settings["views"][0]["sessions"]
+
+
+# 3. Missing within grace — settings unchanged
+def test_prune_within_grace_period_leaves_settings_unchanged():
+    """A key with first_missed_at = now - grace + 1 is within grace; nothing changes."""
+    settings = _pruning_settings()
+    # first missed 1 second before grace expires
+    first_missed_at = _T0 - _GRACE + 1.0
+    pruning_state = {"first_missed_at": {"dev1:work-a": first_missed_at}}
+
+    _, ps, changed = prune_stale_keys(
+        settings,
+        live_keys={"dev1:work-b", "dev1:hidden", "dev1:personal-a"},
+        pruning_state=pruning_state,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+
+    assert changed is False
+    assert "dev1:work-a" in settings["views"][0]["sessions"]
+    assert ps["first_missed_at"]["dev1:work-a"] == first_missed_at  # unchanged
+
+
+# 4. Missing past grace — key removed from hidden_sessions AND view.sessions
+def test_prune_past_grace_removes_key_from_settings():
+    """A key missing for > grace_seconds is removed from hidden_sessions and every view."""
+    settings = {
+        "hidden_sessions": ["dev1:stale", "dev1:live-hidden"],
+        "views": [
+            {"name": "Work", "sessions": ["dev1:stale", "dev1:live-work"]},
+            {"name": "Personal", "sessions": ["dev1:stale", "dev1:live-personal"]},
+        ],
+    }
+    # stale key has been missing for grace + 1 second
+    pruning_state = {"first_missed_at": {"dev1:stale": _T0 - _GRACE - 1.0}}
+
+    updated, ps, changed = prune_stale_keys(
+        settings,
+        live_keys={"dev1:live-hidden", "dev1:live-work", "dev1:live-personal"},
+        pruning_state=pruning_state,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+
+    assert changed is True
+    # Removed from hidden_sessions
+    assert "dev1:stale" not in updated["hidden_sessions"]
+    assert "dev1:live-hidden" in updated["hidden_sessions"]
+    # Removed from every view
+    assert "dev1:stale" not in updated["views"][0]["sessions"]
+    assert "dev1:live-work" in updated["views"][0]["sessions"]
+    assert "dev1:stale" not in updated["views"][1]["sessions"]
+    assert "dev1:live-personal" in updated["views"][1]["sessions"]
+    # Bookkeeping entry dropped
+    assert "dev1:stale" not in ps["first_missed_at"]
+
+
+# 5. Re-appearance after partial absence clears bookkeeping
+def test_prune_reappearance_clears_bookkeeping():
+    """A key that was missing (has bookkeeping) but returns to live_keys is forgiven."""
+    settings = _pruning_settings()
+    # dev1:work-a was previously seen as missing
+    pruning_state = {"first_missed_at": {"dev1:work-a": _T0 - 3600.0}}
+
+    _, ps, changed = prune_stale_keys(
+        settings,
+        live_keys={
+            "dev1:work-a",  # back!
+            "dev1:work-b",
+            "dev1:hidden",
+            "dev1:personal-a",
+        },
+        pruning_state=pruning_state,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+
+    assert "dev1:work-a" not in ps["first_missed_at"]
+    assert changed is False
+    # key is still in settings (was never pruned)
+    assert "dev1:work-a" in settings["views"][0]["sessions"]
+
+
+# 6. Bookkeeping GC — stale entries not in settings are dropped
+def test_prune_gc_drops_orphaned_bookkeeping_entries():
+    """A first_missed_at entry whose key is no longer in settings is garbage-collected."""
+    settings = _pruning_settings()
+    # "dev1:ghost" is in bookkeeping but not in any settings list
+    pruning_state = {
+        "first_missed_at": {
+            "dev1:ghost": _T0 - 100.0,  # not in settings, not expired
+        }
+    }
+
+    _, ps, changed = prune_stale_keys(
+        settings,
+        live_keys={"dev1:work-a", "dev1:work-b", "dev1:hidden", "dev1:personal-a"},
+        pruning_state=pruning_state,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+
+    # Orphaned entry removed — it was never in settings
+    assert "dev1:ghost" not in ps["first_missed_at"]
+    assert changed is False
+
+
+# 7. settings_changed flag is True only when a key is actually dropped
+def test_prune_changed_flag_is_false_when_nothing_pruned():
+    """settings_changed is False when all keys are live or within grace."""
+    settings = _pruning_settings()
+    _, _, changed = prune_stale_keys(
+        settings,
+        live_keys={"dev1:work-a", "dev1:work-b", "dev1:hidden", "dev1:personal-a"},
+        pruning_state={"first_missed_at": {}},
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+    assert changed is False
+
+
+def test_prune_changed_flag_is_true_when_key_dropped():
+    """settings_changed is True when at least one key is actually removed."""
+    settings = {
+        "hidden_sessions": ["dev1:stale"],
+        "views": [],
+    }
+    pruning_state = {"first_missed_at": {"dev1:stale": _T0 - _GRACE - 1.0}}
+
+    _, _, changed = prune_stale_keys(
+        settings,
+        live_keys=set(),
+        pruning_state=pruning_state,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+    assert changed is True
+
+
+# 8. Idempotent on stable state
+def test_prune_idempotent_on_stable_state():
+    """Calling prune twice with the same inputs produces the same result the second time."""
+    # First call: some keys are missing → bookkeeping recorded, settings unchanged.
+    settings_a = _pruning_settings()
+    ps_a: dict = {"first_missed_at": {}}
+
+    settings_a, ps_a, changed_a = prune_stale_keys(
+        settings_a,
+        live_keys={"dev1:hidden", "dev1:personal-a"},  # work keys missing
+        pruning_state=ps_a,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+    assert changed_a is False  # within grace (first miss)
+
+    # Second call with SAME now — bookkeeping is the same, nothing changes.
+    import copy
+
+    settings_b = copy.deepcopy(settings_a)
+    ps_b = copy.deepcopy(ps_a)
+
+    settings_b, ps_b, changed_b = prune_stale_keys(
+        settings_b,
+        live_keys={"dev1:hidden", "dev1:personal-a"},
+        pruning_state=ps_b,
+        grace_seconds=_GRACE,
+        now=_T0,  # same now — grace not expired
+    )
+
+    assert changed_b is False
+    assert settings_b == settings_a
+    assert ps_b == ps_a
+
+
+# 9. Does NOT touch unrelated keys or memberships
+def test_prune_does_not_touch_unrelated_keys():
+    """Pruning one stale key leaves all live keys and other view memberships intact."""
+    settings = {
+        "hidden_sessions": ["dev1:stale", "dev1:live-hidden"],
+        "views": [
+            {
+                "name": "Work",
+                "sessions": ["dev1:stale", "dev1:live-work-1", "dev1:live-work-2"],
+            },
+        ],
+    }
+    pruning_state = {"first_missed_at": {"dev1:stale": _T0 - _GRACE - 1.0}}
+
+    updated, _, changed = prune_stale_keys(
+        settings,
+        live_keys={"dev1:live-hidden", "dev1:live-work-1", "dev1:live-work-2"},
+        pruning_state=pruning_state,
+        grace_seconds=_GRACE,
+        now=_T0,
+    )
+
+    assert changed is True
+    # Stale key gone
+    assert "dev1:stale" not in updated["hidden_sessions"]
+    assert "dev1:stale" not in updated["views"][0]["sessions"]
+    # Live keys untouched
+    assert "dev1:live-hidden" in updated["hidden_sessions"]
+    assert "dev1:live-work-1" in updated["views"][0]["sessions"]
+    assert "dev1:live-work-2" in updated["views"][0]["sessions"]

muxplex/views.py

@@ -1,5 +1,6 @@
 """
-Views invariant enforcement, visibility filtering, and validation for muxplex.
+Views invariant enforcement, visibility filtering, validation, and stale-key
+pruning for muxplex.
 
 Schema v2 semantics (see docs/plans/2026-05-17-hidden-state-redesign-design.md):
 - "hidden" is a property of a session, determined by membership in
@@ -16,6 +17,8 @@ Other invariants:
   `enforce_mutual_exclusion`.
 """
 
+import time
+
 RESERVED_VIEW_NAMES = frozenset({"all", "hidden"})
 MAX_VIEW_NAME_LENGTH = 30
 
@@ -282,3 +285,127 @@ def validate_view_name(name: str, existing_views: list[dict]) -> str | None:
     if trimmed in existing_names:
         return f"A view named '{trimmed}' already exists"
     return None
+
+
+# ---------------------------------------------------------------------------
+# Stale key pruning (Phase 4)
+#
+# Each device independently tracks which session keys it has failed to observe,
+# and prunes them from its own settings once the grace period expires.
+#
+# CRITICAL: pruning bookkeeping (first-missed-at timestamps) does NOT sync.
+# The bookkeeping lives in ~/.config/muxplex/pruning.json (see pruning.py).
+# The prune ACTION (removing keys from views/hidden_sessions) IS a normal
+# settings write and syncs via the existing LWW mechanism.
+# ---------------------------------------------------------------------------
+
+
+def prune_stale_keys(
+    settings: dict,
+    live_keys: set[str],
+    *,
+    pruning_state: dict | None = None,
+    grace_seconds: float = 86400.0,  # 24 hours
+    now: float | None = None,
+) -> tuple[dict, dict, bool]:
+    """Drop session keys that have been missing past the grace period.
+
+    Args:
+        settings: settings dict (will be mutated if pruning happens).
+        live_keys: the set of session keys that currently exist (live).
+        pruning_state: local bookkeeping dict of the form
+            {"first_missed_at": {key: timestamp}}. Defaults to an empty
+            structure if None. Mutated in place.
+        grace_seconds: how long a key may be missing before it's pruned.
+        now: current time (for testing). Defaults to time.time().
+
+    Returns:
+        (settings, pruning_state, settings_changed) — settings_changed is True
+        iff any key was actually removed from view.sessions or hidden_sessions.
+
+    Behavior:
+      1. For each key in settings.hidden_sessions or any view.sessions:
+         - If key in live_keys: drop the key from pruning_state["first_missed_at"]
+           (it's alive, nothing to prune).
+         - If key not in live_keys:
+             - If first_missed_at[key] is absent, record now.
+             - Else if now - first_missed_at[key] >= grace_seconds, remove the
+               key from hidden_sessions and from every view's sessions list;
+               drop the bookkeeping entry for it.
+             - Else: leave both alone (within grace).
+      2. The pruning_state dict is the source of truth for "when did we first
+         miss this key" — never check live_keys against pruning_state's keys
+         that aren't actually in stored settings (clean up bookkeeping for
+         keys that are no longer referenced anywhere).
+
+    Federation note: each device only knows its own live sessions natively.
+    If remote-device keys are also tracked in hidden_sessions or view.sessions,
+    this function will start their grace timer on the pruning device.  Because
+    the grace period is 24 hours by default, brief sync gaps will not cause
+    thrash.  A remote device's keys are also pruned by that remote device from
+    its own perspective; a deletion syncs back via the LWW mechanism.
+    """
+    if pruning_state is None:
+        pruning_state = {}
+    if "first_missed_at" not in pruning_state:
+        pruning_state["first_missed_at"] = {}
+
+    first_missed: dict[str, float] = pruning_state["first_missed_at"]
+
+    if now is None:
+        now = time.time()
+
+    # Collect all session keys currently referenced in settings.
+    all_settings_keys: set[str] = set()
+    for key in settings.get("hidden_sessions") or []:
+        all_settings_keys.add(key)
+    for view in settings.get("views") or []:
+        for key in view.get("sessions") or []:
+            all_settings_keys.add(key)
+
+    settings_changed = False
+
+    # Evaluate each referenced key.
+    for key in all_settings_keys:
+        if key in live_keys:
+            # Session is alive — clear any stale bookkeeping for it.
+            first_missed.pop(key, None)
+        else:
+            # Session is currently missing from live_keys.
+            if key not in first_missed:
+                # First time we notice it's missing — start the clock.
+                first_missed[key] = now
+            elif now - first_missed[key] >= grace_seconds:
+                # Grace period expired — remove the key from settings.
+                hidden = settings.get("hidden_sessions") or []
+                if key in hidden:
+                    settings["hidden_sessions"] = [k for k in hidden if k != key]
+                    settings_changed = True
+                for view in settings.get("views") or []:
+                    view_sessions = view.get("sessions") or []
+                    if key in view_sessions:
+                        view["sessions"] = [k for k in view_sessions if k != key]
+                        settings_changed = True
+                # Drop the bookkeeping entry now that the key is gone.
+                del first_missed[key]
+            # else: still within grace — leave settings and bookkeeping alone.
+
+    # Garbage-collect bookkeeping entries that no longer correspond to any key
+    # in settings (they may have been removed by an external edit, a peer sync,
+    # or a previous prune cycle that ran on another device).  This prevents
+    # first_missed_at from accumulating forever.
+    #
+    # Recompute the live settings key set AFTER the pruning loop above, so
+    # keys that were just pruned don't count as "referenced".
+    current_settings_keys: set[str] = set()
+    for key in settings.get("hidden_sessions") or []:
+        current_settings_keys.add(key)
+    for view in settings.get("views") or []:
+        for key in view.get("sessions") or []:
+            current_settings_keys.add(key)
+
+    for key in list(first_missed):
+        if key not in current_settings_keys:
+            del first_missed[key]
+
+    return settings, pruning_state, settings_changed