fix(extractors): store state in brain DB, not JSON files (closes #63)

cashew_cli.py

@@ -388,7 +388,7 @@ def cmd_ingest(args):
 
     # Initialize registry and register extractors
     data_dir = str(Path(db_path).parent)
-    registry = ExtractorRegistry(data_dir=data_dir)
+    registry = ExtractorRegistry(data_dir=data_dir, db_path=db_path)
     registry.register(ObsidianExtractor())
     registry.register(SessionExtractor())
     registry.register(MarkdownDirExtractor())

core/extractors.py

@@ -70,15 +70,18 @@ def post_ingest_hook(self, source_path: str, db_path: str, nodes_created: int):
 
 class ExtractorRegistry:
     """Registry for extractor plugins.
-    
+
     Extractors register themselves, then the orchestrator can
     discover and run them.
     """
 
-    def __init__(self, data_dir: str = "./data"):
+    def __init__(self, data_dir: str = "./data", db_path: Optional[str] = None):
         self._extractors: Dict[str, BaseExtractor] = {}
         self._state_dir = os.path.join(data_dir, "extractor_state")
         os.makedirs(self._state_dir, exist_ok=True)
+        self._db_path = db_path
+        if db_path:
+            self._ensure_state_table(db_path)
 
     def register(self, extractor: BaseExtractor):
         """Register an extractor plugin."""
@@ -240,27 +243,98 @@ def _ingest_nodes(self, nodes: List[Dict[str, Any]], db_path: str,
         conn.close()
         return created
 
+    def _ensure_state_table(self, db_path: str):
+        """Create extractor_state table if it doesn't exist."""
+        import sqlite3
+        try:
+            conn = sqlite3.connect(db_path)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS extractor_state (
+                    extractor_name TEXT PRIMARY KEY,
+                    state_json TEXT NOT NULL,
+                    updated_at TEXT NOT NULL
+                )
+            """)
+            conn.commit()
+            conn.close()
+        except Exception as e:
+            logger.error(f"Failed to create extractor_state table: {e}")
+
     def _state_path(self, name: str) -> str:
         return os.path.join(self._state_dir, f"{name}.json")
 
     def _save_state(self, name: str, state: Dict[str, Any]):
-        """Persist extractor state to JSON."""
+        """Persist extractor state. Uses DB when available, JSON otherwise."""
         if not state:
             return
+        if self._db_path:
+            import sqlite3
+            try:
+                conn = sqlite3.connect(self._db_path)
+                conn.execute("""
+                    INSERT OR REPLACE INTO extractor_state
+                        (extractor_name, state_json, updated_at)
+                    VALUES (?, ?, ?)
+                """, (name, json.dumps(state), datetime.now().isoformat()))
+                conn.commit()
+                conn.close()
+                return
+            except Exception as e:
+                logger.error(f"Failed to save state for '{name}' to DB: {e}")
+                # Fall through to JSON fallback
+        # JSON fallback (no db_path configured)
         try:
             with open(self._state_path(name), 'w') as f:
                 json.dump(state, f, indent=2)
         except (IOError, TypeError) as e:
             logger.error(f"Failed to save state for '{name}': {e}")
 
     def _load_state(self, name: str) -> Optional[Dict[str, Any]]:
-        """Load persisted extractor state."""
+        """Load persisted extractor state.
+
+        Priority order:
+          1. DB row (if db_path is configured)
+          2. Legacy JSON file shim (imported on first load, then discarded)
+        """
+        if self._db_path:
+            import sqlite3
+            try:
+                self._ensure_state_table(self._db_path)
+                conn = sqlite3.connect(self._db_path)
+                cursor = conn.cursor()
+                cursor.execute(
+                    "SELECT state_json FROM extractor_state WHERE extractor_name = ?",
+                    (name,)
+                )
+                row = cursor.fetchone()
+                conn.close()
+                if row:
+                    try:
+                        return json.loads(row[0])
+                    except json.JSONDecodeError as e:
+                        logger.error(f"Corrupt DB state for '{name}': {e}")
+                        return None
+                # No DB row yet — check for legacy JSON file and import it
+                legacy_state = self._load_legacy_json_state(name)
+                if legacy_state:
+                    logger.info(
+                        f"Importing legacy JSON state for '{name}' into DB"
+                    )
+                    self._save_state(name, legacy_state)
+                return legacy_state
+            except Exception as e:
+                logger.error(f"Failed to load state for '{name}' from DB: {e}")
+                # Fall through to JSON fallback
+        return self._load_legacy_json_state(name)
+
+    def _load_legacy_json_state(self, name: str) -> Optional[Dict[str, Any]]:
+        """Load state from legacy JSON file (backwards-compat shim)."""
         path = self._state_path(name)
         if not os.path.exists(path):
             return None
         try:
             with open(path, 'r') as f:
                 return json.load(f)
         except (IOError, json.JSONDecodeError) as e:
-            logger.error(f"Failed to load state for '{name}': {e}")
+            logger.error(f"Failed to load legacy JSON state for '{name}': {e}")
             return None

tests/test_extractors.py

@@ -4,6 +4,7 @@
 """
 
 import json
+import os
 import sqlite3
 import tempfile
 import unittest
@@ -664,14 +665,8 @@ def test_duplicate_registration(self):
         with self.assertRaises(ValueError):
             self.registry.register(extractor2)
 
-    def test_state_persistence(self):
-        """Test that extractor state is persisted."""
-        # Create a test file
-        test_file = Path(self.data_dir) / "test.md"
-        test_file.write_text("This is test content with sufficient length for extraction processing.")
-
-        # Create test database
-        db_path = Path(self.data_dir) / "test.db"
+    def _make_db(self, db_path):
+        """Create a minimal test database."""
         conn = sqlite3.connect(db_path)
         conn.execute('''
             CREATE TABLE thought_nodes (
@@ -694,21 +689,105 @@ def test_state_persistence(self):
         ''')
         conn.commit()
         conn.close()
-
+
+    def test_state_persistence_json(self):
+        """Legacy JSON state persistence (no db_path)."""
+        test_file = Path(self.data_dir) / "test.md"
+        test_file.write_text("This is test content with sufficient length for extraction processing.")
+
+        db_path = Path(self.data_dir) / "test.db"
+        self._make_db(db_path)
+
         extractor = MarkdownDirExtractor()
         self.registry.register(extractor)
-        
+
         # Run the extractor to save state
-        result = self.registry.run("markdown", str(test_file.parent), None, str(db_path))
-        
-        # Create new registry and extractor
+        self.registry.run("markdown", str(test_file.parent), None, str(db_path))
+
+        # Create new registry (no db_path -> JSON path)
         new_registry = ExtractorRegistry(self.data_dir)
         new_extractor = MarkdownDirExtractor()
         new_registry.register(new_extractor)
-
-        # State should be restored - check that the file is in processed state
+
+        self.assertIn("test.md", new_extractor._processed)
+
+    def test_state_persistence_db(self):
+        """State is stored in DB and survives a fresh registry with db_path."""
+        test_file = Path(self.data_dir) / "test.md"
+        test_file.write_text("This is test content with sufficient length for extraction processing.")
+
+        db_path = Path(self.data_dir) / "test.db"
+        self._make_db(db_path)
+
+        registry = ExtractorRegistry(self.data_dir, db_path=str(db_path))
+        registry.register(MarkdownDirExtractor())
+        registry.run("markdown", str(test_file.parent), None, str(db_path))
+
+        # State must be in DB
+        conn = sqlite3.connect(db_path)
+        row = conn.execute(
+            "SELECT state_json FROM extractor_state WHERE extractor_name = 'markdown'"
+        ).fetchone()
+        conn.close()
+        self.assertIsNotNone(row)
+        state = json.loads(row[0])
+        processed = state.get("processed", {})
+        self.assertTrue(any("test.md" in k for k in processed))
+
+        # New registry (same db_path) should restore state
+        new_registry = ExtractorRegistry(self.data_dir, db_path=str(db_path))
+        new_extractor = MarkdownDirExtractor()
+        new_registry.register(new_extractor)
         self.assertIn("test.md", new_extractor._processed)
 
+    def test_state_db_lifecycle_matches_db(self):
+        """Wiping the DB also wipes extractor state (the core bug fix)."""
+        test_file = Path(self.data_dir) / "test.md"
+        test_file.write_text("This is test content with sufficient length for extraction processing.")
+
+        db_path = Path(self.data_dir) / "test.db"
+        self._make_db(db_path)
+
+        registry = ExtractorRegistry(self.data_dir, db_path=str(db_path))
+        registry.register(MarkdownDirExtractor())
+        registry.run("markdown", str(test_file.parent), None, str(db_path))
+
+        # Simulate wiping the DB (delete file, recreate fresh)
+        os.remove(db_path)
+        self._make_db(db_path)
+
+        # Fresh registry on the wiped DB must NOT see any state
+        new_registry = ExtractorRegistry(self.data_dir, db_path=str(db_path))
+        new_extractor = MarkdownDirExtractor()
+        new_registry.register(new_extractor)
+        self.assertNotIn("test.md", new_extractor._processed)
+
+    def test_legacy_json_shim(self):
+        """Legacy JSON state is imported into DB on first load."""
+        db_path = Path(self.data_dir) / "test.db"
+        self._make_db(db_path)
+
+        # Write a legacy JSON state file directly (format matches get_state())
+        state_dir = Path(self.data_dir) / "extractor_state"
+        state_dir.mkdir(exist_ok=True)
+        legacy_state = {"processed": {"old_file.md": 12345.0}}
+        (state_dir / "markdown.json").write_text(json.dumps(legacy_state))
+
+        # Registry with db_path should pick it up and import to DB
+        registry = ExtractorRegistry(self.data_dir, db_path=str(db_path))
+        extractor = MarkdownDirExtractor()
+        registry.register(extractor)
+
+        self.assertIn("old_file.md", extractor._processed)
+
+        # And it should now be in the DB
+        conn = sqlite3.connect(db_path)
+        row = conn.execute(
+            "SELECT state_json FROM extractor_state WHERE extractor_name = 'markdown'"
+        ).fetchone()
+        conn.close()
+        self.assertIsNotNone(row)
+
 
 class TestCLIIntegration(unittest.TestCase):
     """Test CLI ingest command integration."""