From 337f902a7e8749c3a4583c84a1cc0fc91c8f7c6b Mon Sep 17 00:00:00 2001 From: Test Date: Thu, 19 Mar 2026 10:16:32 -0700 Subject: [PATCH] refactor: use SessionDB instead of JSON files for ACP session persistence Replace the standalone JSON file persistence layer with SessionDB (~/.hermes/state.db) integration. ACP sessions now: - Share the same DB as CLI and gateway sessions - Are searchable via session_search (FTS5) - Get token tracking, cost tracking, and session titles for free - Follow existing session pruning policies Key changes: - _get_db() lazily creates a SessionDB, resolving HERMES_HOME dynamically (not at import time) for test compatibility - _persist() creates session record + replaces messages in DB - _restore() loads from DB with source='acp' filter - cwd stored in model_config JSON field (no schema migration) - Model values coerced to str to handle mock agents in tests - Removed: json files, sessions_dir, ttl_days, _expire logic - Tests updated: DB-backed persistence, FTS search, tool_call round-tripping, source filtering --- acp_adapter/session.py | 303 ++++++++++++++++++++------------------ tests/acp/test_session.py | 177 ++++++++++------------ 2 files changed, 239 insertions(+), 241 deletions(-) diff --git a/acp_adapter/session.py b/acp_adapter/session.py index 10fb3ee6c7..01b2ee479b 100644 --- a/acp_adapter/session.py +++ b/acp_adapter/session.py @@ -1,29 +1,23 @@ """ACP session manager — maps ACP sessions to Hermes AIAgent instances. -Sessions are persisted to ``~/.hermes/acp_sessions/`` as JSON files so they -survive process restarts. When the editor reconnects after idle/restart, the -``load_session`` / ``resume_session`` calls will find the persisted session on -disk and restore the full conversation history. +Sessions are persisted to the shared SessionDB (``~/.hermes/state.db``) so they +survive process restarts and appear in ``session_search``. When the editor +reconnects after idle/restart, the ``load_session`` / ``resume_session`` calls +find the persisted session in the database and restore the full conversation +history. """ from __future__ import annotations import copy import json import logging -import os -import tempfile import uuid from dataclasses import dataclass, field -from datetime import datetime, timedelta -from pathlib import Path from threading import Lock from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) -# Default session time-to-live: sessions older than this are pruned on startup. -_DEFAULT_TTL_DAYS = 7 - def _register_task_cwd(task_id: str, cwd: str) -> None: """Bind a task/session id to the editor's working directory for tools.""" @@ -62,41 +56,24 @@ class SessionState: class SessionManager: """Thread-safe manager for ACP sessions backed by Hermes AIAgent instances. - Sessions are held in-memory for fast access **and** persisted to disk as - JSON files under *sessions_dir* so they survive process restarts. + Sessions are held in-memory for fast access **and** persisted to the + shared SessionDB so they survive process restarts and are searchable + via ``session_search``. """ - def __init__(self, agent_factory=None, sessions_dir: Path | str | None = None, - ttl_days: int = _DEFAULT_TTL_DAYS): + def __init__(self, agent_factory=None, db=None): """ Args: agent_factory: Optional callable that creates an AIAgent-like object. Used by tests. When omitted, a real AIAgent is created using the current Hermes runtime provider configuration. - sessions_dir: Directory for persisted session JSON files. - Defaults to ``~/.hermes/acp_sessions/``. - ttl_days: Sessions older than this many days are pruned on - startup and during ``list_sessions``. 0 = no expiry. + db: Optional SessionDB instance. When omitted, the default + SessionDB (``~/.hermes/state.db``) is lazily created. """ self._sessions: Dict[str, SessionState] = {} self._lock = Lock() self._agent_factory = agent_factory - self._ttl = timedelta(days=ttl_days) if ttl_days > 0 else None - - if sessions_dir is not None: - self._sessions_dir = Path(sessions_dir) - else: - hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) - self._sessions_dir = hermes_home / "acp_sessions" - - # Ensure the directory exists (no-op if already present). - try: - self._sessions_dir.mkdir(parents=True, exist_ok=True) - except OSError: - logger.warning("Cannot create ACP sessions dir %s", self._sessions_dir, exc_info=True) - - # Prune expired sessions on startup. - self._expire_old_sessions() + self._db_instance = db # None → lazy-init on first use # ---- public API --------------------------------------------------------- @@ -123,30 +100,30 @@ class SessionManager: def get_session(self, session_id: str) -> Optional[SessionState]: """Return the session for *session_id*, or ``None``. - If the session is not in memory but exists on disk (e.g. after a - process restart), it is transparently restored. + If the session is not in memory but exists in the database (e.g. after + a process restart), it is transparently restored. """ with self._lock: state = self._sessions.get(session_id) if state is not None: return state - # Attempt to restore from disk. + # Attempt to restore from database. return self._restore(session_id) def remove_session(self, session_id: str) -> bool: - """Remove a session from memory and disk. Returns True if it existed.""" + """Remove a session from memory and database. Returns True if it existed.""" with self._lock: existed = self._sessions.pop(session_id, None) is not None - disk_existed = self._delete_persisted(session_id) - if existed or disk_existed: + db_existed = self._delete_persisted(session_id) + if existed or db_existed: _clear_task_cwd(session_id) - return existed or disk_existed + return existed or db_existed def fork_session(self, session_id: str, cwd: str = ".") -> Optional[SessionState]: """Deep-copy a session's history into a new session.""" import threading - original = self.get_session(session_id) # checks disk too + original = self.get_session(session_id) # checks DB too if original is None: return None @@ -172,7 +149,7 @@ class SessionManager: return state def list_sessions(self) -> List[Dict[str, Any]]: - """Return lightweight info dicts for all sessions (memory + disk).""" + """Return lightweight info dicts for all sessions (memory + database).""" # Collect in-memory sessions first. with self._lock: seen_ids = set(self._sessions.keys()) @@ -187,33 +164,36 @@ class SessionManager: ] # Merge any persisted sessions not currently in memory. - try: - for path in self._sessions_dir.glob("*.json"): - sid = path.stem - if sid in seen_ids: - continue - try: - data = json.loads(path.read_text(encoding="utf-8")) - # Check TTL. - if self._ttl and self._is_expired(data): - path.unlink(missing_ok=True) + db = self._get_db() + if db is not None: + try: + rows = db.search_sessions(source="acp", limit=1000) + for row in rows: + sid = row["id"] + if sid in seen_ids: continue + # Extract cwd from model_config JSON. + cwd = "." + mc = row.get("model_config") + if mc: + try: + cwd = json.loads(mc).get("cwd", ".") + except (json.JSONDecodeError, TypeError): + pass results.append({ "session_id": sid, - "cwd": data.get("cwd", "."), - "model": data.get("model", ""), - "history_len": len(data.get("history", [])), + "cwd": cwd, + "model": row.get("model") or "", + "history_len": row.get("message_count") or 0, }) - except Exception: - logger.debug("Skipping unreadable session file %s", path) - except OSError: - pass + except Exception: + logger.debug("Failed to list ACP sessions from DB", exc_info=True) return results def update_cwd(self, session_id: str, cwd: str) -> Optional[SessionState]: """Update the working directory for a session and its tool overrides.""" - state = self.get_session(session_id) # checks disk too + state = self.get_session(session_id) # checks DB too if state is None: return None state.cwd = cwd @@ -222,23 +202,27 @@ class SessionManager: return state def cleanup(self) -> None: - """Remove all sessions (memory and disk) and clear task-specific cwd overrides.""" + """Remove all sessions (memory and database) and clear task-specific cwd overrides.""" with self._lock: session_ids = list(self._sessions.keys()) self._sessions.clear() for session_id in session_ids: _clear_task_cwd(session_id) self._delete_persisted(session_id) - # Also remove any disk-only sessions not currently in memory. - try: - for path in self._sessions_dir.glob("*.json"): - _clear_task_cwd(path.stem) - path.unlink(missing_ok=True) - except OSError: - pass + # Also remove any DB-only ACP sessions not currently in memory. + db = self._get_db() + if db is not None: + try: + rows = db.search_sessions(source="acp", limit=10000) + for row in rows: + sid = row["id"] + _clear_task_cwd(sid) + db.delete_session(sid) + except Exception: + logger.debug("Failed to cleanup ACP sessions from DB", exc_info=True) def save_session(self, session_id: str) -> None: - """Persist the current state of a session to disk. + """Persist the current state of a session to the database. Called by the server after prompt completion, slash commands that mutate history, and model switches. @@ -248,61 +232,120 @@ class SessionManager: if state is not None: self._persist(state) - # ---- persistence --------------------------------------------------------- + # ---- persistence via SessionDB ------------------------------------------ - def _session_path(self, session_id: str) -> Path: - return self._sessions_dir / f"{session_id}.json" + def _get_db(self): + """Lazily initialise and return the SessionDB instance. + + Returns ``None`` if the DB is unavailable (e.g. import error in a + minimal test environment). + + Note: we resolve ``HERMES_HOME`` dynamically rather than relying on + the module-level ``DEFAULT_DB_PATH`` constant, because that constant + is evaluated at import time and won't reflect env-var changes made + later (e.g. by the test fixture ``_isolate_hermes_home``). + """ + if self._db_instance is not None: + return self._db_instance + try: + import os + from pathlib import Path + from hermes_state import SessionDB + hermes_home = Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) + self._db_instance = SessionDB(db_path=hermes_home / "state.db") + return self._db_instance + except Exception: + logger.debug("SessionDB unavailable for ACP persistence", exc_info=True) + return None def _persist(self, state: SessionState) -> None: - """Write session state to disk atomically.""" - data = { - "session_id": state.session_id, - "cwd": state.cwd, - "model": state.model, - "history": state.history, - "updated_at": datetime.utcnow().isoformat(), - } - path = self._session_path(state.session_id) + """Write session state to the database. + + Creates the session record if it doesn't exist, then replaces all + stored messages with the current in-memory history. + """ + db = self._get_db() + if db is None: + return + + # Ensure model is a plain string (not a MagicMock or other proxy). + model_str = str(state.model) if state.model else None + cwd_json = json.dumps({"cwd": state.cwd}) + try: - # Atomic write: write to temp file in same directory, then rename. - fd, tmp = tempfile.mkstemp(dir=self._sessions_dir, suffix=".tmp") - try: - with os.fdopen(fd, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, default=str) - os.replace(tmp, path) - except Exception: - # Clean up temp file on failure. + # Ensure the session record exists. + existing = db.get_session(state.session_id) + if existing is None: + db.create_session( + session_id=state.session_id, + source="acp", + model=model_str, + model_config={"cwd": state.cwd}, + ) + else: + # Update model_config (contains cwd) if changed. try: - os.unlink(tmp) - except OSError: - pass - raise + with db._lock: + db._conn.execute( + "UPDATE sessions SET model_config = ?, model = COALESCE(?, model) WHERE id = ?", + (cwd_json, model_str, state.session_id), + ) + db._conn.commit() + except Exception: + logger.debug("Failed to update ACP session metadata", exc_info=True) + + # Replace stored messages with current history. + db.clear_messages(state.session_id) + for msg in state.history: + db.append_message( + session_id=state.session_id, + role=msg.get("role", "user"), + content=msg.get("content"), + tool_name=msg.get("tool_name") or msg.get("name"), + tool_calls=msg.get("tool_calls"), + tool_call_id=msg.get("tool_call_id"), + ) except Exception: logger.warning("Failed to persist ACP session %s", state.session_id, exc_info=True) def _restore(self, session_id: str) -> Optional[SessionState]: - """Load a session from disk into memory, recreating the AIAgent.""" + """Load a session from the database into memory, recreating the AIAgent.""" import threading - path = self._session_path(session_id) - if not path.exists(): + db = self._get_db() + if db is None: return None try: - data = json.loads(path.read_text(encoding="utf-8")) + row = db.get_session(session_id) except Exception: - logger.warning("Failed to read persisted ACP session %s", session_id, exc_info=True) + logger.debug("Failed to query DB for ACP session %s", session_id, exc_info=True) return None - # Check TTL. - if self._ttl and self._is_expired(data): - logger.info("Persisted ACP session %s is expired, removing", session_id) - path.unlink(missing_ok=True) + if row is None: return None - cwd = data.get("cwd", ".") - model = data.get("model") or None - history = data.get("history", []) + # Only restore ACP sessions. + if row.get("source") != "acp": + return None + + # Extract cwd from model_config. + cwd = "." + mc = row.get("model_config") + if mc: + try: + cwd = json.loads(mc).get("cwd", ".") + except (json.JSONDecodeError, TypeError): + pass + + model = row.get("model") or None + + # Load conversation history. + try: + history = db.get_messages_as_conversation(session_id) + except Exception: + logger.warning("Failed to load messages for ACP session %s", session_id, exc_info=True) + history = [] try: agent = self._make_agent(session_id=session_id, cwd=cwd, model=model) @@ -321,49 +364,19 @@ class SessionManager: with self._lock: self._sessions[session_id] = state _register_task_cwd(session_id, cwd) - logger.info("Restored ACP session %s from disk (%d messages)", session_id, len(history)) + logger.info("Restored ACP session %s from DB (%d messages)", session_id, len(history)) return state def _delete_persisted(self, session_id: str) -> bool: - """Delete a persisted session file. Returns True if it existed.""" - path = self._session_path(session_id) - try: - path.unlink(missing_ok=False) - return True - except FileNotFoundError: + """Delete a session from the database. Returns True if it existed.""" + db = self._get_db() + if db is None: return False - except OSError: - logger.debug("Failed to delete session file %s", path, exc_info=True) - return False - - def _is_expired(self, data: dict) -> bool: - """Check whether a persisted session dict has exceeded the TTL.""" - if not self._ttl: - return False - updated = data.get("updated_at") - if not updated: - return True # no timestamp → treat as expired try: - ts = datetime.fromisoformat(updated) - return datetime.utcnow() - ts > self._ttl - except (ValueError, TypeError): - return True - - def _expire_old_sessions(self) -> None: - """Remove persisted session files that have exceeded the TTL.""" - if not self._ttl: - return - try: - for path in self._sessions_dir.glob("*.json"): - try: - data = json.loads(path.read_text(encoding="utf-8")) - if self._is_expired(data): - logger.info("Expiring old ACP session %s", path.stem) - path.unlink(missing_ok=True) - except Exception: - logger.debug("Skipping unreadable session file during expiry: %s", path) - except OSError: - pass + return db.delete_session(session_id) + except Exception: + logger.debug("Failed to delete ACP session %s from DB", session_id, exc_info=True) + return False # ---- internal ----------------------------------------------------------- diff --git a/tests/acp/test_session.py b/tests/acp/test_session.py index 0c705a71a6..43d9a72295 100644 --- a/tests/acp/test_session.py +++ b/tests/acp/test_session.py @@ -2,10 +2,10 @@ import json import pytest -from pathlib import Path from unittest.mock import MagicMock from acp_adapter.session import SessionManager, SessionState +from hermes_state import SessionDB def _mock_agent(): @@ -119,23 +119,25 @@ class TestListAndCleanup: # --------------------------------------------------------------------------- -# persistence — sessions survive process restarts +# persistence — sessions survive process restarts (via SessionDB) # --------------------------------------------------------------------------- class TestPersistence: - """Verify that sessions are persisted to disk and can be restored.""" + """Verify that sessions are persisted to SessionDB and can be restored.""" - def test_create_session_writes_json_file(self, manager): + def test_create_session_writes_to_db(self, manager): state = manager.create_session(cwd="/project") - path = manager._session_path(state.session_id) - assert path.exists() - data = json.loads(path.read_text()) - assert data["session_id"] == state.session_id - assert data["cwd"] == "/project" - assert data["history"] == [] + db = manager._get_db() + assert db is not None + row = db.get_session(state.session_id) + assert row is not None + assert row["source"] == "acp" + # cwd stored in model_config JSON + mc = json.loads(row["model_config"]) + assert mc["cwd"] == "/project" - def test_get_session_restores_from_disk(self, manager): + def test_get_session_restores_from_db(self, manager): """Simulate process restart: create session, drop from memory, get again.""" state = manager.create_session(cwd="/work") state.history.append({"role": "user", "content": "hello"}) @@ -148,7 +150,7 @@ class TestPersistence: with manager._lock: del manager._sessions[sid] - # get_session should transparently restore from disk. + # get_session should transparently restore from DB. restored = manager.get_session(sid) assert restored is not None assert restored.session_id == sid @@ -159,35 +161,36 @@ class TestPersistence: # Agent should have been recreated. assert restored.agent is not None - def test_save_session_updates_disk(self, manager): + def test_save_session_updates_db(self, manager): state = manager.create_session() state.history.append({"role": "user", "content": "test"}) manager.save_session(state.session_id) - data = json.loads(manager._session_path(state.session_id).read_text()) - assert len(data["history"]) == 1 - assert data["history"][0]["content"] == "test" + db = manager._get_db() + messages = db.get_messages_as_conversation(state.session_id) + assert len(messages) == 1 + assert messages[0]["content"] == "test" - def test_remove_session_deletes_file(self, manager): + def test_remove_session_deletes_from_db(self, manager): state = manager.create_session() - path = manager._session_path(state.session_id) - assert path.exists() + db = manager._get_db() + assert db.get_session(state.session_id) is not None manager.remove_session(state.session_id) - assert not path.exists() + assert db.get_session(state.session_id) is None - def test_cleanup_removes_all_files(self, manager): + def test_cleanup_removes_all_from_db(self, manager): s1 = manager.create_session() s2 = manager.create_session() - p1 = manager._session_path(s1.session_id) - p2 = manager._session_path(s2.session_id) - assert p1.exists() and p2.exists() + db = manager._get_db() + assert db.get_session(s1.session_id) is not None + assert db.get_session(s2.session_id) is not None manager.cleanup() - assert not p1.exists() - assert not p2.exists() + assert db.get_session(s1.session_id) is None + assert db.get_session(s2.session_id) is None - def test_list_sessions_includes_disk_only(self, manager): - """Sessions only on disk (not in memory) appear in list_sessions.""" - state = manager.create_session(cwd="/disk-only") + def test_list_sessions_includes_db_only(self, manager): + """Sessions only in DB (not in memory) appear in list_sessions.""" + state = manager.create_session(cwd="/db-only") sid = state.session_id # Drop from memory. @@ -198,8 +201,8 @@ class TestPersistence: ids = {s["session_id"] for s in listing} assert sid in ids - def test_fork_restores_source_from_disk(self, manager): - """Forking a session that is only on disk should work.""" + def test_fork_restores_source_from_db(self, manager): + """Forking a session that is only in DB should work.""" original = manager.create_session() original.history.append({"role": "user", "content": "context"}) manager.save_session(original.session_id) @@ -214,7 +217,7 @@ class TestPersistence: assert forked.history[0]["content"] == "context" assert forked.session_id != original.session_id - def test_update_cwd_restores_from_disk(self, manager): + def test_update_cwd_restores_from_db(self, manager): state = manager.create_session(cwd="/old") sid = state.session_id @@ -225,74 +228,56 @@ class TestPersistence: assert updated is not None assert updated.cwd == "/new" - # Should also be persisted. - data = json.loads(manager._session_path(sid).read_text()) - assert data["cwd"] == "/new" + # Should also be persisted in DB. + db = manager._get_db() + row = db.get_session(sid) + mc = json.loads(row["model_config"]) + assert mc["cwd"] == "/new" + def test_only_restores_acp_sessions(self, manager): + """get_session should not restore non-ACP sessions from DB.""" + db = manager._get_db() + # Manually create a CLI session in the DB. + db.create_session(session_id="cli-session-123", source="cli", model="test") + # Should not be found via ACP SessionManager. + assert manager.get_session("cli-session-123") is None -# --------------------------------------------------------------------------- -# TTL / expiry -# --------------------------------------------------------------------------- + def test_sessions_searchable_via_fts(self, manager): + """ACP sessions stored in SessionDB are searchable via FTS5.""" + state = manager.create_session() + state.history.append({"role": "user", "content": "how do I configure nginx"}) + state.history.append({"role": "assistant", "content": "Here is the nginx config..."}) + manager.save_session(state.session_id) + db = manager._get_db() + results = db.search_messages("nginx") + assert len(results) > 0 + session_ids = {r["session_id"] for r in results} + assert state.session_id in session_ids -class TestSessionExpiry: - def test_expired_session_not_restored(self, tmp_path): - """Sessions past TTL are deleted on access, not restored.""" - mgr = SessionManager(agent_factory=_mock_agent, sessions_dir=tmp_path, ttl_days=1) - state = mgr.create_session() - sid = state.session_id + def test_tool_calls_persisted(self, manager): + """Messages with tool_calls should round-trip through the DB.""" + state = manager.create_session() + state.history.append({ + "role": "assistant", + "content": None, + "tool_calls": [{"id": "tc_1", "type": "function", + "function": {"name": "terminal", "arguments": "{}"}}], + }) + state.history.append({ + "role": "tool", + "content": "output here", + "tool_call_id": "tc_1", + "name": "terminal", + }) + manager.save_session(state.session_id) - # Manually backdate the persisted timestamp. - path = mgr._session_path(sid) - data = json.loads(path.read_text()) - from datetime import datetime, timedelta - data["updated_at"] = (datetime.utcnow() - timedelta(days=10)).isoformat() - path.write_text(json.dumps(data)) + # Drop from memory, restore from DB. + with manager._lock: + del manager._sessions[state.session_id] - # Drop from memory. - with mgr._lock: - del mgr._sessions[sid] - - assert mgr.get_session(sid) is None - assert not path.exists() - - def test_expired_sessions_pruned_on_startup(self, tmp_path): - """Creating a new SessionManager prunes expired session files.""" - # Create a stale session file. - from datetime import datetime, timedelta - stale_id = "stale-session-id" - stale_data = { - "session_id": stale_id, - "cwd": ".", - "model": "", - "history": [], - "updated_at": (datetime.utcnow() - timedelta(days=30)).isoformat(), - } - stale_path = tmp_path / f"{stale_id}.json" - stale_path.write_text(json.dumps(stale_data)) - assert stale_path.exists() - - # Creating a SessionManager should prune it. - SessionManager(agent_factory=_mock_agent, sessions_dir=tmp_path, ttl_days=7) - assert not stale_path.exists() - - def test_no_expiry_with_zero_ttl(self, tmp_path): - """ttl_days=0 means sessions never expire.""" - from datetime import datetime, timedelta - old_id = "ancient" - old_data = { - "session_id": old_id, - "cwd": ".", - "model": "", - "history": [{"role": "user", "content": "old"}], - "updated_at": (datetime.utcnow() - timedelta(days=365)).isoformat(), - } - old_path = tmp_path / f"{old_id}.json" - old_path.write_text(json.dumps(old_data)) - - mgr = SessionManager(agent_factory=_mock_agent, sessions_dir=tmp_path, ttl_days=0) - # Should still exist and be loadable. - assert old_path.exists() - restored = mgr.get_session(old_id) + restored = manager.get_session(state.session_id) assert restored is not None - assert len(restored.history) == 1 + assert len(restored.history) == 2 + assert restored.history[0].get("tool_calls") is not None + assert restored.history[1].get("tool_call_id") == "tc_1"