#!/usr/bin/env python3
"""
Git project state → Brain index pipeline.
Ingests repository metadata, branch state, and recent commits for active repos.
"""
import json
import subprocess
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
KNOWN_REPOS = {
'hermes-agent': {
'path': '/home/vincent/.hermes/hermes-agent',
'description': 'Hermes Agent framework — AI assistant system',
'category': 'ai-agent',
},
'llama.cpp': {
'path': '/home/vincent/llama.cpp',
'description': 'LLM inference engine — local model running',
'category': 'ai-inference',
},
'freqtrade': {
'path': '/home/vincent/freqtrade',
'description': 'Crypto trading bot framework',
'category': 'trading',
},
'ComfyUI': {
'path': '/home/vincent/ComfyUI',
'description': 'AI image/video generation pipeline',
'category': 'ai-generation',
},
}
def run_git(repo_path, *args):
"""Run a git command safely."""
try:
r = subprocess.run(
['git', '-C', repo_path] + list(args),
capture_output=True, text=True, timeout=10,
)
return r.stdout.strip()
except Exception as e:
return ''
def purge_existing():
"""Remove previously ingested git events."""
init_brain()
events = query_events(event_type='knowledge_ingest', tool='git_index', limit=10000)
if not events:
print(" No existing git 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_git_repos():
"""Parse and ingest all known git repositories."""
init_brain()
print("Purging existing git events...")
purged = purge_existing()
if purged:
print(f" Purged {purged} events")
print(f"\nIngesting {len(KNOWN_REPOS)} repositories...")
count = 0
for repo_name, repo_info in KNOWN_REPOS.items():
repo_path = repo_info['path']
if not Path(repo_path).exists():
print(f" Skipping {repo_name} — path not found")
continue
# Get branch
branch = run_git(repo_path, 'branch', '--show-current')
# Get last commit
last_commit = run_git(repo_path, 'log', '-1', '--format=%h|%s|%ci')
commit_hash, commit_msg, commit_date = (last_commit.split('|') + ['', '', ''])[:3] if last_commit else ('', '', '')
# Get file count (fast)
files_out = run_git(repo_path, 'rev-list', '--count', 'HEAD')
file_count = 0
try:
file_count = int(run_git(repo_path, 'ls-tree', '-r', '--name-only', 'HEAD').count('\n') + 1)
except:
pass
# Get commit count
try:
commit_count = int(files_out) if files_out else 0
except ValueError:
commit_count = 0
# Get status (dirty check)
status = run_git(repo_path, 'status', '--porcelain')
is_dirty = bool(status)
# Get remote URL
remote = run_git(repo_path, 'remote', 'get-url', 'origin')
# Build tags
tags = [
'git',
'project',
repo_info['category'],
repo_name.replace('-', '_'),
]
if is_dirty:
tags.append('modified')
context = {
'type': 'git_repo',
'name': repo_name,
'path': repo_path,
'branch': branch,
'last_commit': commit_hash,
'last_message': commit_msg,
'last_date': commit_date,
'file_count': file_count,
'commit_count': commit_count,
'is_dirty': is_dirty,
'remote': remote,
'description': repo_info['description'],
'category': repo_info['category'],
}
log_event(
event_type='knowledge_ingest',
tool='git_index',
args={'repo': repo_name, 'branch': branch},
context=context,
tags=tags,
skip_synapse=True,
)
count += 1
print(f" [{count}] {repo_name}: {branch} ({commit_msg[:50]}...)")
return count
def test_queries():
"""Verify git index with queries."""
init_brain()
queries = [
("hermes agent", "Should find hermes-agent repo"),
("freqtrade crypto", "Should find freqtrade"),
("llama inference", "Should find llama.cpp"),
("ComfyUI image", "Should find ComfyUI"),
("git branch", "Should find repo state"),
]
print("\n=== TEST QUERIES ===")
for query, expected in queries:
results = query_events(
event_type='knowledge_ingest',
tool='git_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', '?')
branch = ctx.get('branch', '?')
dirty = " (dirty)" if ctx.get('is_dirty') else ""
print(f" - {name}@{branch}{dirty}")
if __name__ == '__main__':
print("=" * 60)
print("Git Project State → Brain Index Pipeline")
print("=" * 60)
count = ingest_git_repos()
print(f"\nTotal ingested: {count}")
test_queries()
print("\nDone.")