diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py index 66163abed05..2e7d8322258 100644 --- a/python/packages/core/agent_framework/_harness/_memory.py +++ b/python/packages/core/agent_framework/_harness/_memory.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import hashlib import json import logging import os @@ -139,6 +140,23 @@ def _atomic_write_text(path: Path, text: str, *, encoding: str = "utf-8") -> Non def _slugify_topic(topic: str) -> str: + normalized = _normalize_topic(topic) + slug = re.sub(r"[^a-z0-9]+", "-", normalized.lower()).strip("-") + if slug == normalized.lower(): + # Already a safe stem; the mapping lost nothing, so the filename + # stays byte-identical to what older versions wrote. + return slug + # The readable slug is a lossy projection (non-ASCII, accents, case, + # punctuation): two distinct topics can share it. Suffix the full digest + # of the normalized topic so byte-distinct topics get distinct files; a + # truncated prefix still collides (e.g. "丮仑一一" and "仺价一一" share + # the 8-hex-char prefix 36749dfe). + digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest() + return f"{slug}-{digest}" if slug else f"memory-topic-{digest}" + + +def _legacy_slugify_topic(topic: str) -> str: + """The pre-digest slug derivation, kept to read stores written by older versions.""" slug = re.sub(r"[^a-z0-9]+", "-", _normalize_topic(topic).lower()).strip("-") return slug or "memory-topic" @@ -605,6 +623,14 @@ def write_topic(self, session: AgentSession, record: MemoryTopicRecord, *, sourc def delete_topic(self, session: AgentSession, *, source_id: str, topic: str) -> None: """Delete one topic memory file.""" + def migrate_misfiled_legacy_topic(self, session: AgentSession, *, source_id: str, topic: str) -> None: + """Move a different topic's legacy-addressed record off this topic's canonical address. + + Only stores with lossy filename-derived addressing can hold misfiled + legacy records; the default store contract has nothing to migrate. + """ + del session, source_id, topic + @abstractmethod def rebuild_index( self, @@ -776,6 +802,48 @@ def _get_state_path(self, session: AgentSession, *, source_id: str) -> Path: def _topic_path(self, session: AgentSession, *, source_id: str, topic: str) -> Path: return self._get_topics_directory(session, source_id=source_id) / f"{_slugify_topic(topic)}.md" + def _legacy_topic_path(self, session: AgentSession, *, source_id: str, topic: str) -> Path | None: + """The path a pre-digest version wrote for this topic, if it differs.""" + legacy = self._get_topics_directory(session, source_id=source_id) / f"{_legacy_slugify_topic(topic)}.md" + return legacy if legacy != self._topic_path(session, source_id=source_id, topic=topic) else None + + def _topic_read_candidates(self, session: AgentSession, *, source_id: str, topic: str) -> list[Path]: + candidates = [self._topic_path(session, source_id=source_id, topic=topic)] + legacy = self._legacy_topic_path(session, source_id=source_id, topic=topic) + if legacy is not None and self._topic_file_matches(legacy, topic=topic): + candidates.append(legacy) + return candidates + + @staticmethod + def _topic_file_matches(path: Path, *, topic: str) -> bool: + """True when the file's stored topic is the requested topic. + + A legacy-named file is only a fallback alias for ``topic`` when the + topic recorded inside the file agrees. The slug strips accents and + non-ASCII text, so a same-stem file can belong to a different topic + and must not be read, migrated, or deleted through the fallback. + """ + try: + record = MemoryTopicRecord.from_markdown(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, ValueError): + return False + return record.topic == _normalize_topic(topic) + + @staticmethod + def _is_unmigrated_legacy_of_other_topic(path: Path, record: MemoryTopicRecord, *, topic: str) -> bool: + """True when ``path`` is a pre-digest file that belongs to a different topic. + + The slug projection is lossy, so a legacy stem like ``caf`` can hold a + file whose stored topic is ``café`` while ``caf`` is requested. Such a + file must be migrated to its own derived name first, never read, + merged, or deleted as the requested topic. + """ + return ( + record.topic != _normalize_topic(topic) + and path.stem == _legacy_slugify_topic(record.topic) + and path.stem != record.slug + ) + @staticmethod def _serialize_json(value: object, *, dumps: JsonDumps) -> str: serialized = dumps(value) @@ -823,27 +891,76 @@ def list_topics(self, session: AgentSession, *, source_id: str) -> list[MemoryTo topic_path.read_text(encoding="utf-8"), fallback_topic=topic_path.stem.replace("-", " "), ) + if topic_path.stem == _legacy_slugify_topic(record.topic) and topic_path.stem != record.slug: + # Not yet migrated: the file still sits at its pre-digest name. Key the + # record by the stem that exists on disk so index pointers and + # get_topic(entry.slug) resolve to it before a rewrite migrates it. + record.slug = topic_path.stem topics.append(record) return sorted(topics, key=lambda record: (record.topic.lower(), record.updated_at)) def get_topic(self, session: AgentSession, *, source_id: str, topic: str) -> MemoryTopicRecord: """Return one topic memory file by topic name or slug.""" - topic_path = self._topic_path(session, source_id=source_id, topic=topic) - if not topic_path.exists(): - raise FileNotFoundError(f"No memory topic named '{topic}' was found for this owner.") - return MemoryTopicRecord.from_markdown(topic_path.read_text(encoding="utf-8"), fallback_topic=topic) + for topic_path in self._topic_read_candidates(session, source_id=source_id, topic=topic): + if topic_path.exists(): + return MemoryTopicRecord.from_markdown(topic_path.read_text(encoding="utf-8"), fallback_topic=topic) + raise FileNotFoundError(f"No memory topic named '{topic}' was found for this owner.") def write_topic(self, session: AgentSession, record: MemoryTopicRecord, *, source_id: str) -> None: """Persist one topic memory file.""" topic_path = self._topic_path(session, source_id=source_id, topic=record.slug) _atomic_write_text(topic_path, f"{record.to_markdown()}\n") + # Absorb pre-digest files into the new name so the topic does not + # list twice. The record's slug is already the derived stem, so the + # legacy name has to be re-derived from the human-readable topic too. + # Never touch the file just written: an explicit slug can collide + # with the legacy stem re-derived from the topic. + legacy_paths = { + path + for value in (record.topic, record.slug) + if (path := self._legacy_topic_path(session, source_id=source_id, topic=value)) is not None + and path != topic_path + } + for legacy_path in legacy_paths: + if legacy_path.exists() and self._topic_file_matches(legacy_path, topic=record.topic): + legacy_path.unlink() def delete_topic(self, session: AgentSession, *, source_id: str, topic: str) -> None: """Delete one topic memory file.""" + candidates: list[Path] = [] + for topic_path in self._topic_read_candidates(session, source_id=source_id, topic=topic): + if not topic_path.exists(): + continue + try: + record = MemoryTopicRecord.from_markdown(topic_path.read_text(encoding="utf-8"), fallback_topic=topic) + except (OSError, UnicodeDecodeError, ValueError): + record = None + if record is not None and self._is_unmigrated_legacy_of_other_topic(topic_path, record, topic=topic): + # A different topic's pre-digest file sits at this stem: it is + # not this topic's file, so it must survive the delete. + continue + candidates.append(topic_path) + if not candidates: + raise FileNotFoundError(f"No memory topic named '{topic}' was found for this owner.") + for topic_path in candidates: + topic_path.unlink() + + def migrate_misfiled_legacy_topic(self, session: AgentSession, *, source_id: str, topic: str) -> None: + """Move a different topic's pre-digest file off this topic's primary path. + + The slug projection is lossy, so a legacy stem like ``caf`` can hold a + file whose stored topic is ``café``. A write for ``caf`` must migrate + that record to its derived name first instead of merging into it. + """ topic_path = self._topic_path(session, source_id=source_id, topic=topic) if not topic_path.exists(): - raise FileNotFoundError(f"No memory topic named '{topic}' was found for this owner.") - topic_path.unlink() + return + try: + record = MemoryTopicRecord.from_markdown(topic_path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, ValueError): + return + if self._is_unmigrated_legacy_of_other_topic(topic_path, record, topic=topic): + self.write_topic(session, record, source_id=source_id) def rebuild_index( self, @@ -1525,6 +1642,11 @@ async def _merge_memory( normalized_topic = _normalize_topic(topic) normalized_memory = _normalize_memory_text(memory) async with self._topic_async_lock(session, normalized_topic): + # get_topic resolves loosely (slugs and legacy aliases included): a + # pre-digest file of another topic can sit at this topic's primary + # path. Migrate such a record first so this write never merges into + # or overwrites a different topic. + self.store.migrate_misfiled_legacy_topic(session, source_id=self.source_id, topic=normalized_topic) try: existing_record = self.store.get_topic(session, source_id=self.source_id, topic=normalized_topic) except FileNotFoundError: diff --git a/python/packages/core/tests/core/test_harness_memory.py b/python/packages/core/tests/core/test_harness_memory.py index d1c7b3dac51..40945d09319 100644 --- a/python/packages/core/tests/core/test_harness_memory.py +++ b/python/packages/core/tests/core/test_harness_memory.py @@ -1005,4 +1005,236 @@ def test_memory_file_store_uses_literal_folders_for_safe_identifiers(tmp_path) - assert parts[1] == "user_alice" +def _topic_store(tmp_path): + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + return store, session + + +def _record(topic: str, memories: list[str]) -> MemoryTopicRecord: + return MemoryTopicRecord( + topic=topic, + summary="", + memories=memories, + updated_at=datetime(2026, 10, 9, tzinfo=timezone.utc).isoformat(), + ) + + +def test_memory_file_store_keeps_distinct_non_ascii_topics_separate(tmp_path) -> None: + """Topics whose slugs collapse to the same stem must not share a file.""" + store, session = _topic_store(tmp_path) + travel = _record("旅行计划", ["护照已续签"]) + food = _record("饮食偏好", ["不吃香菜"]) + + store.write_topic(session, travel, source_id=DEFAULT_MEMORY_SOURCE_ID) + store.write_topic(session, food, source_id=DEFAULT_MEMORY_SOURCE_ID) + + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="旅行计划").memories == ["护照已续签"] + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="饮食偏好").memories == ["不吃香菜"] + + store.delete_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="饮食偏好") + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="旅行计划").memories == ["护照已续签"] + with pytest.raises(FileNotFoundError): + store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="饮食偏好") + + +def test_memory_file_store_keeps_accent_folded_topics_separate(tmp_path) -> None: + """Accents the slug strips still distinguish topics, as does case folding.""" + store, session = _topic_store(tmp_path) + store.write_topic(session, _record("café", ["a"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + store.write_topic(session, _record("cafè", ["b"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + store.write_topic(session, _record("My Topic", ["c"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café").memories == ["a"] + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="cafè").memories == ["b"] + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="My Topic").memories == ["c"] + + +def test_memory_file_store_reads_and_absorbs_pre_digest_topic_files(tmp_path) -> None: + """Files written before the digest suffix stay readable and migrate on rewrite.""" + store, session = _topic_store(tmp_path) + record = _record("café", ["legacy fact"]) + legacy_path = store._get_topics_directory(session, source_id=DEFAULT_MEMORY_SOURCE_ID) / "caf.md" + legacy_path.parent.mkdir(parents=True, exist_ok=True) + legacy_path.write_text(f"{record.to_markdown()}\n", encoding="utf-8") + + loaded = store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café") + assert loaded.memories == ["legacy fact"] + + store.write_topic(session, _record("café", ["new fact"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + assert not legacy_path.exists() + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café").memories == ["new fact"] + + +def test_memory_file_store_deletes_pre_digest_topic_file(tmp_path) -> None: + """Deleting a topic removes its legacy-named file as well.""" + store, session = _topic_store(tmp_path) + record = _record("旅行计划", ["legacy fact"]) + legacy_path = store._get_topics_directory(session, source_id=DEFAULT_MEMORY_SOURCE_ID) / "memory-topic.md" + legacy_path.parent.mkdir(parents=True, exist_ok=True) + legacy_path.write_text(f"{record.to_markdown()}\n", encoding="utf-8") + + store.delete_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="旅行计划") + assert not legacy_path.exists() + with pytest.raises(FileNotFoundError): + store.delete_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="旅行计划") + + +async def test_memory_context_provider_write_migrates_misfiled_legacy_first(tmp_path) -> None: + """``write_memory("caf")`` must not merge into a legacy ``caf.md`` that holds ``café``.""" + store, session = _topic_store(tmp_path) + record = _record("café", ["legacy accent fact"]) + legacy_path = store._get_topics_directory(session, source_id=DEFAULT_MEMORY_SOURCE_ID) / "caf.md" + legacy_path.parent.mkdir(parents=True, exist_ok=True) + legacy_path.write_text(f"{record.to_markdown()}\n", encoding="utf-8") + provider = MemoryContextProvider(store=store, recent_turns=0, max_extractions=0) + + updated = await provider._merge_memory( + session=session, + topic="caf", + memory="ascii fact", + now=datetime(2026, 10, 16, tzinfo=timezone.utc), + ) + + # The write created a distinct `caf` record instead of appending to `café`. + assert updated.topic == "caf" + assert updated.memories == ["ascii fact"] + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="caf").memories == ["ascii fact"] + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café").memories == ["legacy accent fact"] + + +def test_memory_file_store_delete_keeps_misfiled_legacy_file(tmp_path) -> None: + """Deleting ``caf`` must not delete a legacy ``caf.md`` that belongs to ``café``.""" + store, session = _topic_store(tmp_path) + record = _record("café", ["legacy accent fact"]) + legacy_path = store._get_topics_directory(session, source_id=DEFAULT_MEMORY_SOURCE_ID) / "caf.md" + legacy_path.parent.mkdir(parents=True, exist_ok=True) + legacy_path.write_text(f"{record.to_markdown()}\n", encoding="utf-8") + + with pytest.raises(FileNotFoundError): + store.delete_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="caf") + assert legacy_path.exists() + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café").memories == ["legacy accent fact"] + + +def test_slugify_topic_uses_full_sha256_digest() -> None: + """Truncated digest prefixes collide; the storage key keeps the full SHA-256.""" + import hashlib + + from agent_framework._harness._memory import _slugify_topic + + first, second = "丮仑一一", "仺价一一" + # Both topics share the former 8-hex-char prefix 36749dfe. + assert hashlib.sha256(first.encode("utf-8")).hexdigest()[:8] == hashlib.sha256(second.encode("utf-8")).hexdigest()[:8] + slug_first, slug_second = _slugify_topic(first), _slugify_topic(second) + assert slug_first != slug_second + assert slug_first == f"memory-topic-{hashlib.sha256(first.encode('utf-8')).hexdigest()}" + assert slug_second == f"memory-topic-{hashlib.sha256(second.encode('utf-8')).hexdigest()}" + + +def test_slugify_topic_keeps_safe_stems_byte_identical() -> None: + """Safe lowercase stems keep their pre-digest names; derived slugs are idempotent.""" + from agent_framework._harness._memory import _slugify_topic + + assert _slugify_topic("my-topic") == "my-topic" + spaced = _slugify_topic("travel plans") + assert spaced.startswith("travel-plans-") and spaced != "travel-plans" + derived = _slugify_topic("café") + assert _slugify_topic(derived) == derived + assert _slugify_topic("旅行计划") != _slugify_topic("饮食偏好") + + +def test_memory_file_store_delete_keeps_unrelated_stem_file(tmp_path) -> None: + """A legacy-named file owned by another topic must survive deleting this one.""" + store, session = _topic_store(tmp_path) + store.write_topic(session, _record("café", ["accent fact"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + store.write_topic(session, _record("caf", ["ascii fact"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + + store.delete_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café") + + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="caf").memories == ["ascii fact"] + with pytest.raises(FileNotFoundError): + store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café") + + +def test_memory_file_store_write_keeps_unrelated_stem_file(tmp_path) -> None: + """Writing an accented topic must not absorb the ASCII topic's own file.""" + store, session = _topic_store(tmp_path) + store.write_topic(session, _record("caf", ["ascii fact"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + store.write_topic(session, _record("café", ["accent fact"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="caf").memories == ["ascii fact"] + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café").memories == ["accent fact"] + + +def test_memory_file_store_write_with_explicit_slug_survives_legacy_cleanup(tmp_path) -> None: + """An explicit slug colliding with the legacy stem must not delete the file just written.""" + store, session = _topic_store(tmp_path) + record = _record("café", ["accent fact"]) + record.slug = "caf" + + store.write_topic(session, record, source_id=DEFAULT_MEMORY_SOURCE_ID) + + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="caf").memories == ["accent fact"] + + +def test_memory_file_store_legacy_fallback_requires_matching_stored_topic(tmp_path) -> None: + """A legacy-named file is a fallback alias only when its stored topic agrees.""" + store, session = _topic_store(tmp_path) + legacy_path = store._get_topics_directory(session, source_id=DEFAULT_MEMORY_SOURCE_ID) / "caf.md" + legacy_path.parent.mkdir(parents=True, exist_ok=True) + legacy_path.write_text(f"{_record('café', ['legacy accent fact']).to_markdown()}\n", encoding="utf-8") + + with pytest.raises(FileNotFoundError): + store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="cafè") + store.write_topic(session, _record("cafè", ["grave accent fact"]), source_id=DEFAULT_MEMORY_SOURCE_ID) + assert legacy_path.exists() + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="cafè").memories == ["grave accent fact"] + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="café").memories == ["legacy accent fact"] + + +def test_memory_file_store_lists_pre_digest_topic_under_its_on_disk_stem(tmp_path) -> None: + """An unmigrated legacy file must be listed under the stem that exists on disk.""" + store, session = _topic_store(tmp_path) + legacy_path = store._get_topics_directory(session, source_id=DEFAULT_MEMORY_SOURCE_ID) / "caf.md" + legacy_path.parent.mkdir(parents=True, exist_ok=True) + legacy_path.write_text(f"{_record('café', ['legacy accent fact']).to_markdown()}\n", encoding="utf-8") + + (entry,) = store.list_topics(session, source_id=DEFAULT_MEMORY_SOURCE_ID) + + assert entry.topic == "café" + assert entry.slug == "caf" + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic=entry.slug).memories == [ + "legacy accent fact" + ] + store.rebuild_index(session, source_id=DEFAULT_MEMORY_SOURCE_ID, line_limit=50, line_length=200) + index_text = store._get_index_path(session, source_id=DEFAULT_MEMORY_SOURCE_ID).read_text(encoding="utf-8") + assert "topics/caf.md" in index_text + # Keying by the on-disk stem must not widen the fallback: a same-stem but + # different topic still does not alias this file. + with pytest.raises(FileNotFoundError): + store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="cafè") + + +async def test_memory_context_provider_loads_unmigrated_legacy_topic(tmp_path) -> None: + """``before_run`` must resolve an index entry whose file still uses its pre-digest name.""" + store, session = _topic_store(tmp_path) + legacy_path = store._get_topics_directory(session, source_id=DEFAULT_MEMORY_SOURCE_ID) / "caf.md" + legacy_path.parent.mkdir(parents=True, exist_ok=True) + legacy_path.write_text(f"{_record('café', ['legacy accent fact']).to_markdown()}\n", encoding="utf-8") + provider = MemoryContextProvider(store=store, recent_turns=0, max_extractions=0) + + from agent_framework._sessions import SessionContext + + context = SessionContext( + session_id=session.session_id, + input_messages=[Message(role="user", contents=["tell me the legacy accent fact"])], + ) + + await provider.before_run(agent=None, session=session, context=context, state={}) + + assert any("legacy accent fact" in message.text for message in context.get_messages()) + + # endregion