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
This commit is contained in:
Test
+158
-145
@@ -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 -----------------------------------------------------------
|
||||
|
||||
|
||||
+81
-96
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user