Merge branch 'pr-3184' into release/stage-batch55

# Conflicts:
#	CHANGELOG.md

CHANGELOG.md

@@ -15,6 +15,8 @@
 - Windows native runs now skip the POSIX-only turn-journal directory fsync instead of raising `AttributeError` for missing `os.O_DIRECTORY` on every submitted turn (#3170).
 - `/api/media` now treats Windows cross-drive `commonpath()` comparisons as non-matches instead of 500ing when media paths and allowed roots live on different drives (#3171).
 
+- 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)
 

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,10 +2,12 @@
 
 import json
 import pathlib
+import time
+from types import SimpleNamespace
 import urllib.error
 import urllib.request
 
-from tests._pytest_port import BASE
+from tests._pytest_port import BASE, TEST_STATE_DIR
 
 
 ROOT = pathlib.Path(__file__).resolve().parent.parent
@@ -41,6 +43,41 @@ def make_session(created):
     return sid
 
 
+
+def inject_hidden_pinned_snapshot(sid="hidden-pinned-snapshot"):
+    """Add a persisted legacy hidden snapshot without touching server memory."""
+    sessions_dir = TEST_STATE_DIR / "sessions"
+    sessions_dir.mkdir(parents=True, exist_ok=True)
+    now = time.time()
+    row = {
+        "session_id": sid,
+        "title": "Hidden pinned snapshot",
+        "workspace": str(TEST_STATE_DIR / "test-workspace"),
+        "model": "test/pin-cap",
+        "created_at": now,
+        "updated_at": now,
+        "last_message_at": now,
+        "message_count": 1,
+        "messages": [{"role": "user", "content": "legacy hidden snapshot"}],
+        "tool_calls": [],
+        "pinned": True,
+        "archived": False,
+        "pre_compression_snapshot": True,
+        "_show_pre_compression_snapshot": False,
+    }
+    (sessions_dir / f"{sid}.json").write_text(json.dumps(row), encoding="utf-8")
+    index_path = sessions_dir / "_index.json"
+    try:
+        index = json.loads(index_path.read_text(encoding="utf-8"))
+    except (FileNotFoundError, json.JSONDecodeError):
+        index = []
+    compact = {k: v for k, v in row.items() if k not in {"messages", "tool_calls"}}
+    index = [item for item in index if item.get("session_id") != sid]
+    index.append(compact)
+    index_path.write_text(json.dumps(index), encoding="utf-8")
+    return sid
+
+
 def test_session_pin_endpoint_caps_pinned_sessions_at_three():
     created = []
     try:
@@ -67,6 +104,64 @@ def test_session_pin_endpoint_caps_pinned_sessions_at_three():
             post("/api/session/delete", {"session_id": sid})
 
 
+def test_session_pin_endpoint_ignores_hidden_snapshot_when_enforcing_cap():
+    created = []
+    hidden_sid = "hidden-pinned-snapshot-quota-route"
+    try:
+        hidden = inject_hidden_pinned_snapshot(hidden_sid)
+        pinned = [make_session(created) for _ in range(2)]
+        for sid in pinned:
+            d, status = post("/api/session/pin", {"session_id": sid, "pinned": True})
+            assert status == 200
+            assert d["session"]["pinned"] is True
+
+        third_visible = make_session(created)
+        d, status = post("/api/session/pin", {"session_id": third_visible, "pinned": True})
+        assert status == 200, d
+        assert d["session"]["pinned"] is True
+        assert hidden not in {third_visible, *pinned}
+    finally:
+        for sid in created:
+            post("/api/session/delete", {"session_id": sid})
+        (TEST_STATE_DIR / "sessions" / f"{hidden_sid}.json").unlink(missing_ok=True)
+        index_path = TEST_STATE_DIR / "sessions" / "_index.json"
+        try:
+            index = json.loads(index_path.read_text(encoding="utf-8"))
+            index = [item for item in index if item.get("session_id") != hidden_sid]
+            index_path.write_text(json.dumps(index), encoding="utf-8")
+        except (FileNotFoundError, json.JSONDecodeError):
+            pass
+
+
+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_field(existing, \"archived\", False)" in ROUTES_PY
+    assert "_session_counts_toward_pin_quota(existing)" 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)