diff --git a/cli.py b/cli.py index a326c93db2..6fadd06a45 100755 --- a/cli.py +++ b/cli.py @@ -1094,6 +1094,16 @@ class HermesCLI: self.conversation_history: List[Dict[str, Any]] = [] self.session_start = datetime.now() self._resumed = False + # Initialize SQLite session store early so /title works before first message + self._session_db = None + try: + from hermes_state import SessionDB + self._session_db = SessionDB() + except Exception: + pass + + # Deferred title: stored in memory until the session is created in the DB + self._pending_title: Optional[str] = None # Session ID: reuse existing one when resuming, otherwise generate fresh if resume: @@ -1181,13 +1191,13 @@ class HermesCLI: if not self._ensure_runtime_credentials(): return False - # Initialize SQLite session store for CLI sessions - self._session_db = None - try: - from hermes_state import SessionDB - self._session_db = SessionDB() - except Exception as e: - logger.debug("SQLite session store not available: %s", e) + # Initialize SQLite session store for CLI sessions (if not already done in __init__) + if self._session_db is None: + try: + from hermes_state import SessionDB + self._session_db = SessionDB() + except Exception as e: + logger.debug("SQLite session store not available: %s", e) # If resuming, validate the session exists and load its history if self._resumed and self._session_db: @@ -1200,8 +1210,11 @@ class HermesCLI: if restored: self.conversation_history = restored msg_count = len([m for m in restored if m.get("role") == "user"]) + title_part = "" + if session_meta.get("title"): + title_part = f" \"{session_meta['title']}\"" _cprint( - f"{_GOLD}↻ Resumed session {_BOLD}{self.session_id}{_RST}{_GOLD} " + f"{_GOLD}↻ Resumed session {_BOLD}{self.session_id}{_RST}{_GOLD}{title_part} " f"({msg_count} user message{'s' if msg_count != 1 else ''}, " f"{len(restored)} total messages){_RST}" ) @@ -1243,6 +1256,15 @@ class HermesCLI: clarify_callback=self._clarify_callback, honcho_session_key=self.session_id, ) + # Apply any pending title now that the session exists in the DB + if self._pending_title and self._session_db: + try: + self._session_db.set_session_title(self.session_id, self._pending_title) + _cprint(f" Session title applied: {self._pending_title}") + self._pending_title = None + except (ValueError, Exception) as e: + _cprint(f" Could not apply pending title: {e}") + self._pending_title = None return True except Exception as e: self.console.print(f"[bold red]Failed to initialize agent: {e}[/]") @@ -2091,6 +2113,47 @@ class HermesCLI: print(" ✨ (◕‿◕)✨ Fresh start! Screen cleared and conversation reset.\n") elif cmd_lower == "/history": self.show_history() + elif cmd_lower.startswith("/title"): + parts = cmd_original.split(maxsplit=1) + if len(parts) > 1: + new_title = parts[1].strip() + if new_title: + if self._session_db: + # Check if session exists in DB yet + session = self._session_db.get_session(self.session_id) + if session: + try: + if self._session_db.set_session_title(self.session_id, new_title): + _cprint(f" Session title set: {new_title}") + else: + _cprint(" Session not found in database.") + except ValueError as e: + _cprint(f" {e}") + else: + # Session not created yet — defer the title + # Check uniqueness proactively + existing = self._session_db.get_session_by_title(new_title) + if existing: + _cprint(f" Title '{new_title}' is already in use by session {existing['id']}") + else: + self._pending_title = new_title + _cprint(f" Session title queued: {new_title} (will be saved on first message)") + else: + _cprint(" Session database not available.") + else: + _cprint(" Usage: /title ") + else: + # Show current title if no argument given + if self._session_db: + session = self._session_db.get_session(self.session_id) + if session and session.get("title"): + _cprint(f" Session title: {session['title']}") + elif self._pending_title: + _cprint(f" Session title (pending): {self._pending_title}") + else: + _cprint(f" No title set. Usage: /title ") + else: + _cprint(" Session database not available.") elif cmd_lower in ("/reset", "/new"): self.reset_conversation() elif cmd_lower.startswith("/model"): diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py index 61c5864fd6..20f01b1748 100644 --- a/hermes_cli/commands.py +++ b/hermes_cli/commands.py @@ -34,6 +34,7 @@ COMMANDS = { "/platforms": "Show gateway/messaging platform status", "/verbose": "Cycle tool progress display: off → new → all → verbose", "/compress": "Manually compress conversation context (flush memories + summarize)", + "/title": "Set a title for the current session (usage: /title My Session Name)", "/usage": "Show token usage for the current session", "/insights": "Show usage insights and analytics (last 30 days)", "/paste": "Check clipboard for an image and attach it", diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 20f33998ad..5ba09c35a8 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -120,16 +120,63 @@ def _resolve_last_cli_session() -> Optional[str]: return None +def _resolve_session_by_name_or_id(name_or_id: str) -> Optional[str]: + """Resolve a session name (title) or ID to a session ID. + + - If it looks like a session ID (contains underscore + hex), try direct lookup first. + - Otherwise, treat it as a title and use resolve_session_by_title (auto-latest). + - Falls back to the other method if the first doesn't match. + """ + try: + from hermes_state import SessionDB + db = SessionDB() + + # Try as exact session ID first + session = db.get_session(name_or_id) + if session: + db.close() + return session["id"] + + # Try as title (with auto-latest for lineage) + session_id = db.resolve_session_by_title(name_or_id) + db.close() + return session_id + except Exception: + pass + return None + + def cmd_chat(args): """Run interactive chat CLI.""" - # Resolve --continue into --resume with the latest CLI session - if getattr(args, "continue_last", False) and not getattr(args, "resume", None): - last_id = _resolve_last_cli_session() - if last_id: - args.resume = last_id + # Resolve --continue into --resume with the latest CLI session or by name + continue_val = getattr(args, "continue_last", None) + if continue_val and not getattr(args, "resume", None): + if isinstance(continue_val, str): + # -c "session name" — resolve by title or ID + resolved = _resolve_session_by_name_or_id(continue_val) + if resolved: + args.resume = resolved + else: + print(f"No session found matching '{continue_val}'.") + print("Use 'hermes sessions list' to see available sessions.") + sys.exit(1) else: - print("No previous CLI session found to continue.") - sys.exit(1) + # -c with no argument — continue the most recent session + last_id = _resolve_last_cli_session() + if last_id: + args.resume = last_id + else: + print("No previous CLI session found to continue.") + sys.exit(1) + + # Resolve --resume by title if it's not a direct session ID + resume_val = getattr(args, "resume", None) + if resume_val: + resolved = _resolve_session_by_name_or_id(resume_val) + if resolved: + args.resume = resolved + # If resolution fails, keep the original value — _init_agent will + # report "Session not found" with the original input # First-run guard: check if any provider is configured before launching if not _has_any_provider_configured(): @@ -1209,8 +1256,9 @@ def main(): Examples: hermes Start interactive chat hermes chat -q "Hello" Single query mode - hermes --continue Resume the most recent session - hermes --resume Resume a specific session + hermes -c Resume the most recent session + hermes -c "my project" Resume a session by name (latest in lineage) + hermes --resume Resume a specific session by ID hermes setup Run setup wizard hermes logout Clear stored authentication hermes model Select default model @@ -1221,6 +1269,7 @@ Examples: hermes -w Start in isolated git worktree hermes gateway install Install as system service hermes sessions list List past sessions + hermes sessions rename ID T Rename/title a session hermes update Update to latest version For more help on a command: @@ -1235,16 +1284,18 @@ For more help on a command: ) parser.add_argument( "--resume", "-r", - metavar="SESSION_ID", + metavar="SESSION", default=None, - help="Resume a previous session by ID (shortcut for: hermes chat --resume ID)" + help="Resume a previous session by ID or title" ) parser.add_argument( "--continue", "-c", dest="continue_last", - action="store_true", - default=False, - help="Resume the most recent CLI session" + nargs="?", + const=True, + default=None, + metavar="SESSION_NAME", + help="Resume a session by name, or the most recent if no name given" ) parser.add_argument( "--worktree", "-w", @@ -1294,9 +1345,11 @@ For more help on a command: chat_parser.add_argument( "--continue", "-c", dest="continue_last", - action="store_true", - default=False, - help="Resume the most recent CLI session" + nargs="?", + const=True, + default=None, + metavar="SESSION_NAME", + help="Resume a session by name, or the most recent if no name given" ) chat_parser.add_argument( "--worktree", "-w", @@ -1696,6 +1749,10 @@ For more help on a command: sessions_stats = sessions_subparsers.add_parser("stats", help="Show session store statistics") + sessions_rename = sessions_subparsers.add_parser("rename", help="Set or change a session's title") + sessions_rename.add_argument("session_id", help="Session ID to rename") + sessions_rename.add_argument("title", nargs="+", help="New title for the session") + def cmd_sessions(args): import json as _json try: @@ -1708,18 +1765,51 @@ For more help on a command: action = args.sessions_action if action == "list": - sessions = db.search_sessions(source=args.source, limit=args.limit) + sessions = db.list_sessions_rich(source=args.source, limit=args.limit) if not sessions: print("No sessions found.") return - print(f"{'ID':<30} {'Source':<12} {'Model':<30} {'Messages':>8} {'Started'}") - print("─" * 100) from datetime import datetime + import time as _time + + def _relative_time(ts): + """Format a timestamp as relative time (e.g., '2h ago', 'yesterday').""" + if not ts: + return "?" + delta = _time.time() - ts + if delta < 60: + return "just now" + elif delta < 3600: + mins = int(delta / 60) + return f"{mins}m ago" + elif delta < 86400: + hours = int(delta / 3600) + return f"{hours}h ago" + elif delta < 172800: + return "yesterday" + elif delta < 604800: + days = int(delta / 86400) + return f"{days}d ago" + else: + return datetime.fromtimestamp(ts).strftime("%Y-%m-%d") + + has_titles = any(s.get("title") for s in sessions) + if has_titles: + print(f"{'Title':<22} {'Preview':<40} {'Last Active':<13} {'ID'}") + print("─" * 100) + else: + print(f"{'Preview':<50} {'Last Active':<13} {'Src':<6} {'ID'}") + print("─" * 90) for s in sessions: - started = datetime.fromtimestamp(s["started_at"]).strftime("%Y-%m-%d %H:%M") if s["started_at"] else "?" - model = (s.get("model") or "?")[:28] - ended = " (ended)" if s.get("ended_at") else "" - print(f"{s['id']:<30} {s['source']:<12} {model:<30} {s['message_count']:>8} {started}{ended}") + last_active = _relative_time(s.get("last_active")) + preview = s.get("preview", "")[:38] if has_titles else s.get("preview", "")[:48] + if has_titles: + title = (s.get("title") or "—")[:20] + sid = s["id"][:20] + print(f"{title:<22} {preview:<40} {last_active:<13} {sid}") + else: + sid = s["id"][:20] + print(f"{preview:<50} {last_active:<13} {s['source']:<6} {sid}") elif action == "export": if args.session_id: @@ -1759,6 +1849,16 @@ For more help on a command: count = db.prune_sessions(older_than_days=days, source=args.source) print(f"Pruned {count} session(s).") + elif action == "rename": + title = " ".join(args.title) + try: + if db.set_session_title(args.session_id, title): + print(f"Session '{args.session_id}' renamed to: {title}") + else: + print(f"Session '{args.session_id}' not found.") + except ValueError as e: + print(f"Error: {e}") + elif action == "stats": total = db.session_count() msgs = db.message_count() @@ -1877,7 +1977,7 @@ For more help on a command: args.toolsets = None args.verbose = False args.resume = None - args.continue_last = False + args.continue_last = None if not hasattr(args, "worktree"): args.worktree = False cmd_chat(args) diff --git a/hermes_state.py b/hermes_state.py index 1d1f951c0a..df266f072f 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -24,7 +24,7 @@ from typing import Dict, Any, List, Optional DEFAULT_DB_PATH = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) / "state.db" -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 4 SCHEMA_SQL = """ CREATE TABLE IF NOT EXISTS schema_version ( @@ -46,6 +46,7 @@ CREATE TABLE IF NOT EXISTS sessions ( tool_call_count INTEGER DEFAULT 0, input_tokens INTEGER DEFAULT 0, output_tokens INTEGER DEFAULT 0, + title TEXT, FOREIGN KEY (parent_session_id) REFERENCES sessions(id) ); @@ -133,7 +134,33 @@ class SessionDB: except sqlite3.OperationalError: pass # Column already exists cursor.execute("UPDATE schema_version SET version = 2") + if current_version < 3: + # v3: add title column to sessions + try: + cursor.execute("ALTER TABLE sessions ADD COLUMN title TEXT") + except sqlite3.OperationalError: + pass # Column already exists + cursor.execute("UPDATE schema_version SET version = 3") + if current_version < 4: + # v4: add unique index on title (NULLs allowed, only non-NULL must be unique) + try: + cursor.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_sessions_title_unique " + "ON sessions(title) WHERE title IS NOT NULL" + ) + except sqlite3.OperationalError: + pass # Index already exists + cursor.execute("UPDATE schema_version SET version = 4") + # Unique title index — always ensure it exists (safe to run after migrations + # since the title column is guaranteed to exist at this point) + try: + cursor.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS idx_sessions_title_unique " + "ON sessions(title) WHERE title IS NOT NULL" + ) + except sqlite3.OperationalError: + pass # Index already exists # FTS5 setup (separate because CREATE VIRTUAL TABLE can't be in executescript with IF NOT EXISTS reliably) try: @@ -219,6 +246,153 @@ class SessionDB: row = cursor.fetchone() return dict(row) if row else None + def set_session_title(self, session_id: str, title: str) -> bool: + """Set or update a session's title. + + Returns True if session was found and title was set. + Raises ValueError if title is already in use by another session. + """ + if title: + # Check uniqueness (allow the same session to keep its own title) + cursor = self._conn.execute( + "SELECT id FROM sessions WHERE title = ? AND id != ?", + (title, session_id), + ) + conflict = cursor.fetchone() + if conflict: + raise ValueError( + f"Title '{title}' is already in use by session {conflict['id']}" + ) + cursor = self._conn.execute( + "UPDATE sessions SET title = ? WHERE id = ?", + (title, session_id), + ) + self._conn.commit() + return cursor.rowcount > 0 + + def get_session_title(self, session_id: str) -> Optional[str]: + """Get the title for a session, or None.""" + cursor = self._conn.execute( + "SELECT title FROM sessions WHERE id = ?", (session_id,) + ) + row = cursor.fetchone() + return row["title"] if row else None + + def get_session_by_title(self, title: str) -> Optional[Dict[str, Any]]: + """Look up a session by exact title. Returns session dict or None.""" + cursor = self._conn.execute( + "SELECT * FROM sessions WHERE title = ?", (title,) + ) + row = cursor.fetchone() + return dict(row) if row else None + + def resolve_session_by_title(self, title: str) -> Optional[str]: + """Resolve a title to a session ID, preferring the latest in a lineage. + + If the exact title exists, returns that session's ID. + If not, searches for "title #N" variants and returns the latest one. + If the exact title exists AND numbered variants exist, returns the + latest numbered variant (the most recent continuation). + """ + # First try exact match + exact = self.get_session_by_title(title) + + # Also search for numbered variants: "title #2", "title #3", etc. + cursor = self._conn.execute( + "SELECT id, title, started_at FROM sessions " + "WHERE title LIKE ? ORDER BY started_at DESC", + (f"{title} #%",), + ) + numbered = cursor.fetchall() + + if numbered: + # Return the most recent numbered variant + return numbered[0]["id"] + elif exact: + return exact["id"] + return None + + def get_next_title_in_lineage(self, base_title: str) -> str: + """Generate the next title in a lineage (e.g., "my session" → "my session #2"). + + Strips any existing " #N" suffix to find the base name, then finds + the highest existing number and increments. + """ + import re + # Strip existing #N suffix to find the true base + match = re.match(r'^(.*?) #(\d+)$', base_title) + if match: + base = match.group(1) + else: + base = base_title + + # Find all existing numbered variants + cursor = self._conn.execute( + "SELECT title FROM sessions WHERE title = ? OR title LIKE ?", + (base, f"{base} #%"), + ) + existing = [row["title"] for row in cursor.fetchall()] + + if not existing: + return base # No conflict, use the base name as-is + + # Find the highest number + max_num = 1 # The unnumbered original counts as #1 + for t in existing: + m = re.match(r'^.* #(\d+)$', t) + if m: + max_num = max(max_num, int(m.group(1))) + + return f"{base} #{max_num + 1}" + + def list_sessions_rich( + self, + source: str = None, + limit: int = 20, + offset: int = 0, + ) -> List[Dict[str, Any]]: + """List sessions with preview (first user message) and last active timestamp. + + Returns dicts with keys: id, source, model, title, started_at, ended_at, + message_count, preview (first 60 chars of first user message), + last_active (timestamp of last message). + """ + if source: + cursor = self._conn.execute( + "SELECT * FROM sessions WHERE source = ? ORDER BY started_at DESC LIMIT ? OFFSET ?", + (source, limit, offset), + ) + else: + cursor = self._conn.execute( + "SELECT * FROM sessions ORDER BY started_at DESC LIMIT ? OFFSET ?", + (limit, offset), + ) + sessions = [dict(row) for row in cursor.fetchall()] + + for s in sessions: + # Get first user message preview + preview_cursor = self._conn.execute( + "SELECT content FROM messages WHERE session_id = ? AND role = 'user' " + "ORDER BY timestamp, id LIMIT 1", + (s["id"],), + ) + preview_row = preview_cursor.fetchone() + if preview_row and preview_row["content"]: + text = preview_row["content"].replace("\n", " ").strip() + s["preview"] = text[:60] + ("..." if len(text) > 60 else "") + else: + s["preview"] = "" + + # Get last message timestamp + last_cursor = self._conn.execute( + "SELECT MAX(timestamp) as last_ts FROM messages WHERE session_id = ?", + (s["id"],), + ) + last_row = last_cursor.fetchone() + s["last_active"] = last_row["last_ts"] if last_row and last_row["last_ts"] else s["started_at"] + + return sessions + # ========================================================================= # Message storage # ========================================================================= diff --git a/run_agent.py b/run_agent.py index 75e3dfc95f..0537dd973c 100644 --- a/run_agent.py +++ b/run_agent.py @@ -2484,6 +2484,8 @@ class AIAgent: if self._session_db: try: + # Propagate title to the new session with auto-numbering + old_title = self._session_db.get_session_title(self.session_id) self._session_db.end_session(self.session_id, "compression") old_session_id = self.session_id self.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}" @@ -2493,6 +2495,13 @@ class AIAgent: model=self.model, parent_session_id=old_session_id, ) + # Auto-number the title for the continuation session + if old_title: + try: + new_title = self._session_db.get_next_title_in_lineage(old_title) + self._session_db.set_session_title(self.session_id, new_title) + except (ValueError, Exception) as e: + logger.debug("Could not propagate title on compression: %s", e) self._session_db.update_system_prompt(self.session_id, new_system_prompt) except Exception as e: logger.debug("Session DB compression split failed: %s", e) diff --git a/tests/hermes_cli/test_commands.py b/tests/hermes_cli/test_commands.py index adbf677b64..3b01eb7b32 100644 --- a/tests/hermes_cli/test_commands.py +++ b/tests/hermes_cli/test_commands.py @@ -11,7 +11,7 @@ EXPECTED_COMMANDS = { "/help", "/tools", "/toolsets", "/model", "/provider", "/prompt", "/personality", "/clear", "/history", "/new", "/reset", "/retry", "/undo", "/save", "/config", "/cron", "/skills", "/platforms", - "/verbose", "/compress", "/usage", "/insights", "/paste", + "/verbose", "/compress", "/title", "/usage", "/insights", "/paste", "/reload-mcp", "/quit", } diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 734db494f9..fef1f49c3c 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -351,6 +351,77 @@ class TestPruneSessions: # Schema and WAL mode # ========================================================================= +# ========================================================================= +# Session title +# ========================================================================= + +class TestSessionTitle: + def test_set_and_get_title(self, db): + db.create_session(session_id="s1", source="cli") + assert db.set_session_title("s1", "My Session") is True + + session = db.get_session("s1") + assert session["title"] == "My Session" + + def test_set_title_nonexistent_session(self, db): + assert db.set_session_title("nonexistent", "Title") is False + + def test_title_initially_none(self, db): + db.create_session(session_id="s1", source="cli") + session = db.get_session("s1") + assert session["title"] is None + + def test_update_title(self, db): + db.create_session(session_id="s1", source="cli") + db.set_session_title("s1", "First Title") + db.set_session_title("s1", "Updated Title") + + session = db.get_session("s1") + assert session["title"] == "Updated Title" + + def test_title_in_search_sessions(self, db): + db.create_session(session_id="s1", source="cli") + db.set_session_title("s1", "Debugging Auth") + db.create_session(session_id="s2", source="cli") + + sessions = db.search_sessions() + titled = [s for s in sessions if s.get("title") == "Debugging Auth"] + assert len(titled) == 1 + assert titled[0]["id"] == "s1" + + def test_title_in_export(self, db): + db.create_session(session_id="s1", source="cli") + db.set_session_title("s1", "Export Test") + db.append_message("s1", role="user", content="Hello") + + export = db.export_session("s1") + assert export["title"] == "Export Test" + + def test_title_with_special_characters(self, db): + db.create_session(session_id="s1", source="cli") + title = "PR #438 — fixing the 'auth' middleware" + db.set_session_title("s1", title) + + session = db.get_session("s1") + assert session["title"] == title + + def test_title_empty_string(self, db): + db.create_session(session_id="s1", source="cli") + db.set_session_title("s1", "") + + session = db.get_session("s1") + assert session["title"] == "" + + def test_title_survives_end_session(self, db): + db.create_session(session_id="s1", source="cli") + db.set_session_title("s1", "Before End") + db.end_session("s1", end_reason="user_exit") + + session = db.get_session("s1") + assert session["title"] == "Before End" + assert session["ended_at"] is not None + + class TestSchemaInit: def test_wal_mode(self, db): cursor = db._conn.execute("PRAGMA journal_mode") @@ -373,4 +444,266 @@ class TestSchemaInit: def test_schema_version(self, db): cursor = db._conn.execute("SELECT version FROM schema_version") version = cursor.fetchone()[0] - assert version == 2 + assert version == 4 + + def test_title_column_exists(self, db): + """Verify the title column was created in the sessions table.""" + cursor = db._conn.execute("PRAGMA table_info(sessions)") + columns = {row[1] for row in cursor.fetchall()} + assert "title" in columns + + def test_migration_from_v2(self, tmp_path): + """Simulate a v2 database and verify migration adds title column.""" + import sqlite3 + + db_path = tmp_path / "migrate_test.db" + conn = sqlite3.connect(str(db_path)) + # Create v2 schema (without title column) + conn.executescript(""" + CREATE TABLE schema_version (version INTEGER NOT NULL); + INSERT INTO schema_version (version) VALUES (2); + + CREATE TABLE sessions ( + id TEXT PRIMARY KEY, + source TEXT NOT NULL, + user_id TEXT, + model TEXT, + model_config TEXT, + system_prompt TEXT, + parent_session_id TEXT, + started_at REAL NOT NULL, + ended_at REAL, + end_reason TEXT, + message_count INTEGER DEFAULT 0, + tool_call_count INTEGER DEFAULT 0, + input_tokens INTEGER DEFAULT 0, + output_tokens INTEGER DEFAULT 0 + ); + + CREATE TABLE messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + content TEXT, + tool_call_id TEXT, + tool_calls TEXT, + tool_name TEXT, + timestamp REAL NOT NULL, + token_count INTEGER, + finish_reason TEXT + ); + """) + conn.execute( + "INSERT INTO sessions (id, source, started_at) VALUES (?, ?, ?)", + ("existing", "cli", 1000.0), + ) + conn.commit() + conn.close() + + # Open with SessionDB — should migrate to v4 + migrated_db = SessionDB(db_path=db_path) + + # Verify migration + cursor = migrated_db._conn.execute("SELECT version FROM schema_version") + assert cursor.fetchone()[0] == 4 + + # Verify title column exists and is NULL for existing sessions + session = migrated_db.get_session("existing") + assert session is not None + assert session["title"] is None + + # Verify we can set title on migrated session + assert migrated_db.set_session_title("existing", "Migrated Title") is True + session = migrated_db.get_session("existing") + assert session["title"] == "Migrated Title" + + migrated_db.close() + + +class TestTitleUniqueness: + """Tests for unique title enforcement and title-based lookups.""" + + def test_duplicate_title_raises(self, db): + """Setting a title already used by another session raises ValueError.""" + db.create_session("s1", "cli") + db.create_session("s2", "cli") + db.set_session_title("s1", "my project") + with pytest.raises(ValueError, match="already in use"): + db.set_session_title("s2", "my project") + + def test_same_session_can_keep_title(self, db): + """A session can re-set its own title without error.""" + db.create_session("s1", "cli") + db.set_session_title("s1", "my project") + # Should not raise — it's the same session + assert db.set_session_title("s1", "my project") is True + + def test_null_titles_not_unique(self, db): + """Multiple sessions can have NULL titles (no constraint violation).""" + db.create_session("s1", "cli") + db.create_session("s2", "cli") + # Both have NULL titles — no error + assert db.get_session("s1")["title"] is None + assert db.get_session("s2")["title"] is None + + def test_get_session_by_title(self, db): + db.create_session("s1", "cli") + db.set_session_title("s1", "refactoring auth") + result = db.get_session_by_title("refactoring auth") + assert result is not None + assert result["id"] == "s1" + + def test_get_session_by_title_not_found(self, db): + assert db.get_session_by_title("nonexistent") is None + + def test_get_session_title(self, db): + db.create_session("s1", "cli") + assert db.get_session_title("s1") is None + db.set_session_title("s1", "my title") + assert db.get_session_title("s1") == "my title" + + def test_get_session_title_nonexistent(self, db): + assert db.get_session_title("nonexistent") is None + + +class TestTitleLineage: + """Tests for title lineage resolution and auto-numbering.""" + + def test_resolve_exact_title(self, db): + db.create_session("s1", "cli") + db.set_session_title("s1", "my project") + assert db.resolve_session_by_title("my project") == "s1" + + def test_resolve_returns_latest_numbered(self, db): + """When numbered variants exist, return the most recent one.""" + import time + db.create_session("s1", "cli") + db.set_session_title("s1", "my project") + time.sleep(0.01) + db.create_session("s2", "cli") + db.set_session_title("s2", "my project #2") + time.sleep(0.01) + db.create_session("s3", "cli") + db.set_session_title("s3", "my project #3") + # Resolving "my project" should return s3 (latest numbered variant) + assert db.resolve_session_by_title("my project") == "s3" + + def test_resolve_exact_numbered(self, db): + """Resolving an exact numbered title returns that specific session.""" + db.create_session("s1", "cli") + db.set_session_title("s1", "my project") + db.create_session("s2", "cli") + db.set_session_title("s2", "my project #2") + # Resolving "my project #2" exactly should return s2 + assert db.resolve_session_by_title("my project #2") == "s2" + + def test_resolve_nonexistent_title(self, db): + assert db.resolve_session_by_title("nonexistent") is None + + def test_next_title_no_existing(self, db): + """With no existing sessions, base title is returned as-is.""" + assert db.get_next_title_in_lineage("my project") == "my project" + + def test_next_title_first_continuation(self, db): + """First continuation after the original gets #2.""" + db.create_session("s1", "cli") + db.set_session_title("s1", "my project") + assert db.get_next_title_in_lineage("my project") == "my project #2" + + def test_next_title_increments(self, db): + """Each continuation increments the number.""" + db.create_session("s1", "cli") + db.set_session_title("s1", "my project") + db.create_session("s2", "cli") + db.set_session_title("s2", "my project #2") + db.create_session("s3", "cli") + db.set_session_title("s3", "my project #3") + assert db.get_next_title_in_lineage("my project") == "my project #4" + + def test_next_title_strips_existing_number(self, db): + """Passing a numbered title strips the number and finds the base.""" + db.create_session("s1", "cli") + db.set_session_title("s1", "my project") + db.create_session("s2", "cli") + db.set_session_title("s2", "my project #2") + # Even when called with "my project #2", it should return #3 + assert db.get_next_title_in_lineage("my project #2") == "my project #3" + + +class TestListSessionsRich: + """Tests for enhanced session listing with preview and last_active.""" + + def test_preview_from_first_user_message(self, db): + db.create_session("s1", "cli") + db.append_message("s1", "system", "You are a helpful assistant.") + db.append_message("s1", "user", "Help me refactor the auth module please") + db.append_message("s1", "assistant", "Sure, let me look at it.") + sessions = db.list_sessions_rich() + assert len(sessions) == 1 + assert "Help me refactor the auth module" in sessions[0]["preview"] + + def test_preview_truncated_at_60(self, db): + db.create_session("s1", "cli") + long_msg = "A" * 100 + db.append_message("s1", "user", long_msg) + sessions = db.list_sessions_rich() + assert len(sessions[0]["preview"]) == 63 # 60 chars + "..." + assert sessions[0]["preview"].endswith("...") + + def test_preview_empty_when_no_user_messages(self, db): + db.create_session("s1", "cli") + db.append_message("s1", "system", "System prompt") + sessions = db.list_sessions_rich() + assert sessions[0]["preview"] == "" + + def test_last_active_from_latest_message(self, db): + import time + db.create_session("s1", "cli") + db.append_message("s1", "user", "Hello") + time.sleep(0.01) + db.append_message("s1", "assistant", "Hi there!") + sessions = db.list_sessions_rich() + # last_active should be close to now (the assistant message) + assert sessions[0]["last_active"] > sessions[0]["started_at"] + + def test_last_active_fallback_to_started_at(self, db): + db.create_session("s1", "cli") + sessions = db.list_sessions_rich() + # No messages, so last_active falls back to started_at + assert sessions[0]["last_active"] == sessions[0]["started_at"] + + def test_rich_list_includes_title(self, db): + db.create_session("s1", "cli") + db.set_session_title("s1", "refactoring auth") + sessions = db.list_sessions_rich() + assert sessions[0]["title"] == "refactoring auth" + + def test_rich_list_source_filter(self, db): + db.create_session("s1", "cli") + db.create_session("s2", "telegram") + sessions = db.list_sessions_rich(source="cli") + assert len(sessions) == 1 + assert sessions[0]["id"] == "s1" + + def test_preview_newlines_collapsed(self, db): + db.create_session("s1", "cli") + db.append_message("s1", "user", "Line one\nLine two\nLine three") + sessions = db.list_sessions_rich() + assert "\n" not in sessions[0]["preview"] + assert "Line one Line two" in sessions[0]["preview"] + + +class TestResolveSessionByNameOrId: + """Tests for the main.py helper that resolves names or IDs.""" + + def test_resolve_by_id(self, db): + db.create_session("test-id-123", "cli") + session = db.get_session("test-id-123") + assert session is not None + assert session["id"] == "test-id-123" + + def test_resolve_by_title_falls_back(self, db): + db.create_session("s1", "cli") + db.set_session_title("s1", "my project") + result = db.resolve_session_by_title("my project") + assert result == "s1"