perplexity/Timmy-time-dashboard
Archived
1
0
Fork 0
forked from Rockachopa/Timmy-time-dashboard
Code Issues Pull Requests 1 Activity

[loop-cycle-62] fix: MEMORY.md corruption and hot memory staleness (#252) (#256)

Browse Source
This commit is contained in:
hermes
2026-03-15 15:01:19 -04:00
parent 7bc355eed6
commit dd34dc064f
4 changed files with 201 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
src/timmy/memory_system.py
View File

@@ -641,6 +641,38 @@ def store_personal_fact(fact: str, agent_id: str | None = None) -> MemoryEntry:
)
def store_last_reflection(reflection: str) -> None:
"""Store the last reflection, replacing any previous one.
Uses a single row with memory_type='reflection' to avoid accumulation.
"""
if not reflection or not reflection.strip():
return
with get_connection() as conn:
# Delete previous reflections — only the latest matters
conn.execute("DELETE FROM memories WHERE memory_type = 'reflection'")
conn.execute(
"""
INSERT INTO memories
(id, content, memory_type, source, created_at)
VALUES (?, ?, 'reflection', 'system', ?)
""",
(str(uuid.uuid4()), reflection.strip(), datetime.now(UTC).isoformat()),
)
conn.commit()
logger.debug("Stored last reflection in DB")
def recall_last_reflection() -> str | None:
"""Recall the most recent reflection, or None if absent."""
with get_connection() as conn:
row = conn.execute(
"SELECT content FROM memories WHERE memory_type = 'reflection' "
"ORDER BY created_at DESC LIMIT 1"
).fetchone()
return row["content"] if row else None
# ───────────────────────────────────────────────────────────────────────────────
# Hot Memory (computed from DB instead of MEMORY.md)
# ───────────────────────────────────────────────────────────────────────────────
@@ -655,13 +687,23 @@ class HotMemory:
self._last_modified: float | None = None
def read(self, force_refresh: bool = False) -> str:
"""Read hot memory — computed view of top facts from DB."""
"""Read hot memory — computed view of top facts + last reflection from DB."""
try:
facts = recall_personal_facts()
lines = ["# Timmy Hot Memory\n"]
if facts:
lines = ["# Timmy Hot Memory\n", "## Known Facts\n"]
lines.append("## Known Facts\n")
for f in facts[:15]:
lines.append(f"- {f}")
# Include the last reflection if available
reflection = recall_last_reflection()
if reflection:
lines.append("\n## Last Reflection\n")
lines.append(reflection)
if len(lines) > 1:
return "\n".join(lines)
except Exception:
pass
@@ -707,12 +749,16 @@ class HotMemory:
new_section = f"## {section}\n\n{content}\n\n"
full_content = full_content[: match.start()] + new_section + full_content[match.end() :]
else:
# Append section before last updated line
# Append section — guard against missing prune marker
insert_point = full_content.rfind("*Prune date:")
new_section = f"## {section}\n\n{content}\n\n"
full_content = (
full_content[:insert_point] + new_section + "\n" + full_content[insert_point:]
)
if insert_point < 0:
# No prune marker — just append at end
full_content = full_content.rstrip() + "\n\n" + new_section
else:
full_content = (
full_content[:insert_point] + new_section + "\n" + full_content[insert_point:]
)
self.path.write_text(full_content)
self._content = full_content

4
src/timmy/thinking.py
View File

@@ -743,7 +743,7 @@ class ThinkingEngine:
Never modifies soul.md. Never crashes the heartbeat.
"""
try:
from timmy.memory_system import memory_system
from timmy.memory_system import store_last_reflection
ts = datetime.fromisoformat(thought.created_at)
local_ts = ts.astimezone()
@@ -754,7 +754,7 @@ class ThinkingEngine:
f"**Seed:** {thought.seed_type}\n"
f"**Thought:** {thought.content[:200]}"
)
memory_system.hot.update_section("Last Reflection", reflection)
store_last_reflection(reflection)
except Exception as exc:
logger.debug("Failed to update memory after thought: %s", exc)

137
tests/timmy/test_memory_system.py
View File

@@ -4,7 +4,12 @@ from unittest.mock import patch
import pytest
from timmy.memory_system import MemorySystem, reset_memory_system
from timmy.memory_system import (
HotMemory,
MemorySystem,
reset_memory_system,
store_last_reflection,
)
@pytest.fixture(autouse=True)
@@ -111,3 +116,133 @@ class TestReadSoul:
content = memory.read_soul()
assert content == ""
class TestHotMemoryCorruptionGuard:
"""Tests for MEMORY.md corruption prevention (#252)."""
def test_update_section_no_truncation_without_prune_marker(self, tmp_path):
"""update_section must not corrupt content when *Prune date:* marker is absent.
Regression test: rfind("*Prune date:") returning -1 caused
full_content[:-1] to chop the last character.
"""
mem_file = tmp_path / "MEMORY.md"
mem_file.write_text("# Timmy Hot Memory\n\n## Known Facts\n\n- fact one\n- fact two\n")
hot = HotMemory()
hot.path = mem_file
# Patch read() to return file content (simulating DB unavailable)
with patch.object(hot, "read", return_value=mem_file.read_text()):
hot.update_section("Last Reflection", "**Time:** 2026-03-15\n**Seed:** freeform")
result = mem_file.read_text()
# Must NOT have any truncation — all original content intact
assert "- fact one" in result
assert "- fact two" in result
assert "## Last Reflection" in result
assert "freeform" in result
# The old bug: last char would get sliced off
assert result.count("fact tw") == 1 # "fact two" not "fact tw"
def test_update_section_replaces_existing_section(self, tmp_path):
"""update_section correctly replaces an existing section."""
content = (
"# Timmy Hot Memory\n\n"
"## Known Facts\n\n- fact one\n\n"
"## Last Reflection\n\nold reflection\n\n"
)
mem_file = tmp_path / "MEMORY.md"
mem_file.write_text(content)
hot = HotMemory()
hot.path = mem_file
with patch.object(hot, "read", return_value=mem_file.read_text()):
hot.update_section("Last Reflection", "new reflection")
result = mem_file.read_text()
assert "new reflection" in result
assert "old reflection" not in result
assert "- fact one" in result
def test_update_section_rejects_empty_content(self, tmp_path):
"""update_section refuses to write empty content."""
mem_file = tmp_path / "MEMORY.md"
mem_file.write_text("# Timmy Hot Memory\n\noriginal\n")
hot = HotMemory()
hot.path = mem_file
hot.update_section("Test", "")
# File should be unchanged
assert mem_file.read_text() == "# Timmy Hot Memory\n\noriginal\n"
def test_update_section_truncates_oversized_content(self, tmp_path):
"""update_section caps content at 2000 chars to prevent bloat."""
mem_file = tmp_path / "MEMORY.md"
mem_file.write_text("# Timmy Hot Memory\n\n")
hot = HotMemory()
hot.path = mem_file
huge = "x" * 3000
with patch.object(hot, "read", return_value=mem_file.read_text()):
hot.update_section("Huge", huge)
result = mem_file.read_text()
assert "[truncated]" in result
class TestStoreLastReflection:
"""Tests for store_last_reflection / recall_last_reflection (#252)."""
def test_store_and_recall_reflection(self, tmp_path):
"""Reflection round-trips through the database."""
db_path = tmp_path / "test_memory.db"
with patch("timmy.memory_system.DB_PATH", db_path):
from timmy.memory_system import recall_last_reflection
store_last_reflection("**Time:** 2026-03-15\n**Seed:** freeform\n**Thought:** test")
result = recall_last_reflection()
assert result is not None
assert "freeform" in result
assert "test" in result
def test_store_replaces_previous_reflection(self, tmp_path):
"""Only the latest reflection is kept — no accumulation."""
db_path = tmp_path / "test_memory.db"
with patch("timmy.memory_system.DB_PATH", db_path):
from timmy.memory_system import recall_last_reflection
store_last_reflection("first reflection")
store_last_reflection("second reflection")
result = recall_last_reflection()
assert "second reflection" in result
# DB should only have one reflection row
import sqlite3
conn = sqlite3.connect(str(db_path))
count = conn.execute(
"SELECT COUNT(*) FROM memories WHERE memory_type = 'reflection'"
).fetchone()[0]
conn.close()
assert count == 1
def test_store_empty_reflection_is_noop(self, tmp_path):
"""Empty reflections are silently ignored."""
db_path = tmp_path / "test_memory.db"
with patch("timmy.memory_system.DB_PATH", db_path):
from timmy.memory_system import recall_last_reflection
store_last_reflection("")
store_last_reflection(" ")
result = recall_last_reflection()
assert result is None

33
tests/timmy/test_thinking.py
View File

@@ -447,37 +447,26 @@ async def test_think_once_graceful_without_soul(tmp_path):
@pytest.mark.asyncio
async def test_think_once_updates_memory_after_thought(tmp_path):
"""Post-hook: MEMORY.md should have a 'Last Reflection' section after thinking."""
"""Post-hook: reflection stored in DB after thinking (#252)."""
engine = _make_engine(tmp_path)
# Create a temp MEMORY.md
memory_md = tmp_path / "MEMORY.md"
memory_md.write_text(
"# Timmy Hot Memory\n\n## Current Status\n\nOperational\n\n---\n\n*Prune date: 2026-04-01*\n"
)
db_path = tmp_path / "test_memory.db"
with (
patch("timmy.thinking.HOT_MEMORY_PATH", memory_md),
patch("timmy.memory_system.HOT_MEMORY_PATH", memory_md),
patch("timmy.memory_system.DB_PATH", db_path),
patch.object(engine, "_call_agent", return_value="The swarm hums with quiet purpose."),
patch.object(engine, "_log_event"),
patch.object(engine, "_broadcast", new_callable=AsyncMock),
):
# Also redirect the HotMemory singleton's path
from timmy.memory_system import memory_system
original_path = memory_system.hot.path
memory_system.hot.path = memory_md
memory_system.hot._content = None # clear cache
try:
thought = await engine.think_once()
finally:
memory_system.hot.path = original_path
thought = await engine.think_once()
assert thought is not None
updated = memory_md.read_text()
assert "Last Reflection" in updated
assert "The swarm hums with quiet purpose" in updated
# Reflection should be in the database, not the file
from timmy.memory_system import recall_last_reflection
with patch("timmy.memory_system.DB_PATH", db_path):
reflection = recall_last_reflection()
assert reflection is not None
assert "The swarm hums with quiet purpose" in reflection
@pytest.mark.asyncio