Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 64 additions & 3 deletions python/packages/core/agent_framework/_harness/_memory.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,21 @@
DEFAULT_MEMORY_CONSOLIDATION_INTERVAL: Final[timedelta] = timedelta(hours=24)
DEFAULT_MEMORY_INDEX_HEADER = "# MEMORY"
DEFAULT_MEMORY_NO_TOPICS_TEXT = "- none yet"

# Suffix marking a genuine memory whose exact text equals the empty-list
# placeholder, so a singleton real memory is distinguishable from a written
# placeholder line on read. Unmarked files keep their previous meaning.
_MEMORY_GENUINE_PLACEHOLDER_SUFFIX = " <!-- genuine -->"
# Header line recording that the memories in this file carry genuine markers.
# Only files containing this line decode markers, so literal marker-shaped
# text written by earlier versions keeps its exact bytes.
_MEMORY_GENUINE_MARKER_LINE = "Encoding: genuine memory markers v1"
# A placeholder followed by one or more genuine markers: the written shape of
# a genuine placeholder-valued memory (one marker) and of literal marker-shaped
# text (every marker beyond the first survives the round trip).
_MEMORY_GENUINE_MARKED_SHAPE = re.compile(
re.escape(DEFAULT_MEMORY_NO_TOPICS_TEXT[2:]) + "(?:" + re.escape(_MEMORY_GENUINE_PLACEHOLDER_SUFFIX) + ")+"
)
DEFAULT_MEMORY_EXTRACTION_PROMPT = """You extract durable memory candidates from an agent transcript delta.

Return only JSON with this exact shape:
Expand Down Expand Up @@ -124,6 +139,13 @@ def _unescape_markdown_line(line: str) -> str:
return line


def _decode_genuine_memory(memory: str) -> str:
"""Strip exactly one genuine marker from a marked memory line."""
if _MEMORY_GENUINE_MARKED_SHAPE.fullmatch(memory):
return memory[: -len(_MEMORY_GENUINE_PLACEHOLDER_SUFFIX)]
return memory


def _slugify_topic(topic: str) -> str:
slug = re.sub(r"[^a-z0-9]+", "-", _normalize_topic(topic).lower()).strip("-")
return slug or "memory-topic"
Expand Down Expand Up @@ -446,14 +468,20 @@ def to_markdown(self) -> str:
# content (or LLM-supplied consolidation output) for a section delimiter.
escaped_summary_lines = [_escape_markdown_line(line) for line in self.summary.splitlines()]
escaped_summary = "\n".join(escaped_summary_lines)
memory_lines = [f"- {_escape_markdown_line(memory)}" for memory in self.memories] or [
encoded_memories = [self._encode_memory_text(memory) for memory in self.memories]
memory_lines = [f"- {_escape_markdown_line(memory)}" for memory in encoded_memories] or [
f"- {DEFAULT_MEMORY_NO_TOPICS_TEXT[2:]}"
]
return "\n".join([
header_lines = [
f"# {self.topic}",
"",
f"Updated: {self.updated_at}",
f"Sessions: {session_line}",
]
if any(encoded != memory for encoded, memory in zip(encoded_memories, self.memories)):
header_lines.append(_MEMORY_GENUINE_MARKER_LINE)
return "\n".join([
*header_lines,
"",
"## Summary",
escaped_summary,
Expand All @@ -462,6 +490,19 @@ def to_markdown(self) -> str:
*memory_lines,
]).rstrip()

@staticmethod
def _encode_memory_text(memory: str) -> str:
"""Mark placeholder-shaped memories so the reader can tell them apart.

The write side emits the bare placeholder line only for an empty memory
list. A real memory whose text equals the placeholder, or already looks
like a marked one, gets one more marker, so the reader can strip a
single marker and still recover the exact original text.
"""
if memory == DEFAULT_MEMORY_NO_TOPICS_TEXT[2:] or _MEMORY_GENUINE_MARKED_SHAPE.fullmatch(memory):
return memory + _MEMORY_GENUINE_PLACEHOLDER_SUFFIX
return memory

@classmethod
def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> MemoryTopicRecord:
"""Parse one topic record from the canonical markdown format.
Expand All @@ -484,6 +525,7 @@ def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> M
summary_lines: list[str] = []
memories: list[str] = []
current_section: str | None = None
genuine_markers = False

for raw_line in markdown.splitlines():
line = raw_line.rstrip()
Expand All @@ -492,6 +534,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 == _MEMORY_GENUINE_MARKER_LINE:
genuine_markers = True
continue
if current_section is None and stripped.startswith("Updated: "):
updated_at = stripped.removeprefix("Updated: ").strip()
continue
Expand All @@ -515,9 +560,25 @@ def from_markdown(cls, markdown: str, *, fallback_topic: str | None = None) -> M
continue
if current_section == "memories" and stripped.startswith("- "):
memory_text = _unescape_markdown_line(stripped[2:].strip())
if memory_text and memory_text != DEFAULT_MEMORY_NO_TOPICS_TEXT[2:]:
if memory_text:
memories.append(memory_text)

# Markers decode only in files whose header records them, so literal
# marker-shaped text written by earlier versions keeps its exact bytes.
# Each marked line loses a single marker: a genuine placeholder-valued
# memory carries one, and literal marker-shaped text keeps every marker
# beyond the first. The sole-placeholder drop then applies only to a
# bare placeholder, the one shape the write side emits for an empty
# memory list.
if genuine_markers:
was_marked = [_MEMORY_GENUINE_MARKED_SHAPE.fullmatch(memory) is not None for memory in memories]
memories = [_decode_genuine_memory(memory) for memory in memories]
else:
was_marked = [False] * len(memories)
placeholder = DEFAULT_MEMORY_NO_TOPICS_TEXT[2:]
if memories == [placeholder] and not was_marked[0]:
memories = []

if topic is None:
raise ValueError("Memory topic markdown is missing a '# <topic>' heading.")
return cls(
Expand Down
124 changes: 124 additions & 0 deletions python/packages/core/tests/core/test_harness_memory.py
Original file line number Diff line number Diff line change
Expand Up @@ -1046,6 +1046,130 @@ def test_memory_file_store_encodes_non_ascii_owner_ids(tmp_path) -> None:
assert roots[0] != roots[1]


def test_memory_topic_record_keeps_a_memory_that_matches_the_placeholder(tmp_path) -> None:
"""A real memory whose text equals the empty-list placeholder must survive a
round trip; only the placeholder written for an empty memory list is dropped."""
record = MemoryTopicRecord(
topic="status",
summary="s",
memories=["none yet", "real fact"],
updated_at="2026-10-26T00:00:00+00:00",
)

parsed = MemoryTopicRecord.from_markdown(record.to_markdown(), fallback_topic="status")

assert parsed.memories == ["none yet", "real fact"]


def test_memory_topic_record_round_trips_a_genuine_singleton_placeholder_memory() -> None:
"""A real memories=[none yet] singleton must not collapse to an empty list.

The write side marks a genuine memory whose exact text equals the
placeholder with a distinguishing suffix, and the reader decodes the mark
before applying the sole-placeholder drop, so only a bare placeholder line
written for an empty list reads back empty.
"""
record = MemoryTopicRecord(
topic="notes",
summary="s",
memories=["none yet"],
updated_at="2026-10-26T00:00:00+00:00",
)
parsed = MemoryTopicRecord.from_markdown(record.to_markdown(), fallback_topic="notes")
assert parsed.memories == ["none yet"]


def test_memory_topic_record_marks_genuine_placeholder_on_write() -> None:
"""The written form of a genuine placeholder-valued memory carries the
distinguishing suffix, so it cannot be confused with a written placeholder."""
record = MemoryTopicRecord(
topic="notes",
summary="s",
memories=["none yet", "real fact"],
updated_at="2026-10-26T00:00:00+00:00",
)
markdown = record.to_markdown()
assert "- none yet <!-- genuine -->" in markdown
assert "- real fact" in markdown


def test_memory_topic_record_keeps_unmarked_placeholder_behavior_for_legacy_files() -> None:
"""Files written before the marker keep their previous meaning: a sole bare
placeholder line still reads back as an empty memory list."""
legacy = "\n".join([
"# notes",
"",
"Updated: 2026-10-26T00:00:00+00:00",
"Sessions: -",
"",
"## Summary",
"s",
"",
"## Memories",
"- none yet",
])
parsed = MemoryTopicRecord.from_markdown(legacy, fallback_topic="notes")
assert parsed.memories == []


def test_memory_topic_record_still_drops_the_placeholder_for_an_empty_list() -> None:
"""An empty memory list writes the placeholder line and reads back empty."""
record = MemoryTopicRecord(topic="empty", summary="s", memories=[], updated_at="2026-10-26T00:00:00+00:00")

parsed = MemoryTopicRecord.from_markdown(record.to_markdown(), fallback_topic="empty")

assert parsed.memories == []


def test_memory_topic_record_round_trips_literal_marker_shaped_text() -> None:
"""A memory whose literal text equals the genuine marker must survive a
round trip instead of collapsing onto the placeholder value."""
record = MemoryTopicRecord(
topic="notes",
summary="s",
memories=["none yet <!-- genuine -->"],
updated_at="2026-10-26T00:00:00+00:00",
)

parsed = MemoryTopicRecord.from_markdown(record.to_markdown(), fallback_topic="notes")

assert parsed.memories == ["none yet <!-- genuine -->"]


def test_memory_topic_record_keeps_the_placeholder_value_and_the_literal_marker_apart() -> None:
"""A genuine placeholder-valued memory and the literal marker text must
round trip as two distinct memories rather than merging into one."""
record = MemoryTopicRecord(
topic="notes",
summary="s",
memories=["none yet", "none yet <!-- genuine -->"],
updated_at="2026-10-26T00:00:00+00:00",
)

parsed = MemoryTopicRecord.from_markdown(record.to_markdown(), fallback_topic="notes")

assert parsed.memories == ["none yet", "none yet <!-- genuine -->"]


def test_memory_topic_record_preserves_literal_marker_text_in_legacy_files() -> None:
"""Files written before the marker keep literal marker-shaped text verbatim,
exactly as the base reader reads them."""
legacy = "\n".join([
"# notes",
"",
"Updated: 2026-10-26T00:00:00+00:00",
"Sessions: -",
"",
"## Summary",
"s",
"",
"## Memories",
"- none yet <!-- genuine -->",
])
parsed = MemoryTopicRecord.from_markdown(legacy, fallback_topic="notes")
assert parsed.memories == ["none yet <!-- genuine -->"]


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")
Expand Down
Loading