Python: make the memory markdown escape invertible for backslash runs

python/packages/core/agent_framework/_harness/_memory.py

@@ -108,19 +108,67 @@ def _normalize_memory_text(memory: str) -> str:
     return normalized
 
 
+_LEADING_BACKSLASHES_BEFORE_HASH: re.Pattern[str] = re.compile(r"^(\\+)#")
+
+# Format marker written into every topic file that uses the invertible escape
+# encoding. Files without it were written by the previous implementation,
+# which left backslash runs of two or more unchanged, and are decoded with the
+# legacy reader.
+_MEMORY_ESCAPE_MARKER = "<!-- agent-framework-memory-escape: v2 -->"
+
+
 def _escape_markdown_line(line: str) -> str:
-    """Escape a markdown line so it cannot be re-interpreted as a heading on a later read."""
-    if line.lstrip().startswith(("#", "\\#")):
-        return f"\\{line}"
+    """Escape a markdown line so it cannot be re-interpreted as a heading on a later read.
+
+    A line starting with ``#`` gets one backslash. A line that already starts
+    with backslashes before ``#`` gets one more than it had, so a reader can
+    always recover the exact original run: prepending a single backslash to an
+    escaped hash produced a two backslash prefix the old reader never stripped,
+    corrupting the round trip.
+    """
+    leading_whitespace_length = len(line) - len(line.lstrip())
+    body = line[leading_whitespace_length:]
+    match = _LEADING_BACKSLASHES_BEFORE_HASH.match(body)
+    if match:
+        run_length = len(match.group(1))
+        escaped_run = "\\" * (2 * run_length + 1)
+        return line[:leading_whitespace_length] + escaped_run + body[run_length:]
+    if body.startswith("#"):
+        return "\\" + line
     return line
 
 
 def _unescape_markdown_line(line: str) -> str:
-    """Reverse :func:`_escape_markdown_line` when reading persisted memory content."""
+    """Reverse :func:`_escape_markdown_line` when reading persisted memory content.
+
+    An odd backslash run before ``#`` is one this module wrote (``n`` originals
+    become ``n + 1``), so the reader keeps ``(run - 1) // 2`` originals. An even
+    run was not produced here and is left untouched.
+    """
+    leading_whitespace_length = len(line) - len(line.lstrip())
+    body = line[leading_whitespace_length:]
+    match = _LEADING_BACKSLASHES_BEFORE_HASH.match(body)
+    if match and len(match.group(1)) % 2 == 1:
+        original_length = (len(match.group(1)) - 1) // 2
+        original_run = "\\" * original_length
+        return line[:leading_whitespace_length] + original_run + body[len(match.group(1)) :]
+    return line
+
+
+def _unescape_markdown_line_legacy(line: str) -> str:
+    r"""Decode a line persisted by the previous escape implementation.
+
+    The previous writer prepended a single backslash to ``#`` and ``\\#``
+    lines and left every longer run verbatim, and the previous reader stripped
+    one backslash only from a body starting ``\\#``. Unversioned files must
+    keep that exact behavior: runs of two or more backslashes are original
+    content, not an encoding, and eating them would corrupt the record on load
+    and then persist the corruption on the next write.
+    """
     leading_whitespace_length = len(line) - len(line.lstrip())
     body = line[leading_whitespace_length:]
     if body.startswith("\\#"):
-        return f"{line[:leading_whitespace_length]}{body[1:]}"
+        return line[:leading_whitespace_length] + body[1:]
     return line
 
 
@@ -452,6 +500,8 @@ def to_markdown(self) -> str:
         return "\n".join([
             f"# {self.topic}",
             "",
+            _MEMORY_ESCAPE_MARKER,
+            "",
             f"Updated: {self.updated_at}",
             f"Sessions: {session_line}",
             "",
@@ -484,10 +534,20 @@ def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> M
         summary_lines: list[str] = []
         memories: list[str] = []
         current_section: str | None = None
+        # The format marker is recognized and consumed only as a standalone
+        # line in the metadata header, before any content section, so the same
+        # text inside a summary or a memory stays content, and a legacy body
+        # line that merely mentions the marker never selects v2 decoding for
+        # the whole file. Files without a header marker were written by the
+        # previous implementation and are decoded exactly as before.
+        escape_v2 = False
 
         for raw_line in markdown.splitlines():
             line = raw_line.rstrip()
             stripped = line.strip()
+            if current_section is None and stripped == _MEMORY_ESCAPE_MARKER:
+                escape_v2 = True
+                continue
             if stripped.startswith("# "):
                 topic = stripped[2:].strip()
                 current_section = None
@@ -511,10 +571,12 @@ def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> M
                 continue
             if current_section == "summary":
                 if stripped:
-                    summary_lines.append(_unescape_markdown_line(stripped))
+                    unescape = _unescape_markdown_line if escape_v2 else _unescape_markdown_line_legacy
+                    summary_lines.append(unescape(stripped))
                 continue
             if current_section == "memories" and stripped.startswith("- "):
-                memory_text = _unescape_markdown_line(stripped[2:].strip())
+                unescape = _unescape_markdown_line if escape_v2 else _unescape_markdown_line_legacy
+                memory_text = unescape(stripped[2:].strip())
                 if memory_text and memory_text != DEFAULT_MEMORY_NO_TOPICS_TEXT[2:]:
                     memories.append(memory_text)
 

python/packages/core/tests/core/test_harness_memory.py

@@ -1046,6 +1046,143 @@ def test_memory_file_store_encodes_non_ascii_owner_ids(tmp_path) -> None:
     assert roots[0] != roots[1]
 
 
+def test_memory_topic_record_round_trips_backslash_hash_lines() -> None:
+    """Lines starting with backslash hash must survive a round trip unchanged.
+
+    Escaping such a line used to add a second backslash the reader never
+    stripped, so every reload corrupted the memory a little more.
+    """
+    record = MemoryTopicRecord(
+        topic="notes",
+        summary="\\# summary line",
+        memories=["\\# literal fact", "# heading shaped memory", "\\\\# two backslashes"],
+        updated_at="2026-10-26T00:00:00+00:00",
+    )
+
+    parsed = MemoryTopicRecord.from_markdown(record.to_markdown(), fallback_topic="notes")
+
+    assert parsed.memories == ["\\# literal fact", "# heading shaped memory", "\\\\# two backslashes"]
+
+
+def test_memory_topic_record_marks_files_with_the_escape_format_version() -> None:
+    """Every written record carries the v2 marker, so a reader can tell the
+    invertible encoding from files written by the previous implementation."""
+    record = MemoryTopicRecord(
+        topic="notes",
+        summary="plain",
+        memories=["fact"],
+        updated_at="2026-10-26T00:00:00+00:00",
+    )
+    markdown = record.to_markdown()
+    assert "<!-- agent-framework-memory-escape: v2 -->" in markdown
+    parsed = MemoryTopicRecord.from_markdown(markdown, fallback_topic="notes")
+    assert parsed.summary == "plain"
+    assert parsed.memories == ["fact"]
+
+
+def test_memory_topic_record_round_trips_every_backslash_run_length() -> None:
+    """Runs from zero through five backslashes before a hash decode exactly."""
+    for run_length in range(6):
+        run = "\\" * run_length
+        record = MemoryTopicRecord(
+            topic="notes",
+            summary=f"{run}# summary",
+            memories=[f"{run}# memory"],
+            updated_at="2026-10-26T00:00:00+00:00",
+        )
+        parsed = MemoryTopicRecord.from_markdown(record.to_markdown(), fallback_topic="notes")
+        assert parsed.summary == f"{run}# summary", run_length
+        assert parsed.memories == [f"{run}# memory"], run_length
+
+
+def test_memory_topic_record_preserves_marker_shaped_summary_text() -> None:
+    """A summary whose content equals the format marker round trips as summary
+    content instead of being consumed as metadata and backfilled from a memory."""
+    record = MemoryTopicRecord(
+        topic="deploy",
+        summary="<!-- agent-framework-memory-escape: v2 -->",
+        memories=["Keep the deployment instructions."],
+        updated_at="2026-10-26T00:00:00+00:00",
+    )
+    parsed = MemoryTopicRecord.from_markdown(record.to_markdown(), fallback_topic="deploy")
+    assert parsed.summary == "<!-- agent-framework-memory-escape: v2 -->"
+    assert parsed.memories == ["Keep the deployment instructions."]
+
+
+def test_memory_topic_record_ignores_marker_text_in_legacy_bodies() -> None:
+    """A legacy memory line that merely mentions the format marker must not
+    select the new decoding for the whole file."""
+    three = chr(92) * 3
+    legacy = "\n".join([
+        "# notes",
+        "",
+        "Updated: 2026-10-26T00:00:00+00:00",
+        "Sessions: -",
+        "",
+        "## Summary",
+        "s",
+        "",
+        "## Memories",
+        "- note about <!-- agent-framework-memory-escape: v2 --> handling",
+        "- " + three + "# heading-ish note",
+    ])
+    parsed = MemoryTopicRecord.from_markdown(legacy, fallback_topic="notes")
+    assert parsed.memories == [
+        "note about <!-- agent-framework-memory-escape: v2 --> handling",
+        three + "# heading-ish note",
+    ]
+
+
+def test_memory_topic_record_preserves_odd_runs_in_unversioned_files() -> None:
+    """Odd backslash runs written by the previous implementation are verbatim
+    original content, not an encoding, and must survive a load unchanged.
+
+    The previous writer left runs of two or more backslashes untouched and its
+    reader stripped one backslash only from a body starting backslash hash;
+    treating odd runs as the new encoding would silently delete characters and
+    persist the corruption on the next write.
+    """
+    three = chr(92) * 3
+    five = chr(92) * 5
+    seven = chr(92) * 7
+    legacy = "\n".join([
+        "# notes",
+        "",
+        "Updated: 2026-10-26T00:00:00+00:00",
+        "Sessions: -",
+        "",
+        "## Summary",
+        three + "# three backslashes",
+        "",
+        "## Memories",
+        "- " + five + "# five backslashes",
+        "- " + seven + "# seven backslashes",
+    ])
+    parsed = MemoryTopicRecord.from_markdown(legacy, fallback_topic="notes")
+    assert parsed.summary == three + "# three backslashes"
+    assert parsed.memories == [five + "# five backslashes", seven + "# seven backslashes"]
+
+
+def test_memory_topic_record_keeps_single_backslash_hash_decode_for_unversioned_files() -> None:
+    """An unversioned backslash hash line decodes exactly as the previous
+    reader did: one backslash stripped, no more."""
+    legacy = "\n".join([
+        "# notes",
+        "",
+        "Updated: 2026-10-26T00:00:00+00:00",
+        "Sessions: -",
+        "",
+        "## Summary",
+        "\\# single",
+        "",
+        "## Memories",
+        "- \\# another single",
+    ])
+    parsed = MemoryTopicRecord.from_markdown(legacy, fallback_topic="notes")
+    assert parsed.summary == "# single"
+    assert parsed.memories == ["# another single"]
+
+
 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")