From 8d4eb109881a1379292f8b94a8adbb846c6a6fe4 Mon Sep 17 00:00:00 2001 From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> Date: Mon, 26 Oct 2026 23:18:33 +0530 Subject: [PATCH 1/4] Python: give memory topic files an injective filename stem Distinct topics can share a readable slug, so a slug-only stem let the second topic silently overwrite the first. The stem now appends a digest of the normalized topic, and the storage layer keeps one authoritative record: lookups validate every candidate against the record it actually holds instead of accepting a filename prefix, so a missing topic can never read, merge into, or delete a longer slug; rewriting a topic that lived at a legacy slug-only file retires the legacy file; index pointer links and provider lookups use the persisted stem; and an explicit slug override is persisted in the record and stays findable by that slug, without letting lookups for other names bleed into the overridden record. Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> --- .../core/agent_framework/_harness/_memory.py | 120 ++++++++++++-- .../core/tests/core/test_harness_memory.py | 154 ++++++++++++++++++ 2 files changed, 264 insertions(+), 10 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py index c3f02f883b5..ea6ad95f244 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 re @@ -129,6 +130,22 @@ def _slugify_topic(topic: str) -> str: return slug or "memory-topic" +def _topic_stem(topic: str, *, slug_override: str | None = None) -> str: + """Return the injective filename stem for a topic. + + Distinct topics can share a readable slug: ``C ++`` and ``C #`` both slugify + to ``c``, so a slug-only stem let the second topic silently overwrite the + first topic's memories. The stem keeps the readable slug for humans and + appends a short digest of the normalized topic, so distinct normalized + topics never share a file (mirroring how ``_storage_key_segment`` keeps + byte-distinct identifiers in separate storage locations). + """ + normalized = _normalize_topic(topic) + slug = _slugify_topic(slug_override if slug_override is not None else normalized) + digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:8] + return f"{slug}-{digest}" + + def _timestamp(value: datetime | None = None) -> str: current = value or datetime.now(timezone.utc) return current.replace(microsecond=0).isoformat() @@ -302,7 +319,9 @@ def from_topic_record(cls, record: MemoryTopicRecord) -> MemoryIndexEntry: """ return cls( topic=record.topic, - slug=record.slug, + # The persisted stem, not the readable slug: distinct topics can + # share a slug, and an index link must point at the real file. + slug=record.storage_stem, summary=_coerce_summary(record.summary, record.memories), updated_at=record.updated_at, ) @@ -373,6 +392,11 @@ def __init__( self.updated_at = updated_at self.session_ids = _dedupe_strings(session_ids or []) + @property + def storage_stem(self) -> str: + """Filename stem this record persists under (slug or explicit override plus digest).""" + return _topic_stem(self.topic, slug_override=self.slug) + def to_dict(self) -> dict[str, Any]: """Serialize the topic record for tool output. @@ -449,9 +473,13 @@ def to_markdown(self) -> str: memory_lines = [f"- {_escape_markdown_line(memory)}" for memory in self.memories] or [ f"- {DEFAULT_MEMORY_NO_TOPICS_TEXT[2:]}" ] + header_lines = [f"# {self.topic}", ""] + if self.slug != _slugify_topic(self.topic): + # Persist an explicit slug override so later lookups by that slug, + # and the index stem, still resolve after a reload. + header_lines += [f"Slug: {self.slug}", ""] return "\n".join([ - f"# {self.topic}", - "", + *header_lines, f"Updated: {self.updated_at}", f"Sessions: {session_line}", "", @@ -479,6 +507,7 @@ def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> M ValueError: If the markdown cannot be parsed into the expected format. """ topic = fallback_topic + slug: str | None = None updated_at = _timestamp() session_ids: list[str] = [] summary_lines: list[str] = [] @@ -492,6 +521,9 @@ def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> M topic = stripped[2:].strip() current_section = None continue + if current_section is None and stripped.startswith("Slug: "): + slug = stripped.removeprefix("Slug: ").strip() + continue if current_section is None and stripped.startswith("Updated: "): updated_at = stripped.removeprefix("Updated: ").strip() continue @@ -522,6 +554,7 @@ def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> M raise ValueError("Memory topic markdown is missing a '# ' heading.") return cls( topic=topic, + slug=slug, summary="\n".join(summary_lines).strip(), memories=memories, updated_at=updated_at, @@ -763,9 +796,67 @@ def _get_index_path(self, session: AgentSession, *, source_id: str) -> Path: def _get_state_path(self, session: AgentSession, *, source_id: str) -> Path: return self._get_memory_root(session, source_id=source_id) / self.state_file_name - def _topic_path(self, session: AgentSession, *, source_id: str, topic: str) -> Path: + def _topic_path( + self, session: AgentSession, *, source_id: str, topic: str, slug_override: str | None = None + ) -> Path: + return ( + self._get_topics_directory(session, source_id=source_id) + / f"{_topic_stem(topic, slug_override=slug_override)}.md" + ) + + def _legacy_topic_path(self, session: AgentSession, *, source_id: str, topic: str) -> Path: + """Filename stem used before the digest suffix (kept for reading old stores).""" return self._get_topics_directory(session, source_id=source_id) / f"{_slugify_topic(topic)}.md" + def _existing_topic_path(self, session: AgentSession, *, source_id: str, topic: str) -> Path | None: + """Return the topic file to read, or ``None`` when no file matches. + + Lookups resolve in order: the digest stem of the given name (the exact + file ``write_topic`` uses for that topic), a legacy slug-only file from + stores written before the digest suffix, then any file whose stem starts + with the readable slug followed by a dash, which serves lookups made by + slug or by a topic that normalizes differently than its stored spelling. + When several topics share a slug, a slug-only lookup is ambiguous by + nature; the file whose recorded topic matches the lookup wins, else the + first stem in sorted order keeps the choice deterministic. + """ + topics_dir = self._get_topics_directory(session, source_id=source_id) + if not topics_dir.is_dir(): + return None + try: + wanted = _normalize_topic(topic) + except ValueError: + return None + # 1. The digest stem of the given name: the exact file write_topic uses. + topic_path = self._topic_path(session, source_id=source_id, topic=topic) + if topic_path.exists(): + return topic_path + # 2. An exact filename match: a legacy slug-only file from stores written + # before the digest suffix, or a caller passing a persisted stem. + exact_path = topics_dir / f"{_slugify_topic(topic)}.md" + if exact_path.exists(): + return exact_path + # 3. Digest-suffixed stems sharing the slug. A prefix glob alone proves + # nothing (a missing topic must never read, merge into, or delete a + # longer unrelated slug), so every candidate is validated against the + # record it actually holds. + prefix = f"{_slugify_topic(topic)}-" + matches = sorted(topics_dir.glob(f"{prefix}*.md")) + if not matches: + return None + for path in matches: + record = MemoryTopicRecord.from_markdown(path.read_text(encoding="utf-8"), fallback_topic=topic) + if record.topic == wanted: + return path + # A record carrying an explicit slug override stays findable by that + # slug; a default record never matches here, so colliding topics such + # as C++ and C# cannot merge through a shared slug. + for path in matches: + record = MemoryTopicRecord.from_markdown(path.read_text(encoding="utf-8"), fallback_topic=topic) + if record.slug == _slugify_topic(topic) and record.slug != _slugify_topic(record.topic): + return path + return None + @staticmethod def _serialize_json(value: object, *, dumps: JsonDumps) -> str: serialized = dumps(value) @@ -818,20 +909,29 @@ def list_topics(self, session: AgentSession, *, source_id: str) -> list[MemoryTo 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(): + topic_path = self._existing_topic_path(session, source_id=source_id, topic=topic) + if topic_path is None: 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) 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) + """Persist one topic memory file. + + The digest-stemmed file is the single authoritative record: when this + topic previously lived at another path (a legacy slug-only file, or a + stem written under a different slug override), that file is retired so + lookups can never surface stale content or resurrect the old record. + """ + previous_path = self._existing_topic_path(session, source_id=source_id, topic=record.topic) + topic_path = self._topic_path(session, source_id=source_id, topic=record.topic, slug_override=record.slug) _atomic_write_text(topic_path, f"{record.to_markdown()}\n") + if previous_path is not None and previous_path != topic_path: + previous_path.unlink(missing_ok=True) def delete_topic(self, session: AgentSession, *, source_id: str, topic: str) -> None: """Delete one topic memory file.""" - topic_path = self._topic_path(session, source_id=source_id, topic=topic) - if not topic_path.exists(): + topic_path = self._existing_topic_path(session, source_id=source_id, topic=topic) + if topic_path is None: raise FileNotFoundError(f"No memory topic named '{topic}' was found for this owner.") topic_path.unlink() diff --git a/python/packages/core/tests/core/test_harness_memory.py b/python/packages/core/tests/core/test_harness_memory.py index 464255d0b43..4e034be2de3 100644 --- a/python/packages/core/tests/core/test_harness_memory.py +++ b/python/packages/core/tests/core/test_harness_memory.py @@ -1046,6 +1046,160 @@ def test_memory_file_store_encodes_non_ascii_owner_ids(tmp_path) -> None: assert roots[0] != roots[1] +def test_memory_file_store_topics_with_the_same_slug_do_not_collide(tmp_path) -> None: + """Distinct topics that share a readable slug must keep separate memory files. + + ``C ++`` and ``C #`` both slugify to ``c``, so a slug-only filename let the + second topic's write destroy the first topic's memories. + """ + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + + first = MemoryTopicRecord( + topic="C++", summary="cpp", memories=["smart pointers"], updated_at="2026-10-26T00:00:00+00:00" + ) + second = MemoryTopicRecord( + topic="C#", summary="csharp", memories=["primary constructors"], updated_at="2026-10-26T00:00:00+00:00" + ) + store.write_topic(session, first, source_id=DEFAULT_MEMORY_SOURCE_ID) + store.write_topic(session, second, source_id=DEFAULT_MEMORY_SOURCE_ID) + + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="C++").memories == ["smart pointers"] + assert store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="C#").memories == ["primary constructors"] + listed = sorted(entry.topic for entry in store.list_topics(session, source_id=DEFAULT_MEMORY_SOURCE_ID)) + assert listed == ["C#", "C++"] + + +def test_memory_file_store_reads_topics_from_the_legacy_slug_filename(tmp_path) -> None: + """Topic files written before the digest suffix stay readable.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + + topics_dir = store._get_topics_directory(session, source_id=DEFAULT_MEMORY_SOURCE_ID) + topics_dir.mkdir(parents=True, exist_ok=True) + (topics_dir / "python.md").write_text("# python\n\n## Memories\n\n- old note\n", encoding="utf-8") + + record = store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="Python") + assert record.memories == ["old note"] + + +def test_memory_lookup_never_bleeds_across_slug_prefixes(tmp_path) -> None: + """A missing short topic must not read, merge into, or delete a longer + topic whose slug starts with the same characters.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + + store.write_topic( + session, + MemoryTopicRecord(topic="Collection", slug=None, summary="a", memories=["x"], updated_at="t"), + source_id="src", + ) + + with pytest.raises(FileNotFoundError): + store.get_topic(session, source_id="src", topic="coll") + + +def test_memory_lookup_validates_identity_for_colliding_slugs(tmp_path) -> None: + """Colliding topics such as C++ and C# resolve to their own records, and a + lookup for a topic that does not exist under a shared slug finds nothing + rather than an arbitrary same-slug file.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + + store.write_topic( + session, + MemoryTopicRecord(topic="C++", slug=None, summary="cpp", memories=["smart pointers"], updated_at="t"), + source_id="src", + ) + store.write_topic( + session, + MemoryTopicRecord(topic="C#", slug=None, summary="cs", memories=["ctors"], updated_at="t"), + source_id="src", + ) + + assert store.get_topic(session, source_id="src", topic="C++").memories == ["smart pointers"] + assert store.get_topic(session, source_id="src", topic="C#").memories == ["ctors"] + with pytest.raises(FileNotFoundError): + store.get_topic(session, source_id="src", topic="C") + + +def test_memory_write_retires_a_legacy_slug_file_for_the_same_topic(tmp_path) -> None: + """Rewriting a topic that lived at a legacy slug-only filename leaves a + single authoritative record: the digest-stemmed file.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + topics_dir = store._get_topics_directory(session, source_id="src") + + legacy = topics_dir / "legacy-topic.md" + legacy.parent.mkdir(parents=True, exist_ok=True) + legacy.write_text( + "# Legacy Topic\n\nUpdated: t0\nSessions: -\n\n## Summary\nold\n\n## Memories\n- old fact\n", + encoding="utf-8", + ) + + store.write_topic( + session, + MemoryTopicRecord(topic="Legacy Topic", slug=None, summary="new", memories=["new fact"], updated_at="t1"), + source_id="src", + ) + + assert not legacy.exists() + record = store.get_topic(session, source_id="src", topic="Legacy Topic") + assert record.memories == ["new fact"] + assert sorted(path.name for path in topics_dir.glob("legacy*.md")) == [f"{record.storage_stem}.md"] + + +def test_memory_index_pointer_links_resolve_to_the_persisted_files(tmp_path) -> None: + """Index entries link the persisted stem, so every pointer line names a + file that exists even when several topics share a readable slug.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + store.write_topic( + session, + MemoryTopicRecord(topic="C++", slug=None, summary="cpp", memories=["m"], updated_at="t"), + source_id="src", + ) + store.write_topic( + session, MemoryTopicRecord(topic="C#", slug=None, summary="cs", memories=["m"], updated_at="t"), source_id="src" + ) + topics_dir = store._get_topics_directory(session, source_id="src") + + entries = store.rebuild_index(session, source_id="src", line_limit=50, line_length=200) + + assert len(entries) == 2 + assert len({entry.slug for entry in entries}) == 2 + for entry in entries: + assert f"(topics/{entry.slug}.md)" in entry.to_pointer_line() + assert (topics_dir / f"{entry.slug}.md").exists() + + +def test_memory_explicit_slug_override_is_persisted_and_findable(tmp_path) -> None: + """A record with an explicit slug override persists under that slug and is + findable by it after a reload, without letting lookups for other names + bleed into the overridden record.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + + store.write_topic( + session, + MemoryTopicRecord(topic="Machine Learning", slug="ml-notes", summary="s", memories=["fact"], updated_at="t"), + source_id="src", + ) + + record = store.get_topic(session, source_id="src", topic="ml-notes") + assert record.topic == "Machine Learning" + assert record.memories == ["fact"] + with pytest.raises(FileNotFoundError): + store.get_topic(session, source_id="src", topic="machine") + + def test_memory_file_store_uses_literal_folders_for_safe_identifiers(tmp_path) -> None: """Safe identifiers are readable on disk rather than opaque encoded segments.""" store = MemoryFileStore(tmp_path, owner_prefix="user_", owner_state_key="owner_id") From cef2b8e95aacc8be4e87e8574c279116891da8ea Mon Sep 17 00:00:00 2001 From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> Date: Thu, 29 Oct 2026 18:01:10 +0530 Subject: [PATCH 2/4] Python: make memory topic stems collision safe and reloadable by slug The topic digest widens to 64 bits and write_topic verifies the occupant of an existing stem before reusing it, extending the digest when another topic already lives there. Slug lookups answer a single unambiguous holder again so default slug reloads work, and consolidation reloads each topic by exact identity instead of by slug. Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> --- .../core/agent_framework/_harness/_memory.py | 157 ++++++++++++------ .../core/tests/core/test_harness_memory.py | 112 +++++++++++++ 2 files changed, 214 insertions(+), 55 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py index ea6ad95f244..0d8c99c26ac 100644 --- a/python/packages/core/agent_framework/_harness/_memory.py +++ b/python/packages/core/agent_framework/_harness/_memory.py @@ -12,7 +12,7 @@ from abc import ABC, abstractmethod from base64 import b32decode from binascii import Error as BinasciiError -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterator, Mapping, Sequence from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, ClassVar, Final, cast @@ -131,19 +131,31 @@ def _slugify_topic(topic: str) -> str: def _topic_stem(topic: str, *, slug_override: str | None = None) -> str: - """Return the injective filename stem for a topic. + """Return the primary filename stem for a topic. Distinct topics can share a readable slug: ``C ++`` and ``C #`` both slugify to ``c``, so a slug-only stem let the second topic silently overwrite the first topic's memories. The stem keeps the readable slug for humans and - appends a short digest of the normalized topic, so distinct normalized - topics never share a file (mirroring how ``_storage_key_segment`` keeps + appends a digest of the normalized topic, so distinct normalized topics + never share a file (mirroring how ``_storage_key_segment`` keeps byte-distinct identifiers in separate storage locations). """ + return next(_topic_stem_candidates(topic, slug_override=slug_override)) + + +def _topic_stem_candidates(topic: str, *, slug_override: str | None = None) -> Iterator[str]: + """Yield the stems a topic may persist under, longest digest last. + + The 64-bit digest makes a collision between distinct topics unrealistic, + and ``MemoryFileStore._writable_topic_path`` walks these candidates and + verifies the occupant of an existing stem before reusing it, so even a + real digest collision cannot make one topic overwrite another. + """ normalized = _normalize_topic(topic) slug = _slugify_topic(slug_override if slug_override is not None else normalized) - digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:8] - return f"{slug}-{digest}" + digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest() + for digest_length in (16, 32, 64): + yield f"{slug}-{digest[:digest_length]}" def _timestamp(value: datetime | None = None) -> str: @@ -362,7 +374,7 @@ class MemoryTopicRecord: memories: list[str] updated_at: str session_ids: list[str] - __slots__ = ("memories", "session_ids", "slug", "summary", "topic", "updated_at") + __slots__ = ("_on_disk_stem", "memories", "session_ids", "slug", "summary", "topic", "updated_at") def __init__( self, @@ -391,11 +403,18 @@ def __init__( self.summary = _coerce_summary(summary, self.memories) self.updated_at = updated_at self.session_ids = _dedupe_strings(session_ids or []) + self._on_disk_stem: str | None = None @property def storage_stem(self) -> str: - """Filename stem this record persists under (slug or explicit override plus digest).""" - return _topic_stem(self.topic, slug_override=self.slug) + """Filename stem this record persists under (slug or explicit override plus digest). + + Set from the file the record was read from or written to, so it stays + truthful even when a digest collision pushed the record onto an + extended stem; records that have not touched disk yet fall back to the + computed stem. + """ + return self._on_disk_stem or _topic_stem(self.topic, slug_override=self.slug) def to_dict(self) -> dict[str, Any]: """Serialize the topic record for tool output. @@ -796,29 +815,21 @@ def _get_index_path(self, session: AgentSession, *, source_id: str) -> Path: def _get_state_path(self, session: AgentSession, *, source_id: str) -> Path: return self._get_memory_root(session, source_id=source_id) / self.state_file_name - def _topic_path( - self, session: AgentSession, *, source_id: str, topic: str, slug_override: str | None = None - ) -> Path: - return ( - self._get_topics_directory(session, source_id=source_id) - / f"{_topic_stem(topic, slug_override=slug_override)}.md" - ) - - def _legacy_topic_path(self, session: AgentSession, *, source_id: str, topic: str) -> Path: - """Filename stem used before the digest suffix (kept for reading old stores).""" - return self._get_topics_directory(session, source_id=source_id) / f"{_slugify_topic(topic)}.md" - def _existing_topic_path(self, session: AgentSession, *, source_id: str, topic: str) -> Path | None: """Return the topic file to read, or ``None`` when no file matches. - Lookups resolve in order: the digest stem of the given name (the exact - file ``write_topic`` uses for that topic), a legacy slug-only file from - stores written before the digest suffix, then any file whose stem starts - with the readable slug followed by a dash, which serves lookups made by - slug or by a topic that normalizes differently than its stored spelling. - When several topics share a slug, a slug-only lookup is ambiguous by - nature; the file whose recorded topic matches the lookup wins, else the - first stem in sorted order keeps the choice deterministic. + Lookups resolve in order: the digest-stemmed candidates for the given + name (the files ``write_topic`` uses for that topic, including a stem + extended past a digest collision), an exact filename match (a legacy + slug-only file from stores written before the digest suffix, or a + persisted stem), then any file whose stem starts with the readable + slug followed by a dash, which serves lookups made by slug or by a + topic that normalizes differently than its stored spelling. Every + digest-stemmed or slug-family candidate is validated against the + record it actually holds, and a slug lookup answers only when the + lookup string is itself a slug with a single claiming record: + default-slug reloads stay working, while colliding topics such as + C++ and C# never merge through the shared slug. """ topics_dir = self._get_topics_directory(session, source_id=source_id) if not topics_dir.is_dir(): @@ -827,34 +838,40 @@ def _existing_topic_path(self, session: AgentSession, *, source_id: str, topic: wanted = _normalize_topic(topic) except ValueError: return None - # 1. The digest stem of the given name: the exact file write_topic uses. - topic_path = self._topic_path(session, source_id=source_id, topic=topic) - if topic_path.exists(): - return topic_path - # 2. An exact filename match: a legacy slug-only file from stores written - # before the digest suffix, or a caller passing a persisted stem. + # 1. The digest-stemmed candidates for this name: the exact files + # write_topic uses, each validated against the record it holds so a + # digest collision with another topic can never answer the lookup. + for candidate in _topic_stem_candidates(topic): + candidate_path = topics_dir / f"{candidate}.md" + if not candidate_path.exists(): + continue + record = MemoryTopicRecord.from_markdown(candidate_path.read_text(encoding="utf-8"), fallback_topic=topic) + if record.topic == wanted: + return candidate_path + # 2. An exact filename match: a legacy slug-only file from stores + # written before the digest suffix, or a caller passing a persisted + # stem such as an index entry slug. exact_path = topics_dir / f"{_slugify_topic(topic)}.md" if exact_path.exists(): return exact_path # 3. Digest-suffixed stems sharing the slug. A prefix glob alone proves # nothing (a missing topic must never read, merge into, or delete a - # longer unrelated slug), so every candidate is validated against the - # record it actually holds. - prefix = f"{_slugify_topic(topic)}-" - matches = sorted(topics_dir.glob(f"{prefix}*.md")) - if not matches: - return None - for path in matches: + # longer unrelated slug), so every candidate is validated against + # the record it actually holds. A slug lookup answers only when the + # lookup string is itself a slug and exactly one record claims it: + # default-slug reloads stay working, while a topic name whose + # slugification merely lands in the family (two CJK topics both + # slugifying to ``memory-topic``) cannot borrow a neighbor's file. + slug = _slugify_topic(topic) + slug_holders: list[Path] = [] + for path in sorted(topics_dir.glob(f"{slug}-*.md")): record = MemoryTopicRecord.from_markdown(path.read_text(encoding="utf-8"), fallback_topic=topic) if record.topic == wanted: return path - # A record carrying an explicit slug override stays findable by that - # slug; a default record never matches here, so colliding topics such - # as C++ and C# cannot merge through a shared slug. - for path in matches: - record = MemoryTopicRecord.from_markdown(path.read_text(encoding="utf-8"), fallback_topic=topic) - if record.slug == _slugify_topic(topic) and record.slug != _slugify_topic(record.topic): - return path + if record.slug == slug: + slug_holders.append(path) + if topic == slug and len(slug_holders) == 1: + return slug_holders[0] return None @staticmethod @@ -904,6 +921,7 @@ def list_topics(self, session: AgentSession, *, source_id: str) -> list[MemoryTo topic_path.read_text(encoding="utf-8"), fallback_topic=topic_path.stem.replace("-", " "), ) + record._on_disk_stem = topic_path.stem topics.append(record) return sorted(topics, key=lambda record: (record.topic.lower(), record.updated_at)) @@ -912,19 +930,45 @@ def get_topic(self, session: AgentSession, *, source_id: str, topic: str) -> Mem topic_path = self._existing_topic_path(session, source_id=source_id, topic=topic) if topic_path is None: 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) + record = MemoryTopicRecord.from_markdown(topic_path.read_text(encoding="utf-8"), fallback_topic=topic) + record._on_disk_stem = topic_path.stem + return record + + def _writable_topic_path(self, session: AgentSession, *, source_id: str, record: MemoryTopicRecord) -> Path: + """Return the path this record must persist under. + + A stem that is free, or already holds this topic, is reused. When + another topic lives at the primary stem (a digest collision between + distinct normalized topics), the digest extends until a free stem is + found, so a write can never silently replace a different topic's + memories. + """ + topics_dir = self._get_topics_directory(session, source_id=source_id) + normalized = _normalize_topic(record.topic) + for candidate in _topic_stem_candidates(record.topic, slug_override=record.slug): + candidate_path = topics_dir / f"{candidate}.md" + if not candidate_path.exists(): + return candidate_path + existing = MemoryTopicRecord.from_markdown( + candidate_path.read_text(encoding="utf-8"), fallback_topic=record.topic + ) + if _normalize_topic(existing.topic) == normalized: + return candidate_path + raise RuntimeError(f"Unable to allocate a distinct memory topic filename for '{record.topic}'.") def write_topic(self, session: AgentSession, record: MemoryTopicRecord, *, source_id: str) -> None: """Persist one topic memory file. The digest-stemmed file is the single authoritative record: when this - topic previously lived at another path (a legacy slug-only file, or a - stem written under a different slug override), that file is retired so - lookups can never surface stale content or resurrect the old record. + topic previously lived at another path (a legacy slug-only file, an + extended stem freed by a neighbor's deletion, or a stem written under a + different slug override), that file is retired so lookups can never + surface stale content or resurrect the old record. """ previous_path = self._existing_topic_path(session, source_id=source_id, topic=record.topic) - topic_path = self._topic_path(session, source_id=source_id, topic=record.topic, slug_override=record.slug) + topic_path = self._writable_topic_path(session, source_id=source_id, record=record) _atomic_write_text(topic_path, f"{record.to_markdown()}\n") + record._on_disk_stem = topic_path.stem if previous_path is not None and previous_path != topic_path: previous_path.unlink(missing_ok=True) @@ -1663,7 +1707,10 @@ async def _run_consolidation( for record in topic_records: async with self._topic_async_lock(session, record.slug): try: - current_record = self.store.get_topic(session, source_id=self.source_id, topic=record.slug) + # Reload by the exact topic identity: a slug can be shared + # by several topics, and dropping one of them here would + # silently skip its consolidation. + current_record = self.store.get_topic(session, source_id=self.source_id, topic=record.topic) except FileNotFoundError: continue consolidated_record, succeeded = await self._consolidate_topic( diff --git a/python/packages/core/tests/core/test_harness_memory.py b/python/packages/core/tests/core/test_harness_memory.py index 4e034be2de3..081dad57e80 100644 --- a/python/packages/core/tests/core/test_harness_memory.py +++ b/python/packages/core/tests/core/test_harness_memory.py @@ -28,6 +28,7 @@ Message, ) from agent_framework import _filesystem as filesystem_module +from agent_framework._harness._memory import _topic_stem from .test_filesystem import COLLIDING_IDENTIFIERS @@ -1200,6 +1201,117 @@ def test_memory_explicit_slug_override_is_persisted_and_findable(tmp_path) -> No store.get_topic(session, source_id="src", topic="machine") +def test_memory_topic_is_reloadable_by_its_default_slug(tmp_path) -> None: + """A normally written topic reloads by its default slug, which callers + holding record.slug rely on.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + + record = MemoryTopicRecord(topic="Travel Preferences", summary="s", memories=["window seat"], updated_at="t") + store.write_topic(session, record, source_id="src") + + reloaded = store.get_topic(session, source_id="src", topic=record.slug) + assert reloaded.topic == "Travel Preferences" + assert reloaded.memories == ["window seat"] + + +def test_memory_topics_colliding_on_the_digest_keep_separate_files(tmp_path) -> None: + """Two topics whose normalized names share a short digest must keep separate files. + + 主题乚丢一 and 主题亊不七 both digest to 74043f1d in their first eight hex + characters and both slugify to memory-topic, so a shorter digest let the + second write silently replace the first topic's memories, and reading the + first topic returned the second topic's fact. + """ + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + + first_topic = "主题乚丢一" + second_topic = "主题亊不七" + store.write_topic( + session, + MemoryTopicRecord(topic=first_topic, summary="a", memories=["fact a"], updated_at="t"), + source_id="src", + ) + store.write_topic( + session, + MemoryTopicRecord(topic=second_topic, summary="b", memories=["fact b"], updated_at="t"), + source_id="src", + ) + + assert store.get_topic(session, source_id="src", topic=first_topic).memories == ["fact a"] + assert store.get_topic(session, source_id="src", topic=second_topic).memories == ["fact b"] + listed = {record.topic: record.memories for record in store.list_topics(session, source_id="src")} + assert listed == {first_topic: ["fact a"], second_topic: ["fact b"]} + + +def test_memory_write_extends_the_stem_when_another_topic_owns_the_primary_stem(tmp_path) -> None: + """A write must never replace a different topic that already lives at the + primary stem; the digest extends until the stem is free.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + topics_dir = store._get_topics_directory(session, source_id="src") + + alpha = MemoryTopicRecord(topic="Alpha Topic", summary="a", memories=["alpha fact"], updated_at="t") + beta = MemoryTopicRecord(topic="Beta Topic", summary="b", memories=["beta fact"], updated_at="t") + primary = topics_dir / f"{_topic_stem('Beta Topic')}.md" + primary.parent.mkdir(parents=True, exist_ok=True) + primary.write_text(f"{alpha.to_markdown()}\n", encoding="utf-8") + + store.write_topic(session, beta, source_id="src") + + reread_alpha = MemoryTopicRecord.from_markdown(primary.read_text(encoding="utf-8")) + assert reread_alpha.memories == ["alpha fact"] + beta_files = [path for path in sorted(topics_dir.glob("beta-topic-*.md")) if path != primary] + assert len(beta_files) == 1 + assert beta_files[0].stem != primary.stem + assert beta.storage_stem == beta_files[0].stem + assert store.get_topic(session, source_id="src", topic="Beta Topic").memories == ["beta fact"] + + entries = store.rebuild_index(session, source_id="src", line_limit=10, line_length=200) + for entry in entries: + assert (topics_dir / f"{entry.slug}.md").exists() + + +async def test_memory_consolidation_processes_topics_whose_slug_differs_from_their_name(tmp_path) -> None: + """Forced consolidation reloads each topic by exact identity, so a topic + whose default slug differs from its name is consolidated rather than + silently skipped.""" + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + consolidation_client = _MemoryHarnessClient( + consolidation_payload={"summary": "Consolidated summary.", "memories": ["Visit Oslo in June."]} + ) + provider = MemoryContextProvider( + store=store, + consolidation_client=consolidation_client, # type: ignore[arg-type] # pyrefly: ignore[bad-argument-type] # ty: ignore[invalid-argument-type] + ) + agent = Agent( + client=_MemoryHarnessClient(), # type: ignore[arg-type] # pyrefly: ignore[bad-argument-type] # ty: ignore[invalid-argument-type] + context_providers=[provider], + default_options=_no_store_options(), + ) + + _, options = await agent._prepare_session_and_messages( # type: ignore[reportPrivateUsage] + session=session, + input_messages=[Message(role="user", contents=["Remember this."])], + ) + write_memory = _tool_by_name(options["tools"], "write_memory") + consolidate_memories = _tool_by_name(options["tools"], "consolidate_memories") + + await write_memory.invoke(arguments={"topic": "Travel Preferences", "memory": "Window seat."}) # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + await consolidate_memories.invoke() # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + + topic = store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="Travel Preferences") + assert topic.summary == "Consolidated summary." + assert topic.memories == ["Visit Oslo in June."] + assert consolidation_client.calls == ["consolidate"] + + def test_memory_file_store_uses_literal_folders_for_safe_identifiers(tmp_path) -> None: """Safe identifiers are readable on disk rather than opaque encoded segments.""" store = MemoryFileStore(tmp_path, owner_prefix="user_", owner_state_key="owner_id") From f1fd7fa1b0f1b48dcac441f4ab23c47454563cc9 Mon Sep 17 00:00:00 2001 From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> Date: Sat, 31 Oct 2026 07:58:40 +0530 Subject: [PATCH 3/4] Python: stamp the memory record stem through a setter Package Checks flagged the store assigning the record's private on disk stem field from outside the class. The stamp now goes through a small setter so the assignment happens inside the record. Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> --- .../packages/core/agent_framework/_harness/_memory.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py index 0d8c99c26ac..176c5ef39b5 100644 --- a/python/packages/core/agent_framework/_harness/_memory.py +++ b/python/packages/core/agent_framework/_harness/_memory.py @@ -416,6 +416,10 @@ def storage_stem(self) -> str: """ return self._on_disk_stem or _topic_stem(self.topic, slug_override=self.slug) + def set_on_disk_stem(self, stem: str) -> None: + """Record the filename stem this record was read from or written to.""" + self._on_disk_stem = stem + def to_dict(self) -> dict[str, Any]: """Serialize the topic record for tool output. @@ -921,7 +925,7 @@ def list_topics(self, session: AgentSession, *, source_id: str) -> list[MemoryTo topic_path.read_text(encoding="utf-8"), fallback_topic=topic_path.stem.replace("-", " "), ) - record._on_disk_stem = topic_path.stem + record.set_on_disk_stem(topic_path.stem) topics.append(record) return sorted(topics, key=lambda record: (record.topic.lower(), record.updated_at)) @@ -931,7 +935,7 @@ def get_topic(self, session: AgentSession, *, source_id: str, topic: str) -> Mem if topic_path is None: raise FileNotFoundError(f"No memory topic named '{topic}' was found for this owner.") record = MemoryTopicRecord.from_markdown(topic_path.read_text(encoding="utf-8"), fallback_topic=topic) - record._on_disk_stem = topic_path.stem + record.set_on_disk_stem(topic_path.stem) return record def _writable_topic_path(self, session: AgentSession, *, source_id: str, record: MemoryTopicRecord) -> Path: @@ -968,7 +972,7 @@ def write_topic(self, session: AgentSession, record: MemoryTopicRecord, *, sourc previous_path = self._existing_topic_path(session, source_id=source_id, topic=record.topic) topic_path = self._writable_topic_path(session, source_id=source_id, record=record) _atomic_write_text(topic_path, f"{record.to_markdown()}\n") - record._on_disk_stem = topic_path.stem + record.set_on_disk_stem(topic_path.stem) if previous_path is not None and previous_path != topic_path: previous_path.unlink(missing_ok=True) From 0a7bbf53deccf282ed4c8b2d49f0ad92cce23135 Mon Sep 17 00:00:00 2001 From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> Date: Tue, 3 Nov 2026 09:32:12 +0530 Subject: [PATCH 4/4] Python: bound memory topic stems to the filename component limit A slug long enough that its digest candidate plus the .md extension overshoots the filesystem component limit made the topic unwritable, with the errno raised before any valid persisted or legacy stem was reached. The slug part of generated candidates is capped so every candidate stays storable, oversized names are skipped rather than probed, and regressions cover a long legacy lookup and a bounded write with an index stem reload. Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> --- .../core/agent_framework/_harness/_memory.py | 36 +++++++++++++--- .../core/tests/core/test_harness_memory.py | 43 +++++++++++++++++++ 2 files changed, 74 insertions(+), 5 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py index 176c5ef39b5..b00d3b7440c 100644 --- a/python/packages/core/agent_framework/_harness/_memory.py +++ b/python/packages/core/agent_framework/_harness/_memory.py @@ -143,16 +143,34 @@ def _topic_stem(topic: str, *, slug_override: str | None = None) -> str: return next(_topic_stem_candidates(topic, slug_override=slug_override)) +# Universal filename component budget: the longest candidate stem plus the +# .md extension and one separator must stay within the common 255 byte limit, +# so the readable slug is capped long before any probe touches the filesystem. +_MAX_TOPIC_SLUG_BYTES: Final[int] = 255 - len(".md") - 1 - 64 + + +def _bounded_slug_part(slug: str) -> str: + """Cap the readable slug so every candidate filename stays storable.""" + encoded = slug.encode("utf-8") + if len(encoded) <= _MAX_TOPIC_SLUG_BYTES: + return slug + return encoded[:_MAX_TOPIC_SLUG_BYTES].decode("utf-8", errors="ignore") + + def _topic_stem_candidates(topic: str, *, slug_override: str | None = None) -> Iterator[str]: """Yield the stems a topic may persist under, longest digest last. The 64-bit digest makes a collision between distinct topics unrealistic, and ``MemoryFileStore._writable_topic_path`` walks these candidates and verifies the occupant of an existing stem before reusing it, so even a - real digest collision cannot make one topic overwrite another. + real digest collision cannot make one topic overwrite another. The slug + part is capped so the longest candidate plus the .md extension stays + within the common filename component limit; probing an oversized name + would raise ENAMETOOLONG before a valid persisted or legacy stem is + reached, making long topics unreadable. """ normalized = _normalize_topic(topic) - slug = _slugify_topic(slug_override if slug_override is not None else normalized) + slug = _bounded_slug_part(_slugify_topic(slug_override if slug_override is not None else normalized)) digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest() for digest_length in (16, 32, 64): yield f"{slug}-{digest[:digest_length]}" @@ -847,6 +865,8 @@ def _existing_topic_path(self, session: AgentSession, *, source_id: str, topic: # digest collision with another topic can never answer the lookup. for candidate in _topic_stem_candidates(topic): candidate_path = topics_dir / f"{candidate}.md" + if len(candidate.encode("utf-8")) + len(".md") > 255: + continue if not candidate_path.exists(): continue record = MemoryTopicRecord.from_markdown(candidate_path.read_text(encoding="utf-8"), fallback_topic=topic) @@ -855,9 +875,13 @@ def _existing_topic_path(self, session: AgentSession, *, source_id: str, topic: # 2. An exact filename match: a legacy slug-only file from stores # written before the digest suffix, or a caller passing a persisted # stem such as an index entry slug. - exact_path = topics_dir / f"{_slugify_topic(topic)}.md" - if exact_path.exists(): - return exact_path + legacy_name = _slugify_topic(topic) + if len(legacy_name.encode("utf-8")) + len(".md") > 255: + pass # An overlong legacy name cannot exist on a limit bound filesystem. + else: + exact_path = topics_dir / f"{legacy_name}.md" + if exact_path.exists(): + return exact_path # 3. Digest-suffixed stems sharing the slug. A prefix glob alone proves # nothing (a missing topic must never read, merge into, or delete a # longer unrelated slug), so every candidate is validated against @@ -951,6 +975,8 @@ def _writable_topic_path(self, session: AgentSession, *, source_id: str, record: normalized = _normalize_topic(record.topic) for candidate in _topic_stem_candidates(record.topic, slug_override=record.slug): candidate_path = topics_dir / f"{candidate}.md" + if len(candidate.encode("utf-8")) + len(".md") > 255: + continue if not candidate_path.exists(): return candidate_path existing = MemoryTopicRecord.from_markdown( diff --git a/python/packages/core/tests/core/test_harness_memory.py b/python/packages/core/tests/core/test_harness_memory.py index 081dad57e80..c062b68afb7 100644 --- a/python/packages/core/tests/core/test_harness_memory.py +++ b/python/packages/core/tests/core/test_harness_memory.py @@ -1201,6 +1201,49 @@ def test_memory_explicit_slug_override_is_persisted_and_findable(tmp_path) -> No store.get_topic(session, source_id="src", topic="machine") +def test_memory_reads_a_legacy_topic_file_with_a_long_slug(tmp_path) -> None: + """A legacy slug only filename near the component limit stays readable. + + Lookup probes digest candidates first; without bounding, the longest + candidate for a long slug overshoots the filesystem component limit before + the legacy name is reached. + """ + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + topics_dir = store._get_topics_directory(session, source_id="src") + + legacy_slug = "a" * 240 + topics_dir.mkdir(parents=True, exist_ok=True) + (topics_dir / f"{legacy_slug}.md").write_text( + f"# {legacy_slug}\n\nUpdated: t\nSessions: -\n\n## Summary\ns\n\n## Memories\n- legacy fact\n", + encoding="utf-8", + ) + + record = store.get_topic(session, source_id="src", topic=legacy_slug) + assert record.memories == ["legacy fact"] + + +def test_memory_long_topic_writes_and_reloads_within_the_filename_limit(tmp_path) -> None: + """A topic whose slug overshoots the component limit still writes and reloads.""" + store = MemoryFileStore(tmp_path, owner_state_key="owner_id") + session = AgentSession(session_id="session-1") + session.state["owner_id"] = "alice" + + long_topic = " ".join(["engineering"] * 48) + written = MemoryTopicRecord(topic=long_topic, summary="s", memories=["kept"], updated_at="t") + store.write_topic(session, written, source_id="src") + + assert len(written.storage_stem.encode("utf-8")) + len(".md") <= 255 + assert store.get_topic(session, source_id="src", topic=long_topic).memories == ["kept"] + # Index entries link the same bounded stem and resolve after a reload. + by_stem = store.get_topic(session, source_id="src", topic=written.storage_stem) + assert by_stem.memories == ["kept"] + entries = store.rebuild_index(session, source_id="src", line_limit=10, line_length=400) + for entry in entries: + assert (store._get_topics_directory(session, source_id="src") / f"{entry.slug}.md").exists() + + def test_memory_topic_is_reloadable_by_its_default_slug(tmp_path) -> None: """A normally written topic reloads by its default slug, which callers holding record.slug rely on."""