diff --git a/python/packages/core/agent_framework/_harness/_memory.py b/python/packages/core/agent_framework/_harness/_memory.py
index c3f02f883b5..feacaa87b44 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,56 @@ 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))
+
+
+# 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
+    # A cut can land right after a separator; trailing separators would pair
+    # with the digest separator into a double dash that a later slugify of the
+    # stem collapses, so the bounded slug stays canonical by dropping them.
+    bounded = encoded[:_MAX_TOPIC_SLUG_BYTES].decode("utf-8", errors="ignore")
+    return bounded.rstrip("-")
+
+
+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. 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 = _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]}"
+
+
 def _timestamp(value: datetime | None = None) -> str:
     current = value or datetime.now(timezone.utc)
     return current.replace(microsecond=0).isoformat()
@@ -302,7 +353,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,
         )
@@ -316,8 +369,25 @@ def to_pointer_line(self, *, max_length: int = DEFAULT_MEMORY_INDEX_LINE_LENGTH)
         Returns:
             One compact pointer line.
         """
-        pointer_line = f"- [{self.topic}](topics/{self.slug}.md): {self.summary}"
-        return _trim_pointer_line(pointer_line, max_length=max_length)
+        # The link destination must survive trimming: a long digest stem can
+        # push the line past the budget, and cutting the destination would
+        # strip the .md suffix and the closing parenthesis, leaving an index
+        # link that opens nothing. Trim the summary first, then the visible
+        # label, never the destination itself.
+        link_tail = f"](topics/{self.slug}.md): "
+        whole_link = f"[{self.topic}{link_tail}"
+        if len("- ") + len(whole_link) + len(self.summary) <= max_length:
+            return f"- {whole_link}{self.summary}"
+        summary_budget = max_length - len("- ") - len(whole_link) - len("...")
+        if summary_budget >= 0:
+            return f"- {whole_link}{self.summary[:summary_budget].rstrip()}..."
+        label_budget = max_length - len("- [") - len(link_tail) - len("...")
+        if label_budget > 0:
+            label = self.topic[:label_budget].rstrip()
+            return f"- [{label}{link_tail}..."
+        # Even the bare link destination busts the budget: keep the full line
+        # rather than emit a link whose destination was cut.
+        return f"- [{self.topic}{link_tail}{self.summary}"
 
     def __eq__(self, other: object) -> bool:
         """Compare two index entries by value."""
@@ -343,7 +413,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 +442,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 +535,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 +569,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 +583,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 +616,7 @@ def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> M
             raise ValueError("Memory topic markdown is missing a '# <topic>' heading.")
         return cls(
             topic=topic,
+            slug=slug,
             summary="\n".join(summary_lines).strip(),
             memories=memories,
             updated_at=updated_at,
@@ -763,8 +858,70 @@ 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 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)
+            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.
+        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
+        #    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,28 +970,84 @@ 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 len(candidate.encode("utf-8")) + len(".md") > 255:
+                continue
+            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
+            and self._path_holds_topic(previous_path, record.topic)
+        ):
+            # A resolution through a legacy slug-only file or a shared slug can
+            # name another topic's record; retiring it would silently destroy
+            # that topic, so only a file holding this very topic is retired.
+            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 or not self._path_holds_topic(topic_path, topic):
             raise FileNotFoundError(f"No memory topic named '{topic}' was found for this owner.")
         topic_path.unlink()
 
+    def _path_holds_topic(self, path: Path, topic: str) -> bool:
+        """Return whether the record in ``path`` is the given topic's own record.
+
+        Digest-stemmed resolutions already validated exact identity; this guards
+        the alias-capable resolutions (legacy slug-only files, shared slugs),
+        where the occupant may be a different topic. The comparison folds case
+        because a legacy heading may spell the topic differently than a lookup.
+        """
+        record = MemoryTopicRecord.from_markdown(path.read_text(encoding="utf-8"), fallback_topic=path.stem)
+        return _normalize_topic(record.topic).casefold() == _normalize_topic(topic).casefold()
+
     def rebuild_index(
         self,
         session: AgentSession,
@@ -1079,7 +1292,21 @@ def _store_async_lock(self, session: AgentSession, *, kind: str, key: str) -> as
             return store_lock
 
     def _topic_async_lock(self, session: AgentSession, topic: str) -> asyncio.Lock:
-        return self._store_async_lock(session, kind="topic", key=_slugify_topic(topic))
+        """Return the per topic lock, keyed on one canonical identity.
+
+        Every supported alias of a topic (its name, its readable slug, or the
+        persisted stem an index link carries) must resolve to the same lock, or
+        a write through one alias can interleave with a consolidation holding
+        another alias's lock and be overwritten by the stale result. The
+        persisted stem is that identity: it names the one file that holds the
+        topic. A name with no file yet locks on its slugified name, which every
+        concurrent creator of that topic also computes.
+        """
+        try:
+            record = self.store.get_topic(session, source_id=self.source_id, topic=topic)
+        except FileNotFoundError:
+            return self._store_async_lock(session, kind="topic", key=_slugify_topic(topic))
+        return self._store_async_lock(session, kind="topic", key=record.storage_stem)
 
     def _state_async_lock(self, session: AgentSession) -> asyncio.Lock:
         return self._store_async_lock(session, kind="state", key="maintenance")
@@ -1561,9 +1788,15 @@ async def _run_consolidation(
 
         success_count = 0
         for record in topic_records:
-            async with self._topic_async_lock(session, record.slug):
+            async with self._topic_async_lock(session, record.storage_stem or record.slug):
                 try:
-                    current_record = self.store.get_topic(session, source_id=self.source_id, topic=record.slug)
+                    # Reload by the persisted stem: it is the one name that
+                    # resolves this exact file for every record shape,
+                    # including records stored under an explicit slug, whose
+                    # topic-derived filename family would miss them.
+                    current_record = self.store.get_topic(
+                        session, source_id=self.source_id, topic=record.storage_stem or 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 464255d0b43..53f0525b17e 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,491 @@ 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_writing_a_shared_slug_name_keeps_the_other_topic(tmp_path) -> None:
+    """Writing a topic whose name shares another topic's slug must not retire it."""
+    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="cee", memories=["letter"], 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 == ["letter"]
+    with pytest.raises(FileNotFoundError):
+        store.delete_topic(session, source_id="src", topic="c2")
+    store.delete_topic(session, source_id="src", topic="c")
+    assert store.get_topic(session, source_id="src", topic="C++").memories == ["smart pointers"]
+
+
+def test_memory_writing_a_new_slug_keeps_an_unrelated_legacy_slug_file(tmp_path) -> None:
+    """A legacy slug-only file holding another topic is never retired by a write."""
+    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 / "c.md"
+    legacy.parent.mkdir(parents=True, exist_ok=True)
+    legacy.write_text(
+        "# C++\n\nUpdated: t\nSessions: -\n\n## Summary\ncpp\n\n## Memories\n- smart pointers\n",
+        encoding="utf-8",
+    )
+
+    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 legacy.exists()
+
+
+def test_memory_topic_truncation_after_a_separator_stays_canonical(tmp_path) -> None:
+    """A slug cut right after a separator must not produce a double dash stem."""
+    store = MemoryFileStore(tmp_path, owner_state_key="owner_id")
+    session = AgentSession(session_id="session-1")
+    session.state["owner_id"] = "alice"
+
+    topic = "a" * 186 + " " + "b" * 10
+    written = MemoryTopicRecord(topic=topic, summary="s", memories=["kept"], updated_at="t")
+    store.write_topic(session, written, source_id="src")
+
+    assert "--" not in written.storage_stem
+    assert store.get_topic(session, source_id="src", topic=topic).memories == ["kept"]
+    entries = store.rebuild_index(session, source_id="src", line_limit=10, line_length=200)
+    for entry in entries:
+        assert (store._get_topics_directory(session, source_id="src") / f"{entry.slug}.md").exists()
+        assert store.get_topic(session, source_id="src", topic=entry.slug).memories == ["kept"]
+
+
+def test_memory_index_pointer_lines_keep_the_full_link_for_long_names(tmp_path) -> None:
+    """Trimming a long pointer line shortens the label and summary, never the link."""
+    entry = MemoryIndexEntry(
+        topic="release " + "notes " * 18,
+        slug="release-notes-" + "f" * 16,
+        summary="quarterly planning review with the product and research teams across every region",
+        updated_at="t",
+    )
+    line = entry.to_pointer_line(max_length=150)
+    assert len(line) <= 150
+    assert line.endswith(f"](topics/{entry.slug}.md): ...") or line.endswith("...")
+    assert f"](topics/{entry.slug}.md)" in line
+    parsed_slug = line.split("(topics/", 1)[1].split(".md)", 1)[0]
+    assert parsed_slug == entry.slug
+
+
+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."""
+    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_records_stored_under_an_explicit_slug(tmp_path) -> None:
+    """Consolidation reloads by the persisted stem, so override slug records are not 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 override.", "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."])],
+    )
+    consolidate_memories = _tool_by_name(options["tools"], "consolidate_memories")
+
+    store.write_topic(
+        session,
+        MemoryTopicRecord(
+            topic="Machine Learning",
+            slug="ml-notes",
+            summary="s",
+            memories=["prefers short answers"],
+            updated_at="t",
+        ),
+        source_id=DEFAULT_MEMORY_SOURCE_ID,
+    )
+    await consolidate_memories.invoke()  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+
+    record = store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="ml-notes")
+    assert record.topic == "Machine Learning"
+    assert record.summary == "Consolidated override."
+    assert consolidation_client.calls == ["consolidate"]
+
+
+async def test_memory_consolidation_does_not_lose_a_write_through_an_index_stem(tmp_path) -> None:
+    """A write through a persisted stem alias during consolidation must survive it."""
+    session = AgentSession(session_id="session-1")
+    session.state["owner_id"] = "alice"
+    store = MemoryFileStore(tmp_path, owner_state_key="owner_id")
+    provider = MemoryContextProvider(store=store)
+    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")
+    await write_memory.invoke(arguments={"topic": "Travel Preferences", "memory": "window seat"})  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]
+    written = store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="Travel Preferences")
+    stem = written.storage_stem
+
+    original_consolidate = provider._consolidate_topic  # type: ignore[reportPrivateUsage]
+    pending_merge: dict[str, object] = {}
+
+    async def consolidate_with_a_write_in_flight(*, client, record, now):  # type: ignore[reportPrivateUsage]
+        # A write arrives through the index stem alias while consolidation is
+        # mid call. With one canonical lock identity it waits for the lock and
+        # then applies on top of the consolidated record; with alias keyed
+        # locks it would complete now and be overwritten by the stale result.
+        import asyncio as _asyncio
+
+        task = _asyncio.create_task(
+            provider._merge_memory(session=session, topic=stem, memory="urgent fact", now=now)  # type: ignore[reportPrivateUsage]
+        )
+        pending_merge["task"] = task
+        await _asyncio.sleep(0)
+        await _asyncio.sleep(0)
+        return record, True
+
+    provider._consolidate_topic = consolidate_with_a_write_in_flight  # type: ignore[reportPrivateUsage]
+    await provider._run_consolidation(  # type: ignore[reportPrivateUsage]
+        client=_MemoryHarnessClient(), session=session, force=True, now=datetime.now(timezone.utc)
+    )
+    await pending_merge["task"]  # type: ignore[misc]
+
+    final_record = store.get_topic(session, source_id=DEFAULT_MEMORY_SOURCE_ID, topic="Travel Preferences")
+    assert "urgent fact" in final_record.memories
+    assert "window seat" in final_record.memories
+    del original_consolidate
+
+
+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")