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

Add periodic memory status auditing to the thinking engine. Every 50
thoughts (configurable via thinking_memory_check_every), the engine now
calls get_memory_status() and logs tier metrics (hot memory lines,
semantic vectors, vault files) both to the application log and to
data/memory_audit.log for persistent tracking.

Fixes #310

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
kimi
2026-03-18 18:28:01 -04:00
parent f4106452d2
commit e3c7e17696
2 changed files with 45 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

@@ -305,6 +305,9 @@ class ThinkingEngine:
# Post-hook: check workspace for new messages from Hermes
await self._check_workspace()
# Post-hook: proactive memory status audit
self._maybe_check_memory_status()
# Post-hook: update MEMORY.md with latest reflection
self._update_memory(thought)
@@ -532,6 +535,47 @@ class ThinkingEngine:
except Exception as exc:
logger.warning("Thought distillation failed: %s", exc)
def _maybe_check_memory_status(self) -> None:
"""Every N thoughts, run a proactive memory status audit and log 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()
# Log summary at INFO level
tier1 = status.get("tier1_hot_memory", {})
tier3 = status.get("tier3_semantic", {})
hot_lines = tier1.get("line_count", "?")
vectors = tier3.get("vector_count", "?")
logger.info(
"Memory audit (thought #%d): hot_memory=%s lines, semantic=%s vectors",
count,
hot_lines,
vectors,
)
# Write to memory_audit.log for persistent tracking
audit_path = Path("data/memory_audit.log")
audit_path.parent.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now(UTC).isoformat(timespec="seconds")
with audit_path.open("a") as f:
f.write(
f"{timestamp} thought={count} "
f"hot_lines={hot_lines} "
f"vectors={vectors} "
f"vault_files={status.get('tier2_vault', {}).get('file_count', '?')}\n"
)
except Exception as exc:
logger.warning("Memory status check failed: %s", exc)
async def _maybe_file_issues(self) -> None:
"""Every N thoughts, classify recent thoughts and file Gitea issues.