fix: protect the active database during history rollback

docs/history-and-rollback.md

@@ -10,7 +10,7 @@ Use `/history` in the terminal UI or the session-history routes listed in [tool-
 
 In the TUI, choose a numbered history node to preview the target. Explicitly confirm with `/rollback NUMBER confirm` after reviewing it. The session head is updated only through the rollback operation. A stale head or a mismatched session is rejected instead of overwriting a newer branch. In the web API, the rollback endpoint is scoped to the requested session and node; authenticate and inspect its preview/result before continuing work.
 
-Rollback restores the recorded files in scope and selects the conversation branch captured by that node. It does not reverse external effects such as messages, network requests, deployments, provider charges, or database changes outside the captured workspace. It also does not imply that shared memory claims, agent credentials, or another session's state have been erased. Check those stores separately.
+Rollback restores the recorded files in scope and selects the conversation branch captured by that node. The active OpenKyrozen SQLite database and its journal files are excluded even when they are inside the workspace; older snapshots cannot restore them over the live database. If an old snapshot would replace the directory containing that database with a file, rollback is rejected. It does not reverse external effects such as messages, network requests, deployments, provider charges, or database changes outside the captured workspace. It also does not imply that shared memory claims, agent credentials, or another session's state have been erased. Check those stores separately.
 
 ## Recovery points and safety
 

openkyrozen/workspace/history.py

@@ -38,7 +38,14 @@ def __init__(self, store: HistoryStore, root: str | os.PathLike[str], state_root
                  *, source_scope_id: str, user_id: str = "local", workspace_id: str = "default",
                  session_id: str):
         self.store = store
-        self.root = Path(root).expanduser().resolve()
+        root_path = Path(root).expanduser().absolute()
+        self.root = root_path.resolve()
+        database = Path(store.path).expanduser().absolute()
+        if database.is_relative_to(root_path):
+            database = self.root / database.relative_to(root_path)
+        self.database_paths = {Path(str(path) + suffix)
+                               for path in (database, database.resolve())
+                               for suffix in ("", "-wal", "-shm", "-journal")}
         self.state_root = Path(state_root).expanduser().resolve()
         self.config_path = self.state_root.parent.parent / ".kyrozen_config.json"
         if not re.fullmatch(r"source-[a-f0-9]{32}", str(source_scope_id)):
@@ -67,7 +74,8 @@ def _excluded(self, path: Path) -> bool:
             relative = path.relative_to(self.root)
         except ValueError:
             return True
-        if ".git" in relative.parts:
+        if (".git" in relative.parts or path in self.database_paths
+                or (path.is_symlink() and any(db.is_relative_to(path) for db in self.database_paths))):
             return True
         try:
             if path == self.state_root or path.is_relative_to(self.state_root):
@@ -314,26 +322,34 @@ def _restore(self, node: dict[str, Any]) -> None:
         manifest = self._manifest(node)
         if not snapshot_workspace.is_dir():
             raise HistoryError("history workspace snapshot is missing")
+        for relative, item in manifest.items():
+            destination = self.root / relative
+            if (not self._excluded(destination) and item.get("kind") != "dir"
+                    and any(db.is_relative_to(destination) for db in self.database_paths)):
+                raise HistoryError("history restore would overwrite the active database directory")
         for base, directories, files in os.walk(self.root, topdown=False, followlinks=False):
             base_path = Path(base)
             for name in files + directories:
                 path = base_path / name
-                if self._excluded(path):
+                if self._excluded(path) or any(db.is_relative_to(path) for db in self.database_paths):
                     continue
                 relative = path.relative_to(self.root).as_posix()
                 if relative not in manifest:
                     self._remove_path(path)
         for relative, item in sorted(manifest.items(), key=lambda pair: (pair[1].get("kind") != "dir", pair[0])):
             destination = self.root / relative
             kind = item.get("kind")
+            if self._excluded(destination):
+                continue
             if kind == "dir":
                 if destination.exists() and not destination.is_dir():
                     self._remove_path(destination)
                 destination.mkdir(parents=True, exist_ok=True)
-                try:
-                    os.chmod(destination, int(item.get("mode", 0o755)), follow_symlinks=False)
-                except (OSError, NotImplementedError):
-                    pass
+                if not any(db.is_relative_to(destination) for db in self.database_paths):
+                    try:
+                        os.chmod(destination, int(item.get("mode", 0o755)), follow_symlinks=False)
+                    except (OSError, NotImplementedError):
+                        pass
             elif kind == "symlink":
                 if destination.exists() or destination.is_symlink():
                     self._remove_path(destination)

openkyrozen/workspace/history_port.py

@@ -1,6 +1,7 @@
 """Repository operations consumed by HistoryStore."""
 from __future__ import annotations
 from typing import Protocol, Any
+from pathlib import Path
 from dataclasses import dataclass
 
 class HistoryError(RuntimeError):
@@ -14,6 +15,7 @@ class TurnToken:
 
 
 class HistoryStore(Protocol):
+    path: Path
     def append_event(self, event_type: str, payload: Any, *, user_id: str='local', workspace_id: str='default', session_id: str | None=None, task_id: str | None=None):
         ...
     def history_head(self, *, user_id: str='local', workspace_id: str='default', session_id: str):

tests/test_history.py

@@ -117,6 +117,81 @@ def test_branching_restore_preserves_files_descendants_memory_tasks_and_git(self
         )
         self.assertEqual(reopened.current()["id"], n1["id"])
 
+    def test_snapshot_excludes_active_database_and_sidecars_inside_workspace(self):
+        db_dir = self.root / "database"
+        db_dir.mkdir()
+        store = EventStore(db_dir / "state.sqlite3")
+        manager = HistoryManager(store, self.root, self.state,
+            source_scope_id=source_scope_id(self.root), session_id="inside")
+        (db_dir / "state.sqlite3-unrelated").write_text("user file")
+        with store.database.connection() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            _, manifest, _ = manager._snapshot("hist_" + "a" * 32)
+            connection.rollback()
+        self.assertNotIn("database/state.sqlite3", manifest)
+        self.assertNotIn("database/state.sqlite3-wal", manifest)
+        self.assertNotIn("database/state.sqlite3-shm", manifest)
+        self.assertIn("database/state.sqlite3-unrelated", manifest)
+
+    def test_restore_ignores_database_in_legacy_snapshot_and_preserves_new_db_directory(self):
+        before_db = self.manager.ensure_root()
+        db_dir = self.root / "database"
+        db_dir.mkdir()
+        store = EventStore(db_dir / "state.sqlite3")
+        manager = HistoryManager(store, self.root, self.state,
+            source_scope_id=source_scope_id(self.root), session_id="inside")
+        # Construct a legacy snapshot that includes the live database.
+        with patch.object(manager, "_excluded", side_effect=lambda path: False):
+            legacy = manager.ensure_root()
+        store.append_event("memory.test", {"content": "keep"})
+        manager._restore(legacy)
+        self.assertEqual(len(store.list_events("memory.test")), 1)
+        # A snapshot from before the database directory existed must not remove it.
+        manager._restore(before_db)
+        self.assertEqual(len(store.list_events("memory.test")), 1)
+
+    @unittest.skipIf(os.name == "nt", "Windows symlinks may require extra privileges")
+    def test_snapshot_and_restore_preserve_database_directory_symlink(self):
+        actual = Path(self.directory.name) / "external-database"
+        actual.mkdir()
+        link = self.root / "database-link"
+        link.symlink_to(actual, target_is_directory=True)
+        store = EventStore(link / "state.sqlite3")
+        manager = HistoryManager(store, self.root, self.state,
+            source_scope_id=source_scope_id(self.root), session_id="inside")
+        baseline = manager.ensure_root()
+        self.assertNotIn("database-link", manager._manifest(baseline))
+        manager._restore(baseline)
+        self.assertTrue(link.is_symlink())
+        store.append_event("memory.test", {"content": "keep"})
+        self.assertEqual(len(store.list_events("memory.test")), 1)
+
+    @unittest.skipIf(os.name == "nt", "POSIX directory modes are not supported on Windows")
+    def test_restore_preserves_database_directory_permissions(self):
+        db_dir = self.root / "database"
+        db_dir.mkdir(mode=0o700)
+        baseline = self.manager.ensure_root()
+        store = EventStore(db_dir / "state.sqlite3")
+        os.chmod(db_dir, 0o750)
+        manager = HistoryManager(store, self.root, self.state,
+            source_scope_id=source_scope_id(self.root), session_id="inside")
+        manager._restore(baseline)
+        self.assertEqual(stat.S_IMODE(db_dir.stat().st_mode), 0o750)
+
+    def test_restore_rejects_file_that_would_replace_database_directory(self):
+        (self.root / "database").write_text("old file")
+        baseline = self.manager.ensure_root()
+        (self.root / "database").unlink()
+        (self.root / "database").mkdir()
+        store = EventStore(self.root / "database" / "state.sqlite3")
+        manager = HistoryManager(store, self.root, self.state,
+            source_scope_id=source_scope_id(self.root), session_id="inside")
+        (self.root / "later.txt").write_text("keep if preflight fails")
+        with self.assertRaisesRegex(HistoryError, "active database directory"):
+            manager._restore(baseline)
+        self.assertTrue(store.path.exists())
+        self.assertTrue((self.root / "later.txt").exists())
+
     def test_scope_and_stale_head_are_rejected(self):
         root = self.manager.ensure_root()
         other = HistoryManager(