#!/usr/bin/env python3
"""Ingest session history and system information into the brain."""
import sys
import os
import json
import sqlite3
from pathlib import Path
from datetime import datetime
# Add brain path
sys.path.insert(0, str(Path.home() / '.hermes' / 'brain'))
from brain import init_brain, log_event
init_brain()
total_ingested = 0
errors = 0
def ingest(name, event_type, searchable_text, tags, source, context=None):
"""Helper to ingest with error tracking."""
global total_ingested, errors
try:
if context is None:
context = {}
context['source'] = source
context['searchable'] = searchable_text[:60000] # 60KB limit
log_event(
event_type=event_type,
tags=tags,
context=context,
skip_synapse=True # Skip for bulk ingest
)
total_ingested += 1
print(f" ✓ {name}")
except Exception as e:
errors += 1
print(f" ✗ {name}: {e}")
# ============================================================
# SESSION HISTORY INGESTION
# ============================================================
def ingest_session_history():
print("\n=== Ingesting Session History ===")
db_path = Path.home() / '.hermes' / 'state.db'
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
# Get session summaries
cursor.execute("SELECT session_id, summary, keywords FROM session_summaries")
summaries = cursor.fetchall()
print(f"Found {len(summaries)} session summaries")
for session_id, summary, keywords in summaries:
# Parse keywords
kw_list = [kw.strip() for kw in (keywords or '').split(',') if kw.strip()]
# Create searchable summary
searchable = f"Session: {session_id}\n\nSummary:\n{summary or 'No summary available.'}"
ingest(
name=f"Session Summary: {session_id[:12]}",
event_type="session_summary",
searchable_text=searchable,
tags=['session', 'history'] + kw_list[:5],
source=f"state_db/session_{session_id}",
context={'session_id': session_id, 'keywords': kw_list[:10]}
)
# Get recent sessions with titles
cursor.execute("""
SELECT id, title, model, source, started_at, ended_at, message_count,
tool_call_count, input_tokens, output_tokens, estimated_cost_usd
FROM sessions
WHERE message_count > 0
ORDER BY started_at DESC
LIMIT 500
""")
sessions = cursor.fetchall()
print(f"Found {len(sessions)} sessions with messages")
# Aggregate session topics
topic_counts = {}
for session in sessions:
sid, title, model, source, started, ended, msg_count, tool_count, input_t, output_t, cost = session
if not title:
continue
# Extract topic from title
topic = title[:50].lower()
topic_counts[topic] = topic_counts.get(topic, 0) + 1
# Ingest significant sessions (more than 5 messages or have tool calls)
if msg_count > 5 or tool_count > 0:
date_str = datetime.fromtimestamp(started).strftime('%Y-%m-%d')
content = f"""Session: {title}
Date: {date_str}
Model: {model}
Source: {source}
Messages: {msg_count}, Tool calls: {tool_count}
Input tokens: {input_t}, Output tokens: {output_t}
"""
ingest(
name=f"Session: {title[:40]}",
event_type="session_record",
searchable_text=content,
tags=['session', 'conversation', date_str[:7]],
source=f"state_db/session_{sid}",
context={
'session_id': sid,
'title': title,
'model': model,
'source': source,
'date': date_str,
'message_count': msg_count,
'tool_call_count': tool_count
}
)
# Ingest topic distribution
top_topics = sorted(topic_counts.items(), key=lambda x: -x[1])[:20]
topic_content = "=== Session Topic Distribution (Top 20) ===\n\n"
for topic, count in top_topics:
topic_content += f"- {topic}: {count} sessions\n"
ingest(
name="Session Topic Distribution",
event_type="session_analytics",
searchable_text=topic_content,
tags=['session', 'analytics', 'topics'],
source="state_db/topic_distribution",
context={'top_topics': top_topics[:10]}
)
conn.close()
# ============================================================
# SYSTEM INFORMATION INGESTION
# ============================================================
def ingest_system_info():
print("\n=== Ingesting System Information ===")
# Installed tools and versions
tools = [
('python3', '--version'),
('pip3', '--version'),
('node', '--version'),
('npm', '--version'),
('git', '--version'),
('docker', '--version'),
('ffmpeg', '-version'),
('weasyprint', '--version'),
('uv', '--version'),
('rsync', '--version'),
('tar', '--version'),
('curl', '--version'),
('wget', '--version'),
]
tool_info = []
for cmd, flag in tools:
try:
import subprocess
result = subprocess.run([cmd, flag], capture_output=True, text=True, timeout=5)
version = result.stdout.split('\n')[0] or result.stderr.split('\n')[0]
path = subprocess.run(['which', cmd], capture_output=True, text=True).stdout.strip()
tool_info.append(f"{cmd}: {version} ({path})")
except Exception as e:
tool_info.append(f"{cmd}: NOT INSTALLED")
content = "=== Installed Tools & Versions ===\n\n" + "\n".join(tool_info)
ingest(
name="Installed Tools Inventory",
event_type="system_info",
searchable_text=content,
tags=['system', 'tools', 'software'],
source="system/tools_inventory",
context={'tools': tool_info}
)
# System specs
import subprocess
# CPU/GPU info
cpu_info = subprocess.run(['lscpu'], capture_output=True, text=True).stdout.split('\n')[:10]
gpu_info = subprocess.run(['nvidia-smi', '--query-gpu=name,driver_version,memory.total', '--format=csv'],
capture_output=True, text=True).stdout.strip()
# Disk info
disk_info = subprocess.run(['df', '-h', '/'], capture_output=True, text=True).stdout.strip()
# Memory info
mem_info = subprocess.run(['free', '-h'], capture_output=True, text=True).stdout.strip()
# OS info
os_info = subprocess.run(['uname', '-a'], capture_output=True, text=True).stdout.strip()
sys_content = f"""=== System Specifications ===
CPU:
{" ".join(cpu_info)}
GPU:
{gpu_info}
Memory:
{mem_info}
Disk:
{disk_info}
OS:
{os_info}
"""
ingest(
name="System Specifications",
event_type="system_info",
searchable_text=sys_content,
tags=['system', 'specs', 'hardware'],
source="system/hardware_specs",
context={'gpu': gpu_info, 'os': os_info}
)
# Docker containers
containers = subprocess.run(['docker', 'ps', '--format', '{{.Names}}\t{{.Status}}\t{{.Ports}}'],
capture_output=True, text=True).stdout.strip()
if containers:
docker_content = f"=== Running Docker Containers ===\n\n{containers}\n"
ingest(
name="Docker Containers",
event_type="system_info",
searchable_text=docker_content,
tags=['system', 'docker', 'containers'],
source="system/docker_containers",
context={'containers': containers.split('\n')[:10]}
)
# Cron jobs
cron_info = subprocess.run(['crontab', '-l'], capture_output=True, text=True).stdout.strip()
if cron_info:
cron_content = f"=== System Cron Jobs ===\n\n{cron_info}\n"
ingest(
name="System Cron Jobs",
event_type="system_info",
searchable_text=cron_content,
tags=['system', 'cron', 'scheduled'],
source="system/cron_jobs",
context={'cron_jobs': cron_info.split('\n')[:10]}
)
# Hermes config
config_path = Path.home() / '.hermes' / 'config.yaml'
if config_path.exists():
with open(config_path) as f:
config_content = f.read()[:20000] # Limit size
ingest(
name="Hermes Configuration",
event_type="system_info",
searchable_text=config_content,
tags=['system', 'hermes', 'config'],
source="hermes/config.yaml",
context={'config_path': str(config_path)}
)
# Projects inventory
projects = [
('bitcoin-trading-bot', 'Trading bot with freqtrade'),
('duck-game', 'Duck game project'),
('freqtrade', 'Freqtrade crypto trading'),
('hermes-agent', 'Hermes AI agent framework'),
('wiki', 'Knowledge base/wiki'),
]
project_info = []
for project, desc in projects:
path = Path.home() / project
if path.exists():
# Count files
file_count = len(list(path.rglob('*')))
# Check for git repo
has_git = (path / '.git').exists()
project_info.append(f"{project}: {desc} ({file_count} files, git: {has_git})")
project_content = "=== Projects Inventory ===\n\n" + "\n".join(project_info)
ingest(
name="Projects Inventory",
event_type="system_info",
searchable_text=project_content,
tags=['system', 'projects'],
source="system/projects_inventory",
context={'projects': project_info}
)
# ============================================================
# RUN INGESTION
# ============================================================
if __name__ == '__main__':
print("=== Brain Ingestion: Sessions & System ===\n")
# Ingest session history
ingest_session_history()
# Ingest system information
ingest_system_info()
# Final stats
print(f"\n=== Ingestion Complete ===")
print(f"Total ingested: {total_ingested}")
print(f"Errors: {errors}")
# Verify search
from brain import query_events
print("\n=== Verification Queries ===")
test_queries = [
("session history", "Session records"),
("system tools", "Installed tools"),
("docker container", "Docker containers"),
("project inventory", "Projects"),
("hermes config", "Configuration"),
]
for query, desc in test_queries:
results = query_events(full_text_search=query, limit=3)
status = "✓" if results else "✗"
print(f" {status} {desc}: {len(results)} matches")
print(f"\nTotal brain events: {len(query_events(limit=10000))}")