diff --git a/src/timmy/memory/consolidation.py b/src/timmy/memory/consolidation.py index d3068cbc..8affca2a 100644 --- a/src/timmy/memory/consolidation.py +++ b/src/timmy/memory/consolidation.py @@ -9,7 +9,7 @@ import re from datetime import UTC, datetime from pathlib import Path -from timmy.memory.crud import recall_last_reflection, recall_personal_facts +from timmy.memory.crud import recall_last_activity_time, recall_last_reflection, recall_personal_facts from timmy.memory.db import HOT_MEMORY_PATH, VAULT_PATH logger = logging.getLogger(__name__) @@ -89,25 +89,41 @@ class HotMemory: """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.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: + if facts or reflection: + last_ts = recall_last_activity_time() + try: + updated_date = datetime.fromisoformat(last_ts).strftime("%Y-%m-%d %H:%M UTC") + except (TypeError, ValueError): + updated_date = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC") + + lines = [ + "# Timmy Hot Memory", + "", + "> Working RAM — always loaded, ~300 lines max, pruned monthly", + f"> Last updated: {updated_date}", + "", + ] + + if facts: + lines.append("## Known Facts") + lines.append("") + for f in facts[:15]: + lines.append(f"- {f}") + + if reflection: + lines.append("") + lines.append("## Last Reflection") + lines.append("") + lines.append(reflection) + return "\n".join(lines) + except Exception: logger.debug("DB context read failed, falling back to file") - # Fallback to file if DB unavailable + # Fallback to file if DB unavailable or empty if self.path.exists(): return self.path.read_text() diff --git a/src/timmy/memory/crud.py b/src/timmy/memory/crud.py index ddd740dd..1d679eba 100644 --- a/src/timmy/memory/crud.py +++ b/src/timmy/memory/crud.py @@ -393,3 +393,12 @@ def recall_last_reflection() -> str | None: "ORDER BY created_at DESC LIMIT 1" ).fetchone() return row["content"] if row else None + + +def recall_last_activity_time() -> str | None: + """Return the ISO timestamp of the most recently stored memory, or None.""" + with get_connection() as conn: + row = conn.execute( + "SELECT created_at FROM memories ORDER BY created_at DESC LIMIT 1" + ).fetchone() + return row["created_at"] if row else None diff --git a/src/timmy/memory_system.py b/src/timmy/memory_system.py index bab814b9..bdd07eee 100644 --- a/src/timmy/memory_system.py +++ b/src/timmy/memory_system.py @@ -27,6 +27,7 @@ from timmy.memory.crud import ( # noqa: F401 get_memory_context, get_memory_stats, prune_memories, + recall_last_activity_time, recall_last_reflection, recall_personal_facts, recall_personal_facts_with_ids, diff --git a/tests/timmy/test_memory_system.py b/tests/timmy/test_memory_system.py index 40590341..1d8e97b4 100644 --- a/tests/timmy/test_memory_system.py +++ b/tests/timmy/test_memory_system.py @@ -287,6 +287,148 @@ class TestJotNote: assert "body is empty" in jot_note("title", " ") +class TestHotMemoryTimestamp: + """Tests for Working RAM auto-updating timestamp (issue #10).""" + + def test_read_includes_last_updated_when_facts_exist(self, tmp_path): + """HotMemory.read() includes a 'Last updated' timestamp when DB has facts.""" + db_path = tmp_path / "memory.db" + + with ( + patch("timmy.memory.db.DB_PATH", db_path), + patch("timmy.memory.crud.get_connection") as mock_conn, + ): + import sqlite3 + from contextlib import contextmanager + + real_conn = sqlite3.connect(str(db_path)) + real_conn.row_factory = sqlite3.Row + real_conn.execute(""" + CREATE TABLE IF NOT EXISTS memories ( + id TEXT PRIMARY KEY, + content TEXT NOT NULL, + memory_type TEXT NOT NULL DEFAULT 'fact', + source TEXT NOT NULL DEFAULT 'agent', + embedding TEXT, metadata TEXT, source_hash TEXT, + agent_id TEXT, task_id TEXT, session_id TEXT, + confidence REAL NOT NULL DEFAULT 0.8, + tags TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL, + last_accessed TEXT, + access_count INTEGER NOT NULL DEFAULT 0 + ) + """) + real_conn.execute( + "INSERT INTO memories (id, content, memory_type, source, created_at) " + "VALUES ('1', 'User prefers dark mode', 'fact', 'system', '2026-03-20T10:00:00+00:00')" + ) + real_conn.commit() + + @contextmanager + def fake_get_connection(): + yield real_conn + + mock_conn.side_effect = fake_get_connection + + hot = HotMemory() + result = hot.read() + + assert "> Last updated:" in result + assert "2026-03-20" in result + assert "User prefers dark mode" in result + + def test_read_timestamp_reflects_most_recent_memory(self, tmp_path): + """The timestamp in HotMemory.read() matches the latest memory's created_at.""" + db_path = tmp_path / "memory.db" + + with patch("timmy.memory.crud.get_connection") as mock_conn: + import sqlite3 + from contextlib import contextmanager + + real_conn = sqlite3.connect(str(db_path)) + real_conn.row_factory = sqlite3.Row + real_conn.execute(""" + CREATE TABLE IF NOT EXISTS memories ( + id TEXT PRIMARY KEY, + content TEXT NOT NULL, + memory_type TEXT NOT NULL DEFAULT 'fact', + source TEXT NOT NULL DEFAULT 'agent', + embedding TEXT, metadata TEXT, source_hash TEXT, + agent_id TEXT, task_id TEXT, session_id TEXT, + confidence REAL NOT NULL DEFAULT 0.8, + tags TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL, + last_accessed TEXT, + access_count INTEGER NOT NULL DEFAULT 0 + ) + """) + # Older fact + real_conn.execute( + "INSERT INTO memories (id, content, memory_type, source, created_at) " + "VALUES ('1', 'old fact', 'fact', 'system', '2026-03-15T08:00:00+00:00')" + ) + # Newer fact — this should be reflected in the timestamp + real_conn.execute( + "INSERT INTO memories (id, content, memory_type, source, created_at) " + "VALUES ('2', 'new fact', 'fact', 'system', '2026-03-23T14:30:00+00:00')" + ) + real_conn.commit() + + @contextmanager + def fake_get_connection(): + yield real_conn + + mock_conn.side_effect = fake_get_connection + + hot = HotMemory() + result = hot.read() + + assert "2026-03-23" in result + assert "> Last updated:" in result + + def test_read_falls_back_to_file_when_db_empty(self, tmp_path): + """HotMemory.read() falls back to MEMORY.md when DB has no facts or reflections.""" + mem_file = tmp_path / "MEMORY.md" + mem_file.write_text("# Timmy Hot Memory\n\n## Current Status\n\nOperational\n") + + with patch("timmy.memory.crud.get_connection") as mock_conn: + import sqlite3 + from contextlib import contextmanager + + db_path = tmp_path / "empty.db" + real_conn = sqlite3.connect(str(db_path)) + real_conn.row_factory = sqlite3.Row + real_conn.execute(""" + CREATE TABLE IF NOT EXISTS memories ( + id TEXT PRIMARY KEY, + content TEXT NOT NULL, + memory_type TEXT NOT NULL DEFAULT 'fact', + source TEXT NOT NULL DEFAULT 'agent', + embedding TEXT, metadata TEXT, source_hash TEXT, + agent_id TEXT, task_id TEXT, session_id TEXT, + confidence REAL NOT NULL DEFAULT 0.8, + tags TEXT NOT NULL DEFAULT '[]', + created_at TEXT NOT NULL, + last_accessed TEXT, + access_count INTEGER NOT NULL DEFAULT 0 + ) + """) + real_conn.commit() + + @contextmanager + def fake_get_connection(): + yield real_conn + + mock_conn.side_effect = fake_get_connection + + hot = HotMemory() + hot.path = mem_file + result = hot.read() + + assert "Operational" in result + assert "> Last updated:" not in result + + class TestLogDecision: """Tests for log_decision() artifact tool."""