#!/usr/bin/env python3
"""
Cron job registry → Brain index pipeline.
Ingests cron job definitions and recent output history.
"""
import json
import os
import sys
from pathlib import Path
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 cron events."""
init_brain()
events = query_events(event_type='knowledge_ingest', tool='cron_index', limit=10000)
if not events:
print(" No existing cron 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:
import sqlite3
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_cron_jobs():
"""Parse and ingest all cron jobs."""
init_brain()
print("Purging existing cron events...")
purged = purge_existing()
if purged:
print(f" Purged {purged} events")
# Load cron jobs
cron_path = Path.home() / '.hermes' / 'cron' / 'jobs.json'
if not cron_path.exists():
print("Error: cron/jobs.json not found")
return 0
with open(cron_path) as f:
cron_data = json.load(f)
jobs = cron_data.get('jobs', [])
print(f"\nIngesting {len(jobs)} cron jobs...")
count = 0
for job in jobs:
jid = job.get('id', '')
name = job.get('name', 'unnamed')
schedule = job.get('schedule', {})
schedule_expr = schedule.get('expr', '') if isinstance(schedule, dict) else str(schedule)
state = job.get('state', 'unknown')
enabled = job.get('enabled', False)
prompt = job.get('prompt', '') or ''
script = job.get('script', '') or ''
deliver = job.get('deliver', '') or ''
skills = job.get('skills', []) or []
last_run = job.get('last_run_at', '') or ''
next_run = job.get('next_run_at', '') or ''
last_status = job.get('last_status', '') or ''
completed = job.get('repeat', {}).get('completed', 0) if isinstance(job.get('repeat'), dict) else 0
# Build tags
tags = ['cron', 'automation', 'job']
if not enabled:
tags.append('disabled')
if deliver:
tags.append(f'deliver:{deliver}')
for skill in (skills or [])[:3]:
tags.append(f'skill:{skill}')
# Extract topic keywords from prompt
prompt_lower = prompt.lower()
if 'backup' in prompt_lower:
tags.append('backup')
if 'memory' in prompt_lower:
tags.append('memory-maintenance')
# Get recent outputs
output_dir = Path.home() / '.hermes' / 'cron' / 'output'
recent_outputs = []
if output_dir.exists():
job_outputs = list(output_dir.glob(f'{jid[:8]}*.txt'))
for ofile in sorted(job_outputs, key=os.path.getmtime, reverse=True)[:3]:
try:
content = ofile.read_text()[:200]
recent_outputs.append({
'file': ofile.name,
'preview': content.strip(),
})
except:
pass
context = {
'type': 'cron_job',
'job_id': jid,
'name': name,
'schedule': schedule_expr,
'state': state,
'enabled': enabled,
'prompt_preview': prompt[:300],
'script': script[:100] if script else '',
'deliver': deliver,
'skills': skills or [],
'last_run': last_run,
'next_run': next_run,
'last_status': last_status,
'completed_count': completed,
'recent_outputs': recent_outputs,
}
log_event(
event_type='knowledge_ingest',
tool='cron_index',
args={'job_id': jid, 'name': name},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
print(f" [{count}] {name} ({state}, {schedule_expr})")
return count
def test_queries():
"""Verify cron index with queries."""
init_brain()
queries = [
("backup hermes", "Should find backup job"),
("memory maintenance", "Should find memory job"),
("cron schedule", "Should find job definitions"),
]
print("\n=== TEST QUERIES ===")
for query, expected in queries:
results = query_events(
event_type='knowledge_ingest',
tool='cron_index',
full_text_search=query,
limit=2,
)
status = "✓" if results else "✗"
print(f" {status} '{query}' → {len(results)} results ({expected})")
if results:
for r in results[:1]:
ctx = r.get('context', {})
name = ctx.get('name', '?')
schedule = ctx.get('schedule', '?')
state = ctx.get('state', '?')
print(f" - {name} ({state}, {schedule})")
if __name__ == '__main__':
print("=" * 60)
print("Cron Job Registry → Brain Index Pipeline")
print("=" * 60)
count = ingest_cron_jobs()
print(f"\nTotal ingested: {count}")
test_queries()
print("\nDone.")