import threading from typing import Any, Optional class WorkingMemory: """ Fast, in-memory volatile storage for EIDOS. Implemented as a thread-safe singleton. """ _instance = None _lock = threading.Lock() def __new__(cls): with cls._lock: if cls._instance is None: cls._instance = super(WorkingMemory, cls).__new__(cls) cls._instance._data = {} return cls._instance def set(self, key: str, value: Any): with self._lock: self._data[key] = value def get(self, key: str, default: Any = None) -> Any: with self._lock: return self._data.get(key, default) def delete(self, key: str): with self._lock: if key in self._data: del self._data[key] def clear(self): with self._lock: self._data.clear() @property def snapshot(self) -> dict: with self._lock: return self._data.copy() def set_topic(self, topic: str): self.set("current_topic", topic) def get_topic(self) -> str: return self.get("current_topic", "General") # Global accessor working_memory = WorkingMemory()