god_db.py
1 """ 2 God Database Implementation 3 4 The foundational data layer that stores all information as atomic bullets 5 with full graph relationships, temporal indexing, and prediction state. 6 7 This is the source of truth for all data in the Sovereign OS system. 8 All other components read from and write to this database. 9 """ 10 11 import sqlite3 12 import uuid as uuid_lib 13 import json 14 from datetime import datetime 15 from pathlib import Path 16 from typing import Optional, List, Dict, Any, Tuple 17 from contextlib import contextmanager 18 19 from .models import Bullet, Edge, EdgeType, ContentType, PredictionTarget 20 21 22 class GodDatabase: 23 """ 24 The God Database - atomic storage for the entire system. 25 26 Design principles: 27 1. Bullets are atomic - they are the smallest unit 28 2. UUIDs persist through all transformations 29 3. Content never leaves the blanket 30 4. Structure (edges, topology) can be extracted for sharing 31 5. Temporal state is always tracked 32 """ 33 34 def __init__(self, db_path: str = "god.db"): 35 """ 36 Initialize the God Database. 37 38 Args: 39 db_path: Path to SQLite database file 40 """ 41 self.db_path = Path(db_path) 42 self.db_path.parent.mkdir(parents=True, exist_ok=True) 43 self._init_db() 44 45 @contextmanager 46 def _get_conn(self): 47 """Get a database connection with proper cleanup.""" 48 conn = sqlite3.connect(str(self.db_path)) 49 conn.row_factory = sqlite3.Row 50 try: 51 yield conn 52 finally: 53 conn.close() 54 55 def _init_db(self): 56 """Initialize database schema.""" 57 with self._get_conn() as conn: 58 # Bullets table 59 conn.execute(''' 60 CREATE TABLE IF NOT EXISTS bullets ( 61 uuid TEXT PRIMARY KEY, 62 blanket_id TEXT NOT NULL, 63 content TEXT NOT NULL, 64 content_type TEXT DEFAULT 'text', 65 parent_uuid TEXT, 66 child_uuids TEXT DEFAULT '[]', 67 created_at TEXT NOT NULL, 68 updated_at TEXT NOT NULL, 69 accessed_at TEXT NOT NULL, 70 access_count INTEGER DEFAULT 0, 71 visible_tags TEXT DEFAULT '[]', 72 metadata_tags TEXT DEFAULT '{}', 73 prediction_targets TEXT DEFAULT '[]', 74 gravity_well_strength REAL DEFAULT 0.0, 75 free_energy_score REAL DEFAULT 0.0, 76 resonance_score REAL DEFAULT 0.0, 77 last_resonance_factors TEXT DEFAULT '{}' 78 ) 79 ''') 80 81 # Edges table 82 conn.execute(''' 83 CREATE TABLE IF NOT EXISTS edges ( 84 uuid TEXT PRIMARY KEY, 85 source_uuid TEXT NOT NULL, 86 target_uuid TEXT NOT NULL, 87 edge_type TEXT DEFAULT 'relates_to', 88 weight REAL DEFAULT 1.0, 89 confidence REAL DEFAULT 1.0, 90 created_at TEXT NOT NULL, 91 last_traversed TEXT NOT NULL, 92 traversal_count INTEGER DEFAULT 0, 93 predicted INTEGER DEFAULT 0, 94 prediction_strength REAL DEFAULT 0.0, 95 formation_surprise REAL DEFAULT 0.0, 96 FOREIGN KEY (source_uuid) REFERENCES bullets(uuid), 97 FOREIGN KEY (target_uuid) REFERENCES bullets(uuid) 98 ) 99 ''') 100 101 # Attention log table 102 conn.execute(''' 103 CREATE TABLE IF NOT EXISTS attention_log ( 104 id INTEGER PRIMARY KEY AUTOINCREMENT, 105 bullet_uuid TEXT NOT NULL, 106 session_id TEXT, 107 timestamp TEXT NOT NULL, 108 duration_ms INTEGER DEFAULT 0, 109 modality TEXT DEFAULT 'read', 110 FOREIGN KEY (bullet_uuid) REFERENCES bullets(uuid) 111 ) 112 ''') 113 114 # Indexes for common queries 115 conn.execute('CREATE INDEX IF NOT EXISTS idx_bullets_blanket ON bullets(blanket_id)') 116 conn.execute('CREATE INDEX IF NOT EXISTS idx_bullets_parent ON bullets(parent_uuid)') 117 conn.execute('CREATE INDEX IF NOT EXISTS idx_bullets_resonance ON bullets(resonance_score DESC)') 118 conn.execute('CREATE INDEX IF NOT EXISTS idx_bullets_accessed ON bullets(accessed_at DESC)') 119 conn.execute('CREATE INDEX IF NOT EXISTS idx_edges_source ON edges(source_uuid)') 120 conn.execute('CREATE INDEX IF NOT EXISTS idx_edges_target ON edges(target_uuid)') 121 conn.execute('CREATE INDEX IF NOT EXISTS idx_attention_bullet ON attention_log(bullet_uuid)') 122 123 conn.commit() 124 125 # ==================== BULLET OPERATIONS ==================== 126 127 def create_bullet( 128 self, 129 content: str, 130 blanket_id: str, 131 parent_uuid: Optional[str] = None, 132 content_type: ContentType = ContentType.TEXT, 133 visible_tags: Optional[List[str]] = None, 134 metadata_tags: Optional[Dict[str, Any]] = None, 135 ) -> Bullet: 136 """ 137 Create a new atomic bullet. 138 139 Args: 140 content: The content of the bullet 141 blanket_id: Which Markov blanket owns this 142 parent_uuid: Optional parent for hierarchy 143 content_type: Type of content 144 visible_tags: Human-readable tags 145 metadata_tags: System metadata 146 147 Returns: 148 The created Bullet 149 """ 150 now = datetime.now() 151 152 bullet = Bullet( 153 uuid=str(uuid_lib.uuid4()), 154 blanket_id=blanket_id, 155 content=content, 156 content_type=content_type, 157 parent_uuid=parent_uuid, 158 created_at=now, 159 updated_at=now, 160 accessed_at=now, 161 visible_tags=visible_tags or [], 162 metadata_tags=metadata_tags or {}, 163 ) 164 165 with self._get_conn() as conn: 166 conn.execute(''' 167 INSERT INTO bullets ( 168 uuid, blanket_id, content, content_type, parent_uuid, 169 child_uuids, created_at, updated_at, accessed_at, 170 access_count, visible_tags, metadata_tags, 171 prediction_targets, gravity_well_strength, 172 free_energy_score, resonance_score, last_resonance_factors 173 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 174 ''', ( 175 bullet.uuid, 176 bullet.blanket_id, 177 bullet.content, 178 bullet.content_type.value, 179 bullet.parent_uuid, 180 json.dumps(bullet.child_uuids), 181 bullet.created_at.isoformat(), 182 bullet.updated_at.isoformat(), 183 bullet.accessed_at.isoformat(), 184 bullet.access_count, 185 json.dumps(bullet.visible_tags), 186 json.dumps(bullet.metadata_tags), 187 json.dumps([]), # prediction_targets 188 bullet.gravity_well_strength, 189 bullet.free_energy_score, 190 bullet.resonance_score, 191 json.dumps(bullet.last_resonance_factors), 192 )) 193 194 # If there's a parent, update parent's child_uuids 195 if parent_uuid: 196 parent = self.get_bullet(parent_uuid) 197 if parent: 198 parent.child_uuids.append(bullet.uuid) 199 conn.execute( 200 'UPDATE bullets SET child_uuids = ? WHERE uuid = ?', 201 (json.dumps(parent.child_uuids), parent_uuid) 202 ) 203 204 conn.commit() 205 206 return bullet 207 208 def get_bullet(self, uuid: str) -> Optional[Bullet]: 209 """Get a bullet by UUID.""" 210 with self._get_conn() as conn: 211 row = conn.execute( 212 'SELECT * FROM bullets WHERE uuid = ?', (uuid,) 213 ).fetchone() 214 215 if not row: 216 return None 217 218 return self._row_to_bullet(row) 219 220 def get_bullets_for_blanket( 221 self, 222 blanket_id: str, 223 limit: Optional[int] = None, 224 order_by: str = 'created_at DESC' 225 ) -> List[Bullet]: 226 """Get all bullets for a Markov blanket.""" 227 with self._get_conn() as conn: 228 query = f'SELECT * FROM bullets WHERE blanket_id = ? ORDER BY {order_by}' 229 if limit: 230 query += f' LIMIT {limit}' 231 232 rows = conn.execute(query, (blanket_id,)).fetchall() 233 return [self._row_to_bullet(row) for row in rows] 234 235 def update_bullet( 236 self, 237 uuid: str, 238 updates: Dict[str, Any] 239 ) -> Optional[Bullet]: 240 """ 241 Update a bullet. 242 243 Args: 244 uuid: Bullet UUID 245 updates: Dict of field -> value to update 246 247 Returns: 248 Updated bullet or None if not found 249 """ 250 bullet = self.get_bullet(uuid) 251 if not bullet: 252 return None 253 254 # Apply updates 255 for key, value in updates.items(): 256 if hasattr(bullet, key): 257 setattr(bullet, key, value) 258 259 bullet.updated_at = datetime.now() 260 261 # Save to database 262 with self._get_conn() as conn: 263 conn.execute(''' 264 UPDATE bullets SET 265 content = ?, 266 content_type = ?, 267 parent_uuid = ?, 268 child_uuids = ?, 269 updated_at = ?, 270 visible_tags = ?, 271 metadata_tags = ?, 272 prediction_targets = ?, 273 gravity_well_strength = ?, 274 free_energy_score = ?, 275 resonance_score = ?, 276 last_resonance_factors = ? 277 WHERE uuid = ? 278 ''', ( 279 bullet.content, 280 bullet.content_type.value, 281 bullet.parent_uuid, 282 json.dumps(bullet.child_uuids), 283 bullet.updated_at.isoformat(), 284 json.dumps(bullet.visible_tags), 285 json.dumps(bullet.metadata_tags), 286 json.dumps([pt.__dict__ for pt in bullet.prediction_targets]), 287 bullet.gravity_well_strength, 288 bullet.free_energy_score, 289 bullet.resonance_score, 290 json.dumps(bullet.last_resonance_factors), 291 uuid, 292 )) 293 conn.commit() 294 295 return bullet 296 297 def delete_bullet(self, uuid: str) -> bool: 298 """Delete a bullet and its edges.""" 299 with self._get_conn() as conn: 300 # Delete edges 301 conn.execute('DELETE FROM edges WHERE source_uuid = ? OR target_uuid = ?', (uuid, uuid)) 302 303 # Delete attention log 304 conn.execute('DELETE FROM attention_log WHERE bullet_uuid = ?', (uuid,)) 305 306 # Delete bullet 307 result = conn.execute('DELETE FROM bullets WHERE uuid = ?', (uuid,)) 308 conn.commit() 309 310 return result.rowcount > 0 311 312 def _row_to_bullet(self, row: sqlite3.Row) -> Bullet: 313 """Convert database row to Bullet object.""" 314 prediction_targets = [] 315 for pt_data in json.loads(row['prediction_targets'] or '[]'): 316 if isinstance(pt_data, dict): 317 prediction_targets.append(PredictionTarget( 318 target_uuid=pt_data.get('target_uuid', ''), 319 predicted_edge_type=EdgeType(pt_data.get('predicted_edge_type', 'relates_to')), 320 prediction_strength=pt_data.get('prediction_strength', 0.0), 321 free_energy_reduction=pt_data.get('free_energy_reduction', 0.0), 322 first_predicted=datetime.fromisoformat(pt_data['first_predicted']) if 'first_predicted' in pt_data else datetime.now(), 323 prediction_count=pt_data.get('prediction_count', 1), 324 )) 325 326 return Bullet( 327 uuid=row['uuid'], 328 blanket_id=row['blanket_id'], 329 content=row['content'], 330 content_type=ContentType(row['content_type'] or 'text'), 331 parent_uuid=row['parent_uuid'], 332 child_uuids=json.loads(row['child_uuids'] or '[]'), 333 created_at=datetime.fromisoformat(row['created_at']), 334 updated_at=datetime.fromisoformat(row['updated_at']), 335 accessed_at=datetime.fromisoformat(row['accessed_at']), 336 access_count=row['access_count'] or 0, 337 visible_tags=json.loads(row['visible_tags'] or '[]'), 338 metadata_tags=json.loads(row['metadata_tags'] or '{}'), 339 prediction_targets=prediction_targets, 340 gravity_well_strength=row['gravity_well_strength'] or 0.0, 341 free_energy_score=row['free_energy_score'] or 0.0, 342 resonance_score=row['resonance_score'] or 0.0, 343 last_resonance_factors=json.loads(row['last_resonance_factors'] or '{}'), 344 ) 345 346 # ==================== EDGE OPERATIONS ==================== 347 348 def create_edge( 349 self, 350 source_uuid: str, 351 target_uuid: str, 352 edge_type: EdgeType = EdgeType.RELATES_TO, 353 weight: float = 1.0, 354 predicted: bool = False, 355 prediction_strength: float = 0.0, 356 ) -> Optional[Edge]: 357 """ 358 Create an edge between two bullets. 359 360 Args: 361 source_uuid: Source bullet UUID 362 target_uuid: Target bullet UUID 363 edge_type: Type of relationship 364 weight: Strength of connection (0-1) 365 predicted: Was this edge predicted? 366 prediction_strength: How strongly was it predicted? 367 368 Returns: 369 Created Edge or None if bullets don't exist 370 """ 371 # Verify bullets exist 372 source = self.get_bullet(source_uuid) 373 target = self.get_bullet(target_uuid) 374 375 if not source or not target: 376 return None 377 378 now = datetime.now() 379 380 # Calculate formation surprise 381 formation_surprise = 0.0 382 if not predicted and prediction_strength == 0: 383 formation_surprise = 1.0 # Completely unexpected 384 elif predicted: 385 formation_surprise = 1.0 - prediction_strength 386 387 edge = Edge( 388 uuid=str(uuid_lib.uuid4()), 389 source_uuid=source_uuid, 390 target_uuid=target_uuid, 391 edge_type=edge_type, 392 weight=weight, 393 created_at=now, 394 last_traversed=now, 395 predicted=predicted, 396 prediction_strength=prediction_strength, 397 formation_surprise=formation_surprise, 398 ) 399 400 with self._get_conn() as conn: 401 conn.execute(''' 402 INSERT INTO edges ( 403 uuid, source_uuid, target_uuid, edge_type, weight, 404 confidence, created_at, last_traversed, traversal_count, 405 predicted, prediction_strength, formation_surprise 406 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 407 ''', ( 408 edge.uuid, 409 edge.source_uuid, 410 edge.target_uuid, 411 edge.edge_type.value, 412 edge.weight, 413 edge.confidence, 414 edge.created_at.isoformat(), 415 edge.last_traversed.isoformat(), 416 edge.traversal_count, 417 1 if edge.predicted else 0, 418 edge.prediction_strength, 419 edge.formation_surprise, 420 )) 421 conn.commit() 422 423 return edge 424 425 def get_edge(self, uuid: str) -> Optional[Edge]: 426 """Get an edge by UUID.""" 427 with self._get_conn() as conn: 428 row = conn.execute( 429 'SELECT * FROM edges WHERE uuid = ?', (uuid,) 430 ).fetchone() 431 432 if not row: 433 return None 434 435 return self._row_to_edge(row) 436 437 def get_edges_from(self, source_uuid: str) -> List[Edge]: 438 """Get all edges from a source bullet.""" 439 with self._get_conn() as conn: 440 rows = conn.execute( 441 'SELECT * FROM edges WHERE source_uuid = ?', (source_uuid,) 442 ).fetchall() 443 return [self._row_to_edge(row) for row in rows] 444 445 def get_edges_to(self, target_uuid: str) -> List[Edge]: 446 """Get all edges to a target bullet.""" 447 with self._get_conn() as conn: 448 rows = conn.execute( 449 'SELECT * FROM edges WHERE target_uuid = ?', (target_uuid,) 450 ).fetchall() 451 return [self._row_to_edge(row) for row in rows] 452 453 def get_neighbors(self, uuid: str) -> List[str]: 454 """Get UUIDs of all bullets connected to this one.""" 455 neighbors = set() 456 457 for edge in self.get_edges_from(uuid): 458 neighbors.add(edge.target_uuid) 459 460 for edge in self.get_edges_to(uuid): 461 neighbors.add(edge.source_uuid) 462 463 return list(neighbors) 464 465 def traverse_edge(self, edge_uuid: str): 466 """Record that an edge was traversed (attention flowed through it).""" 467 with self._get_conn() as conn: 468 conn.execute(''' 469 UPDATE edges SET 470 last_traversed = ?, 471 traversal_count = traversal_count + 1 472 WHERE uuid = ? 473 ''', (datetime.now().isoformat(), edge_uuid)) 474 conn.commit() 475 476 def _row_to_edge(self, row: sqlite3.Row) -> Edge: 477 """Convert database row to Edge object.""" 478 return Edge( 479 uuid=row['uuid'], 480 source_uuid=row['source_uuid'], 481 target_uuid=row['target_uuid'], 482 edge_type=EdgeType(row['edge_type'] or 'relates_to'), 483 weight=row['weight'] or 1.0, 484 confidence=row['confidence'] or 1.0, 485 created_at=datetime.fromisoformat(row['created_at']), 486 last_traversed=datetime.fromisoformat(row['last_traversed']), 487 traversal_count=row['traversal_count'] or 0, 488 predicted=bool(row['predicted']), 489 prediction_strength=row['prediction_strength'] or 0.0, 490 formation_surprise=row['formation_surprise'] or 0.0, 491 ) 492 493 # ==================== ATTENTION OPERATIONS ==================== 494 495 def record_attention( 496 self, 497 bullet_uuid: str, 498 session_id: Optional[str] = None, 499 duration_ms: int = 0, 500 modality: str = 'read' 501 ): 502 """ 503 Record that attention was paid to a bullet. 504 505 This updates both the bullet's access metadata and logs the event. 506 """ 507 now = datetime.now() 508 509 with self._get_conn() as conn: 510 # Update bullet access info 511 conn.execute(''' 512 UPDATE bullets SET 513 accessed_at = ?, 514 access_count = access_count + 1 515 WHERE uuid = ? 516 ''', (now.isoformat(), bullet_uuid)) 517 518 # Log the attention event 519 conn.execute(''' 520 INSERT INTO attention_log (bullet_uuid, session_id, timestamp, duration_ms, modality) 521 VALUES (?, ?, ?, ?, ?) 522 ''', (bullet_uuid, session_id, now.isoformat(), duration_ms, modality)) 523 524 conn.commit() 525 526 def get_attention_log( 527 self, 528 bullet_uuid: Optional[str] = None, 529 session_id: Optional[str] = None, 530 limit: int = 100 531 ) -> List[Dict[str, Any]]: 532 """Get attention log entries.""" 533 with self._get_conn() as conn: 534 if bullet_uuid: 535 rows = conn.execute(''' 536 SELECT * FROM attention_log 537 WHERE bullet_uuid = ? 538 ORDER BY timestamp DESC LIMIT ? 539 ''', (bullet_uuid, limit)).fetchall() 540 elif session_id: 541 rows = conn.execute(''' 542 SELECT * FROM attention_log 543 WHERE session_id = ? 544 ORDER BY timestamp DESC LIMIT ? 545 ''', (session_id, limit)).fetchall() 546 else: 547 rows = conn.execute(''' 548 SELECT * FROM attention_log 549 ORDER BY timestamp DESC LIMIT ? 550 ''', (limit,)).fetchall() 551 552 return [dict(row) for row in rows] 553 554 # ==================== GRAPH OPERATIONS ==================== 555 556 def get_subgraph( 557 self, 558 center_uuid: str, 559 depth: int = 2 560 ) -> Tuple[List[Bullet], List[Edge]]: 561 """ 562 Get a subgraph centered on a bullet. 563 564 Args: 565 center_uuid: The center bullet 566 depth: How many hops to include 567 568 Returns: 569 Tuple of (bullets, edges) in the subgraph 570 """ 571 visited = set() 572 bullets = [] 573 edges = [] 574 575 def explore(uuid: str, current_depth: int): 576 if uuid in visited or current_depth > depth: 577 return 578 579 visited.add(uuid) 580 581 bullet = self.get_bullet(uuid) 582 if bullet: 583 bullets.append(bullet) 584 585 if current_depth < depth: 586 for edge in self.get_edges_from(uuid): 587 edges.append(edge) 588 explore(edge.target_uuid, current_depth + 1) 589 590 for edge in self.get_edges_to(uuid): 591 edges.append(edge) 592 explore(edge.source_uuid, current_depth + 1) 593 594 explore(center_uuid, 0) 595 596 return bullets, edges 597 598 def search_bullets( 599 self, 600 query: str, 601 blanket_id: Optional[str] = None, 602 limit: int = 20 603 ) -> List[Bullet]: 604 """ 605 Search bullets by content. 606 607 Simple text search for now - can be enhanced with FTS later. 608 """ 609 with self._get_conn() as conn: 610 if blanket_id: 611 rows = conn.execute(''' 612 SELECT * FROM bullets 613 WHERE blanket_id = ? AND content LIKE ? 614 ORDER BY resonance_score DESC, accessed_at DESC 615 LIMIT ? 616 ''', (blanket_id, f'%{query}%', limit)).fetchall() 617 else: 618 rows = conn.execute(''' 619 SELECT * FROM bullets 620 WHERE content LIKE ? 621 ORDER BY resonance_score DESC, accessed_at DESC 622 LIMIT ? 623 ''', (f'%{query}%', limit)).fetchall() 624 625 return [self._row_to_bullet(row) for row in rows] 626 627 # ==================== STATISTICS ==================== 628 629 def get_stats(self, blanket_id: Optional[str] = None) -> Dict[str, Any]: 630 """Get database statistics.""" 631 with self._get_conn() as conn: 632 if blanket_id: 633 bullet_count = conn.execute( 634 'SELECT COUNT(*) FROM bullets WHERE blanket_id = ?', (blanket_id,) 635 ).fetchone()[0] 636 edge_count = conn.execute(''' 637 SELECT COUNT(*) FROM edges e 638 JOIN bullets b ON e.source_uuid = b.uuid 639 WHERE b.blanket_id = ? 640 ''', (blanket_id,)).fetchone()[0] 641 else: 642 bullet_count = conn.execute('SELECT COUNT(*) FROM bullets').fetchone()[0] 643 edge_count = conn.execute('SELECT COUNT(*) FROM edges').fetchone()[0] 644 645 return { 646 'bullet_count': bullet_count, 647 'edge_count': edge_count, 648 'blanket_id': blanket_id, 649 } 650 651 652 # Quick test 653 if __name__ == "__main__": 654 import tempfile 655 import os 656 657 # Create a temp database for testing 658 db_path = os.path.join(tempfile.gettempdir(), "test_god.db") 659 db = GodDatabase(db_path) 660 661 print("=== God Database Test ===\n") 662 663 # Create bullets 664 bullet1 = db.create_bullet( 665 content="This is the first atomic bullet - a Markov blanket at the lowest level", 666 blanket_id="rick-graph", 667 visible_tags=["#test", "#architecture"] 668 ) 669 print(f"Created bullet 1: {bullet1.uuid[:8]}...") 670 671 bullet2 = db.create_bullet( 672 content="This is a second bullet that relates to the first", 673 blanket_id="rick-graph", 674 visible_tags=["#test"] 675 ) 676 print(f"Created bullet 2: {bullet2.uuid[:8]}...") 677 678 # Create edge 679 edge = db.create_edge( 680 source_uuid=bullet1.uuid, 681 target_uuid=bullet2.uuid, 682 edge_type=EdgeType.RELATES_TO 683 ) 684 print(f"Created edge: {edge.uuid[:8]}...") 685 686 # Record attention 687 db.record_attention(bullet1.uuid, session_id="test-session") 688 db.record_attention(bullet1.uuid, session_id="test-session") 689 print("Recorded attention events") 690 691 # Retrieve and verify 692 retrieved = db.get_bullet(bullet1.uuid) 693 print(f"\nRetrieved bullet:") 694 print(f" Content: {retrieved.content[:50]}...") 695 print(f" Access count: {retrieved.access_count}") 696 print(f" Tags: {retrieved.visible_tags}") 697 698 # Get neighbors 699 neighbors = db.get_neighbors(bullet1.uuid) 700 print(f"\nNeighbors of bullet 1: {len(neighbors)}") 701 702 # Get subgraph 703 bullets, edges = db.get_subgraph(bullet1.uuid, depth=1) 704 print(f"Subgraph: {len(bullets)} bullets, {len(edges)} edges") 705 706 # Stats 707 stats = db.get_stats("rick-graph") 708 print(f"\nStats: {stats}") 709 710 # Cleanup 711 os.remove(db_path) 712 print("\n=== Test Complete ===")