feat: session naming with unique titles, auto-lineage, rich listing, resume by name

- Schema v4: unique title index, migration from v2/v3
- set/get/resolve session titles with uniqueness enforcement
- Auto-lineage: context compression auto-numbers titles (Task -> Task #2 -> Task #3)
- resolve_session_by_title: auto-latest finds most recent continuation
- list_sessions_rich: preview (first 60 chars) + last_active timestamp
- CLI: -c accepts optional name arg (hermes -c 'my project')
- CLI: /title command with deferred mode (set before session exists)
- CLI: sessions list shows Title, Preview, Last Active, ID
- 27 new tests (1844 total passing)
This commit is contained in:
teknium1 2026-03-08 15:20:29 -07:00
parent 4d53b7ccaa
commit 60b6abefd9
7 changed files with 716 additions and 36 deletions

79
cli.py
View file

@ -1094,6 +1094,16 @@ class HermesCLI:
self.conversation_history: List[Dict[str, Any]] = [] self.conversation_history: List[Dict[str, Any]] = []
self.session_start = datetime.now() self.session_start = datetime.now()
self._resumed = False 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 # Session ID: reuse existing one when resuming, otherwise generate fresh
if resume: if resume:
@ -1181,13 +1191,13 @@ class HermesCLI:
if not self._ensure_runtime_credentials(): if not self._ensure_runtime_credentials():
return False return False
# Initialize SQLite session store for CLI sessions # Initialize SQLite session store for CLI sessions (if not already done in __init__)
self._session_db = None if self._session_db is None:
try: try:
from hermes_state import SessionDB from hermes_state import SessionDB
self._session_db = SessionDB() self._session_db = SessionDB()
except Exception as e: except Exception as e:
logger.debug("SQLite session store not available: %s", e) logger.debug("SQLite session store not available: %s", e)
# If resuming, validate the session exists and load its history # If resuming, validate the session exists and load its history
if self._resumed and self._session_db: if self._resumed and self._session_db:
@ -1200,8 +1210,11 @@ class HermesCLI:
if restored: if restored:
self.conversation_history = restored self.conversation_history = restored
msg_count = len([m for m in restored if m.get("role") == "user"]) 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( _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"({msg_count} user message{'s' if msg_count != 1 else ''}, "
f"{len(restored)} total messages){_RST}" f"{len(restored)} total messages){_RST}"
) )
@ -1243,6 +1256,15 @@ class HermesCLI:
clarify_callback=self._clarify_callback, clarify_callback=self._clarify_callback,
honcho_session_key=self.session_id, 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 return True
except Exception as e: except Exception as e:
self.console.print(f"[bold red]Failed to initialize agent: {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") print(" ✨ (◕‿◕)✨ Fresh start! Screen cleared and conversation reset.\n")
elif cmd_lower == "/history": elif cmd_lower == "/history":
self.show_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 <your session 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 <your session title>")
else:
_cprint(" Session database not available.")
elif cmd_lower in ("/reset", "/new"): elif cmd_lower in ("/reset", "/new"):
self.reset_conversation() self.reset_conversation()
elif cmd_lower.startswith("/model"): elif cmd_lower.startswith("/model"):

View file

@ -34,6 +34,7 @@ COMMANDS = {
"/platforms": "Show gateway/messaging platform status", "/platforms": "Show gateway/messaging platform status",
"/verbose": "Cycle tool progress display: off → new → all → verbose", "/verbose": "Cycle tool progress display: off → new → all → verbose",
"/compress": "Manually compress conversation context (flush memories + summarize)", "/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", "/usage": "Show token usage for the current session",
"/insights": "Show usage insights and analytics (last 30 days)", "/insights": "Show usage insights and analytics (last 30 days)",
"/paste": "Check clipboard for an image and attach it", "/paste": "Check clipboard for an image and attach it",

View file

@ -120,16 +120,63 @@ def _resolve_last_cli_session() -> Optional[str]:
return None 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): def cmd_chat(args):
"""Run interactive chat CLI.""" """Run interactive chat CLI."""
# Resolve --continue into --resume with the latest CLI session # Resolve --continue into --resume with the latest CLI session or by name
if getattr(args, "continue_last", False) and not getattr(args, "resume", None): continue_val = getattr(args, "continue_last", None)
last_id = _resolve_last_cli_session() if continue_val and not getattr(args, "resume", None):
if last_id: if isinstance(continue_val, str):
args.resume = last_id # -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: else:
print("No previous CLI session found to continue.") # -c with no argument — continue the most recent session
sys.exit(1) 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 # First-run guard: check if any provider is configured before launching
if not _has_any_provider_configured(): if not _has_any_provider_configured():
@ -1209,8 +1256,9 @@ def main():
Examples: Examples:
hermes Start interactive chat hermes Start interactive chat
hermes chat -q "Hello" Single query mode hermes chat -q "Hello" Single query mode
hermes --continue Resume the most recent session hermes -c Resume the most recent session
hermes --resume <session_id> Resume a specific session hermes -c "my project" Resume a session by name (latest in lineage)
hermes --resume <session_id> Resume a specific session by ID
hermes setup Run setup wizard hermes setup Run setup wizard
hermes logout Clear stored authentication hermes logout Clear stored authentication
hermes model Select default model hermes model Select default model
@ -1221,6 +1269,7 @@ Examples:
hermes -w Start in isolated git worktree hermes -w Start in isolated git worktree
hermes gateway install Install as system service hermes gateway install Install as system service
hermes sessions list List past sessions hermes sessions list List past sessions
hermes sessions rename ID T Rename/title a session
hermes update Update to latest version hermes update Update to latest version
For more help on a command: For more help on a command:
@ -1235,16 +1284,18 @@ For more help on a command:
) )
parser.add_argument( parser.add_argument(
"--resume", "-r", "--resume", "-r",
metavar="SESSION_ID", metavar="SESSION",
default=None, 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( parser.add_argument(
"--continue", "-c", "--continue", "-c",
dest="continue_last", dest="continue_last",
action="store_true", nargs="?",
default=False, const=True,
help="Resume the most recent CLI session" default=None,
metavar="SESSION_NAME",
help="Resume a session by name, or the most recent if no name given"
) )
parser.add_argument( parser.add_argument(
"--worktree", "-w", "--worktree", "-w",
@ -1294,9 +1345,11 @@ For more help on a command:
chat_parser.add_argument( chat_parser.add_argument(
"--continue", "-c", "--continue", "-c",
dest="continue_last", dest="continue_last",
action="store_true", nargs="?",
default=False, const=True,
help="Resume the most recent CLI session" default=None,
metavar="SESSION_NAME",
help="Resume a session by name, or the most recent if no name given"
) )
chat_parser.add_argument( chat_parser.add_argument(
"--worktree", "-w", "--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_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): def cmd_sessions(args):
import json as _json import json as _json
try: try:
@ -1708,18 +1765,51 @@ For more help on a command:
action = args.sessions_action action = args.sessions_action
if action == "list": 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: if not sessions:
print("No sessions found.") print("No sessions found.")
return return
print(f"{'ID':<30} {'Source':<12} {'Model':<30} {'Messages':>8} {'Started'}")
print("" * 100)
from datetime import datetime 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: for s in sessions:
started = datetime.fromtimestamp(s["started_at"]).strftime("%Y-%m-%d %H:%M") if s["started_at"] else "?" last_active = _relative_time(s.get("last_active"))
model = (s.get("model") or "?")[:28] preview = s.get("preview", "")[:38] if has_titles else s.get("preview", "")[:48]
ended = " (ended)" if s.get("ended_at") else "" if has_titles:
print(f"{s['id']:<30} {s['source']:<12} {model:<30} {s['message_count']:>8} {started}{ended}") 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": elif action == "export":
if args.session_id: 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) count = db.prune_sessions(older_than_days=days, source=args.source)
print(f"Pruned {count} session(s).") 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": elif action == "stats":
total = db.session_count() total = db.session_count()
msgs = db.message_count() msgs = db.message_count()
@ -1877,7 +1977,7 @@ For more help on a command:
args.toolsets = None args.toolsets = None
args.verbose = False args.verbose = False
args.resume = None args.resume = None
args.continue_last = False args.continue_last = None
if not hasattr(args, "worktree"): if not hasattr(args, "worktree"):
args.worktree = False args.worktree = False
cmd_chat(args) cmd_chat(args)

View file

@ -24,7 +24,7 @@ from typing import Dict, Any, List, Optional
DEFAULT_DB_PATH = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) / "state.db" DEFAULT_DB_PATH = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) / "state.db"
SCHEMA_VERSION = 2 SCHEMA_VERSION = 4
SCHEMA_SQL = """ SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS schema_version ( CREATE TABLE IF NOT EXISTS schema_version (
@ -46,6 +46,7 @@ CREATE TABLE IF NOT EXISTS sessions (
tool_call_count INTEGER DEFAULT 0, tool_call_count INTEGER DEFAULT 0,
input_tokens INTEGER DEFAULT 0, input_tokens INTEGER DEFAULT 0,
output_tokens INTEGER DEFAULT 0, output_tokens INTEGER DEFAULT 0,
title TEXT,
FOREIGN KEY (parent_session_id) REFERENCES sessions(id) FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
); );
@ -133,7 +134,33 @@ class SessionDB:
except sqlite3.OperationalError: except sqlite3.OperationalError:
pass # Column already exists pass # Column already exists
cursor.execute("UPDATE schema_version SET version = 2") 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) # FTS5 setup (separate because CREATE VIRTUAL TABLE can't be in executescript with IF NOT EXISTS reliably)
try: try:
@ -219,6 +246,153 @@ class SessionDB:
row = cursor.fetchone() row = cursor.fetchone()
return dict(row) if row else None 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 # Message storage
# ========================================================================= # =========================================================================

View file

@ -2484,6 +2484,8 @@ class AIAgent:
if self._session_db: if self._session_db:
try: 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") self._session_db.end_session(self.session_id, "compression")
old_session_id = self.session_id old_session_id = self.session_id
self.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}" 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, model=self.model,
parent_session_id=old_session_id, 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) self._session_db.update_system_prompt(self.session_id, new_system_prompt)
except Exception as e: except Exception as e:
logger.debug("Session DB compression split failed: %s", e) logger.debug("Session DB compression split failed: %s", e)

View file

@ -11,7 +11,7 @@ EXPECTED_COMMANDS = {
"/help", "/tools", "/toolsets", "/model", "/provider", "/prompt", "/help", "/tools", "/toolsets", "/model", "/provider", "/prompt",
"/personality", "/clear", "/history", "/new", "/reset", "/retry", "/personality", "/clear", "/history", "/new", "/reset", "/retry",
"/undo", "/save", "/config", "/cron", "/skills", "/platforms", "/undo", "/save", "/config", "/cron", "/skills", "/platforms",
"/verbose", "/compress", "/usage", "/insights", "/paste", "/verbose", "/compress", "/title", "/usage", "/insights", "/paste",
"/reload-mcp", "/quit", "/reload-mcp", "/quit",
} }

View file

@ -351,6 +351,77 @@ class TestPruneSessions:
# Schema and WAL mode # 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: class TestSchemaInit:
def test_wal_mode(self, db): def test_wal_mode(self, db):
cursor = db._conn.execute("PRAGMA journal_mode") cursor = db._conn.execute("PRAGMA journal_mode")
@ -373,4 +444,266 @@ class TestSchemaInit:
def test_schema_version(self, db): def test_schema_version(self, db):
cursor = db._conn.execute("SELECT version FROM schema_version") cursor = db._conn.execute("SELECT version FROM schema_version")
version = cursor.fetchone()[0] 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"