diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py index c3f02f883b..176c5ef39b 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 @@ -11,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 @@ -129,6 +130,34 @@ def _slugify_topic(topic: str) -> str: return slug or "memory-topic" +def _topic_stem(topic: str, *, slug_override: str | None = None) -> str: + """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 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() + for digest_length in (16, 32, 64): + yield f"{slug}-{digest[:digest_length]}" + + def _timestamp(value: datetime | None = None) -> str: current = value or datetime.now(timezone.utc) return current.replace(microsecond=0).isoformat() @@ -302,7 +331,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, ) @@ -343,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, @@ -372,6 +403,22 @@ 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). + + 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 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. @@ -449,9 +496,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 +530,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 +544,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 +577,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,8 +819,64 @@ 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: - 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-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(): + return None + try: + wanted = _normalize_topic(topic) + except ValueError: + return None + # 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. 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 + if record.slug == slug: + slug_holders.append(path) + if topic == slug and len(slug_holders) == 1: + return slug_holders[0] + return None @staticmethod def _serialize_json(value: object, *, dumps: JsonDumps) -> str: @@ -813,25 +925,61 @@ 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.set_on_disk_stem(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(): + 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.set_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.""" - 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, 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._writable_topic_path(session, source_id=source_id, record=record) _atomic_write_text(topic_path, f"{record.to_markdown()}\n") + 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) 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() @@ -1563,7 +1711,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 464255d0b4..081dad57e8 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 @@ -1046,6 +1047,271 @@ 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_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")