diff --git a/acp_adapter/server.py b/acp_adapter/server.py
index 1081104e92bc..92788988b02c 100644
--- a/acp_adapter/server.py
+++ b/acp_adapter/server.py
@@ -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
diff --git a/acp_adapter/session.py b/acp_adapter/session.py
index 0f5b2428e1d7..10fb3ee6c7d6 100644
--- a/acp_adapter/session.py
+++ b/acp_adapter/session.py
@@ -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 -----------------------------------------------------------
 
diff --git a/tests/acp/test_session.py b/tests/acp/test_session.py
index 79cbcf534965..0c705a71a63c 100644
--- a/tests/acp/test_session.py
+++ b/tests/acp/test_session.py
@@ -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