diff --git a/code_puppy/command_line/autosave_menu.py b/code_puppy/command_line/autosave_menu.py index 65000e316..2e3a664ab 100644 --- a/code_puppy/command_line/autosave_menu.py +++ b/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 diff --git a/code_puppy/session_storage.py b/code_puppy/session_storage.py index 62c6bfad1..d5507ff29 100644 --- a/code_puppy/session_storage.py +++ b/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], diff --git a/tests/command_line/test_autosave_menu.py b/tests/command_line/test_autosave_menu.py index 484cdcdf8..d248d4801 100644 --- a/tests/command_line/test_autosave_menu.py +++ b/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.""" diff --git a/tests/test_session_storage.py b/tests/test_session_storage.py index 339f9dc26..6e47072bf 100644 --- a/tests/test_session_storage.py +++ b/tests/test_session_storage.py @@ -9,9 +9,13 @@ from code_puppy.session_storage import ( cleanup_sessions, + is_pinned, list_sessions, + load_pins, load_session, + save_pins, save_session, + toggle_pin, ) @@ -81,3 +85,90 @@ def test_cleanup_sessions(tmp_path: Path, history: List[str], token_estimator): assert removed == ["session_earliest"] remaining = list_sessions(tmp_path) assert sorted(remaining) == sorted(["session_middle", "session_latest"]) + + +def test_load_pins_empty_when_no_file(tmp_path: Path): + assert load_pins(tmp_path) == set() + + +def test_toggle_pin_round_trip(tmp_path: Path, history: List[str], token_estimator): + # toggle_pin only pins sessions that actually exist on disk. + save_session( + history=history, + session_name="sess", + base_dir=tmp_path, + timestamp="2024-01-01T00:00:00", + token_estimator=token_estimator, + ) + + assert is_pinned(tmp_path, "sess") is False + + # First toggle pins it. + assert toggle_pin(tmp_path, "sess") is True + assert is_pinned(tmp_path, "sess") is True + assert load_pins(tmp_path) == {"sess"} + + # Second toggle unpins it. + assert toggle_pin(tmp_path, "sess") is False + assert is_pinned(tmp_path, "sess") is False + assert load_pins(tmp_path) == set() + + +def test_toggle_pin_refuses_missing_session(tmp_path: Path): + # Pinning a session that doesn't exist is a no-op (no ghost pins). + assert toggle_pin(tmp_path, "does_not_exist") is False + assert load_pins(tmp_path) == set() + + +def test_toggle_pin_prunes_stale_pins( + tmp_path: Path, history: List[str], token_estimator +): + # A pin left over for a now-deleted session gets pruned on next toggle. + save_session( + history=history, + session_name="real", + base_dir=tmp_path, + timestamp="2024-01-01T00:00:00", + token_estimator=token_estimator, + ) + # Seed the registry with a ghost entry directly. + save_pins(tmp_path, {"ghost", "real"}) + + # Toggling the real session prunes the ghost in the same write. + toggle_pin(tmp_path, "real") # was pinned -> now unpinned + assert load_pins(tmp_path) == set() + + +def test_save_and_load_pins(tmp_path: Path): + save_pins(tmp_path, {"a", "b", "c"}) + assert load_pins(tmp_path) == {"a", "b", "c"} + + +def test_load_pins_handles_corrupt_file(tmp_path: Path): + from code_puppy.session_storage import get_pins_path + + get_pins_path(tmp_path).write_text("not valid json{{", encoding="utf-8") + assert load_pins(tmp_path) == set() + + +def test_cleanup_sessions_protects_pinned( + tmp_path: Path, history: List[str], token_estimator +): + session_names = ["session_earliest", "session_middle", "session_latest"] + for index, name in enumerate(session_names): + metadata = save_session( + history=history, + session_name=name, + base_dir=tmp_path, + timestamp="2024-01-01T00:00:00", + token_estimator=token_estimator, + ) + os.utime(metadata.pickle_path, (0, index)) + + # Pin the oldest session - it must survive cleanup. + toggle_pin(tmp_path, "session_earliest") + + removed = cleanup_sessions(tmp_path, 2) + assert "session_earliest" not in removed + remaining = list_sessions(tmp_path) + assert "session_earliest" in remaining