my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Mnemosyne Plugin Architecture
#3	===============================
#4	Extensible plugin system for Mnemosyne memory operations.
#5
#6	Plugins can hook into memory lifecycle events:
#7	- on_remember: called when a memory is stored
#8	- on_recall: called when a memory is recalled
#9	- on_consolidate: called during sleep/consolidation
#10	- on_invalidate: called when a memory is invalidated
#11
#12	Plugin discovery loads plugins from ~/.hermes/mnemosyne/plugins/
#13	and built-in plugins are always available.
#14	"""
#15
#16	import abc
#17	import importlib
#18	import importlib.util
#19	import inspect
#20	import json
#21	import logging
#22	import sys
#23	import time
#24	from datetime import datetime
#25	from pathlib import Path
#26	from typing import Any, Callable, Dict, List, Optional, Type
#27
#28	# Plugin directory under ~/.hermes
#29	DEFAULT_PLUGIN_DIR = Path.home() / ".hermes" / "mnemosyne" / "plugins"
#30
#31	logger = logging.getLogger(__name__)
#32
#33
#34	class MnemosynePlugin(abc.ABC):
#35	"""
#36	Base class for all Mnemosyne plugins.
#37
#38	Subclasses must implement the four lifecycle hooks.
#39	Each hook receives the relevant data and can perform
#40	side effects (logging, metrics, filtering, etc.).
#41	"""
#42
#43	name: str = ""
#44	version: str = "1.0.0"
#45	enabled: bool = True
#46
#47	def __init__(self, config: Dict[str, Any] = None):
#48	self.config = config or {}
#49	self._initialized = False
#50
#51	def initialize(self) -> None:
#52	"""Called once when the plugin is loaded."""
#53	self._initialized = True
#54
#55	def shutdown(self) -> None:
#56	"""Called once when the plugin is unloaded."""
#57	self._initialized = False
#58
#59	@abc.abstractmethod
#60	def on_remember(self, memory: Dict[str, Any]) -> None:
#61	"""Called when a memory is stored."""
#62	...
#63
#64	@abc.abstractmethod
#65	def on_recall(self, memory: Dict[str, Any]) -> None:
#66	"""Called when a memory is recalled."""
#67	...
#68
#69	@abc.abstractmethod
#70	def on_consolidate(self, summary: Dict[str, Any]) -> None:
#71	"""Called during sleep/consolidation."""
#72	...
#73
#74	@abc.abstractmethod
#75	def on_invalidate(self, memory_id: str) -> None:
#76	"""Called when a memory is invalidated."""
#77	...
#78
#79	def to_dict(self) -> Dict[str, Any]:
#80	"""Serialize plugin metadata."""
#81	return {
#82	"name": self.name,
#83	"version": self.version,
#84	"enabled": self.enabled,
#85	"initialized": self._initialized,
#86	"config": self.config,
#87	}
#88
#89
#90	class LoggingPlugin(MnemosynePlugin):
#91	"""
#92	Built-in plugin that logs all memory operations.
#93	"""
#94
#95	name = "logging"
#96	version = "1.0.0"
#97
#98	def __init__(self, config: Dict[str, Any] = None):
#99	super().__init__(config)
#100	self.log_level = self.config.get("log_level", "INFO")
#101	self._memory_log: List[Dict[str, Any]] = []
#102	self._max_entries = self.config.get("max_entries", 10000)
#103
#104	def on_remember(self, memory: Dict[str, Any]) -> None:
#105	entry = {
#106	"event": "remember",
#107	"timestamp": datetime.now().isoformat(),
#108	"memory_id": memory.get("id"),
#109	"content_preview": self._preview(memory.get("content", "")),
#110	}
#111	self._append(entry)
#112	logger.log(getattr(logging, self.log_level, logging.INFO),
#113	"[LoggingPlugin] remember: %s", entry)
#114
#115	def on_recall(self, memory: Dict[str, Any]) -> None:
#116	entry = {
#117	"event": "recall",
#118	"timestamp": datetime.now().isoformat(),
#119	"memory_id": memory.get("id"),
#120	"content_preview": self._preview(memory.get("content", "")),
#121	}
#122	self._append(entry)
#123	logger.log(getattr(logging, self.log_level, logging.INFO),
#124	"[LoggingPlugin] recall: %s", entry)
#125
#126	def on_consolidate(self, summary: Dict[str, Any]) -> None:
#127	entry = {
#128	"event": "consolidate",
#129	"timestamp": datetime.now().isoformat(),
#130	"summary_preview": self._preview(summary.get("summary", "")),
#131	"source_count": len(summary.get("source_wm_ids", [])),
#132	}
#133	self._append(entry)
#134	logger.log(getattr(logging, self.log_level, logging.INFO),
#135	"[LoggingPlugin] consolidate: %s", entry)
#136
#137	def on_invalidate(self, memory_id: str) -> None:
#138	entry = {
#139	"event": "invalidate",
#140	"timestamp": datetime.now().isoformat(),
#141	"memory_id": memory_id,
#142	}
#143	self._append(entry)
#144	logger.log(getattr(logging, self.log_level, logging.INFO),
#145	"[LoggingPlugin] invalidate: %s", entry)
#146
#147	def _preview(self, content: str, max_len: int = 80) -> str:
#148	if len(content) <= max_len:
#149	return content
#150	return content[:max_len] + "..."
#151
#152	def _append(self, entry: Dict[str, Any]) -> None:
#153	self._memory_log.append(entry)
#154	if len(self._memory_log) > self._max_entries:
#155	self._memory_log.pop(0)
#156
#157	def get_log(self) -> List[Dict[str, Any]]:
#158	"""Return the in-memory log entries."""
#159	return list(self._memory_log)
#160
#161	def clear_log(self) -> None:
#162	"""Clear the in-memory log."""
#163	self._memory_log.clear()
#164
#165
#166	class MetricsPlugin(MnemosynePlugin):
#167	"""
#168	Built-in plugin that collects performance metrics for memory operations.
#169	"""
#170
#171	name = "metrics"
#172	version = "1.0.0"
#173
#174	def __init__(self, config: Dict[str, Any] = None):
#175	super().__init__(config)
#176	self._counters: Dict[str, int] = {
#177	"remember": 0,
#178	"recall": 0,
#179	"consolidate": 0,
#180	"invalidate": 0,
#181	}
#182	self._timings: Dict[str, List[float]] = {
#183	"remember": [],
#184	"recall": [],
#185	"consolidate": [],
#186	"invalidate": [],
#187	}
#188	self._max_timing_samples = self.config.get("max_timing_samples", 1000)
#189
#190	def on_remember(self, memory: Dict[str, Any]) -> None:
#191	self._counters["remember"] += 1
#192
#193	def on_recall(self, memory: Dict[str, Any]) -> None:
#194	self._counters["recall"] += 1
#195
#196	def on_consolidate(self, summary: Dict[str, Any]) -> None:
#197	self._counters["consolidate"] += 1
#198
#199	def on_invalidate(self, memory_id: str) -> None:
#200	self._counters["invalidate"] += 1
#201
#202	def record_timing(self, event: str, duration_ms: float) -> None:
#203	"""Record the duration of an operation."""
#204	if event not in self._timings:
#205	self._timings[event] = []
#206	self._timings[event].append(duration_ms)
#207	if len(self._timings[event]) > self._max_timing_samples:
#208	self._timings[event].pop(0)
#209
#210	def get_counters(self) -> Dict[str, int]:
#211	"""Return event counters."""
#212	return dict(self._counters)
#213
#214	def get_timings(self, event: str) -> List[float]:
#215	"""Return timing samples for an event."""
#216	return list(self._timings.get(event, []))
#217
#218	def get_average_timing(self, event: str) -> Optional[float]:
#219	"""Return average timing for an event."""
#220	samples = self._timings.get(event, [])
#221	if not samples:
#222	return None
#223	return sum(samples) / len(samples)
#224
#225	def reset(self) -> None:
#226	"""Reset all counters and timings."""
#227	for key in self._counters:
#228	self._counters[key] = 0
#229	for key in self._timings:
#230	self._timings[key].clear()
#231
#232	def get_summary(self) -> Dict[str, Any]:
#233	"""Return a summary of all metrics."""
#234	summary = {
#235	"counters": self.get_counters(),
#236	"averages": {},
#237	}
#238	for event in self._timings:
#239	avg = self.get_average_timing(event)
#240	summary["averages"][event] = avg
#241	return summary
#242
#243
#244	class FilterPlugin(MnemosynePlugin):
#245	"""
#246	Built-in plugin that filters memories based on custom rules.
#247
#248	Rules are callables registered via add_rule().
#249	Each rule receives the memory dict and returns True to allow,
#250	False to block. Blocked memories are tracked but not passed
#251	to downstream plugins.
#252	"""
#253
#254	name = "filter"
#255	version = "1.0.0"
#256
#257	def __init__(self, config: Dict[str, Any] = None):
#258	super().__init__(config)
#259	self._rules: List[Callable[[Dict[str, Any]], bool]] = []
#260	self._blocked: List[Dict[str, Any]] = []
#261	self._max_blocked = self.config.get("max_blocked", 1000)
#262
#263	def add_rule(self, rule: Callable[[Dict[str, Any]], bool]) -> None:
#264	"""Register a filtering rule."""
#265	self._rules.append(rule)
#266
#267	def remove_rule(self, rule: Callable[[Dict[str, Any]], bool]) -> None:
#268	"""Unregister a filtering rule."""
#269	if rule in self._rules:
#270	self._rules.remove(rule)
#271
#272	def clear_rules(self) -> None:
#273	"""Remove all filtering rules."""
#274	self._rules.clear()
#275
#276	def on_remember(self, memory: Dict[str, Any]) -> None:
#277	if not self._passes(memory):
#278	self._block(memory)
#279
#280	def on_recall(self, memory: Dict[str, Any]) -> None:
#281	if not self._passes(memory):
#282	self._block(memory)
#283
#284	def on_consolidate(self, summary: Dict[str, Any]) -> None:
#285	if not self._passes(summary):
#286	self._block(summary)
#287
#288	def on_invalidate(self, memory_id: str) -> None:
#289	pass
#290
#291	def _passes(self, item: Dict[str, Any]) -> bool:
#292	for rule in self._rules:
#293	try:
#294	if not rule(item):
#295	return False
#296	except Exception:
#297	return False
#298	return True
#299
#300	def _block(self, item: Dict[str, Any]) -> None:
#301	self._blocked.append({
#302	"timestamp": datetime.now().isoformat(),
#303	"item": item,
#304	})
#305	if len(self._blocked) > self._max_blocked:
#306	self._blocked.pop(0)
#307
#308	def get_blocked(self) -> List[Dict[str, Any]]:
#309	"""Return all blocked items."""
#310	return list(self._blocked)
#311
#312	def is_blocked(self, memory_id: str) -> bool:
#313	"""Check if a memory ID has been blocked."""
#314	for entry in self._blocked:
#315	item = entry.get("item", {})
#316	if item.get("id") == memory_id:
#317	return True
#318	return False
#319
#320
#321	class PluginManager:
#322	"""
#323	Register, load, and manage Mnemosyne plugins.
#324
#325	Supports:
#326	- Built-in plugins (LoggingPlugin, MetricsPlugin, FilterPlugin)
#327	- External plugins discovered from ~/.hermes/mnemosyne/plugins/
#328	- Manual registration of plugin classes
#329	"""
#330
#331	def __init__(self, plugin_dir: Path = None):
#332	self._registry: Dict[str, Type[MnemosynePlugin]] = {}
#333	self._instances: Dict[str, MnemosynePlugin] = {}
#334	self._plugin_dir = plugin_dir or DEFAULT_PLUGIN_DIR
#335
#336	# Register built-in plugins
#337	self.register_plugin("logging", LoggingPlugin)
#338	self.register_plugin("metrics", MetricsPlugin)
#339	self.register_plugin("filter", FilterPlugin)
#340
#341	def register_plugin(self, name: str, plugin_class: Type[MnemosynePlugin]) -> None:
#342	"""
#343	Register a plugin class by name.
#344
#345	Args:
#346	name: Unique identifier for the plugin.
#347	plugin_class: Subclass of MnemosynePlugin.
#348
#349	Raises:
#350	TypeError: If plugin_class is not a subclass of MnemosynePlugin.
#351	ValueError: If name is already registered.
#352	"""
#353	if not inspect.isclass(plugin_class) or not issubclass(plugin_class, MnemosynePlugin):
#354	raise TypeError(f"plugin_class must be a subclass of MnemosynePlugin, got {plugin_class}")
#355	if name in self._registry:
#356	raise ValueError(f"Plugin '{name}' is already registered")
#357	self._registry[name] = plugin_class
#358
#359	def load_plugin(self, name: str, config: Dict[str, Any] = None) -> MnemosynePlugin:
#360	"""
#361	Instantiate and initialize a registered plugin.
#362
#363	Args:
#364	name: Registered plugin name.
#365	config: Optional configuration dict passed to the plugin.
#366
#367	Returns:
#368	The initialized plugin instance.
#369
#370	Raises:
#371	ValueError: If the plugin is not registered.
#372	RuntimeError: If the plugin is already loaded.
#373	"""
#374	if name not in self._registry:
#375	raise ValueError(f"Plugin '{name}' is not registered")
#376	if name in self._instances:
#377	raise RuntimeError(f"Plugin '{name}' is already loaded")
#378
#379	plugin_class = self._registry[name]
#380	instance = plugin_class(config=config or {})
#381	instance.initialize()
#382	self._instances[name] = instance
#383	logger.info("Loaded plugin: %s v%s", instance.name, instance.version)
#384	return instance
#385
#386	def unload_plugin(self, name: str) -> None:
#387	"""
#388	Cleanup and remove a loaded plugin.
#389
#390	Args:
#391	name: Loaded plugin name.
#392
#393	Raises:
#394	ValueError: If the plugin is not loaded.
#395	"""
#396	if name not in self._instances:
#397	raise ValueError(f"Plugin '{name}' is not loaded")
#398	instance = self._instances.pop(name)
#399	instance.shutdown()
#400	logger.info("Unloaded plugin: %s", name)
#401
#402	def list_plugins(self) -> List[Dict[str, Any]]:
#403	"""
#404	List all registered plugins with their load status.
#405
#406	Returns:
#407	List of dicts with keys: name, class, loaded, instance.
#408	"""
#409	result = []
#410	for name, plugin_class in self._registry.items():
#411	loaded = name in self._instances
#412	result.append({
#413	"name": name,
#414	"class": plugin_class.__name__,
#415	"loaded": loaded,
#416	"instance": self._instances.get(name),
#417	})
#418	return result
#419
#420	def get_plugin(self, name: str) -> Optional[MnemosynePlugin]:
#421	"""Return a loaded plugin instance, or None if not loaded."""
#422	return self._instances.get(name)
#423
#424	def is_loaded(self, name: str) -> bool:
#425	"""Check if a plugin is currently loaded."""
#426	return name in self._instances
#427
#428	def is_registered(self, name: str) -> bool:
#429	"""Check if a plugin class is registered."""
#430	return name in self._registry
#431
#432	def load_all(self, configs: Dict[str, Dict[str, Any]] = None) -> List[MnemosynePlugin]:
#433	"""
#434	Load all registered plugins.
#435
#436	Args:
#437	configs: Optional mapping of plugin name -> config dict.
#438
#439	Returns:
#440	List of loaded plugin instances.
#441	"""
#442	configs = configs or {}
#443	loaded = []
#444	for name in self._registry:
#445	if name not in self._instances:
#446	instance = self.load_plugin(name, config=configs.get(name, {}))
#447	loaded.append(instance)
#448	return loaded
#449
#450	def unload_all(self) -> None:
#451	"""Unload all loaded plugins."""
#452	for name in list(self._instances.keys()):
#453	self.unload_plugin(name)
#454
#455	def discover_plugins(self) -> List[str]:
#456	"""
#457	Discover and register plugins from the plugin directory.
#458
#459	Scans ~/.hermes/mnemosyne/plugins/ for Python files and
#460	registers any MnemosynePlugin subclasses found.
#461
#462	Returns:
#463	List of newly registered plugin names.
#464	"""
#465	discovered: List[str] = []
#466	if not self._plugin_dir.exists():
#467	return discovered
#468
#469	for file_path in self._plugin_dir.glob("*.py"):
#470	if file_path.name.startswith("_"):
#471	continue
#472	try:
#473	spec = importlib.util.spec_from_file_location(
#474	file_path.stem, str(file_path)
#475	)
#476	if spec is None or spec.loader is None:
#477	continue
#478	module = importlib.util.module_from_spec(spec)
#479	sys.modules[file_path.stem] = module
#480	spec.loader.exec_module(module)
#481
#482	for attr_name in dir(module):
#483	obj = getattr(module, attr_name)
#484	if (
#485	inspect.isclass(obj)
#486	and issubclass(obj, MnemosynePlugin)
#487	and obj is not MnemosynePlugin
#488	and not obj.__name__.startswith("_")
#489	):
#490	plugin_name = getattr(obj, "name", None) or obj.__name__.lower()
#491	if plugin_name not in self._registry:
#492	self.register_plugin(plugin_name, obj)
#493	discovered.append(plugin_name)
#494	except Exception as exc:
#495	logger.warning("Failed to load plugin from %s: %s", file_path, exc)
#496
#497	return discovered
#498
#499	def notify_remember(self, memory: Dict[str, Any]) -> None:
#500	"""Notify all loaded plugins of a remember event."""
#501	for instance in self._instances.values():
#502	if instance.enabled:
#503	try:
#504	instance.on_remember(memory)
#505	except Exception as exc:
#506	logger.error("Plugin %s on_remember error: %s", instance.name, exc)
#507
#508	def notify_recall(self, memory: Dict[str, Any]) -> None:
#509	"""Notify all loaded plugins of a recall event."""
#510	for instance in self._instances.values():
#511	if instance.enabled:
#512	try:
#513	instance.on_recall(memory)
#514	except Exception as exc:
#515	logger.error("Plugin %s on_recall error: %s", instance.name, exc)
#516
#517	def notify_consolidate(self, summary: Dict[str, Any]) -> None:
#518	"""Notify all loaded plugins of a consolidate event."""
#519	for instance in self._instances.values():
#520	if instance.enabled:
#521	try:
#522	instance.on_consolidate(summary)
#523	except Exception as exc:
#524	logger.error("Plugin %s on_consolidate error: %s", instance.name, exc)
#525
#526	def notify_invalidate(self, memory_id: str) -> None:
#527	"""Notify all loaded plugins of an invalidate event."""
#528	for instance in self._instances.values():
#529	if instance.enabled:
#530	try:
#531	instance.on_invalidate(memory_id)
#532	except Exception as exc:
#533	logger.error("Plugin %s on_invalidate error: %s", instance.name, exc)
#534
#535	def __enter__(self):
#536	return self
#537
#538	def __exit__(self, exc_type, exc_val, exc_tb):
#539	self.unload_all()
#540	return False
#541
#542
#543	# Global plugin manager instance for convenience
#544	_default_manager: Optional[PluginManager] = None
#545
#546
#547	def get_manager() -> PluginManager:
#548	"""Get or create the global PluginManager instance."""
#549	global _default_manager
#550	if _default_manager is None:
#551	_default_manager = PluginManager()
#552	return _default_manager
#553
#554
#555	def reset_manager() -> None:
#556	"""Reset the global PluginManager (useful in tests)."""
#557	global _default_manager
#558	if _default_manager is not None:
#559	_default_manager.unload_all()
#560	_default_manager = None
#561

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public