diff --git a/plugins/KnowledgeEngine/LongTermMemory/README.md b/plugins/KnowledgeEngine/LongTermMemory/README.md index 7ca7e54a1..90c322202 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/README.md +++ b/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 `, `!memory list [page]`, `!memory forget `, `!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 ` when explicitly inspecting hidden lifecycle states. ## Episode Lifecycle diff --git a/plugins/KnowledgeEngine/LongTermMemory/components/knowledge_engine/memory_engine.py b/plugins/KnowledgeEngine/LongTermMemory/components/knowledge_engine/memory_engine.py index 6d930edd3..0cd63e328 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/components/knowledge_engine/memory_engine.py +++ b/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), diff --git a/plugins/KnowledgeEngine/LongTermMemory/components/tools/forget.py b/plugins/KnowledgeEngine/LongTermMemory/components/tools/forget.py index 76d76f7a1..5ed646d3d 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/components/tools/forget.py +++ b/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}, ) diff --git a/plugins/KnowledgeEngine/LongTermMemory/components/tools/recall_memory.py b/plugins/KnowledgeEngine/LongTermMemory/components/tools/recall_memory.py index 41f6fd6c0..8f7da9d32 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/components/tools/recall_memory.py +++ b/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) diff --git a/plugins/KnowledgeEngine/LongTermMemory/components/tools/recall_memory.yaml b/plugins/KnowledgeEngine/LongTermMemory/components/tools/recall_memory.yaml index 1a25a9e40..adeb40d6a 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/components/tools/recall_memory.yaml +++ b/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 diff --git a/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.py b/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.py index 3f26e6969..f9df5b2f9 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.py +++ b/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, diff --git a/plugins/KnowledgeEngine/LongTermMemory/components/tools/update_profile.py b/plugins/KnowledgeEngine/LongTermMemory/components/tools/update_profile.py index 1ce014551..2f82dd544 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/components/tools/update_profile.py +++ b/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, diff --git a/plugins/KnowledgeEngine/LongTermMemory/store/memory_store.py b/plugins/KnowledgeEngine/LongTermMemory/store/memory_store.py index 39eb9a286..2f5d47d18 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/store/memory_store.py +++ b/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 diff --git a/plugins/KnowledgeEngine/LongTermMemory/tests/store/test_memory_store_episode_lifecycle.py b/plugins/KnowledgeEngine/LongTermMemory/tests/store/test_memory_store_episode_lifecycle.py index 51f647b8a..828e57074 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/tests/store/test_memory_store_episode_lifecycle.py +++ b/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", diff --git a/plugins/KnowledgeEngine/LongTermMemory/tests/tools/test_memory_audit_tools.py b/plugins/KnowledgeEngine/LongTermMemory/tests/tools/test_memory_audit_tools.py index 69ef1ee2a..627174eb6 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/tests/tools/test_memory_audit_tools.py +++ b/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") diff --git a/tests/e2e/test_longterm_memory_processor.py b/tests/e2e/test_longterm_memory_processor.py new file mode 100644 index 000000000..89468fa70 --- /dev/null +++ b/tests/e2e/test_longterm_memory_processor.py @@ -0,0 +1,237 @@ +"""Real memory plugin/Host/storage with scripted LocalAgent model responses.""" + +from __future__ import annotations + +import asyncio +import json +import shutil +import threading +import uuid +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +import pytest + +from tests.e2e import test_local_runner_fake_provider as local + +pytest_plugins = ['tests.e2e.test_local_runner_fake_provider'] +pytestmark = [pytest.mark.e2e, pytest.mark.local_agent] + +MEMORY_TOOLS = ['remember', 'recall_memory', 'update_profile', 'forget'] + + +@pytest.fixture +def memory_embedding_url(): + """Serve deterministic embeddings over the actual provider HTTP protocol.""" + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + request = json.loads(self.rfile.read(int(self.headers['Content-Length']))) + payload = json.dumps( + { + 'object': 'list', + 'model': request['model'], + 'data': [ + {'object': 'embedding', 'index': index, 'embedding': [0.1, 0.2, 0.3]} + for index, _ in enumerate(request['input']) + ], + 'usage': {'prompt_tokens': 1, 'total_tokens': 1}, + } + ).encode() + self.send_response(200) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f'http://127.0.0.1:{server.server_port}/v1' + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def test_local_agent_memory_tools_without_pipeline( + local_agent_e2e_tmpdir, + local_agent_e2e_config_path, + local_agent_runtime_process, + memory_embedding_url, + monkeypatch, +): + """Read/write across runs and Host restart without Pipeline or KB grants.""" + from langbot_plugin.api.entities.builtin.provider import message as provider_message + from langbot_plugin.runtime.plugin.mgr import PluginInstallSource + import yaml + from langbot.pkg.vector.vdbs import chroma + + # Each app fixture needs a distinct Chroma cache identity across cwd changes. + persistent_client = chroma.PersistentClient + monkeypatch.setattr(chroma, 'PersistentClient', lambda *, path: persistent_client(path=str(Path(path).resolve()))) + + config = yaml.safe_load(local_agent_e2e_config_path.read_text()) + config['api']['host'] = '127.0.0.1' + local_agent_e2e_config_path.write_text(yaml.safe_dump(config)) + + plugin_root = Path(__file__).parents[2] / 'plugins' / 'KnowledgeEngine' / 'LongTermMemory' + archive = Path(shutil.make_archive(str(local_agent_e2e_tmpdir / 'long-term-memory'), 'zip', plugin_root)) + counter = 0 + + async def call(ap, model, name, params, *, group='group-a', speaker='alice', bot='bot-a', tools=None): + nonlocal counter + counter += 1 + model.queue_llm_responses( + provider_message.Message( + role='assistant', + content='', + tool_calls=[ + provider_message.ToolCall( + id=f'memory-call-{counter}', + type='function', + function=provider_message.FunctionCall(name=name, arguments=json.dumps(params)), + ) + ], + ), + 'Memory tool turn complete.', + ) + event = local._event( + event_id=f'memory-event-{uuid.uuid4()}', + conversation_id=f'memory-turn-{uuid.uuid4()}', + text=f'Call {name} for this memory test.', + ) + event.bot_id = bot + event.actor.actor_id = speaker + event.delivery.reply_target = {'target_type': 'group', 'target_id': group} + binding = local._binding( + binding_id='memory-processor', + allowed_tool_names=MEMORY_TOOLS if tools is None else tools, + allowed_kb_uuids=[], + runner_config={'knowledge-bases': [], 'box-enabled': False, 'tool-execution-mode': 'serial'}, + ) + outputs = await local._run_runner(ap, event, binding) + assert local._content_text(outputs[-1].content) == 'Memory tool turn complete.' + assert not ap.query_pool.cached_queries + results = [m for m in model._invoke_payloads[-1]['messages'] if m.role == 'tool'] + assert results, 'LocalAgent must receive an actual tool result before its final model call' + return local._message_text(results[-1]) + + async def write_and_read(ap): + model = await local._inject_fake_llm_model(ap) + await ap.plugin_connector.install_plugin(PluginInstallSource.LOCAL, {'plugin_file': archive.read_bytes()}) + for _ in range(60): + if 'remember' in {tool.metadata.name for tool in await ap.plugin_connector.list_tools()}: + break + await asyncio.sleep(1) + else: + pytest.fail('LongTermMemory tools were not discovered') + + context = await ap.plugin_connector._current_execution_context() + provider = await ap.provider_service.create_provider( + context, + { + 'name': 'Memory test local embedding', + 'requester': 'openai-chat-completions', + 'base_url': memory_embedding_url, + 'api_keys': ['local-test-key'], + }, + ) + embedding = await ap.embedding_models_service.create_embedding_model( + context, + { + 'name': 'memory-test-embedding', + 'provider_uuid': provider, + 'extra_args': {}, + }, + ) + engines = await ap.plugin_connector.list_knowledge_engines() + engine = next(item['plugin_id'] for item in engines if 'LongTermMemory' in item['plugin_id']) + kb = await ap.rag_mgr.create_knowledge_base( + context, + name='Processor memory', + knowledge_engine_plugin_id=engine, + creation_settings={'embedding_model_uuid': embedding, 'isolation': 'session'}, + ) + + stored = await call(ap, model, 'remember', {'content': 'The release rendezvous is at Cedar Pier on Friday.'}) + assert stored.startswith('Remembered:'), stored + updated = await call( + ap, + model, + 'update_profile', + { + 'field': 'preferences', + 'action': 'add', + 'value': 'brief technical answers', + }, + ) + assert updated.startswith('Speaker profile updated.'), updated + updated = await call( + ap, + model, + 'update_profile', + { + 'scope': 'session', + 'field': 'notes', + 'action': 'set', + 'value': 'Release coordination group', + }, + ) + assert updated.startswith('Session profile updated.'), updated + result = json.loads(await call(ap, model, 'recall_memory', {'query': 'release rendezvous'})) + assert result['episodes'][0]['content'] == 'The release rendezvous is at Cedar Pier on Friday.' + assert result['episodes'][0]['sender_id'] == 'alice' + assert 'brief technical answers' in result['speaker_profile'] + assert 'Release coordination group' in result['session_profile'] + # Retain an old-format record to exercise deletion after an upgrade. + items, _ = await ap.rag_runtime_service.vector_list(context, kb.uuid, limit=1) + metadata = dict(items[0]['metadata']) + metadata.pop('episode_id') + await ap.rag_runtime_service.vector_upsert( + context, + kb.uuid, + [[0.1, 0.2, 0.3]], + [items[0]['id']], + metadata=[metadata], + documents=[result['episodes'][0]['content']], + ) + return result['episodes'][0]['id'] + + episode_id = local._run_local_agent_probe(local_agent_e2e_tmpdir, write_and_read) + + async def read_after_restart(ap): + model = await local._inject_fake_llm_model(ap) + result = json.loads(await call(ap, model, 'recall_memory', {'query': 'release rendezvous'})) + assert result['episodes'][0]['id'] == episode_id + assert 'brief technical answers' in result['speaker_profile'] + + bob = json.loads(await call(ap, model, 'recall_memory', {'query': 'release rendezvous'}, speaker='bob')) + assert bob['speaker_profile'] == '' + assert bob['episodes'][0]['id'] == episode_id + assert 'Release coordination group' in bob['session_profile'] + for kwargs in ({'group': 'group-b'}, {'bot': 'bot-b'}): + assert ( + await call(ap, model, 'recall_memory', {'query': 'release rendezvous'}, **kwargs) + == 'No relevant memories found.' + ) + + denied = await call(ap, model, 'remember', {'content': 'Unauthorized memory'}, tools=['recall_memory']) + assert 'not authorized' in denied + unchanged = json.loads(await call(ap, model, 'recall_memory', {'query': 'release rendezvous'})) + assert len(unchanged['episodes']) == 1 + + await call(ap, model, 'forget', {'episode_id': episode_id}, group='group-b') + still_present = json.loads(await call(ap, model, 'recall_memory', {'query': 'release rendezvous'})) + assert still_present['episodes'][0]['id'] == episode_id + deleted = await call(ap, model, 'forget', {'episode_id': episode_id}) + assert deleted == f'Deleted episode {episode_id}.' + profiles_only = json.loads(await call(ap, model, 'recall_memory', {'query': 'release rendezvous'})) + assert profiles_only['episodes'] == [] + assert 'brief technical answers' in profiles_only['speaker_profile'] + + local._run_local_agent_probe(local_agent_e2e_tmpdir, read_after_restart)