diff --git a/plugins/KnowledgeEngine/LongTermMemory/README.md b/plugins/KnowledgeEngine/LongTermMemory/README.md index 90c322202..4fa537c05 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/README.md +++ b/plugins/KnowledgeEngine/LongTermMemory/README.md @@ -24,7 +24,7 @@ Long-term memory plugin for LangBot with a dual-layer design: - Provides `!memory audit [page|export]` to inspect scoped memory change history - Provides `!memory consolidate preview|run` for preview-first scoped L2 cleanup - Provides optional `!memory candidates` and `!memory candidate accept|reject` review workflow for extracted memory candidates -- Automatically supersedes related older episodes when a correction/fact-update/clarification is stored +- Supersedes explicitly identified older episodes when a correction names their IDs; similarity or tags alone never hide a fact ## Overall Design @@ -312,7 +312,7 @@ L2 episodes carry a lifecycle `status` metadata field: - `archived`: retained for history, excluded from normal recall - `deleted`: reserved for tombstones or future audit workflows -Existing records without `status` are treated as `active`, so no migration is required. Automatic recall and `recall_memory` only return active episodes by default. Correction-style writes mark similar older active episodes as `superseded`, which keeps them inspectable through explicit management commands without letting them compete with current memory. +Existing records without `status` are treated as `active`, so no migration is required. Automatic recall and `recall_memory` only return active episodes by default. Correction writes can pass `supersedes` with IDs obtained from recall. Only active records in the current memory scope and from the same speaker can be replaced; invalid targets fail before any write. Tags and vector similarity do not imply replacement. Explicitly replaced records become `superseded`, which keeps them inspectable through explicit management commands without letting them compete with current memory. ## Audit Log @@ -326,9 +326,9 @@ Use `!memory show ` to inspect one episode, including status, sender ## Memory Consolidation -Use `!memory consolidate preview` to inspect cleanup candidates for the current scope. Preview never modifies memory. It reports candidate episode IDs, a deterministic summary episode proposal, possible profile updates, episodes that would be archived, and risk notes. +Use `!memory consolidate preview` to inspect cleanup candidates for the current scope. Preview never modifies memory. It reports candidate episode IDs, older active records to retain, superseded records to archive, and risk notes. -Use `!memory consolidate run` only after setting `consolidation_enabled=true` on the memory knowledge base. Run applies only the current `user_key` scope. It can archive selected active or superseded episodes and write one summary episode tagged `consolidation` and `summary`. Every archive and summary write creates scoped audit entries. Profile update application is disabled by default through `consolidation_apply_profile_updates=false`. +Use `!memory consolidate run` only after setting `consolidation_enabled=true` on the memory knowledge base. Run applies only the current `user_key` scope. It archives only records already marked superseded. Active records remain searchable regardless of age or importance; a count/tag summary cannot preserve their facts. No summary episode is generated. Archive writes create scoped audit entries. Profile update application is disabled by default through `consolidation_apply_profile_updates=false`. Relevant knowledge base settings: @@ -600,3 +600,10 @@ We welcome contributions! Feel free to: - Share your ideas and feedback Star the repo if you find it useful! + +### Memory quality guarantees + +- Retrieve the relevant episode first, then pass its ID in `remember.supersedes` only for an explicit correction of the same fact or event by the same speaker. Omit targets when uncertain. Original content and importance are preserved for historical inspection. +- Consolidation is conservative archival, not semantic synthesis. It neither replaces active facts with statistics nor copies them into derived summaries. Existing archived records and previously generated summaries are not migrated automatically. +- Recall ranks the complete bounded candidate set before taking `top_k`; exact matches beyond the backend's first `top_k` results can still win. Candidates remain bounded by the existing fetch window, so this is not a guarantee of exhaustive retrieval. +- Deterministic regression cases cover correction ownership, unrelated similar facts, retention across consolidation, and ranking at the candidate boundary. They do not measure real-model extraction or answer quality on LongMemEval. diff --git a/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.py b/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.py index f9df5b2f9..4a4ee3c3b 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.py +++ b/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.py @@ -30,9 +30,7 @@ class Remember(Tool): sorted(params.keys()), ) bot_uuid = session.bot_uuid or "" - session_key, user_key, kb_id, _, config = await store.resolve_user_context( - session, bot_uuid - ) + session_key, user_key, kb_id, _, config = await store.resolve_user_context(session, bot_uuid) if not kb_id: return "Error: no memory knowledge base configured. Create one first." @@ -59,16 +57,20 @@ class Remember(Tool): len(str(content)), ) - episode = await store.add_episode( - collection_id=kb_id, - embedding_model_uuid=embedding_model_uuid, - user_key=user_key, - content=content, - tags=tags, - importance=importance, - sender_id=sender_id, - bot_uuid=bot_uuid, - ) + try: + episode = await store.add_episode( + collection_id=kb_id, + embedding_model_uuid=embedding_model_uuid, + user_key=user_key, + content=content, + tags=tags, + importance=importance, + sender_id=sender_id, + bot_uuid=bot_uuid, + supersedes=params.get("supersedes"), + ) + except ValueError as exc: + return f"Error: {exc}" logger.info( "[LongTermMemory] remember stored: query_id=%s episode_id=%s user_key=%s content_len=%s", diff --git a/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.yaml b/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.yaml index 00f7842aa..48d840748 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.yaml +++ b/plugins/KnowledgeEngine/LongTermMemory/components/tools/remember.yaml @@ -21,6 +21,13 @@ spec: items: type: string description: Short category tags for this memory (e.g. "travel", "work", "schedule"). + supersedes: + type: array + maxItems: 20 + items: + type: string + minLength: 1 + description: IDs of active memories explicitly replaced by this correction. Retrieve them first; they must belong to the current speaker and memory scope. Omit when unsure. importance: type: integer description: How important this memory is, from 1 (low) to 5 (critical). Default 2. @@ -58,6 +65,7 @@ spec: If the user explicitly corrects, retracts, or revises an earlier statement: - store the correction as an episodic memory with clear wording - describe the new statement as a correction, not as an unrelated new fact + - retrieve the old memory and pass its ID in supersedes only when the same speaker explicitly replaces the same fact or event; similarity alone is insufficient - when appropriate, add tags such as ["correction"], ["profile-update"], or ["preference-change"] - if the correction changes the user's current stable profile, also update that profile through "update_profile" - when updating a stable preference or trait that may change later, prefer giving update_profile a reusable fact_key such as "coffee_preference" @@ -74,6 +82,7 @@ spec: Parameters: - content (required): What to remember. Be concise and factual. Always attribute to the speaker. - tags (optional): Category tags like ["travel", "work"]. + - supersedes (optional): IDs of explicitly replaced memories. Tags alone never replace older memories. - importance (optional): 1-5 scale, default 2. Use 4-5 for critical facts. execution: diff --git a/plugins/KnowledgeEngine/LongTermMemory/store/memory_store.py b/plugins/KnowledgeEngine/LongTermMemory/store/memory_store.py index 2f5d47d18..043c05d07 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/store/memory_store.py +++ b/plugins/KnowledgeEngine/LongTermMemory/store/memory_store.py @@ -12,8 +12,6 @@ from store.serialization import FENCE_KEY_PREFIX, BindingWrites, serialized_writ logger = logging.getLogger(__name__) -_SUPERSEDE_TAGS = {"correction", "fact-update", "clarification"} - def _installation_scope(_store, _args, _kwargs): """Serialization identity for a write that carries no memory-scope identity. @@ -1225,99 +1223,6 @@ class MemoryStore: # ======================== L2: episodes (ChromaDB vector) ======================== - _SUPERSEDE_IMPORTANCE_FACTOR = 0.1 # multiplied onto old importance - - @serialized_write - async def _auto_supersede( - self, - collection_id: str, - embedding_model_uuid: str, - query_vector: list[float], - user_key: str, - new_episode_id: str, - similarity_threshold: float = 0.85, - max_candidates: int = 5, - ) -> int: - """Find similar older episodes and mark them as superseded. - - Superseding means re-upserting the old vector with: - - importance reduced by _SUPERSEDE_IMPORTANCE_FACTOR - - 'superseded_by' metadata field set to new_episode_id - - 'status' metadata field set to 'superseded' - - Returns the number of superseded episodes. - """ - filters: dict[str, Any] = {"user_key": user_key} - results = await self.plugin.vector_search( - collection_id=collection_id, - query_vector=query_vector, - top_k=max_candidates + 1, # +1 because the new episode itself may appear - filters=filters, - ) - - superseded = 0 - for r in results: - rid = r.get("id", "") - if rid == new_episode_id: - continue - meta = r.get("metadata", {}) - # Already superseded — skip - if ( - meta.get("superseded_by") - or self.episode_status_from_metadata(meta) - != self.EPISODE_STATUS_ACTIVE - ): - continue - - # Check similarity: distance is cosine distance (lower = more similar) - distance = r.get("distance", 1.0) - similarity = 1.0 - distance - if similarity < similarity_threshold: - continue - - # Re-upsert with reduced importance and superseded_by marker - old_importance = int(meta.get("importance", "2")) - new_importance = max( - 1, - int(old_importance * self._SUPERSEDE_IMPORTANCE_FACTOR), - ) - meta["importance"] = str(new_importance) - meta["superseded_by"] = new_episode_id - meta["status"] = self.EPISODE_STATUS_SUPERSEDED - - # We need the original vector; re-embed from content - old_content = meta.get("content", "") - if not old_content: - continue - old_vectors = await self.plugin.invoke_embedding( - embedding_model_uuid, - [old_content], - ) - - await self._writes.mutate( - self.plugin, - collection_id, - lambda: self.plugin.vector_upsert( - collection_id=collection_id, - vectors=old_vectors, - ids=[rid], - metadata=[meta], - documents=[old_content], - ), - "Host vector_upsert outcome unknown", - ) - superseded += 1 - logger.info( - "[LongTermMemory] auto_supersede: episode %s superseded by %s (similarity=%.3f, importance %s->%s)", - rid, - new_episode_id, - similarity, - old_importance, - new_importance, - ) - - return superseded - @serialized_write async def add_episode( self, @@ -1331,8 +1236,26 @@ class MemoryStore: sender_id: str = "", sender_name: str = "", bot_uuid: str = "", + supersedes: list[str] | None = None, ) -> dict[str, Any]: - """Store an episodic memory into vector DB.""" + """Store an episode, optionally replacing explicitly identified facts.""" + if supersedes is None: + supersedes = [] + if ( + not isinstance(supersedes, list) + or len(supersedes) > 20 + or any(not isinstance(item, str) or not item.strip() for item in supersedes) + ): + raise ValueError("supersedes must be a list of at most 20 non-empty episode IDs") + supersedes = list(dict.fromkeys(supersedes)) + for old_id in supersedes: + old = await self.get_episode_by_id(collection_id, old_id, user_key) + if ( + old is None + or old["status"] != self.EPISODE_STATUS_ACTIVE + or str(old.get("sender_id", "")) != str(sender_id) + ): + raise ValueError("Replacement targets must be active memories in the current scope and speaker") episode_id = uuid.uuid4().hex[:12] timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) importance = max(1, min(5, importance)) @@ -1382,25 +1305,15 @@ class MemoryStore: timestamp, ) - # Auto-supersede: when the new episode is a correction / fact-update / - # clarification, search for similar older episodes in the same scope and - # mark them as superseded by lowering their importance. - if _SUPERSEDE_TAGS & set(tags): - try: - await self._auto_supersede( - collection_id=collection_id, - embedding_model_uuid=embedding_model_uuid, - query_vector=vectors[0], - user_key=user_key, - new_episode_id=episode_id, - similarity_threshold=0.85, - ) - except Exception: - logger.warning( - "[LongTermMemory] auto_supersede failed for episode %s, skipping", - episode_id, - exc_info=True, - ) + for old_id in supersedes: + await self.update_episode_status( + collection_id, + embedding_model_uuid, + old_id, + user_key, + self.EPISODE_STATUS_SUPERSEDED, + superseded_by=episode_id, + ) return { "id": episode_id, @@ -1544,27 +1457,24 @@ class MemoryStore: meta = r.get("metadata", {}) if not self.episode_status_included(meta, include_statuses): continue - exact_score = ( - self._metadata_exact_match_score(query, rid, meta) - if exact_match_boost - else 0.0 + exact_score = self._metadata_exact_match_score(query, rid, meta) if exact_match_boost else 0.0 + episodes.append( + { + "id": rid, + "content": meta.get("content", ""), + "tags": meta.get("tags", "").split(",") if meta.get("tags") else [], + "importance": int(meta.get("importance", "2")), + "timestamp": meta.get("timestamp", ""), + "sender_id": meta.get("sender_id", ""), + "sender_name": meta.get("sender_name", ""), + "source": meta.get("source", ""), + "status": self.episode_status_from_metadata(meta), + "superseded_by": meta.get("superseded_by", ""), + "score": r.get("score"), + "_exact_match_score": exact_score, + } ) - episodes.append({ - "id": rid, - "content": meta.get("content", ""), - "tags": meta.get("tags", "").split(",") if meta.get("tags") else [], - "importance": int(meta.get("importance", "2")), - "timestamp": meta.get("timestamp", ""), - "sender_id": meta.get("sender_id", ""), - "sender_name": meta.get("sender_name", ""), - "source": meta.get("source", ""), - "status": self.episode_status_from_metadata(meta), - "superseded_by": meta.get("superseded_by", ""), - "score": r.get("score"), - "_exact_match_score": exact_score, - }) - if len(episodes) >= top_k: - break + seen_ids.add(rid) if exact_match_boost: episodes.sort( @@ -1852,34 +1762,6 @@ class MemoryStore: now_dt = now or datetime.now(timezone.utc) return max(0.0, (now_dt - ts.astimezone(timezone.utc)).total_seconds() / 86400.0) - @staticmethod - def _summarize_consolidation_candidates(candidates: list[dict[str, Any]]) -> str: - if not candidates: - return "" - - tag_counts: dict[str, int] = {} - sender_counts: dict[str, int] = {} - for episode in candidates: - for tag in episode.get("tags", []): - tag_counts[tag] = tag_counts.get(tag, 0) + 1 - sender = episode.get("sender_name") or episode.get("sender_id") or "" - if sender: - sender_counts[sender] = sender_counts.get(sender, 0) + 1 - - top_tags = sorted(tag_counts.items(), key=lambda item: item[1], reverse=True)[:5] - top_senders = sorted(sender_counts.items(), key=lambda item: item[1], reverse=True)[:3] - parts = [f"Consolidated {len(candidates)} older or hidden memory episodes."] - if top_tags: - parts.append( - "Top tags: " + ", ".join(f"{tag} ({count})" for tag, count in top_tags) - ) - if top_senders: - parts.append( - "Speakers: " - + ", ".join(f"{sender} ({count})" for sender, count in top_senders) - ) - return " ".join(parts) - async def preview_consolidation( self, collection_id: str, @@ -1907,6 +1789,8 @@ class MemoryStore: importance = int(episode.get("importance", 2) or 2) age_days = self._age_days(str(episode.get("timestamp", "") or ""), now) reasons: list[str] = [] + if episode.get("source") == "consolidation": + continue if status == self.EPISODE_STATUS_SUPERSEDED: reasons.append("already superseded") elif status == self.EPISODE_STATUS_ARCHIVED: @@ -1926,7 +1810,7 @@ class MemoryStore: candidate["consolidation_reasons"] = reasons candidate["age_days"] = age_days candidates.append(candidate) - if status != self.EPISODE_STATUS_ARCHIVED: + if status == self.EPISODE_STATUS_SUPERSEDED: archive_ids.append(episode["id"]) if len(candidates) >= max_candidates: break @@ -1938,16 +1822,15 @@ class MemoryStore: if not apply_profile_updates: risk_notes.append("Profile update application is disabled by default.") if archive_ids: - risk_notes.append("Run will archive selected active/superseded episodes in this scope only.") + risk_notes.append("Run will archive superseded episodes only; active facts are retained.") - summary_episode = "" - if len(candidates) >= 2: - summary_episode = self._summarize_consolidation_candidates(candidates) + if any(episode.get("status") == self.EPISODE_STATUS_ACTIVE for episode in candidates): + risk_notes.append("Active facts are retained: statistical summaries cannot replace their content.") return { "candidate_episode_ids": [episode["id"] for episode in candidates], "candidates": candidates, - "summary_episode": summary_episode, + "summary_episode": "", "profile_updates": [] if not apply_profile_updates else [], "episodes_to_archive": archive_ids, "risk_notes": risk_notes, @@ -1966,7 +1849,7 @@ class MemoryStore: max_candidates: int = 20, apply_profile_updates: bool = False, ) -> dict[str, Any]: - """Apply scoped consolidation by writing a summary and archiving candidates.""" + """Archive superseded episodes while retaining active source facts.""" preview = await self.preview_consolidation( collection_id=collection_id, user_key=user_key, @@ -1986,22 +1869,10 @@ class MemoryStore: if episode: archived.append(episode_id) - summary = None - if preview.get("summary_episode"): - summary = await self.add_episode( - collection_id=collection_id, - embedding_model_uuid=embedding_model_uuid, - user_key=user_key, - content=preview["summary_episode"], - tags=["consolidation", "summary"], - importance=2, - source="consolidation", - ) - return { "preview": preview, "archived_episode_ids": archived, - "summary_episode": summary, + "summary_episode": None, "profile_updates_applied": [], } @@ -2095,6 +1966,8 @@ class MemoryStore: episode_id: str, user_key: str, status: str, + *, + superseded_by: str = "", ) -> dict[str, Any] | None: """Update an episode lifecycle status while preserving metadata.""" normalized_status = self.normalize_episode_status_value(status) @@ -2111,6 +1984,8 @@ class MemoryStore: content = episode.get("content", "") meta["content"] = content meta["status"] = normalized_status + if superseded_by: + meta["superseded_by"] = superseded_by if normalized_status == self.EPISODE_STATUS_ACTIVE: meta.pop("superseded_by", None) 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 828e57074..15aa16a18 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 @@ -178,7 +178,7 @@ async def test_include_superseded_can_inspect_hidden_records(): @pytest.mark.asyncio -async def test_auto_supersede_marks_old_episode_status(): +async def test_explicit_replacement_marks_old_episode_status(): plugin = FakeVectorPlugin() plugin.records = { "old-1": _record("old-1", status="active", content="User prefers tea"), @@ -191,6 +191,7 @@ async def test_auto_supersede_marks_old_episode_status(): user_key="user-1", content="Correction: user prefers coffee", tags=["correction"], + supersedes=["old-1"], ) assert plugin.records["old-1"]["metadata"]["status"] == "superseded" @@ -382,8 +383,8 @@ async def test_consolidation_preview_does_not_modify_records(): ) assert set(preview["candidate_episode_ids"]) == {"old-low", "hidden"} - assert set(preview["episodes_to_archive"]) == {"old-low", "hidden"} - assert preview["summary_episode"] + assert set(preview["episodes_to_archive"]) == {"hidden"} + assert preview["summary_episode"] == "" assert plugin.records["old-low"]["metadata"]["status"] == "active" assert plugin.records["hidden"]["metadata"]["status"] == "superseded" assert "other" in plugin.records @@ -391,7 +392,7 @@ async def test_consolidation_preview_does_not_modify_records(): @pytest.mark.asyncio -async def test_apply_consolidation_archives_scoped_candidates_and_writes_summary(): +async def test_apply_consolidation_archives_only_superseded_records(): plugin = FakeVectorPlugin() plugin.records = { "old-low": _record( @@ -415,14 +416,11 @@ async def test_apply_consolidation_archives_scoped_candidates_and_writes_summary max_candidates=20, ) - assert set(result["archived_episode_ids"]) == {"old-low", "hidden"} - assert plugin.records["old-low"]["metadata"]["status"] == "archived" + assert set(result["archived_episode_ids"]) == {"hidden"} + assert plugin.records["old-low"]["metadata"]["status"] == "active" assert plugin.records["hidden"]["metadata"]["status"] == "archived" assert plugin.records["other"]["metadata"]["status"] == "superseded" - summary = result["summary_episode"] - assert summary is not None - assert summary["source"] == "consolidation" - assert {"consolidation", "summary"} == set(summary["tags"]) + assert result["summary_episode"] is None class FakeStorageVectorPlugin(FakeVectorPlugin): @@ -497,3 +495,69 @@ async def test_accept_l2_candidate_uses_normal_episode_write_path(): assert episode["source"] == "candidate" assert episode["sender_id"] == "u-1" assert episode["id"] in plugin.records + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tag", ["correction", "fact-update", "clarification"]) +async def test_similar_correction_without_explicit_target_preserves_old_fact(tag): + plugin = FakeVectorPlugin() + plugin.records["old"] = _record("old", content="Alice visits Shanghai on Friday") + store = MemoryStore(plugin) + await store.add_episode("kb", "emb", "user-1", "Bob visits Shanghai on Friday", tags=[tag]) + assert plugin.records["old"]["metadata"]["status"] == "active" + assert len(plugin.records) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("invalid", ["missing", "other-speaker", "other-scope", "archived"]) +async def test_replacement_rejects_invalid_target_before_any_write(invalid): + plugin = FakeVectorPlugin() + record = _record("old") + record["metadata"]["sender_id"] = "alice" + if invalid == "other-speaker": + record["metadata"]["sender_id"] = "bob" + elif invalid == "other-scope": + record["metadata"]["user_key"] = "other" + elif invalid == "archived": + record["metadata"]["status"] = "archived" + if invalid != "missing": + plugin.records["old"] = record + store = MemoryStore(plugin) + with pytest.raises(ValueError, match="Replacement targets"): + await store.add_episode("kb", "emb", "user-1", "Corrected fact", sender_id="alice", supersedes=["old"]) + assert plugin.upserts == [] + + +@pytest.mark.asyncio +async def test_exact_match_after_backend_top_k_boundary_is_still_ranked(): + plugin = FakeVectorPlugin() + plugin.records = {f"noise-{i}": _record(f"noise-{i}", content="Unrelated note") for i in range(49)} + plugin.records["answer"] = _record("answer", content="Release rendezvous is at Cedar Pier") + result = await MemoryStore(plugin).search_episodes("kb", "emb", "release rendezvous", user_key="user-1", top_k=1) + assert [episode["id"] for episode in result] == ["answer"] + + +@pytest.mark.asyncio +async def test_consolidation_preserves_old_active_facts_and_keeps_obsolete_facts_hidden(): + plugin = FakeVectorPlugin() + plugin.records["active"] = _record("active", content="Alice chose PostgreSQL because of JSON queries") + plugin.records["outdated"] = _record("outdated", status="superseded", content="Alice chose MySQL") + store = MemoryStore(plugin) + result = await store.apply_consolidation("kb", "emb", "user-1", min_age_days=0) + assert result["summary_episode"] is None + assert result["archived_episode_ids"] == ["outdated"] + recalled = await store.search_episodes("kb", "emb", "database choice", user_key="user-1") + assert [episode["content"] for episode in recalled] == ["Alice chose PostgreSQL because of JSON queries"] + assert plugin.records["active"]["metadata"]["status"] == "active" + second = await store.apply_consolidation("kb", "emb", "user-1", min_age_days=0) + assert second["summary_episode"] is None + assert second["archived_episode_ids"] == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("targets", ["old", [""], [1], ["old"] * 21]) +async def test_replacement_target_input_is_validated_before_writing(targets): + plugin = FakeVectorPlugin() + with pytest.raises(ValueError, match="supersedes"): + await MemoryStore(plugin).add_episode("kb", "emb", "user-1", "New fact", supersedes=targets) + assert plugin.upserts == [] diff --git a/plugins/KnowledgeEngine/LongTermMemory/tests/store/test_memory_store_serialization.py b/plugins/KnowledgeEngine/LongTermMemory/tests/store/test_memory_store_serialization.py index 486aa82e1..115b2a526 100644 --- a/plugins/KnowledgeEngine/LongTermMemory/tests/store/test_memory_store_serialization.py +++ b/plugins/KnowledgeEngine/LongTermMemory/tests/store/test_memory_store_serialization.py @@ -434,7 +434,7 @@ async def test_nested_serialized_episode_write_does_not_self_lock(): importance=2, ) - # add_episode nests _auto_supersede for the same collection scope, which must + # add_episode can nest update_episode_status for the same collection scope, which must # not wait on the lock its own caller already holds. created = await asyncio.wait_for( store.add_episode( @@ -443,6 +443,7 @@ async def test_nested_serialized_episode_write_does_not_self_lock(): user_key="user-1", content="Alice lives in Munich", tags=["correction"], + supersedes=[original["id"]], importance=3, ), timeout=5, diff --git a/tests/e2e/test_longterm_memory_processor.py b/tests/e2e/test_longterm_memory_processor.py index 89468fa70..137e3bfb5 100644 --- a/tests/e2e/test_longterm_memory_processor.py +++ b/tests/e2e/test_longterm_memory_processor.py @@ -234,4 +234,32 @@ def test_local_agent_memory_tools_without_pipeline( assert profiles_only['episodes'] == [] assert 'brief technical answers' in profiles_only['speaker_profile'] + await call(ap, model, 'remember', {'content': 'Alice travels to Shanghai on Friday.'}, group='corrections') + await call( + ap, model, 'remember', {'content': 'Bob travels to Shanghai on Friday.'}, group='corrections', speaker='bob' + ) + original = json.loads( + await call(ap, model, 'recall_memory', {'query': 'Shanghai', 'speaker_id': 'alice'}, group='corrections') + ) + original_id = original['episodes'][0]['id'] + correction = { + 'content': 'Alice corrected her trip: Shanghai on Saturday.', + 'tags': ['correction'], + 'supersedes': [original_id], + } + refused = await call(ap, model, 'remember', correction, group='corrections', speaker='bob') + assert refused.startswith('Error: Replacement targets'), refused + replaced = await call(ap, model, 'remember', correction, group='corrections') + assert replaced.startswith('Remembered:'), replaced + current = json.loads(await call(ap, model, 'recall_memory', {'query': 'Shanghai'}, group='corrections')) + assert {item['content'] for item in current['episodes']} == { + 'Bob travels to Shanghai on Friday.', + correction['content'], + } + historical = json.loads( + await call(ap, model, 'recall_memory', {'query': 'Shanghai', 'status': 'superseded'}, group='corrections') + ) + assert historical['episodes'][0]['id'] == original_id + assert historical['episodes'][0]['superseded_by'] in {item['id'] for item in current['episodes']} + local._run_local_agent_probe(local_agent_e2e_tmpdir, read_after_restart)