fix: exclude hidden snapshots from pin quota

CHANGELOG.md

@@ -3,6 +3,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Hidden pre-compression snapshots no longer keep stale pin state or count toward the visible pinned-session quota (#3181).
+
 ## [v0.51.171] — 2026-05-30 — Release EQ (stage-batch53 — tool-output card badge + Neon opt-in skin)
 
 ### Added

api/routes.py

@@ -99,6 +99,25 @@ def _session_field(session, field, default=None):
     return getattr(session, field, default)
 
 
+def _session_counts_toward_pin_quota(session) -> bool:
+    """Return True when a pinned session should consume visible pin quota."""
+    if not _session_field(session, "pinned", False):
+        return False
+    if _session_field(session, "archived", False):
+        return False
+    if isinstance(session, dict):
+        row = session
+    elif hasattr(session, "compact"):
+        row = session.compact()
+    else:
+        row = {
+            "pre_compression_snapshot": _session_field(session, "pre_compression_snapshot", False),
+            "source_tag": _session_field(session, "source_tag", None),
+            "default_hidden": _session_field(session, "default_hidden", False),
+        }
+    return not _hide_from_default_sidebar(row)
+
+
 # ── Profile-scoped session/project filtering (#1611, #1614) ────────────────
 #
 # Sessions and projects are stored in the WebUI sidecar without per-row
@@ -2730,6 +2749,7 @@ from api.models import (
     merge_session_messages_append_only,
     _session_message_merge_key,
     _is_empty_partial_activity_message,
+    _hide_from_default_sidebar,
     prune_session_from_index,
     ensure_cron_project,
     is_cron_session,
@@ -6509,7 +6529,7 @@ def handle_post(handler, parsed) -> bool:
             # so must run outside our own LOCK acquire below).
             persisted_pinned_ids = {
                 _session_field(existing, "session_id", None) for existing in all_sessions()
-                if _session_field(existing, "pinned", False) and not _session_field(existing, "archived", False)
+                if _session_counts_toward_pin_quota(existing)
             }
             with LOCK:
                 # Final authoritative count: merge persisted-pinned with the
@@ -6519,7 +6539,7 @@ def handle_post(handler, parsed) -> bool:
                 pinned_ids = set(persisted_pinned_ids)
                 pinned_ids.update(
                     sid for sid, existing in SESSIONS.items()
-                    if getattr(existing, "pinned", False) and not getattr(existing, "archived", False)
+                    if _session_counts_toward_pin_quota(existing)
                 )
                 pinned_ids.discard(body["session_id"])
                 pinned_sessions_limit = int(load_settings().get("pinned_sessions_limit", 3) or 3)

api/streaming.py

@@ -2286,8 +2286,10 @@ def _preserve_pre_compression_snapshot(s, old_sid: str) -> None:
             # pre-existing parent_session_id lineage.
             saved_sid = s.session_id
             saved_snapshot = bool(getattr(s, 'pre_compression_snapshot', False))
+            saved_pinned = bool(getattr(s, 'pinned', False))
             s.session_id = old_sid
             s.pre_compression_snapshot = True
+            s.pinned = False
             # Stage-359 / PR #2295: clear runtime stream-state fields on the
             # archived snapshot so the sidebar does not reopen the parent as
             # a permanently-running session while the child already holds the
@@ -2314,6 +2316,7 @@ def _preserve_pre_compression_snapshot(s, old_sid: str) -> None:
             finally:
                 s.session_id = saved_sid
                 s.pre_compression_snapshot = saved_snapshot
+                s.pinned = saved_pinned
                 s.active_stream_id = saved_active_stream_id
                 s.pending_user_message = saved_pending_user_message
                 s.pending_attachments = saved_pending_attachments
@@ -2326,6 +2329,7 @@ def _preserve_pre_compression_snapshot(s, old_sid: str) -> None:
         snapshot = Session.load(old_sid)
         if snapshot:
             snapshot.pre_compression_snapshot = True
+            snapshot.pinned = False
             # Stage-359 Opus SHOULD-FIX: clear runtime fields on the loaded
             # snapshot too. If the disk snapshot was last persisted while the
             # parent was live, it could carry a stale active_stream_id /

tests/test_compression_snapshot_runtime_clear.py

@@ -9,6 +9,7 @@ class FakeSession:
         self.session_id = "new_session"
         self.parent_session_id = "original_parent"
         self.pre_compression_snapshot = False
+        self.pinned = True
         self.active_stream_id = "live-stream"
         self.pending_user_message = "current prompt"
         self.pending_attachments = [{"name": "file.txt"}]
@@ -21,6 +22,7 @@ class FakeSession:
             "session_id": self.session_id,
             "parent_session_id": self.parent_session_id,
             "pre_compression_snapshot": self.pre_compression_snapshot,
+            "pinned": self.pinned,
             "active_stream_id": self.active_stream_id,
             "pending_user_message": self.pending_user_message,
             "pending_attachments": list(self.pending_attachments),
@@ -43,6 +45,7 @@ def test_preserve_pre_compression_snapshot_clears_runtime_fields_while_restoring
         "session_id": "old_session",
         "parent_session_id": "original_parent",
         "pre_compression_snapshot": True,
+        "pinned": False,
         "active_stream_id": None,
         "pending_user_message": None,
         "pending_attachments": [],
@@ -52,6 +55,7 @@ def test_preserve_pre_compression_snapshot_clears_runtime_fields_while_restoring
     }
     assert session.session_id == "new_session"
     assert session.pre_compression_snapshot is False
+    assert session.pinned is True
     assert session.active_stream_id == "live-stream"
     assert session.pending_user_message == "current prompt"
     assert session.pending_attachments == [{"name": "file.txt"}]
@@ -59,6 +63,7 @@ def test_preserve_pre_compression_snapshot_clears_runtime_fields_while_restoring
 
     saved = json.loads((tmp_path / "old_session.json").read_text(encoding="utf-8"))
     assert saved["pre_compression_snapshot"] is True
+    assert saved["pinned"] is False
     assert saved["active_stream_id"] is None
     assert saved["pending_user_message"] is None
     assert saved["pending_attachments"] == []
@@ -77,6 +82,7 @@ def test_preserve_pre_compression_snapshot_load_and_mark_branch_clears_runtime_f
             {"role": "user", "content": "newer prompt already on disk"},
         ],
         "pre_compression_snapshot": True,
+        "pinned": True,
         "active_stream_id": "stale-stream",
         "pending_user_message": "stale prompt",
         "pending_attachments": [{"name": "stale.txt"}],
@@ -91,6 +97,7 @@ def test_preserve_pre_compression_snapshot_load_and_mark_branch_clears_runtime_f
     saved = json.loads((tmp_path / "old_session.json").read_text(encoding="utf-8"))
     assert saved["messages"] == old_payload["messages"]
     assert saved["pre_compression_snapshot"] is True
+    assert saved["pinned"] is False
     assert saved["active_stream_id"] is None
     assert saved["pending_user_message"] is None
     assert saved["pending_attachments"] == []

tests/test_issue2508_session_pin_cap.py

@@ -2,6 +2,7 @@
 
 import json
 import pathlib
+from types import SimpleNamespace
 import urllib.error
 import urllib.request
 
@@ -67,6 +68,35 @@ def test_session_pin_endpoint_caps_pinned_sessions_at_three():
             post("/api/session/delete", {"session_id": sid})
 
 
+def test_hidden_pre_compression_snapshot_does_not_count_toward_pin_quota():
+    from api.routes import _session_counts_toward_pin_quota
+
+    assert _session_counts_toward_pin_quota({
+        "session_id": "hidden-snapshot",
+        "pinned": True,
+        "archived": False,
+        "pre_compression_snapshot": True,
+    }) is False
+    assert _session_counts_toward_pin_quota({
+        "session_id": "visible-session",
+        "pinned": True,
+        "archived": False,
+        "pre_compression_snapshot": False,
+    }) is True
+
+
+def test_hidden_in_memory_snapshot_does_not_count_toward_pin_quota():
+    from api.routes import _session_counts_toward_pin_quota
+
+    snapshot = SimpleNamespace(
+        session_id="hidden-memory-snapshot",
+        pinned=True,
+        archived=False,
+        pre_compression_snapshot=True,
+    )
+    assert _session_counts_toward_pin_quota(snapshot) is False
+
+
 def test_session_pin_cap_has_backend_and_frontend_guards():
     assert 'pinned_ids = {' in ROUTES_PY
     assert 'pinned_ids.update(' in ROUTES_PY

tests/test_issue2821_session_pin_state_sync.py

@@ -41,9 +41,10 @@ def test_session_field_helper_reads_dicts_and_objects():
 
 
 def test_pin_limit_snapshot_counts_index_dict_entries():
+    assert "def _session_counts_toward_pin_quota(session)" in ROUTES_PY
     assert "_session_field(existing, \"session_id\", None)" in ROUTES_PY
-    assert "_session_field(existing, \"pinned\", False)" in ROUTES_PY
+    assert "_session_counts_toward_pin_quota(existing)" in ROUTES_PY
-    assert "_session_field(existing, \"archived\", False)" in ROUTES_PY
+    assert "_hide_from_default_sidebar(row)" in ROUTES_PY
     start = ROUTES_PY.find("persisted_pinned_ids = {")
     assert start != -1, "persisted pin snapshot not found"
     end = ROUTES_PY.find("with LOCK:", start)