Add sleep_all() to consolidate all users with pending conversations

README.md

@@ -114,6 +114,17 @@ Run the full River Algorithm consolidation pipeline:
 
 Returns a summary dict with counts of actions taken.
 
+#### `r.sleep_all()`
+
+Run `sleep()` for every user that has unprocessed conversations since the last call.
+Returns a dict mapping each `user_id` to its `sleep()` summary.
+
+```python
+results = r.sleep_all()
+for user_id, summary in results.items():
+    print(f"{user_id}: {summary}")
+```
+
 #### `r.get_profile(user_id)`
 
 Returns all active profile facts as a list of dicts.
@@ -259,6 +270,16 @@ r = Riverse(
 
 返回操作计数的摘要字典。
 
+#### `r.sleep_all()`
+
+对所有有未处理对话的用户运行 `sleep()`。返回一个将每个 `user_id` 映射到其 `sleep()` 摘要的字典。
+
+```python
+results = r.sleep_all()
+for user_id, summary in results.items():
+    print(f"{user_id}: {summary}")
+```
+
 #### `r.get_profile(user_id)`
 
 返回所有活跃的画像事实列表。
@@ -393,6 +414,16 @@ River Algorithmの完全な統合パイプラインを実行：
 
 アクション数のサマリー辞書を返します。
 
+#### `r.sleep_all()`
+
+未処理の会話があるすべてのユーザーに対して `sleep()` を実行します。各 `user_id` をその `sleep()` サマリーにマッピングする辞書を返します。
+
+```python
+results = r.sleep_all()
+for user_id, summary in results.items():
+    print(f"{user_id}: {summary}")
+```
+
 #### `r.get_profile(user_id)`
 
 すべてのアクティブなプロファイル事実をリストで返します。

riverse/client.py

@@ -66,6 +66,16 @@ def sleep(self, user_id: str = "default") -> dict:
             config=self.config,
         )
 
+    def sleep_all(self) -> dict[str, dict]:
+        """Run sleep() for every user that has unprocessed conversations.
+
+        Returns a dict mapping each user_id to its sleep() summary.
+        """
+        results: dict[str, dict] = {}
+        for user_id in self.storage.get_dirty_user_ids():
+            results[user_id] = self.sleep(user_id)
+        return results
+
     def search(self, query: str, user_id: str = "default", top_k: int = 10) -> list[dict]:
         """Search memory for relevant profile facts, events, and observations."""
         results = []

riverse/storage/base.py

@@ -19,6 +19,11 @@ def get_unprocessed_conversations(self, user_id: str) -> dict[str, list[dict]]:
         """Return {session_id: [messages]} for unprocessed conversations."""
         ...
 
+    @abstractmethod
+    def get_dirty_user_ids(self) -> list[str]:
+        """Return user_ids that have unprocessed conversations."""
+        ...
+
     @abstractmethod
     def mark_processed(self, message_ids: list[int]) -> None:
         ...

riverse/storage/sqlite.py

@@ -76,6 +76,17 @@ def get_unprocessed_conversations(self, user_id: str) -> dict[str, list[dict]]:
         finally:
             conn.close()
 
+    def get_dirty_user_ids(self) -> list[str]:
+        conn = self._conn()
+        try:
+            rows = conn.execute(
+                "SELECT DISTINCT user_id FROM conversations WHERE processed = 0 "
+                "ORDER BY user_id",
+            ).fetchall()
+            return [row["user_id"] for row in rows]
+        finally:
+            conn.close()
+
     def mark_processed(self, message_ids: list[int]) -> None:
         if not message_ids:
             return