feat: CLI session bridge - read CLI sessions from agent SQLite store

Read CLI sessions from the agent's state.db and surface them in the
WebUI sidebar alongside local sessions, with read-only display and
import-on-click to avoid data duplication.

Key changes:
- get_cli_sessions(): reads sessions list via parameterized SQL,
  wrapped in sqlite3 context manager (no connection leaks)
- get_cli_session_messages(): reads messages for a CLI session
  via parameterized SQL, also context-managed
- GET /api/sessions: merges WebUI + CLI sessions with dedup
  (WebUI takes priority on same session_id)
- GET /api/session: falls back to CLI store if not a WebUI session
- POST /api/session/import_cli: imports a CLI session into the
  WebUI store (idempotent, no duplicates on re-import)
- Imported sessions use get_last_workspace() for the workspace field
  (not a hardcoded string) and carry the active profile tag
- CSS: .cli-session with ::after 'cli' indicator (no theme changes)

Fixes review feedback:
- SQLite connections use 'with' context managers (no leaks)
- Workspace uses real path via get_last_workspace()
- Profile awareness via api.profiles.get_active_profile_name()
- Parameterized SQL queries throughout (no injection risk)
- Graceful fallback when sqlite3 or state.db is missing
This commit is contained in:
Thad Reber
2026-04-03 19:54:54 -07:00
parent 846565484b
commit cabda6b77a
3 changed files with 237 additions and 7 deletions

View File

@@ -192,3 +192,130 @@ def load_projects():
def save_projects(projects):
"""Write project list to disk."""
PROJECTS_FILE.write_text(json.dumps(projects, ensure_ascii=False, indent=2), encoding='utf-8')
def import_cli_session(session_id, title, messages, model='unknown', profile=None):
"""Create a new WebUI session populated with CLI messages.
Returns the Session object.
"""
s = Session(
session_id=session_id,
title=title,
workspace=get_last_workspace(),
model=model,
messages=messages,
profile=profile,
)
s.save()
return s
# ── CLI session bridge ──────────────────────────────────────────────────────
def get_cli_sessions():
"""Read CLI sessions from the agent's SQLite store and return them as
dicts in a format the WebUI sidebar can render alongside local sessions.
Returns empty list if the SQLite DB is missing, the sqlite3 module is
unavailable, or any error occurs -- the bridge is purely additive and never
crashes the WebUI.
"""
import os
cli_sessions = []
try:
import sqlite3
except ImportError:
return cli_sessions
hermes_home = Path(os.getenv('HERMES_HOME', str(HOME / '.hermes'))).expanduser()
db_path = hermes_home / 'state.db'
if not db_path.exists():
return cli_sessions
# Try to resolve the active CLI profile so imported sessions integrate
# with the WebUI profile filter (available since Sprint 22).
try:
from api.profiles import get_active_profile_name
_cli_profile = get_active_profile_name()
except ImportError:
_cli_profile = None # older agent -- fall back to no profile
try:
with sqlite3.connect(str(db_path)) as conn:
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute("""
SELECT s.id, s.title, s.model, s.message_count,
s.started_at, s.source, s.profile,
MAX(m.timestamp) AS last_activity
FROM sessions s
LEFT JOIN messages m ON m.session_id = s.id
GROUP BY s.id
ORDER BY COALESCE(MAX(m.timestamp), s.started_at) DESC
LIMIT 200
""")
for row in cur.fetchall():
sid = row['id']
raw_ts = row['last_activity'] or row['started_at']
# Prefer the CLI session's own profile from the DB; fall back to
# the active CLI profile so sidebar filtering works either way.
profile = row.get('profile') or _cli_profile
cli_sessions.append({
'session_id': sid,
'title': row['title'] or 'CLI Session',
'workspace': str(get_last_workspace()),
'model': row['model'] or 'unknown',
'message_count': row['message_count'] or 0,
'created_at': row['started_at'],
'updated_at': raw_ts,
'pinned': False,
'archived': False,
'project_id': None,
'profile': profile,
'source_tag': 'cli',
'is_cli_session': True,
})
except Exception:
# DB schema changed, locked, or corrupted -- silently degrade
return []
return cli_sessions
def get_cli_session_messages(sid):
"""Read messages for a single CLI session from the SQLite store.
Returns a list of {role, content, timestamp} dicts.
Returns empty list on any error.
"""
import os
try:
import sqlite3
except ImportError:
return []
hermes_home = Path(os.getenv('HERMES_HOME', str(HOME / '.hermes'))).expanduser()
db_path = hermes_home / 'state.db'
if not db_path.exists():
return []
try:
with sqlite3.connect(str(db_path)) as conn:
conn.row_factory = sqlite3.Row
cur = conn.cursor()
cur.execute("""
SELECT role, content, timestamp
FROM messages
WHERE session_id = ?
ORDER BY timestamp ASC
""", (sid,))
msgs = []
for row in cur.fetchall():
msgs.append({
'role': row['role'],
'content': row['content'],
'timestamp': row['timestamp'],
})
except Exception:
return []
return msgs