fix: check memory status proactively during thought tracking
All checks were successful
Tests / lint (pull_request) Successful in 5s
Tests / test (pull_request) Successful in 51s

Adds periodic memory status checks every 50 thoughts (configurable via
thinking_memory_check_every) to prevent unmonitored memory bloat during
long thinking sessions. Follows the same interval pattern as distill
and issue-filing post-hooks.

Fixes #310

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
kimi
2026-03-18 18:14:48 -04:00
parent f5a570c56d
commit 3dc97a9580
3 changed files with 99 additions and 0 deletions

View File

@@ -244,6 +244,7 @@ class Settings(BaseSettings):
thinking_interval_seconds: int = 300 # 5 minutes between thoughts
thinking_distill_every: int = 10 # distill facts from thoughts every Nth thought
thinking_issue_every: int = 20 # file Gitea issues from thoughts every Nth thought
thinking_memory_check_every: int = 50 # check memory status every Nth thought
# ── Gitea Integration ─────────────────────────────────────────────
# Local Gitea instance for issue tracking and self-improvement.

View File

@@ -296,6 +296,9 @@ class ThinkingEngine:
thought = self._store_thought(content, seed_type)
self._last_thought_id = thought.id
# Post-hook: check memory status periodically
self._maybe_check_memory()
# Post-hook: distill facts from recent thoughts periodically
await self._maybe_distill()
@@ -515,6 +518,35 @@ class ThinkingEngine:
result = memory_write(fact.strip(), context_type="fact")
logger.info("Distilled fact: %s%s", fact[:60], result[:40])
def _maybe_check_memory(self) -> None:
"""Every N thoughts, check memory status and log it.
Prevents unmonitored memory bloat during long thinking sessions
by periodically calling get_memory_status and logging the results.
"""
try:
interval = settings.thinking_memory_check_every
if interval <= 0:
return
count = self.count_thoughts()
if count == 0 or count % interval != 0:
return
from timmy.tools_intro import get_memory_status
status = get_memory_status()
hot = status.get("tier1_hot_memory", {})
vault = status.get("tier2_vault", {})
logger.info(
"Memory status check (thought #%d): hot_memory=%d lines, vault=%d files",
count,
hot.get("line_count", 0),
vault.get("file_count", 0),
)
except Exception as exc:
logger.warning("Memory status check failed: %s", exc)
async def _maybe_distill(self) -> None:
"""Every N thoughts, extract lasting insights and store as facts."""
try:

View File

@@ -1074,3 +1074,69 @@ def test_parse_facts_invalid_json(tmp_path):
"""Totally invalid text with no JSON array should return empty list."""
engine = _make_engine(tmp_path)
assert engine._parse_facts_response("no json here at all") == []
# ---------------------------------------------------------------------------
# Memory status check
# ---------------------------------------------------------------------------
def test_maybe_check_memory_fires_at_interval(tmp_path):
"""_maybe_check_memory should call get_memory_status every N thoughts."""
engine = _make_engine(tmp_path)
# Store exactly 50 thoughts to hit the default interval
for i in range(50):
engine._store_thought(f"Thought {i}.", "freeform")
with (
patch("timmy.thinking.settings") as mock_settings,
patch(
"timmy.tools_intro.get_memory_status",
return_value={
"tier1_hot_memory": {"line_count": 42},
"tier2_vault": {"file_count": 5},
},
) as mock_status,
):
mock_settings.thinking_memory_check_every = 50
engine._maybe_check_memory()
mock_status.assert_called_once()
def test_maybe_check_memory_skips_between_intervals(tmp_path):
"""_maybe_check_memory should not fire when count is not a multiple of interval."""
engine = _make_engine(tmp_path)
# Store 30 thoughts — not a multiple of 50
for i in range(30):
engine._store_thought(f"Thought {i}.", "freeform")
with (
patch("timmy.thinking.settings") as mock_settings,
patch(
"timmy.tools_intro.get_memory_status",
) as mock_status,
):
mock_settings.thinking_memory_check_every = 50
engine._maybe_check_memory()
mock_status.assert_not_called()
def test_maybe_check_memory_graceful_on_error(tmp_path):
"""_maybe_check_memory should not crash if get_memory_status fails."""
engine = _make_engine(tmp_path)
for i in range(50):
engine._store_thought(f"Thought {i}.", "freeform")
with (
patch("timmy.thinking.settings") as mock_settings,
patch(
"timmy.tools_intro.get_memory_status",
side_effect=Exception("boom"),
),
):
mock_settings.thinking_memory_check_every = 50
# Should not raise
engine._maybe_check_memory()