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

feat: heartbeat memory hooks — pre-recall and post-update

Browse Source 
Wire MEMORY.md + soul.md into the thinking loop so each heartbeat
is grounded in identity and recent context, breaking repetitive loops.

Pre-hook: _load_memory_context() reads hot memory first (changes each
cycle) then soul.md (stable identity), truncated to 1500 chars.

Post-hook: _update_memory() writes a "Last Reflection" section to
MEMORY.md after each thought so the next cycle has fresh context.

soul.md is read-only from the heartbeat — never modified by it.
All hooks degrade gracefully and never crash the heartbeat.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Trip T
2026-03-11 20:54:13 -04:00
parent ea2dbdb4b5
commit 6a7875e05f
3 changed files with 233 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/timmy/memory_system.py
View File

@@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
PROJECT_ROOT = Path(__file__).parent.parent.parent
HOT_MEMORY_PATH = PROJECT_ROOT / "MEMORY.md"
VAULT_PATH = PROJECT_ROOT / "memory"
SOUL_PATH = VAULT_PATH / "self" / "soul.md"
HANDOFF_PATH = VAULT_PATH / "notes" / "last-session-handoff.md"
@@ -433,6 +434,15 @@ class MemorySystem:
return "\n".join(summary_parts) if summary_parts else ""
def read_soul(self) -> str:
"""Read soul.md — Timmy's core identity. Returns empty string if missing."""
try:
if SOUL_PATH.exists():
return SOUL_PATH.read_text()
except Exception as exc:
logger.debug("Failed to read soul.md: %s", exc)
return ""
def get_system_context(self) -> str:
"""Get full context for system prompt injection.

59
src/timmy/thinking.py
View File

@@ -26,6 +26,7 @@ from datetime import UTC, datetime
from pathlib import Path
from config import settings
from timmy.memory_system import HOT_MEMORY_PATH, SOUL_PATH
logger = logging.getLogger(__name__)
@@ -102,6 +103,8 @@ _OBSERVATION_SEEDS = [
_THINKING_PROMPT = """You are Timmy, an AI agent pondering in your own mind. This is your private thought \
thread — no one is watching. Think freely, deeply, honestly.
{memory_context}
Guidelines for richer thinking:
- Ground abstract ideas in something concrete: a recent task, an observation, a specific moment.
- Vary your metaphors — don't reuse the same imagery across thoughts.
@@ -187,8 +190,10 @@ class ThinkingEngine:
seed_type, seed_context = self._gather_seed()
continuity = self._build_continuity_context()
memory_context = self._load_memory_context()
prompt = _THINKING_PROMPT.format(
memory_context=memory_context,
seed_context=seed_context,
continuity_context=continuity,
)
@@ -206,6 +211,9 @@ class ThinkingEngine:
thought = self._store_thought(content.strip(), seed_type)
self._last_thought_id = thought.id
# Post-hook: update MEMORY.md with latest reflection
self._update_memory(thought)
# Log to swarm event system
self._log_event(thought)
@@ -271,6 +279,57 @@ class ThinkingEngine:
# ── Private helpers ──────────────────────────────────────────────────
def _load_memory_context(self) -> str:
"""Pre-hook: load MEMORY.md + soul.md for the thinking prompt.
Hot memory first (changes each cycle), soul second (stable identity).
Returns a combined string truncated to ~1500 chars.
Graceful on any failure — returns empty string.
"""
parts: list[str] = []
try:
if HOT_MEMORY_PATH.exists():
hot = HOT_MEMORY_PATH.read_text().strip()
if hot:
parts.append(hot)
except Exception as exc:
logger.debug("Failed to read MEMORY.md: %s", exc)
try:
if SOUL_PATH.exists():
soul = SOUL_PATH.read_text().strip()
if soul:
parts.append(soul)
except Exception as exc:
logger.debug("Failed to read soul.md: %s", exc)
if not parts:
return ""
combined = "\n\n---\n\n".join(parts)
if len(combined) > 1500:
combined = combined[:1500] + "\n... [truncated]"
return combined
def _update_memory(self, thought: Thought) -> None:
"""Post-hook: update MEMORY.md 'Last Reflection' section with latest thought.
Never modifies soul.md. Never crashes the heartbeat.
"""
try:
from timmy.memory_system import memory_system
ts = datetime.fromisoformat(thought.created_at)
time_str = ts.strftime("%Y-%m-%d %H:%M")
reflection = (
f"**Time:** {time_str}\n"
f"**Seed:** {thought.seed_type}\n"
f"**Thought:** {thought.content[:200]}"
)
memory_system.hot.update_section("Last Reflection", reflection)
except Exception as exc:
logger.debug("Failed to update memory after thought: %s", exc)
def _gather_seed(self) -> tuple[str, str]:
"""Pick a seed type and gather relevant context.

164
tests/timmy/test_thinking.py
View File

@@ -360,6 +360,170 @@ async def test_think_once_chains_thoughts(tmp_path):
assert t3.parent_id == t2.id
# ---------------------------------------------------------------------------
# Memory hooks (pre-recall / post-update)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_think_once_prompt_includes_memory_context(tmp_path):
"""Pre-hook: the prompt sent to _call_agent should include MEMORY.md content."""
engine = _make_engine(tmp_path)
# Create a temp MEMORY.md with recognisable content
memory_md = tmp_path / "MEMORY.md"
memory_md.write_text("# Timmy Hot Memory\n\n## Current Status\n\n**Unique-marker-alpha**\n")
captured_prompts = []
def capture_agent(prompt):
captured_prompts.append(prompt)
return "A grounded thought."
with (
patch("timmy.thinking.HOT_MEMORY_PATH", memory_md),
patch.object(engine, "_call_agent", side_effect=capture_agent),
patch.object(engine, "_log_event"),
patch.object(engine, "_update_memory"),
patch.object(engine, "_broadcast", new_callable=AsyncMock),
):
thought = await engine.think_once()
assert thought is not None
assert len(captured_prompts) == 1
assert "Unique-marker-alpha" in captured_prompts[0]
@pytest.mark.asyncio
async def test_think_once_prompt_includes_soul(tmp_path):
"""Pre-hook: the prompt should include soul.md content when it exists."""
engine = _make_engine(tmp_path)
# Create temp soul.md
soul_dir = tmp_path / "memory" / "self"
soul_dir.mkdir(parents=True)
soul_md = soul_dir / "soul.md"
soul_md.write_text("# Soul\n\nI am Timmy. Soul-marker-beta.\n")
captured_prompts = []
def capture_agent(prompt):
captured_prompts.append(prompt)
return "A soulful thought."
with (
patch("timmy.thinking.SOUL_PATH", soul_md),
patch.object(engine, "_call_agent", side_effect=capture_agent),
patch.object(engine, "_log_event"),
patch.object(engine, "_update_memory"),
patch.object(engine, "_broadcast", new_callable=AsyncMock),
):
thought = await engine.think_once()
assert thought is not None
assert len(captured_prompts) == 1
assert "Soul-marker-beta" in captured_prompts[0]
@pytest.mark.asyncio
async def test_think_once_graceful_without_soul(tmp_path):
"""Pre-hook: think_once works fine when soul.md doesn't exist."""
engine = _make_engine(tmp_path)
nonexistent = tmp_path / "no_such_soul.md"
with (
patch("timmy.thinking.SOUL_PATH", nonexistent),
patch.object(engine, "_call_agent", return_value="Still thinking."),
patch.object(engine, "_log_event"),
patch.object(engine, "_update_memory"),
patch.object(engine, "_broadcast", new_callable=AsyncMock),
):
thought = await engine.think_once()
assert thought is not None
assert thought.content == "Still thinking."
@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."""
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"
)
with (
patch("timmy.thinking.HOT_MEMORY_PATH", memory_md),
patch("timmy.memory_system.HOT_MEMORY_PATH", memory_md),
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
assert thought is not None
updated = memory_md.read_text()
assert "Last Reflection" in updated
assert "The swarm hums with quiet purpose" in updated
@pytest.mark.asyncio
async def test_think_once_never_writes_soul(tmp_path):
"""Post-hook: soul.md must never be modified by the heartbeat."""
engine = _make_engine(tmp_path)
soul_dir = tmp_path / "memory" / "self"
soul_dir.mkdir(parents=True)
soul_md = soul_dir / "soul.md"
original_content = "# Soul\n\nI am Timmy. Immutable identity.\n"
soul_md.write_text(original_content)
with (
patch("timmy.thinking.SOUL_PATH", soul_md),
patch.object(engine, "_call_agent", return_value="A deep reflection."),
patch.object(engine, "_log_event"),
patch.object(engine, "_broadcast", new_callable=AsyncMock),
):
await engine.think_once()
assert soul_md.read_text() == original_content
@pytest.mark.asyncio
async def test_think_once_memory_update_graceful_on_failure(tmp_path):
"""Post-hook: if memory update fails, thought is still stored and returned."""
engine = _make_engine(tmp_path)
# Point at a read-only path to force write failure
bad_memory = tmp_path / "readonly" / "MEMORY.md"
# Don't create the parent dir — write will fail
with (
patch("timmy.thinking.HOT_MEMORY_PATH", bad_memory),
patch.object(engine, "_call_agent", return_value="Resilient thought."),
patch.object(engine, "_log_event"),
patch.object(engine, "_broadcast", new_callable=AsyncMock),
):
thought = await engine.think_once()
assert thought is not None
assert thought.content == "Resilient thought."
assert engine.count_thoughts() == 1
# ---------------------------------------------------------------------------
# Dashboard route
# ---------------------------------------------------------------------------