mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-10-20 05:15:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix: persist ACP sessions to disk so they survive process restarts

Browse source 
The ACP adapter stored sessions entirely in-memory. When the editor
restarted the ACP subprocess (idle timeout, crash, system sleep/wake,
editor restart), all sessions were lost. The editor's load_session /
resume_session calls would fail to find the session, forcing a new
empty session and losing all conversation history.

Changes:
- SessionManager now persists each session as a JSON file under
  ~/.hermes/acp_sessions/<session_id>.json
- get_session() transparently restores from disk when not in memory
- update_cwd(), fork_session(), list_sessions() all check disk
- server.py calls save_session() after prompt completion, /reset,
  /compact, and model switches
- cleanup() and remove_session() delete disk files too
- Sessions have a 7-day TTL; expired sessions are pruned on startup
- Atomic writes via tempfile + os.replace to prevent corruption
- 11 new tests covering persistence, disk restoration, and TTL expiry
This commit is contained in:
Test 2026-03-19 10:03:46 -07:00
parent 35558dadf4
commit 0e9138173e
3 changed files with 440 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

6
acp_adapter/server.py
View file

@ -304,6 +304,8 @@ class HermesACPAgent(acp.Agent):
if result.get("messages"):
state.history = result["messages"]
# Persist updated history so sessions survive process restarts.
self.session_manager.save_session(session_id)
final_response = result.get("final_response", "")
if final_response and conn:
@ -400,6 +402,7 @@ class HermesACPAgent(acp.Agent):
cwd=state.cwd,
model=new_model,
)
self.session_manager.save_session(state.session_id)
provider_label = target_provider or getattr(state.agent, "provider", "auto")
logger.info("Session %s: model switched to %s", state.session_id, new_model)
return f"Model switched to: {new_model}\nProvider: {provider_label}"
@ -444,6 +447,7 @@ class HermesACPAgent(acp.Agent):
def _cmd_reset(self, args: str, state: SessionState) -> str:
state.history.clear()
self.session_manager.save_session(state.session_id)
return "Conversation history cleared."
def _cmd_compact(self, args: str, state: SessionState) -> str:
@ -453,6 +457,7 @@ class HermesACPAgent(acp.Agent):
agent = state.agent
if hasattr(agent, "compress_context"):
agent.compress_context(state.history)
self.session_manager.save_session(state.session_id)
return f"Context compressed. Messages: {len(state.history)}"
return "Context compression not available for this agent."
except Exception as e:
@ -475,5 +480,6 @@ class HermesACPAgent(acp.Agent):
cwd=state.cwd,
model=model_id,
)
self.session_manager.save_session(session_id)
logger.info("Session %s: model switched to %s", session_id, model_id)
return None

281
acp_adapter/session.py
View file

@ -1,15 +1,29 @@
"""ACP session manager — maps ACP sessions to Hermes AIAgent instances."""
"""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.
"""
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."""
@ -46,18 +60,43 @@ class SessionState:
class SessionManager:
"""Thread-safe manager for ACP sessions backed by Hermes AIAgent instances."""
"""Thread-safe manager for ACP sessions backed by Hermes AIAgent instances.
def __init__(self, agent_factory=None):
Sessions are held in-memory for fast access **and** persisted to disk as
JSON files under *sessions_dir* so they survive process restarts.
"""
def __init__(self, agent_factory=None, sessions_dir: Path | str | None = None,
ttl_days: int = _DEFAULT_TTL_DAYS):
"""
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.
"""
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()
# ---- public API ---------------------------------------------------------
@ -77,54 +116,67 @@ class SessionManager:
with self._lock:
self._sessions[session_id] = state
_register_task_cwd(session_id, cwd)
self._persist(state)
logger.info("Created ACP session %s (cwd=%s)", session_id, cwd)
return state
def get_session(self, session_id: str) -> Optional[SessionState]:
"""Return the session for *session_id*, or ``None``."""
"""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.
"""
with self._lock:
return self._sessions.get(session_id)
state = self._sessions.get(session_id)
if state is not None:
return state
# Attempt to restore from disk.
return self._restore(session_id)
def remove_session(self, session_id: str) -> bool:
"""Remove a session. Returns True if it existed."""
"""Remove a session from memory and disk. Returns True if it existed."""
with self._lock:
existed = self._sessions.pop(session_id, None) is not None
if existed:
disk_existed = self._delete_persisted(session_id)
if existed or disk_existed:
_clear_task_cwd(session_id)
return existed
return existed or disk_existed
def fork_session(self, session_id: str, cwd: str = ".") -> Optional[SessionState]:
"""Deep-copy a session's history into a new session."""
import threading
with self._lock:
original = self._sessions.get(session_id)
if original is None:
return None
original = self.get_session(session_id) # checks disk too
if original is None:
return None
new_id = str(uuid.uuid4())
agent = self._make_agent(
session_id=new_id,
cwd=cwd,
model=original.model or None,
)
state = SessionState(
session_id=new_id,
agent=agent,
cwd=cwd,
model=getattr(agent, "model", original.model) or original.model,
history=copy.deepcopy(original.history),
cancel_event=threading.Event(),
)
new_id = str(uuid.uuid4())
agent = self._make_agent(
session_id=new_id,
cwd=cwd,
model=original.model or None,
)
state = SessionState(
session_id=new_id,
agent=agent,
cwd=cwd,
model=getattr(agent, "model", original.model) or original.model,
history=copy.deepcopy(original.history),
cancel_event=threading.Event(),
)
with self._lock:
self._sessions[new_id] = state
_register_task_cwd(new_id, cwd)
self._persist(state)
logger.info("Forked ACP session %s -> %s", session_id, new_id)
return state
def list_sessions(self) -> List[Dict[str, Any]]:
"""Return lightweight info dicts for all sessions."""
"""Return lightweight info dicts for all sessions (memory + disk)."""
# Collect in-memory sessions first.
with self._lock:
return [
seen_ids = set(self._sessions.keys())
results = [
{
"session_id": s.session_id,
"cwd": s.cwd,
@ -134,23 +186,184 @@ class SessionManager:
for s in self._sessions.values()
]
# 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)
continue
results.append({
"session_id": sid,
"cwd": data.get("cwd", "."),
"model": data.get("model", ""),
"history_len": len(data.get("history", [])),
})
except Exception:
logger.debug("Skipping unreadable session file %s", path)
except OSError:
pass
return results
def update_cwd(self, session_id: str, cwd: str) -> Optional[SessionState]:
"""Update the working directory for a session and its tool overrides."""
with self._lock:
state = self._sessions.get(session_id)
if state is None:
return None
state.cwd = cwd
state = self.get_session(session_id) # checks disk too
if state is None:
return None
state.cwd = cwd
_register_task_cwd(session_id, cwd)
self._persist(state)
return state
def cleanup(self) -> None:
"""Remove all sessions and clear task-specific cwd overrides."""
"""Remove all sessions (memory and disk) 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
def save_session(self, session_id: str) -> None:
"""Persist the current state of a session to disk.
Called by the server after prompt completion, slash commands that
mutate history, and model switches.
"""
with self._lock:
state = self._sessions.get(session_id)
if state is not None:
self._persist(state)
# ---- persistence ---------------------------------------------------------
def _session_path(self, session_id: str) -> Path:
return self._sessions_dir / f"{session_id}.json"
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)
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.
try:
os.unlink(tmp)
except OSError:
pass
raise
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."""
import threading
path = self._session_path(session_id)
if not path.exists():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except Exception:
logger.warning("Failed to read persisted 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)
return None
cwd = data.get("cwd", ".")
model = data.get("model") or None
history = data.get("history", [])
try:
agent = self._make_agent(session_id=session_id, cwd=cwd, model=model)
except Exception:
logger.warning("Failed to recreate agent for ACP session %s", session_id, exc_info=True)
return None
state = SessionState(
session_id=session_id,
agent=agent,
cwd=cwd,
model=model or getattr(agent, "model", "") or "",
history=history,
cancel_event=threading.Event(),
)
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))
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:
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
# ---- internal -----------------------------------------------------------

188
tests/acp/test_session.py
View file

@ -1,15 +1,21 @@
"""Tests for acp_adapter.session — SessionManager and SessionState."""
import json
import pytest
from pathlib import Path
from unittest.mock import MagicMock
from acp_adapter.session import SessionManager, SessionState
def _mock_agent():
return MagicMock(name="MockAIAgent")
@pytest.fixture()
def manager():
"""SessionManager with a mock agent factory (avoids needing API keys)."""
return SessionManager(agent_factory=lambda: MagicMock(name="MockAIAgent"))
return SessionManager(agent_factory=_mock_agent)
# ---------------------------------------------------------------------------
@ -110,3 +116,183 @@ class TestListAndCleanup:
assert manager.get_session(state.session_id) is None
# Removing again returns False
assert manager.remove_session(state.session_id) is False
# ---------------------------------------------------------------------------
# persistence — sessions survive process restarts
# ---------------------------------------------------------------------------
class TestPersistence:
"""Verify that sessions are persisted to disk and can be restored."""
def test_create_session_writes_json_file(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"] == []
def test_get_session_restores_from_disk(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"})
state.history.append({"role": "assistant", "content": "hi there"})
manager.save_session(state.session_id)
sid = state.session_id
# Drop from in-memory store (simulates process restart).
with manager._lock:
del manager._sessions[sid]
# get_session should transparently restore from disk.
restored = manager.get_session(sid)
assert restored is not None
assert restored.session_id == sid
assert restored.cwd == "/work"
assert len(restored.history) == 2
assert restored.history[0]["content"] == "hello"
assert restored.history[1]["content"] == "hi there"
# Agent should have been recreated.
assert restored.agent is not None
def test_save_session_updates_disk(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"
def test_remove_session_deletes_file(self, manager):
state = manager.create_session()
path = manager._session_path(state.session_id)
assert path.exists()
manager.remove_session(state.session_id)
assert not path.exists()
def test_cleanup_removes_all_files(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()
manager.cleanup()
assert not p1.exists()
assert not p2.exists()
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")
sid = state.session_id
# Drop from memory.
with manager._lock:
del manager._sessions[sid]
listing = manager.list_sessions()
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."""
original = manager.create_session()
original.history.append({"role": "user", "content": "context"})
manager.save_session(original.session_id)
# Drop original from memory.
with manager._lock:
del manager._sessions[original.session_id]
forked = manager.fork_session(original.session_id, cwd="/fork")
assert forked is not None
assert len(forked.history) == 1
assert forked.history[0]["content"] == "context"
assert forked.session_id != original.session_id
def test_update_cwd_restores_from_disk(self, manager):
state = manager.create_session(cwd="/old")
sid = state.session_id
with manager._lock:
del manager._sessions[sid]
updated = manager.update_cwd(sid, "/new")
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"
# ---------------------------------------------------------------------------
# TTL / expiry
# ---------------------------------------------------------------------------
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
# 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.
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)
assert restored is not None
assert len(restored.history) == 1