feat: clear memory (#239)

Author: evan-ak

src/memu/app/crud.py

@@ -76,6 +76,27 @@ async def list_memory_categories(
             raise RuntimeError(msg)
         return response
 
+    async def clear_memory(
+        self,
+        where: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        ctx = self._get_context()
+        store = self._get_database()
+        where_filters = self._normalize_where(where)
+
+        state: WorkflowState = {
+            "ctx": ctx,
+            "store": store,
+            "where": where_filters,
+        }
+
+        result = await self._run_workflow("crud_clear_memory", state)
+        response = cast(dict[str, Any] | None, result.get("response"))
+        if response is None:
+            msg = "Clear memory workflow failed to produce a response"
+            raise RuntimeError(msg)
+        return response
+
     def _build_list_memory_items_workflow(self) -> list[WorkflowStep]:
         steps = [
             WorkflowStep(
@@ -98,7 +119,7 @@ def _build_list_memory_items_workflow(self) -> list[WorkflowStep]:
         return steps
 
     @staticmethod
-    def _list_list_memory_items_initial_keys() -> set[str]:
+    def _list_list_memories_initial_keys() -> set[str]:
         return {
             "ctx",
             "store",
@@ -126,6 +147,51 @@ def _build_list_memory_categories_workflow(self) -> list[WorkflowStep]:
         ]
         return steps
 
+    def _build_clear_memory_workflow(self) -> list[WorkflowStep]:
+        steps = [
+            WorkflowStep(
+                step_id="clear_memory_categories",
+                role="delete_memories",
+                handler=self._crud_clear_memory_categories,
+                requires={"ctx", "store", "where"},
+                produces={"deleted_categories"},
+                capabilities={"db"},
+            ),
+            WorkflowStep(
+                step_id="clear_memory_items",
+                role="delete_memories",
+                handler=self._crud_clear_memory_items,
+                requires={"ctx", "store", "where"},
+                produces={"deleted_items"},
+                capabilities={"db"},
+            ),
+            WorkflowStep(
+                step_id="clear_memory_resources",
+                role="delete_memories",
+                handler=self._crud_clear_memory_resources,
+                requires={"ctx", "store", "where"},
+                produces={"deleted_resources"},
+                capabilities={"db"},
+            ),
+            WorkflowStep(
+                step_id="build_response",
+                role="emit",
+                handler=self._crud_build_clear_memory_response,
+                requires={"ctx", "store", "deleted_categories", "deleted_items", "deleted_resources"},
+                produces={"response"},
+                capabilities=set(),
+            ),
+        ]
+        return steps
+
+    @staticmethod
+    def _list_clear_memories_initial_keys() -> set[str]:
+        return {
+            "ctx",
+            "store",
+            "where",
+        }
+
     def _normalize_where(self, where: Mapping[str, Any] | None) -> dict[str, Any]:
         """Validate and clean the `where` scope filters against the configured user model."""
         if not where:
@@ -177,6 +243,39 @@ def _crud_build_list_categories_response(self, state: WorkflowState, step_contex
         state["response"] = response
         return state
 
+    def _crud_clear_memory_categories(self, state: WorkflowState, step_context: Any) -> WorkflowState:
+        where_filters = state.get("where") or {}
+        store = state["store"]
+        deleted = store.memory_category_repo.clear_categories(where_filters)
+        state["deleted_categories"] = deleted
+        return state
+
+    def _crud_clear_memory_items(self, state: WorkflowState, step_context: Any) -> WorkflowState:
+        where_filters = state.get("where") or {}
+        store = state["store"]
+        deleted = store.memory_item_repo.clear_items(where_filters)
+        state["deleted_items"] = deleted
+        return state
+
+    def _crud_clear_memory_resources(self, state: WorkflowState, step_context: Any) -> WorkflowState:
+        where_filters = state.get("where") or {}
+        store = state["store"]
+        deleted = store.resource_repo.clear_resources(where_filters)
+        state["deleted_resources"] = deleted
+        return state
+
+    def _crud_build_clear_memory_response(self, state: WorkflowState, step_context: Any) -> WorkflowState:
+        deleted_categories = state.get("deleted_categories", {})
+        deleted_items = state.get("deleted_items", {})
+        deleted_resources = state.get("deleted_resources", {})
+        response = {
+            "deleted_categories": [self._model_dump_without_embeddings(cat) for cat in deleted_categories.values()],
+            "deleted_items": [self._model_dump_without_embeddings(item) for item in deleted_items.values()],
+            "deleted_resources": [self._model_dump_without_embeddings(res) for res in deleted_resources.values()],
+        }
+        state["response"] = response
+        return state
+
     async def create_memory_item(
         self,
         *,

src/memu/app/service.py

@@ -277,7 +277,7 @@ def _register_pipelines(self) -> None:
         patch_delete_initial_keys = CRUDMixin._list_delete_memory_item_initial_keys()
         self._pipelines.register("patch_delete", patch_delete_workflow, initial_state_keys=patch_delete_initial_keys)
         crud_list_items_workflow = self._build_list_memory_items_workflow()
-        crud_list_memories_initial_keys = CRUDMixin._list_list_memory_items_initial_keys()
+        crud_list_memories_initial_keys = CRUDMixin._list_list_memories_initial_keys()
         self._pipelines.register(
             "crud_list_memory_items", crud_list_items_workflow, initial_state_keys=crud_list_memories_initial_keys
         )
@@ -287,6 +287,11 @@ def _register_pipelines(self) -> None:
             crud_list_categories_workflow,
             initial_state_keys=crud_list_memories_initial_keys,
         )
+        crud_clear_memory_workflow = self._build_clear_memory_workflow()
+        crud_clear_memory_initial_keys = CRUDMixin._list_clear_memories_initial_keys()
+        self._pipelines.register(
+            "crud_clear_memory", crud_clear_memory_workflow, initial_state_keys=crud_clear_memory_initial_keys
+        )
 
     async def _run_workflow(self, workflow_name: str, initial_state: WorkflowState) -> WorkflowState:
         """Execute a workflow through the configured runner backend."""

src/memu/database/inmemory/repositories/memory_category_repo.py

@@ -23,6 +23,15 @@ def list_categories(self, where: Mapping[str, Any] | None = None) -> dict[str, M
             return dict(self.categories)
         return {cid: cat for cid, cat in self.categories.items() if matches_where(cat, where)}
 
+    def clear_categories(self, where: Mapping[str, Any] | None = None) -> dict[str, MemoryCategory]:
+        if not where:
+            matches = self.categories.copy()
+            self.categories.clear()
+            return matches
+        matches = {cid: cat for cid, cat in self.categories.items() if matches_where(cat, where)}
+        self.categories = {cid: cat for cid, cat in self.categories.items() if cid not in matches}
+        return matches
+
     def get_or_create_category(
         self, *, name: str, description: str, embedding: list[float], user_data: dict[str, Any]
     ) -> MemoryCategory:

src/memu/database/inmemory/repositories/memory_item_repo.py

@@ -22,6 +22,15 @@ def list_items(self, where: Mapping[str, Any] | None = None) -> dict[str, Memory
             return dict(self.items)
         return {mid: item for mid, item in self.items.items() if matches_where(item, where)}
 
+    def clear_items(self, where: Mapping[str, Any] | None = None) -> dict[str, MemoryItem]:
+        if not where:
+            matches = self.items.copy()
+            self.items.clear()
+            return matches
+        matches = {mid: item for mid, item in self.items.items() if matches_where(item, where)}
+        self.items = {mid: item for mid, item in self.items.items() if mid not in matches}
+        return matches
+
     def create_item(
         self,
         *,

src/memu/database/inmemory/repositories/resource_repo.py

@@ -21,6 +21,15 @@ def list_resources(self, where: Mapping[str, Any] | None = None) -> dict[str, Re
             return dict(self.resources)
         return {rid: res for rid, res in self.resources.items() if matches_where(res, where)}
 
+    def clear_resources(self, where: Mapping[str, Any] | None = None) -> dict[str, Resource]:
+        if not where:
+            matches = self.resources.copy()
+            self.resources.clear()
+            return matches
+        matches = {rid: res for rid, res in self.resources.items() if matches_where(res, where)}
+        self.resources = {rid: res for rid, res in self.resources.items() if rid not in matches}
+        return matches
+
     def create_resource(
         self,
         *,

src/memu/database/postgres/repositories/memory_category_repo.py

@@ -37,6 +37,31 @@ def list_categories(self, where: Mapping[str, Any] | None = None) -> dict[str, M
                 result[cat.id] = cat
         return result
 
+    def clear_categories(self, where: Mapping[str, Any] | None = None) -> dict[str, MemoryCategory]:
+        from sqlmodel import delete, select
+
+        filters = self._build_filters(self._sqla_models.MemoryCategory, where)
+        with self._sessions.session() as session:
+            # First get the objects to delete
+            rows = session.scalars(select(self._sqla_models.MemoryCategory).where(*filters)).all()
+            deleted: dict[str, MemoryCategory] = {}
+            for row in rows:
+                row.embedding = self._normalize_embedding(row.embedding)
+                deleted[row.id] = row
+
+            if not deleted:
+                return {}
+
+            # Delete from database
+            session.exec(delete(self._sqla_models.MemoryCategory).where(*filters))
+            session.commit()
+
+            # Clean up cache
+            for cat_id in deleted:
+                self.categories.pop(cat_id, None)
+
+        return deleted
+
     def get_or_create_category(
         self,
         *,

src/memu/database/postgres/repositories/memory_item_repo.py

@@ -51,6 +51,31 @@ def list_items(self, where: Mapping[str, Any] | None = None) -> dict[str, Memory
                 result[item.id] = item
         return result
 
+    def clear_items(self, where: Mapping[str, Any] | None = None) -> dict[str, MemoryItem]:
+        from sqlmodel import delete, select
+
+        filters = self._build_filters(self._sqla_models.MemoryItem, where)
+        with self._sessions.session() as session:
+            # First get the objects to delete
+            rows = session.scalars(select(self._sqla_models.MemoryItem).where(*filters)).all()
+            deleted: dict[str, MemoryItem] = {}
+            for row in rows:
+                row.embedding = self._normalize_embedding(row.embedding)
+                deleted[row.id] = row
+
+            if not deleted:
+                return {}
+
+            # Delete from database
+            session.exec(delete(self._sqla_models.MemoryItem).where(*filters))
+            session.commit()
+
+            # Clean up cache
+            for item_id in deleted:
+                self.items.pop(item_id, None)
+
+        return deleted
+
     def create_item(
         self,
         *,

src/memu/database/postgres/repositories/resource_repo.py

@@ -37,6 +37,31 @@ def list_resources(self, where: Mapping[str, Any] | None = None) -> dict[str, Re
                 result[res.id] = res
         return result
 
+    def clear_resources(self, where: Mapping[str, Any] | None = None) -> dict[str, Resource]:
+        from sqlmodel import delete, select
+
+        filters = self._build_filters(self._sqla_models.Resource, where)
+        with self._sessions.session() as session:
+            # First get the objects to delete
+            rows = session.scalars(select(self._sqla_models.Resource).where(*filters)).all()
+            deleted: dict[str, Resource] = {}
+            for row in rows:
+                row.embedding = self._normalize_embedding(row.embedding)
+                deleted[row.id] = row
+
+            if not deleted:
+                return {}
+
+            # Delete from database
+            session.exec(delete(self._sqla_models.Resource).where(*filters))
+            session.commit()
+
+            # Clean up cache
+            for res_id in deleted:
+                self.resources.pop(res_id, None)
+
+        return deleted
+
     def create_resource(
         self,
         *,

src/memu/database/repositories/memory_category.py

@@ -14,6 +14,8 @@ class MemoryCategoryRepo(Protocol):
 
     def list_categories(self, where: Mapping[str, Any] | None = None) -> dict[str, MemoryCategory]: ...
 
+    def clear_categories(self, where: Mapping[str, Any] | None = None) -> dict[str, MemoryCategory]: ...
+
     def get_or_create_category(
         self, *, name: str, description: str, embedding: list[float], user_data: dict[str, Any]
     ) -> MemoryCategory: ...

src/memu/database/repositories/memory_item.py

@@ -16,6 +16,8 @@ def get_item(self, item_id: str) -> MemoryItem | None: ...
 
     def list_items(self, where: Mapping[str, Any] | None = None) -> dict[str, MemoryItem]: ...
 
+    def clear_items(self, where: Mapping[str, Any] | None = None) -> dict[str, MemoryItem]: ...
+
     def create_item(
         self,
         *,