my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Memory Bank Isolation
#3	======================
#4
#5	Provides named memory banks that are fully isolated at the database level.
#6	Each bank gets its own SQLite file under:
#7	~/.hermes/mnemosyne/data/banks/<bank_name>/mnemosyne.db
#8
#9	This enables:
#10	- Multi-tenant deployments (one bank per user/tenant)
#11	- Domain separation (work vs personal vs project-specific memories)
#12	- Testing isolation (test bank doesn't pollute production)
#13	- Compliance boundaries (sensitive data in dedicated banks)
#14
#15	API:
#16	BankManager.create_bank("work")
#17	BankManager.list_banks() -> ["default", "work", "personal"]
#18	BankManager.delete_bank("work")
#19	BankManager.bank_exists("work") -> bool
#20
#21	Mnemosyne(bank="work") # All operations isolated to work bank
#22	"""
#23
#24	import os
#25	import shutil
#26	import sqlite3
#27	from pathlib import Path
#28	from typing import List, Optional
#29
#30	# On Fly.io and other ephemeral VMs, only ~/.hermes is persisted.
#31	DEFAULT_DATA_DIR = Path.home() / ".hermes" / "mnemosyne" / "data"
#32	BANKS_DIR = DEFAULT_DATA_DIR / "banks"
#33
#34	if os.environ.get("MNEMOSYNE_DATA_DIR"):
#35	DEFAULT_DATA_DIR = Path(os.environ.get("MNEMOSYNE_DATA_DIR"))
#36	BANKS_DIR = DEFAULT_DATA_DIR / "banks"
#37
#38
#39	def _default_data_dir() -> Path:
#40	"""Return the current default data directory, honoring runtime env changes."""
#41	if os.environ.get("MNEMOSYNE_DATA_DIR"):
#42	return Path(os.environ["MNEMOSYNE_DATA_DIR"])
#43	return DEFAULT_DATA_DIR
#44
#45
#46	class BankManager:
#47	"""
#48	Manage named memory banks.
#49
#50	Each bank is a self-contained directory with its own SQLite database,
#51	enabling complete isolation between banks.
#52	"""
#53
#54	def __init__(self, data_dir: Path = None):
#55	self.data_dir = data_dir or _default_data_dir()
#56	self.banks_dir = self.data_dir / "banks"
#57	self.banks_dir.mkdir(parents=True, exist_ok=True)
#58
#59	def create_bank(self, name: str) -> Path:
#60	"""
#61	Create a new memory bank.
#62
#63	Args:
#64	name: Bank name. Must be alphanumeric with hyphens/underscores.
#65
#66	Returns:
#67	Path to the bank's database file.
#68
#69	Raises:
#70	ValueError: If name is invalid or already exists.
#71	"""
#72	self._validate_name(name)
#73	bank_dir = self.banks_dir / name
#74	if bank_dir.exists():
#75	raise ValueError(f"Bank '{name}' already exists")
#76	bank_dir.mkdir(parents=True)
#77	# Initialize the database by creating it
#78	db_path = bank_dir / "mnemosyne.db"
#79	conn = sqlite3.connect(str(db_path))
#80	conn.execute("SELECT 1")
#81	conn.close()
#82	return db_path
#83
#84	def delete_bank(self, name: str, force: bool = False) -> bool:
#85	"""
#86	Delete a memory bank and all its data.
#87
#88	Args:
#89	name: Bank name to delete.
#90	force: If False, refuses to delete the 'default' bank.
#91
#92	Returns:
#93	True if deleted, False if bank didn't exist.
#94
#95	Raises:
#96	ValueError: If trying to delete 'default' without force=True.
#97	"""
#98	if name == "default" and not force:
#99	raise ValueError("Cannot delete 'default' bank without force=True")
#100	bank_dir = self.banks_dir / name
#101	if not bank_dir.exists():
#102	return False
#103	shutil.rmtree(bank_dir)
#104	return True
#105
#106	def list_banks(self) -> List[str]:
#107	"""Return list of all existing bank names."""
#108	if not self.banks_dir.exists():
#109	return ["default"]
#110	banks = [d.name for d in self.banks_dir.iterdir() if d.is_dir()]
#111	# Ensure 'default' is always present
#112	if "default" not in banks:
#113	banks.insert(0, "default")
#114	return sorted(banks)
#115
#116	def bank_exists(self, name: str) -> bool:
#117	"""Check if a bank exists."""
#118	if name == "default":
#119	return True
#120	return (self.banks_dir / name).is_dir()
#121
#122	def get_bank_db_path(self, name: str) -> Path:
#123	"""
#124	Get the database path for a bank.
#125
#126	The 'default' bank uses the legacy path (data_dir/mnemosyne.db).
#127	All other banks use banks_dir/<name>/mnemosyne.db.
#128	"""
#129	if name == "default" or not name:
#130	return self.data_dir / "mnemosyne.db"
#131	return self.banks_dir / name / "mnemosyne.db"
#132
#133	def rename_bank(self, old_name: str, new_name: str) -> Path:
#134	"""
#135	Rename a bank.
#136
#137	Args:
#138	old_name: Existing bank name.
#139	new_name: New bank name.
#140
#141	Returns:
#142	Path to the new bank's database.
#143
#144	Raises:
#145	ValueError: If old_name doesn't exist or new_name is taken.
#146	"""
#147	if old_name == "default":
#148	raise ValueError("Cannot rename 'default' bank")
#149	self._validate_name(new_name)
#150	old_dir = self.banks_dir / old_name
#151	new_dir = self.banks_dir / new_name
#152	if not old_dir.exists():
#153	raise ValueError(f"Bank '{old_name}' does not exist")
#154	if new_dir.exists():
#155	raise ValueError(f"Bank '{new_name}' already exists")
#156	old_dir.rename(new_dir)
#157	return new_dir / "mnemosyne.db"
#158
#159	def get_bank_stats(self, name: str) -> dict:
#160	"""
#161	Get statistics for a bank.
#162
#163	Returns dict with: exists, db_path, db_size_bytes.
#164	"""
#165	db_path = self.get_bank_db_path(name)
#166	exists = db_path.exists()
#167	size = db_path.stat().st_size if exists else 0
#168	return {
#169	"name": name,
#170	"exists": exists,
#171	"db_path": str(db_path),
#172	"db_size_bytes": size,
#173	}
#174
#175	def _validate_name(self, name: str):
#176	"""Validate bank name format."""
#177	if not name:
#178	raise ValueError("Bank name cannot be empty")
#179	if name == "default":
#180	return # 'default' is always valid
#181	if not all(c.isalnum() or c in "-_" for c in name):
#182	raise ValueError(f"Invalid bank name '{name}'. Use alphanumeric, hyphens, underscores only.")
#183	if len(name) > 64:
#184	raise ValueError(f"Bank name '{name}' exceeds 64 characters")
#185
#186
#187	# Module-level convenience functions
#188	def create_bank(name: str, data_dir: Path = None) -> Path:
#189	"""Create a new memory bank."""
#190	return BankManager(data_dir).create_bank(name)
#191
#192
#193	def delete_bank(name: str, data_dir: Path = None, force: bool = False) -> bool:
#194	"""Delete a memory bank."""
#195	return BankManager(data_dir).delete_bank(name, force=force)
#196
#197
#198	def list_banks(data_dir: Path = None) -> List[str]:
#199	"""List all memory banks."""
#200	return BankManager(data_dir).list_banks()
#201
#202
#203	def bank_exists(name: str, data_dir: Path = None) -> bool:
#204	"""Check if a bank exists."""
#205	return BankManager(data_dir).bank_exists(name)
#206

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public