#!/usr/bin/env python3 """ sunaiva_session_capture.py — Grov-style Stop hook for Genesis Captures session wisdom and writes to Sunaiva vault for next session injection. Reads the hook event from stdin (Claude Code Stop event JSON), extracts key entities/decisions/insights from the session summary, and persists them to the Sunaiva PostgreSQL vault. Architecture inspired by Grov (Apache 2.0): https://github.com/TonyStef/Grov """ import sys import json import os import subprocess import tempfile import psycopg2 import uuid from datetime import datetime, timezone VAULT_ID = "41f4785c-e9e0-43ac-b6e8-eedb571fba57" # Kinan's Genesis vault DB = { "host": "postgresql-genesis-u50607.vm.elestio.app", "port": 25432, "user": "postgres", "password": "CiBjh6LM7Yuqkq-jo2r7eQDw", "database": "postgres", "connect_timeout": 5, "keepalives": 1, "keepalives_idle": 10, } GENESIS_ROOT = "/mnt/e/genesis-system" def read_hook_event(): """Read the Stop hook event from stdin if available.""" try: if not sys.stdin.isatty(): raw = sys.stdin.read(4096) if raw: return json.loads(raw) except Exception: pass return {} def extract_session_entities(event: dict) -> list[dict]: """Extract key entities from the session event for vault storage.""" entities = [] now = datetime.now(timezone.utc).isoformat() # Extract from stop_hook_active or session metadata if present session_id = event.get("session_id", "unknown") transcript = event.get("transcript_path", "") # Try to extract from MEMORY.md (our current state document) memory_path = f"{GENESIS_ROOT}/memory/MEMORY.md" try: with open(memory_path) as f: memory_content = f.read() # Store a snapshot of current session state entities.append({ "name": f"session_{now[:10]}_state", "type": "session_snapshot", "context": memory_content[:500], "source": session_id, }) except Exception: pass # Record the session itself as a conversation entities.append({ "name": f"Genesis Session {now[:10]}", "type": "session", "context": f"Session completed at {now}. ID: {session_id}", "source": session_id, }) return entities def write_to_vault(entities: list[dict]) -> None: """Write extracted entities to Sunaiva vault.""" if not entities: return try: conn = psycopg2.connect(**DB) cur = conn.cursor() now = datetime.now(timezone.utc) for entity in entities: import json as _json ctx_json = _json.dumps([entity.get("context", "")[:2000]]) # Upsert entity — context column is JSONB cur.execute(""" INSERT INTO sunaiva.vault_entities (id, vault_id, name, type, context, source_conversation, first_seen, last_seen) VALUES (%s, %s, %s, %s, %s::jsonb, %s, %s, %s) ON CONFLICT DO NOTHING """, ( str(uuid.uuid4()), VAULT_ID, entity["name"][:255], entity.get("type", "fact"), ctx_json, entity.get("source", "session_capture"), now.isoformat(), now.isoformat(), )) conn.commit() conn.close() print(f"[sunaiva_capture] Stored {len(entities)} entities to vault", file=sys.stderr) except Exception as e: print(f"[sunaiva_capture] DB write failed: {e}", file=sys.stderr) def write_to_rlm(event: dict, entities: list[dict]) -> None: """Write session to genesis_rlm schema via rlm_adapter.""" try: import sys as _sys _sys.path.insert(0, GENESIS_ROOT + "/core") import rlm_adapter session_id = event.get("session_id", "unknown") now_str = datetime.now(timezone.utc).isoformat() session_summary = f"Stop-hook capture at {now_str}. Entities captured: {len(entities)}." rlm_adapter.record_session( session_id=session_id, summary=session_summary, entities_touched=len(entities), decisions_made=0, cost_usd=0.0, handoff_file=event.get("transcript_path", ""), ended=True, ) for entity in entities: eid = rlm_adapter.upsert_entity( entity_type="SESSION", name=entity["name"][:500], description=entity.get("context", "")[:2000], properties={"source_hook": "sunaiva_session_capture", "session_id": session_id}, tags=["session", entity.get("type", "fact")], surprise_score=0.5, source="sunaiva_stop_hook", ) rlm_adapter.add_event( event_type="session_entity_captured", entity_id=eid, session_id=session_id, payload={"entity_name": entity["name"], "entity_type": entity.get("type")}, ) print(f"[sunaiva_capture] RLM session recorded: {session_id}", file=sys.stderr) except Exception as e: print(f"[sunaiva_capture] RLM write failed (non-fatal): {e}", file=sys.stderr) def run_background_capture(event: dict) -> None: """Run the actual DB writes in a detached background subprocess. Writes event to a temp file then spawns a worker process. The hook itself returns in <100ms. """ try: queue_dir = "/mnt/e/genesis-system/data/hook_queue" os.makedirs(queue_dir, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S_%f") queue_file = os.path.join(queue_dir, f"session_capture_{ts}.json") with open(queue_file, "w") as f: json.dump(event, f) log_file = os.path.join(queue_dir, "sunaiva_capture.log") worker_script = f""" import sys, json, os sys.path.insert(0, '{GENESIS_ROOT}/.claude/hooks') sys.path.insert(0, '{GENESIS_ROOT}/core') queue_file = {repr(queue_file)} try: with open(queue_file) as fh: event = json.load(fh) from sunaiva_session_capture import extract_session_entities, write_to_vault, write_to_rlm entities = extract_session_entities(event) write_to_vault(entities) write_to_rlm(event, entities) except Exception as e: print(f'[bg_capture] error: {{e}}', file=sys.stderr) finally: try: os.unlink(queue_file) except Exception: pass """ subprocess.Popen( [sys.executable, "-c", worker_script], stdout=subprocess.DEVNULL, stderr=open(log_file, "a"), start_new_session=True, ) except Exception as e: print(f"[sunaiva_capture] bg spawn failed: {e}", file=sys.stderr) def main(): """Hook entry point — returns immediately, spawns background worker for DB I/O.""" if "--background" in sys.argv: # Running as background worker (legacy path, not used) event = read_hook_event() entities = extract_session_entities(event) write_to_vault(entities) write_to_rlm(event, entities) return event = read_hook_event() run_background_capture(event) # Return immediately — Claude Code does not block on Stop hooks print(json.dumps({})) if __name__ == "__main__": main()