fix(LongTermMemory): support memory tools in Agent processors (#2843)

* fix(LongTermMemory): support memory tools in Agent processors

* docs(LongTermMemory): clarify tool and Pipeline lifecycle access

* test(LongTermMemory): isolate persistent Chroma clients across fixtures

plugins/KnowledgeEngine/LongTermMemory/README.md

@@ -8,7 +8,7 @@ Long-term memory plugin for LangBot with a dual-layer design:
 ## What It Does
 
 - Exposes a `remember` tool for episodic memory writes
-- Exposes a `recall_memory` tool for active episodic memory lookup with controlled filters
+- Exposes a `recall_memory` tool to read session/speaker profiles and look up episodic memories with controlled filters
 - Exposes an `update_profile` tool for stable profile updates
 - Exposes a `forget` tool for agent-initiated deletion of specific episodic memories
 - Injects profile memory and current speaker identity through an EventListener
@@ -118,9 +118,12 @@ This plugin intentionally stays close to LangBot's existing extension points ins
 
 - L1 profile is stored in plugin storage as JSON
 - L2 episodic memory is stored in the vector database
-- Memory retrieval is enabled per pipeline by attaching this plugin's KnowledgeEngine
+- Agent processors access memory by authorizing this plugin's tools; attaching the memory KB to a Pipeline is not required.
+- Pipeline event-based automatic injection still requires attaching this plugin's KnowledgeEngine.
 - The plugin currently assumes a single memory KB per plugin instance and isolates memory by metadata
 
+Tool access uses the Host-provided session identity and the existing authorized Host tool API. LocalAgent calls that API directly; external runners can expose it through their LangBot asset gateway. Authorizing memory tools does not enable the Pipeline lifecycle hooks: automatic prompt injection and response-based candidate extraction still use the existing Pipeline event listeners.
+
 The current implementation is built around the existing LangBot and SDK APIs. If LangBot later adds more explicit memory-oriented APIs, session identity APIs, or KB registration APIs, the plugin could be simplified, but the current architecture would still remain valid.
 
 ### Vector Database Backend Compatibility
@@ -287,12 +290,17 @@ That is intentional:
    - `isolation`
    - optional `recency_half_life_days`
    - optional `auto_recall_top_k`
-4. Let the agent use:
+4. Authorize these tools in the Agent processor or Pipeline:
    - `remember` for events, plans, and episodic facts
-   - `recall_memory` for active memory lookup when automatic recall is insufficient
+   - `recall_memory` to read the current session/speaker profiles and search episodic memory
    - `update_profile` for stable preferences and profile data
    - `forget` to delete a specific episodic memory by ID
-5. Run `!memory health` in a pipeline that includes the memory KB. Treat `ERROR` as a setup problem before relying on scoped recall.
+
+   `recall_memory` returns a JSON object containing `session_profile`, `speaker_profile`
+   (formatted profile text), and `episodes`. Tools use the Host-provided bot, chat,
+   and current-speaker identity. They do not require a Pipeline-bound Query.
+
+5. Run the Memory Console health check before relying on scoped recall. For Pipeline automatic injection, also attach the memory KB and run `!memory health` in that Pipeline.
 6. Use `!memory`, `!memory profile`, `!memory search <query>`, `!memory list [page]`, `!memory forget <id>`, `!memory export`, and `!memory consolidate preview` to inspect behavior. `!memory list` and `!memory search` show active episodes by default; add `--include-superseded`, `--include-archived`, or `--status <status>` when explicitly inspecting hidden lifecycle states.
 
 ## Episode Lifecycle

plugins/KnowledgeEngine/LongTermMemory/components/knowledge_engine/memory_engine.py

@@ -468,6 +468,7 @@ class LongTermMemoryEngine(KnowledgeEngine):
             batch_texts.append(content)
             batch_ids.append(mid)
             batch_metas.append({
+                "episode_id": mid,
                 "content": content,
                 "tags": ",".join(tags) if isinstance(tags, list) else str(tags),
                 "importance": str(importance),

plugins/KnowledgeEngine/LongTermMemory/components/tools/forget.py

@@ -5,7 +5,6 @@ from typing import Any
 
 from langbot_plugin.api.definition.components.tool.tool import Tool
 from langbot_plugin.api.entities.builtin.provider import session as provider_session
-from langbot_plugin.api.proxies.query_based_api import QueryBasedAPIProxy
 
 logger = logging.getLogger(__name__)
 
@@ -18,12 +17,7 @@ class Forget(Tool):
         query_id: int,
     ) -> str:
         store = self.plugin.memory_store
-        api = QueryBasedAPIProxy(
-            query_id=query_id,
-            plugin_runtime_handler=self.plugin.plugin_runtime_handler,
-        )
-        bot_uuid = await api.get_bot_uuid()
-        query_vars = await api.get_query_vars()
+        bot_uuid = session.bot_uuid or ""
         session_key, user_key, kb_id, _, config = await store.resolve_user_context(
             session, bot_uuid
         )
@@ -48,6 +42,8 @@ class Forget(Tool):
             episode_id=episode_id,
             user_key=user_key,
         )
+        if not deleted:
+            return "No matching memory found in the current scope."
         await store.append_audit_entry(
             scope_key=session_key,
             user_key=user_key,
@@ -55,8 +51,7 @@ class Forget(Tool):
             target_type="episode",
             target_id=str(episode_id),
             summary=f"Deleted episode {episode_id}",
-            sender_id=str(query_vars.get("sender_id", "") or ""),
-            sender_name=str(query_vars.get("sender_name", "") or ""),
+            sender_id=str(session.sender_id or ""),
             query_id=query_id,
             metadata={"kb_id": kb_id, "deleted": deleted},
         )

plugins/KnowledgeEngine/LongTermMemory/components/tools/recall_memory.py

@@ -6,7 +6,6 @@ from typing import Any
 
 from langbot_plugin.api.definition.components.tool.tool import Tool
 from langbot_plugin.api.entities.builtin.provider import session as provider_session
-from langbot_plugin.api.proxies.query_based_api import QueryBasedAPIProxy
 
 logger = logging.getLogger(__name__)
 
@@ -26,27 +25,19 @@ class RecallMemory(Tool):
         query_id: int,
     ) -> str:
         store = self.plugin.memory_store
-        api = QueryBasedAPIProxy(
-            query_id=query_id,
-            plugin_runtime_handler=self.plugin.plugin_runtime_handler,
-        )
         logger.info(
             "[LongTermMemory] recall_memory called: query_id=%s params_keys=%s",
             query_id,
             sorted(params.keys()),
         )
-        bot_uuid = await api.get_bot_uuid()
-        _, user_key, kb_id, _isolation, config = await store.resolve_user_context(
+        bot_uuid = session.bot_uuid or ""
+        session_key, user_key, kb_id, _isolation, config = await store.resolve_user_context(
             session, bot_uuid
         )
 
         if not kb_id:
             return "Error: no memory knowledge base configured. Create one first."
 
-        pipeline_kbs = await api.list_pipeline_knowledge_bases()
-        if not any(kb.get("uuid") == kb_id for kb in pipeline_kbs):
-            return "Error: memory knowledge base is not configured for the current pipeline."
-
         embedding_model_uuid = config.get("embedding_model_uuid", "")
         if not embedding_model_uuid:
             return "Error: no embedding model configured in knowledge base."
@@ -140,7 +131,9 @@ class RecallMemory(Tool):
         except ValueError as exc:
             return f"Error: {exc}"
 
-        if not episodes:
+        session_profile = await store.load_session_profile(session_key)
+        speaker_profile = await store.load_speaker_profile(session_key, str(session.sender_id or ""))
+        if not episodes and not any(store.has_profile_data(profile) for profile in (session_profile, speaker_profile)):
             logger.info(
                 "[LongTermMemory] recall_memory no results: query_id=%s kb_id=%s",
                 query_id,
@@ -154,4 +147,8 @@ class RecallMemory(Tool):
             kb_id,
             len(episodes),
         )
-        return json.dumps(episodes, ensure_ascii=False)
+        return json.dumps({
+            "session_profile": store.format_profile_text(session_profile),
+            "speaker_profile": store.format_profile_text(speaker_profile),
+            "episodes": episodes,
+        }, ensure_ascii=False)

plugins/KnowledgeEngine/LongTermMemory/components/tools/recall_memory.yaml

@@ -6,8 +6,8 @@ metadata:
     en_US: RecallMemory
     zh_Hans: 主动回忆
   description:
-    en_US: Search long-term episodic memories with controlled filters such as speaker and time range.
-    zh_Hans: 使用受控过滤条件（如说话人、时间范围）主动搜索长期情景记忆。
+    en_US: Read current session and speaker profiles, and search episodic memories with controlled filters.
+    zh_Hans: 读取当前会话和发言人的档案，并使用受控过滤条件搜索情景记忆。
 
 spec:
   parameters:
@@ -49,7 +49,9 @@ spec:
       - query
 
   llm_prompt: |
-    Search long-term episodic memories when the default automatic memory recall is not sufficient.
+    Read long-term memory for the current conversation and speaker.
+    Returns session_profile and speaker_profile text together with matching episodes.
+    Independent Agent processors can use this tool without a Pipeline or automatic memory injection.
 
     Use this tool when you need to actively inspect memory with a specific retrieval intent, such as:
     - recalling what a specific person said

plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.py

@@ -5,7 +5,6 @@ from typing import Any
 
 from langbot_plugin.api.definition.components.tool.tool import Tool
 from langbot_plugin.api.entities.builtin.provider import session as provider_session
-from langbot_plugin.api.proxies.query_based_api import QueryBasedAPIProxy
 
 logger = logging.getLogger(__name__)
 
@@ -25,17 +24,12 @@ class Remember(Tool):
         query_id: int,
     ) -> str:
         store = self.plugin.memory_store
-        api = QueryBasedAPIProxy(
-            query_id=query_id,
-            plugin_runtime_handler=self.plugin.plugin_runtime_handler,
-        )
         logger.info(
             "[LongTermMemory] remember called: query_id=%s params_keys=%s",
             query_id,
             sorted(params.keys()),
         )
-        bot_uuid = await api.get_bot_uuid()
-        query_vars = await api.get_query_vars()
+        bot_uuid = session.bot_uuid or ""
         session_key, user_key, kb_id, _, config = await store.resolve_user_context(
             session, bot_uuid
         )
@@ -43,10 +37,6 @@ class Remember(Tool):
         if not kb_id:
             return "Error: no memory knowledge base configured. Create one first."
 
-        pipeline_kbs = await api.list_pipeline_knowledge_bases()
-        if not any(kb.get("uuid") == kb_id for kb in pipeline_kbs):
-            return "Error: memory knowledge base is not configured for the current pipeline."
-
         embedding_model_uuid = config.get("embedding_model_uuid", "")
         if not embedding_model_uuid:
             return "Error: no embedding model configured in knowledge base."
@@ -57,8 +47,7 @@ class Remember(Tool):
 
         tags = params.get("tags", [])
         importance = params.get("importance", 2)
-        sender_id = str(query_vars.get("sender_id", "") or "")
-        sender_name = str(query_vars.get("sender_name", "") or "")
+        sender_id = str(session.sender_id or "")
         logger.info(
             "[LongTermMemory] remember storing episode: query_id=%s kb_id=%s user_key=%s sender_id=%s importance=%s tags=%s content_len=%s",
             query_id,
@@ -78,7 +67,6 @@ class Remember(Tool):
             tags=tags,
             importance=importance,
             sender_id=sender_id,
-            sender_name=sender_name,
             bot_uuid=bot_uuid,
         )
 
@@ -97,7 +85,6 @@ class Remember(Tool):
             target_id=episode["id"],
             summary=self._preview_text(str(content)),
             sender_id=sender_id,
-            sender_name=sender_name,
             query_id=query_id,
             metadata={
                 "kb_id": kb_id,

plugins/KnowledgeEngine/LongTermMemory/components/tools/update_profile.py

@@ -5,7 +5,6 @@ from typing import Any
 
 from langbot_plugin.api.definition.components.tool.tool import Tool
 from langbot_plugin.api.entities.builtin.provider import session as provider_session
-from langbot_plugin.api.proxies.query_based_api import QueryBasedAPIProxy
 
 logger = logging.getLogger(__name__)
 
@@ -94,14 +93,8 @@ class UpdateProfile(Tool):
         if not isinstance(previous_value, str):
             return "Error: previous_value must be a string."
 
-        api = QueryBasedAPIProxy(
-            query_id=query_id,
-            plugin_runtime_handler=self.plugin.plugin_runtime_handler,
-        )
-        bot_uuid = await api.get_bot_uuid()
-        query_vars = await api.get_query_vars()
-        sender_id = str(query_vars.get("sender_id", "") or "")
-        sender_name = str(query_vars.get("sender_name", "") or "")
+        bot_uuid = session.bot_uuid or ""
+        sender_id = str(session.sender_id or "")
 
         session_key, user_key, kb_id, _isolation, _ = await store.resolve_user_context(
             session, bot_uuid
@@ -109,10 +102,6 @@ class UpdateProfile(Tool):
         if not kb_id:
             return "Error: no memory knowledge base configured. Create one first."
 
-        pipeline_kbs = await api.list_pipeline_knowledge_bases()
-        if not any(kb.get("uuid") == kb_id for kb in pipeline_kbs):
-            return "Error: memory knowledge base is not configured for the current pipeline."
-
         target_scope = self._infer_scope(field, scope)
         logger.info(
             "[LongTermMemory] update_profile resolved scope: query_id=%s target_scope=%s session_key=%s sender_id=%s value_len=%s",
@@ -165,7 +154,6 @@ class UpdateProfile(Tool):
                 f"{f' ({fact_key})' if fact_key else ''}: {self._preview_text(str(value))}"
             ),
             sender_id=sender_id,
-            sender_name=sender_name,
             query_id=query_id,
             metadata={
                 "field": field,

plugins/KnowledgeEngine/LongTermMemory/store/memory_store.py

@@ -1348,6 +1348,7 @@ class MemoryStore:
         )
 
         metadata = {
+            "episode_id": episode_id,
             "content": content,
             "tags": ",".join(tags),
             "importance": str(importance),
@@ -1752,6 +1753,7 @@ class MemoryStore:
             episode_id = uuid.uuid4().hex[:12]
 
             metadata = {
+                "episode_id": episode_id,
                 "content": content,
                 "tags": ",".join(tags),
                 "importance": str(importance),
@@ -2011,21 +2013,36 @@ class MemoryStore:
         user_key: str,
     ) -> int:
         """Delete a single episode by its ID, scoped to user_key for safety."""
+        episode = await self.get_episode_by_id(collection_id, episode_id, user_key)
+        if episode is None:
+            return 0
+        if episode["metadata"].get("episode_id") != episode_id:
+            # Legacy records need a filterable ID before exact, scoped deletion.
+            kb = await self.get_kb_config()
+            embedding_model_uuid = kb[1].get("embedding_model_uuid", "") if kb else ""
+            if not embedding_model_uuid:
+                raise ValueError("No embedding model configured for legacy memory deletion")
+            await self.update_episode_status(
+                collection_id, embedding_model_uuid, episode_id, user_key, episode["status"]
+            )
         filters: dict[str, Any] = {
             "$and": [
                 {"user_key": user_key},
+                {"episode_id": episode_id},
             ]
         }
-        return await self._writes.mutate(
+        await self._writes.mutate(
             self.plugin,
             collection_id,
             lambda: self.plugin.vector_delete(
                 collection_id=collection_id,
-                file_ids=[episode_id],
                 filters=filters,
             ),
             "Host vector_delete outcome unknown",
         )
+        if await self.get_episode_by_id(collection_id, episode_id, user_key) is not None:
+            raise RuntimeError("Memory deletion did not remove the episode")
+        return 1
 
     def _episode_from_vector_item(self, item: dict[str, Any]) -> dict[str, Any]:
         meta = item.get("metadata", {})
@@ -2090,6 +2107,7 @@ class MemoryStore:
             return None
 
         meta = dict(episode.get("metadata", {}))
+        meta["episode_id"] = episode_id
         content = episode.get("content", "")
         meta["content"] = content
         meta["status"] = normalized_status

plugins/KnowledgeEngine/LongTermMemory/tests/store/test_memory_store_episode_lifecycle.py

@@ -89,6 +89,7 @@ def _record(
     content: str | None = None,
 ) -> dict:
     metadata = {
+        "episode_id": item_id,
         "content": content or f"memory {item_id}",
         "tags": "",
         "importance": "2",

plugins/KnowledgeEngine/LongTermMemory/tests/tools/test_memory_audit_tools.py

@@ -5,21 +5,13 @@ import json
 import pytest
 
 from components.tools.remember import Remember
+from components.tools.recall_memory import RecallMemory
+from components.tools.update_profile import UpdateProfile
+from components.tools.forget import Forget
 from langbot_plugin.api.entities.builtin.provider.session import Session, LauncherTypes
 from store.memory_store import MemoryStore
 
 
-class FakeAPI:
-    async def get_bot_uuid(self) -> str:
-        return "bot-1"
-
-    async def get_query_vars(self) -> dict:
-        return {"sender_id": "u-1", "sender_name": "Alice"}
-
-    async def list_pipeline_knowledge_bases(self) -> list[dict]:
-        return [{"uuid": "kb-1"}]
-
-
 class FakePlugin:
     def __init__(self):
         self.storage: dict[str, bytes] = {}
@@ -43,16 +35,43 @@ class FakePlugin:
 
     async def vector_upsert(self, collection_id, vectors, ids, metadata=None, documents=None):
         for index, item_id in enumerate(ids):
-            self.records[item_id] = {"metadata": metadata[index], "document": documents[index]}
+            self.records[item_id] = {
+                "id": item_id,
+                "metadata": metadata[index],
+                "document": documents[index],
+                "score": 1.0,
+            }
+
+    async def vector_search(self, collection_id, query_vector, top_k=5, filters=None, **kwargs):
+        return [item for item in self.records.values() if self._matches(item, filters)][:top_k]
+
+    async def vector_list(self, collection_id, filters=None, limit=20, offset=0):
+        items = [item for item in self.records.values() if self._matches(item, filters)]
+        return {"items": items[offset : offset + limit], "total": len(items)}
+
+    async def vector_delete(self, collection_id, file_ids=None, filters=None):
+        ids = [
+            key
+            for key in (file_ids or self.records)
+            if key in self.records and self._matches(self.records[key], filters)
+        ]
+        for key in ids:
+            del self.records[key]
+        return len(ids)
+
+    @classmethod
+    def _matches(cls, item, filters):
+        if filters and "$and" in filters:
+            return all(cls._matches(item, condition) for condition in filters["$and"])
+        return all(item["metadata"].get(key) == value for key, value in (filters or {}).items())
 
 
 def _session() -> Session:
-    return Session(launcher_type=LauncherTypes.GROUP, launcher_id="1")
+    return Session(launcher_type=LauncherTypes.GROUP, launcher_id="1", bot_uuid="bot-1", sender_id="u-1")
 
 
 @pytest.mark.asyncio
-async def test_remember_writes_audit_entry(monkeypatch):
-    monkeypatch.setattr("components.tools.remember.QueryBasedAPIProxy", lambda **_: FakeAPI())
+async def test_remember_writes_audit_entry():
     tool = Remember()
     plugin = FakePlugin()
     plugin.storage["kb_configs"] = json.dumps(
@@ -74,3 +93,96 @@ async def test_remember_writes_audit_entry(monkeypatch):
     assert entries[0]["sender_id"] == "u-1"
     assert entries[0]["query_id"] == 123
     assert "quiet summaries" in entries[0]["summary"]
+
+
+@pytest.mark.asyncio
+async def test_processor_memory_round_trip_and_scope_isolation():
+    plugin = FakePlugin()
+    plugin.storage["kb_configs"] = json.dumps(
+        {"kb-1": {"embedding_model_uuid": "emb-1", "isolation": "session"}}
+    ).encode()
+
+    async def call(tool_class, params, session=None):
+        tool = tool_class()
+        tool.plugin = plugin
+        return await tool.call(params, session or _session(), query_id=123)
+
+    assert (await call(Remember, {"content": "Release on Friday"})).startswith("Remembered:")
+    assert (
+        await call(UpdateProfile, {"field": "preferences", "action": "add", "value": "concise replies"})
+    ).startswith("Speaker profile updated.")
+    recalled = json.loads(await call(RecallMemory, {"query": "release"}))
+    assert recalled["episodes"][0]["content"] == "Release on Friday"
+    assert "concise replies" in recalled["speaker_profile"]
+
+    other_speaker = _session().model_copy(update={"sender_id": "u-2"})
+    other = json.loads(await call(RecallMemory, {"query": "release"}, other_speaker))
+    assert other["speaker_profile"] == ""
+    assert other["episodes"][0]["content"] == "Release on Friday"
+
+    for update in ({"launcher_id": "2"}, {"bot_uuid": "bot-2"}):
+        assert (
+            await call(RecallMemory, {"query": "release"}, _session().model_copy(update=update))
+            == "No relevant memories found."
+        )
+
+    episode_id = recalled["episodes"][0]["id"]
+    await call(Forget, {"episode_id": episode_id})
+    assert json.loads(await call(RecallMemory, {"query": "release"}))["episodes"] == []
+
+
+@pytest.mark.asyncio
+async def test_processor_profile_recall_with_no_episodes():
+    plugin = FakePlugin()
+    plugin.storage["kb_configs"] = json.dumps(
+        {"kb-1": {"embedding_model_uuid": "emb-1", "isolation": "session"}}
+    ).encode()
+    await plugin.memory_store.update_session_profile_field("bot-1:group_1", "notes", "set", "Release coordination")
+    tool = RecallMemory()
+    tool.plugin = plugin
+    recalled = json.loads(await tool.call({"query": "group purpose"}, _session(), query_id=123))
+    assert "Release coordination" in recalled["session_profile"]
+    assert recalled["episodes"] == []
+
+
+@pytest.mark.asyncio
+async def test_forget_legacy_episode_preserves_other_scopes():
+    plugin = FakePlugin()
+    plugin.storage["kb_configs"] = json.dumps(
+        {"kb-1": {"embedding_model_uuid": "emb-1", "isolation": "session"}}
+    ).encode()
+
+    # Model the legacy Host contract: file_ids address document metadata, not vector IDs.
+    async def delete(collection_id, file_ids=None, filters=None):
+        assert file_ids is None
+        ids = [key for key, item in plugin.records.items() if plugin._matches(item, filters)]
+        for key in ids:
+            del plugin.records[key]
+        return 0  # Chroma acknowledges deletion without returning a count.
+
+    plugin.vector_delete = delete
+    plugin.records = {
+        "legacy": {"id": "legacy", "metadata": {"user_key": "bot-1:group_1", "content": "old fact"}},
+        "other": {"id": "other", "metadata": {"user_key": "bot-2:group_1", "content": "other fact"}},
+    }
+    tool = Forget()
+    tool.plugin = plugin
+    assert await tool.call({"episode_id": "other"}, _session(), 123) == "No matching memory found in the current scope."
+    assert await tool.call({"episode_id": "legacy"}, _session(), 123) == "Deleted episode legacy."
+    assert set(plugin.records) == {"other"}
+
+
+@pytest.mark.asyncio
+async def test_forget_does_not_report_success_when_backend_keeps_record():
+    plugin = FakePlugin()
+    plugin.records["kept"] = {
+        "id": "kept",
+        "metadata": {"episode_id": "kept", "user_key": "bot-1:group_1", "content": "still here"},
+    }
+
+    async def no_delete(**kwargs):
+        return 0
+
+    plugin.vector_delete = no_delete
+    with pytest.raises(RuntimeError, match="did not remove"):
+        await plugin.memory_store.delete_episode_by_id("kb-1", "kept", "bot-1:group_1")