Fix pending document processing

.github/workflows/ci.yml

@@ -19,7 +19,7 @@ jobs:
           python-version: "3.12"
           cache: pip
       - name: Install test dependencies
-        run: pip install fastapi pydantic-settings sqlalchemy loguru pytest
+        run: pip install fastapi pydantic-settings sqlalchemy loguru watchdog pytest
       - name: Run regression tests
         run: python -m pytest tests -q
       - name: Parse PowerShell scripts

app/database.py

@@ -1,6 +1,5 @@
 from sqlalchemy import create_engine
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker, Session
+from sqlalchemy.orm import declarative_base, sessionmaker, Session
 from loguru import logger
 import os
 from .config import get_settings

app/main.py

@@ -154,33 +154,19 @@ async def ensure_defaults():
     async def start_file_watcher():
         await asyncio.sleep(3)  # Wait for database initialization
         try:
-            from .services.file_watcher import FileWatcherHandler
-            from .services.document_processor import DocumentProcessor
-            from watchdog.observers import Observer
-
-            db = SessionLocal()
-            try:
-                settings = get_settings(db)
-                staging_path = Path(settings.staging_folder)
-
-                # Ensure staging folder exists
-                staging_path.mkdir(parents=True, exist_ok=True)
-
-                # Create document processor and file watcher
-                processor = DocumentProcessor(db)
-                event_handler = FileWatcherHandler(processor, settings, db)
-
-                # Set up observer
-                global file_watcher
-                file_watcher = Observer()
-                file_watcher.schedule(event_handler, str(staging_path), recursive=False)
-                file_watcher.start()
-
-                print(f"📁 File watcher started monitoring: {staging_path}")
-
-            finally:
-                db.close()
-
+            from .services.file_watcher import FileWatcher
+
+            watcher = FileWatcher()
+            # FileWatcher.start also recovers files that arrived before the
+            # observer was ready. Run the initial scan off the event loop
+            # because OCR can be CPU intensive.
+            await asyncio.to_thread(watcher.start)
+            if not watcher.is_running:
+                raise RuntimeError("File watcher did not start")
+
+            global file_watcher
+            file_watcher = watcher
+            print(f"📁 File watcher started monitoring: {watcher.settings.staging_folder}")
         except Exception as e:
             print(f"⚠️  Could not start file watcher: {e}")
 

app/services/file_watcher.py

@@ -14,8 +14,7 @@
 class FileWatcherHandler(FileSystemEventHandler):
     """Handler for file system events"""
 
-    def __init__(self, processor: DocumentProcessor, settings, db: Session = None):
-        self.processor = processor
+    def __init__(self, settings):
         self.settings = settings
         # Track recently processed files to prevent duplicates
         self._recent_files = defaultdict(datetime)
@@ -59,10 +58,13 @@ def on_created(self, event):
 
         logger.info(f"New file detected: {file_path.name}")
 
-        # Process the file in background
+        # Create the processor with the same live session used for this job.
+        # Keeping a startup-time processor leaked closed SQLAlchemy sessions
+        # into later AI/settings operations.
         try:
             with SessionLocal() as db:
-                result = self.processor.process_file(file_path, db)
+                processor = DocumentProcessor(db)
+                result = processor.process_file(file_path, db)
                 if result:
                     logger.info(f"Successfully processed file: {file_path.name}")
                 else:
@@ -90,7 +92,6 @@ class FileWatcher:
     def __init__(self, db: Session = None):
         self.settings = None
         self.observer = Observer()
-        self.processor = None
         self.is_running = False
         self._initialized = False
 
@@ -100,7 +101,6 @@ def _ensure_initialized(self):
             try:
                 with SessionLocal() as db:
                     self.settings = get_settings(db)
-                    self.processor = DocumentProcessor(db)
                 self._initialized = True
             except Exception as e:
                 logger.error(f"Failed to initialize file watcher: {e}")
@@ -115,7 +115,7 @@ def start(self):
             logger.warning(f"Staging folder does not exist: {staging_path}")
             return
 
-        handler = FileWatcherHandler(self.processor, self.settings)
+        handler = FileWatcherHandler(self.settings)
         self.observer.schedule(handler, str(staging_path), recursive=False)
 
         try:
@@ -157,7 +157,8 @@ def _process_existing_files(self):
 
                         try:
                             with SessionLocal() as db:
-                                self.processor.process_file(file_path, db)
+                                processor = DocumentProcessor(db)
+                                processor.process_file(file_path, db)
                         except Exception as e:
                             logger.error(f"Failed to process existing file {file_path.name}: {e}")
 

tests/test_file_watcher.py

@@ -0,0 +1,77 @@
+import importlib
+import sys
+import types
+from pathlib import Path
+from types import SimpleNamespace
+
+
+class FakeSession:
+    def __init__(self):
+        self.active = False
+
+    def __enter__(self):
+        self.active = True
+        return self
+
+    def __exit__(self, *_args):
+        self.active = False
+
+
+def load_file_watcher(monkeypatch):
+    processed = []
+
+    class FakeProcessor:
+        def __init__(self, db):
+            assert db.active
+            self.db = db
+
+        def process_file(self, path, db):
+            assert db is self.db
+            assert db.active
+            processed.append(path)
+            return object()
+
+    processor_module = types.ModuleType("app.services.document_processor")
+    processor_module.DocumentProcessor = FakeProcessor
+    monkeypatch.setitem(sys.modules, "app.services.document_processor", processor_module)
+
+    sys.modules.pop("app.services.file_watcher", None)
+    module = importlib.import_module("app.services.file_watcher")
+    monkeypatch.setattr(module, "SessionLocal", FakeSession)
+    monkeypatch.setattr(module.time, "sleep", lambda _seconds: None)
+    return module, processed
+
+
+def test_created_file_uses_a_live_database_session(tmp_path, monkeypatch):
+    module, processed = load_file_watcher(monkeypatch)
+    uploaded_file = tmp_path / "invoice.pdf"
+    uploaded_file.write_bytes(b"document")
+    settings = SimpleNamespace(allowed_extensions_list=["pdf"])
+    handler = module.FileWatcherHandler(settings)
+
+    handler.on_created(SimpleNamespace(is_directory=False, src_path=str(uploaded_file)))
+
+    assert processed == [uploaded_file]
+
+
+def test_start_recovers_files_that_precede_the_observer(tmp_path, monkeypatch):
+    module, _processed = load_file_watcher(monkeypatch)
+
+    class FakeObserver:
+        def schedule(self, *_args, **_kwargs):
+            pass
+
+        def start(self):
+            pass
+
+    watcher = module.FileWatcher()
+    watcher.observer = FakeObserver()
+    watcher.settings = SimpleNamespace(staging_folder=str(tmp_path), allowed_extensions_list=["pdf"])
+    watcher._initialized = True
+    recovered = []
+    monkeypatch.setattr(watcher, "_process_existing_files", lambda: recovered.append(True))
+
+    watcher.start()
+
+    assert watcher.is_running is True
+    assert recovered == [True]