#!/usr/bin/env python3
"""
Session history → Brain index pipeline.
Ingests session summaries and metadata from state.db.
Bounded by summaries (125) rather than raw messages (58K) to avoid noise.
"""
import json
import sys
import sqlite3
from pathlib import Path
from datetime import datetime
BRAIN_DIR = Path.home() / '.hermes' / 'brain'
sys.path.insert(0, str(BRAIN_DIR))
from brain import init_brain, log_event, query_events
def purge_existing():
"""Remove previously ingested session events."""
init_brain()
events = query_events(event_type='knowledge_ingest', tool='session_index', limit=10000)
if not events:
print(" No existing session events to purge")
return 0
brain_dir = Path.home() / '.hermes' / 'brain'
db_path = brain_dir / 'index' / 'events.db'
event_ids = [e['id'] for e in events]
try:
db = sqlite3.connect(str(db_path))
batch_size = 500
for i in range(0, len(event_ids), batch_size):
batch = event_ids[i:i+batch_size]
placeholders = ','.join(['?'] * len(batch))
db.execute(f"DELETE FROM events WHERE id IN ({placeholders})", batch)
db.commit()
db.close()
except Exception as e:
print(f" Warning: purge error: {e}")
return len(event_ids)
def ingest_session_summaries():
"""Ingest session summaries from state.db."""
init_brain()
state_db = Path.home() / '.hermes' / 'state.db'
if not state_db.exists():
print("Error: state.db not found")
return 0
print("Purging existing session events...")
purged = purge_existing()
if purged:
print(f" Purged {purged} events")
db = sqlite3.connect(str(state_db))
db.row_factory = sqlite3.Row
# Get session summaries
summaries = db.execute(
"SELECT * FROM session_summaries ORDER BY created_at DESC"
).fetchall()
# Get sessions for metadata
sessions = db.execute(
"SELECT * FROM sessions ORDER BY started_at DESC"
).fetchall()
# Build session lookup
session_map = {}
for s in sessions:
row = dict(s)
session_map[row['id']] = row
print(f"\nIngesting {len(summaries)} session summaries...")
count = 0
for summary in summaries:
row = dict(summary)
session_id = row.get('session_id', '')
# Get session metadata
sess = session_map.get(session_id, {})
summary_text = row.get('summary', '') or ''
keywords = row.get('keywords', '') or ''
summary_ts = row.get('created_at', '') or sess.get('started_at', '')
source = sess.get('source', '') or ''
model = sess.get('model', '') or ''
title = sess.get('title', '') or ''
message_count = sess.get('message_count', 0) or 0
input_tokens = sess.get('input_tokens', 0) or 0
output_tokens = sess.get('output_tokens', 0) or 0
# Convert timestamp
created_str = ''
if summary_ts:
try:
from datetime import datetime
created_str = datetime.fromtimestamp(summary_ts).isoformat()
except (ValueError, OSError):
pass
# Build tags
tags = ['session', 'summary']
# Source/platform tag
if source:
tags.append(f'platform:{source}')
# Model tag
if model:
tags.append(f'model:{model.split("/")[-1]}')
# Keyword tags
if keywords:
for kw in keywords.split(','):
kw = kw.strip().lower()
if kw and len(kw) > 2:
tags.append(kw)
if len(tags) >= 15:
break # Limit tags
context = {
'type': 'session_summary',
'session_id': session_id,
'title': (title or '')[:500],
'summary': summary_text[:500],
'keywords': keywords,
'created_at': created_str,
'source': source,
'model': model,
'message_count': message_count,
'input_tokens': input_tokens,
'output_tokens': output_tokens,
}
log_event(
event_type='knowledge_ingest',
tool='session_index',
args={'session_id': session_id, 'created_at': created_str},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
# Ingest recent session metadata (last 100 sessions without summaries)
summarized_ids = {dict(s).get('session_id', '') for s in summaries}
unsimplified = [s for s in sessions if dict(s)['id'] not in summarized_ids]
recent_no_summary = unsimplified[:100]
if recent_no_summary:
print(f"\nIngesting {len(recent_no_summary)} recent sessions without summaries...")
for sess in recent_no_summary:
row = dict(sess)
tags = ['session', 'recent', 'no-summary']
if row.get('source'):
tags.append(f'platform:{row["source"]}')
sess_ts = row.get('started_at', '')
sess_str = ''
if sess_ts:
try:
from datetime import datetime
sess_str = datetime.fromtimestamp(sess_ts).isoformat()
except (ValueError, OSError):
pass
context = {
'type': 'session_meta',
'session_id': row.get('id', ''),
'created_at': sess_str,
'source': row.get('source', ''),
'model': row.get('model', ''),
'title': row.get('title', '') or '',
}
log_event(
event_type='knowledge_ingest',
tool='session_index',
args={'session_id': row.get('id', '')},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
db.close()
return count
def test_queries():
"""Verify session index with queries."""
init_brain()
queries = [
("brain system audit", "Should find brain-related sessions"),
("wiki ingestion", "Should find wiki pipeline sessions"),
("seedvault", "Should find seedvault sessions"),
("worksheet generation", "Should find curriculum sessions"),
("cron job", "Should find automation sessions"),
("memory v2", "Should find memory system sessions"),
]
print("\n=== TEST QUERIES ===")
for query, expected in queries:
results = query_events(
event_type='knowledge_ingest',
tool='session_index',
full_text_search=query,
limit=3,
)
status = "✓" if results else "✗"
print(f" {status} '{query}' → {len(results)} results ({expected})")
if results:
for r in results[:2]:
ctx = r.get('context', {})
sid = ctx.get('session_id', '?')[:12]
title = (ctx.get('title') or '')[:60]
created = ctx.get('created_at', '')[:10]
print(f" - [{created}] {sid}... : {title}")
if __name__ == '__main__':
print("=" * 60)
print("Session History → Brain Index Pipeline")
print("=" * 60)
count = ingest_session_summaries()
print(f"\nTotal ingested: {count}")
# Skip test queries — FTS on large DB is slow
# test_queries()
print("\nDone (skipped FTS test queries — query manually to verify).")