feat: pin sessions in the /resume picker

code_puppy/command_line/autosave_menu.py

@@ -6,6 +6,7 @@
 
 import asyncio
 import json
+import logging
 import sys
 from datetime import datetime
 from io import StringIO
@@ -33,9 +34,16 @@
 )
 from code_puppy.callbacks import on_prompt_toolkit_style
 from code_puppy.config import AUTOSAVE_DIR
-from code_puppy.session_storage import list_sessions, load_session
+from code_puppy.session_storage import (
+    list_sessions,
+    load_pins,
+    load_session,
+    toggle_pin,
+)
 from code_puppy.tools.command_runner import set_awaiting_user_input
 
+logger = logging.getLogger(__name__)
+
 PAGE_SIZE = 15  # Sessions per page
 
 
@@ -49,6 +57,33 @@ def _get_session_metadata(base_dir: Path, session_name: str) -> dict:
         return {}
 
 
+def _order_entries(
+    entries: List[Tuple[str, dict]],
+) -> List[Tuple[str, dict]]:
+    """Order entries for display: pinned first, then most-recent-first.
+
+    Pin state is read from each entry's ``metadata['pinned']`` flag
+    (populated by :func:`_get_session_entries`). Pulled out as a standalone
+    helper so the initial load and the live pin-toggle share one ordering rule.
+    """
+
+    def sort_key(entry):
+        _, metadata = entry
+        timestamp = metadata.get("timestamp")
+        if timestamp:
+            try:
+                return datetime.fromisoformat(timestamp)
+            except ValueError:
+                return datetime.min
+        return datetime.min
+
+    pinned = [e for e in entries if e[1].get("pinned")]
+    rest = [e for e in entries if not e[1].get("pinned")]
+    pinned.sort(key=sort_key, reverse=True)
+    rest.sort(key=sort_key, reverse=True)
+    return pinned + rest
+
+
 def _get_session_entries(base_dir: Path) -> List[Tuple[str, dict]]:
     """Get all sessions with their metadata, most recent first."""
     try:
@@ -57,27 +92,19 @@ def _get_session_entries(base_dir: Path) -> List[Tuple[str, dict]]:
         return []
 
     entries = []
+    pins = load_pins(base_dir)
 
     for name in sessions:
         try:
             metadata = _get_session_metadata(base_dir, name)
         except (FileNotFoundError, PermissionError):
             metadata = {}
+        # Stash pin state in the (display-only) metadata dict so the renderer
+        # can show a marker without re-reading the registry per row.
+        metadata["pinned"] = name in pins
         entries.append((name, metadata))
 
-    # Sort by timestamp (most recent first)
-    def sort_key(entry):
-        _, metadata = entry
-        timestamp = metadata.get("timestamp")
-        if timestamp:
-            try:
-                return datetime.fromisoformat(timestamp)
-            except ValueError:
-                return datetime.min
-        return datetime.min
-
-    entries.sort(key=sort_key, reverse=True)
-    return entries
+    return _order_entries(entries)
 
 
 def _extract_last_user_message(history: list) -> str:
@@ -236,10 +263,16 @@ def _render_menu_panel(
             label = f"{time_str} \u2022 {msg_count} msgs ({session_name})"
 
         # Highlight selected item
-        if is_selected:
-            lines.append(("class:tui.selected", f" > {label}"))
+        is_session_pinned = bool(metadata.get("pinned"))
+        row_style = "class:tui.selected" if is_selected else "class:tui.muted"
+        caret = " > " if is_selected else "   "
+        lines.append((row_style, caret))
+        # Pin marker column (keeps rows aligned when not pinned).
+        if is_session_pinned:
+            lines.append(("class:tui.success", "* "))
         else:
-            lines.append(("class:tui.muted", f"   {label}"))
+            lines.append(("", "  "))
+        lines.append((row_style, label))
 
         lines.append(("", "\n"))
 
@@ -259,6 +292,8 @@ def _render_menu_panel(
         lines.append(("", "Browse msgs\n"))
         lines.append(("class:tui.help-key", "  /   "))
         lines.append(("", "Search content\n"))
+        lines.append(("class:tui.help-key", "  p   "))
+        lines.append(("", "Pin/Unpin\n"))
     lines.append(("class:tui.help-key", "  Enter  "))
     lines.append(("", "Load\n"))
     lines.append(("class:tui.help-key", "  Ctrl+C "))
@@ -383,6 +418,10 @@ def _render_preview_panel(base_dir: Path, entry: Optional[Tuple[str, dict]]) ->
     lines.append(("", session_name))
     lines.append(("", "\n"))
 
+    if metadata.get("pinned"):
+        lines.append(("class:tui.success", "  * Pinned"))
+        lines.append(("", "\n"))
+
     timestamp = metadata.get("timestamp", "unknown")
     try:
         dt = datetime.fromisoformat(timestamp)
@@ -579,8 +618,10 @@ def _filter_entries(needle: str) -> List[Tuple[str, dict]]:
         this is safe to invoke from an ``asyncio.to_thread`` worker.
         """
         if not needle:
-            return list(entries)
-        return [e for e in entries if entry_matches(e, needle, content_index, base_dir)]
+            return _order_entries(list(entries))
+        return _order_entries(
+            [e for e in entries if entry_matches(e, needle, content_index, base_dir)]
+        )
 
     def _apply_filter_result(filtered: List[Tuple[str, dict]]) -> None:
         """Apply a filter result to picker state. Must run on the main thread."""
@@ -756,7 +797,13 @@ def _(event):
                 message_idx[0] = 0  # Start at most recent
                 update_display()
             except Exception:
-                pass  # Silently fail if can't load
+                # Can't load this session (e.g. unpicklable or missing).
+                # Stay in list mode; log so the failure is debuggable.
+                logger.warning(
+                    "Failed to load session %r for browsing",
+                    session_name,
+                    exc_info=True,
+                )
 
     @kb.add("escape")
     def _(event):
@@ -842,6 +889,43 @@ def _alpha(event, _c=_append):
                 search_buffer[0] += _c
                 update_display()
 
+    @kb.add("p")
+    def _(event):
+        """Pin/unpin the selected session -- unless we're typing a search."""
+        # While typing a search query, 'p' is just a character. This binding
+        # is registered after the alphabet loop so it owns the 'p' key.
+        if in_search_mode[0]:
+            search_buffer[0] += "p"
+            update_display()
+            return
+        if browse_mode[0]:
+            return  # Pinning only makes sense in the session list
+        entry = get_current_entry()
+        if not entry:
+            return
+        session_name, metadata = entry
+        try:
+            new_state = toggle_pin(base_dir, session_name)
+        except Exception:
+            # Don't crash the picker on a write hiccup; log so it's debuggable.
+            logger.warning(
+                "Failed to toggle pin for session %r",
+                session_name,
+                exc_info=True,
+            )
+            return
+        # Reflect the new flag on the shared metadata dict, then re-order the
+        # visible list (honoring any active search) and keep this session
+        # selected so the cursor follows it to its new position.
+        metadata["pinned"] = new_state
+        update_visible_entries()
+        for i, (name, _meta) in enumerate(visible_entries[0]):
+            if name == session_name:
+                selected_idx[0] = i
+                current_page[0] = get_page_for_index(i, PAGE_SIZE)
+                break
+        update_display()
+
     @kb.add("c-c")
     def _(event):
         result[0] = None

code_puppy/session_storage.py

@@ -9,7 +9,9 @@
 from __future__ import annotations
 
 import json
+import os
 import pickle
+import tempfile
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable, List
@@ -20,6 +22,30 @@ def _safe_loads(data: bytes) -> Any:
     return pickle.loads(data)  # noqa: S301
 
 
+def _atomic_write_text(path: Path, text: str) -> None:
+    """Write ``text`` to ``path`` atomically and durably.
+
+    Writes to a uniquely-named temp file in the same directory, flushes and
+    fsyncs it, then atomically ``replace()``s the target. The unique temp name
+    (vs. a fixed ``.tmp``) avoids races and lost updates between concurrent
+    writers, and the fsync gives us crash safety.
+    """
+    ensure_directory(path.parent)
+    fd, tmp_name = tempfile.mkstemp(
+        prefix=f"{path.name}.", suffix=".tmp", dir=str(path.parent)
+    )
+    tmp_path = Path(tmp_name)
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(text)
+            handle.flush()
+            os.fsync(handle.fileno())
+        tmp_path.replace(path)
+    finally:
+        # If replace() succeeded the temp file is gone; otherwise clean it up.
+        tmp_path.unlink(missing_ok=True)
+
+
 _LEGACY_SIGNED_HEADER = b"CPSESSION\x01"
 _LEGACY_SIGNATURE_SIZE = (
     32  # legacy signature bytes, retained only for backward-compat parsing
@@ -138,6 +164,72 @@ def list_sessions(base_dir: Path) -> List[str]:
     return sorted(path.stem for path in base_dir.glob("*.pkl"))
 
 
+# ---------------------------------------------------------------------------
+# Pinned sessions
+# ---------------------------------------------------------------------------
+# Pin state is kept in a dedicated registry file rather than each session's
+# *_meta.json. The meta files get rewritten on every autosave, which would
+# silently clobber the pin flag on the active session. A single small registry
+# keeps the pin state stable and decoupled (one source of truth).
+
+PINS_FILENAME = "pinned_sessions.json"
+
+
+def get_pins_path(base_dir: Path) -> Path:
+    """Return the path to the pinned-sessions registry file."""
+    return base_dir / PINS_FILENAME
+
+
+def load_pins(base_dir: Path) -> set[str]:
+    """Load the set of pinned session names. Returns empty set on any issue."""
+    pins_path = get_pins_path(base_dir)
+    try:
+        with pins_path.open("r", encoding="utf-8") as pins_file:
+            data = json.load(pins_file)
+    except (FileNotFoundError, json.JSONDecodeError, OSError):
+        return set()
+    if isinstance(data, list):
+        return {str(name) for name in data}
+    return set()
+
+
+def save_pins(base_dir: Path, pins: set[str]) -> None:
+    """Persist the set of pinned session names atomically and durably."""
+    ensure_directory(base_dir)
+    _atomic_write_text(get_pins_path(base_dir), json.dumps(sorted(pins), indent=2))
+
+
+def is_pinned(base_dir: Path, session_name: str) -> bool:
+    """Return True if the given session is currently pinned."""
+    return session_name in load_pins(base_dir)
+
+
+def toggle_pin(base_dir: Path, session_name: str) -> bool:
+    """Toggle the pin state for a session. Returns the new state (True=pinned).
+
+    Also prunes any pins whose sessions no longer exist on disk, and refuses to
+    pin a session that doesn't exist, so the registry never accumulates ghosts.
+    """
+    pins = load_pins(base_dir)
+    existing = set(list_sessions(base_dir))
+
+    # Drop stale pins for sessions that have since been deleted.
+    pins &= existing
+
+    if session_name in pins:
+        pins.discard(session_name)
+        new_state = False
+    elif session_name in existing:
+        pins.add(session_name)
+        new_state = True
+    else:
+        # Session doesn't exist - nothing to pin. Still persist any pruning.
+        new_state = False
+
+    save_pins(base_dir, pins)
+    return new_state
+
+
 def cleanup_sessions(base_dir: Path, max_sessions: int) -> List[str]:
     if max_sessions <= 0:
         return []
@@ -149,6 +241,12 @@ def cleanup_sessions(base_dir: Path, max_sessions: int) -> List[str]:
     if len(candidate_paths) <= max_sessions:
         return []
 
+    # Never auto-delete pinned sessions - that's the whole point of pinning.
+    pinned = load_pins(base_dir)
+    candidate_paths = [p for p in candidate_paths if p.stem not in pinned]
+    if len(candidate_paths) <= max_sessions:
+        return []
+
     sorted_candidates = sorted(
         ((path.stat().st_mtime, path) for path in candidate_paths),
         key=lambda item: item[0],

tests/command_line/test_autosave_menu.py

@@ -515,7 +515,9 @@ def test_with_permission_denied_access(self):
                 entries = _get_session_entries(Path("/protected/path"))
                 # Should handle permission errors gracefully
                 assert len(entries) == 1
-                assert entries[0][1] == {}  # metadata should be empty due to error
+                # Real metadata failed to load; only the always-present pin
+                # flag remains (defaults to False when not pinned).
+                assert entries[0][1] == {"pinned": False}
 
     def test_console_output_and_ansi_sequences(self):
         """Test that console output includes proper ANSI sequences."""