feature: Add recover stale jobs

.env.example

@@ -78,6 +78,32 @@ RETRIEVAL_TOP_K=5
 CHUNK_DURATION_SECONDS=60
 CHUNK_OVERLAP_SECONDS=15
 
+# HyDE (Hypothetical Document Embeddings) — improves retrieval by generating
+# a synthetic transcript paragraph that matches the query before embedding.
+# Disabled by default to save time & cost; enable when retrieval quality is critical.
+USE_HYDE=false
+
+# ── Cross-Encoder Reranker ────────────────────────────────────────────────────
+# After vector store retrieval, re-rank results with a cross-encoder model.
+# This significantly improves result quality but adds ~50-100ms per query.
+USE_RERANKER=True
+# Model name for the cross-encoder (must be a sentence-transformers CrossEncoder model)
+RERANKER_MODEL=cross-encoder/ms-marco-MiniLM-L-6-v2
+# How many results to fetch from vector store before reranking
+RERANKER_TOP_K=10
+
+# ── MMR Diversity ─────────────────────────────────────────────────────────────
+# Maximal Marginal Relevance — ensures diverse results instead of near-identical chunks.
+# 0 = only diversity, 1 = only relevance. 0.5 is a good balance.
+USE_MMR=True
+MMR_LAMBDA=0.5
+
+# ── Multi-Query Expansion ──────────────────────────────────────────────────────
+# Generate multiple rephrasings of the question to improve recall.
+# Disabled by default: costs 1 extra LLM call + N extra vector searches per query.
+USE_MULTI_QUERY=false
+MULTI_QUERY_COUNT=3
+
 # ── Live Stream ───────────────────────────────────────────────────────────────
 LIVE_STREAM_SEGMENT_SECONDS=60
 
@@ -86,6 +112,3 @@ UI_API_BASE_URL=http://localhost:8000
 # HTTP request timeout in seconds. Increase this when using local Ollama with long videos.
 # Default: 120, Recommended for Ollama: 600 (10 minutes)
 UI_REQUEST_TIMEOUT=120
-
-# Query transformation technique
-USE_HYDE=True

.gitignore

@@ -62,4 +62,7 @@ yarn-debug.log*
 yarn-error.log*
 .next/
 .nuxt/
-dist/
+dist/
+
+// Files
+Dev-*.md

api/main.py

@@ -19,13 +19,40 @@
 from storage.database import init_db
 
 
+def _recover_stale_jobs() -> None:
+    """
+    Mark jobs that were in progress when the server last stopped as errored.
+    BackgroundTasks do not survive a restart, so these jobs can never complete.
+    """
+    from storage.database import IngestJob, SessionLocal
+
+    stale_statuses = ("queued", "ingesting", "transcribing", "indexing")
+    db = SessionLocal()
+    try:
+        stale_jobs = (
+            db.query(IngestJob).filter(IngestJob.status.in_(stale_statuses)).all()
+        )
+        for job in stale_jobs:
+            job.status = "error"
+            job.error_message = "Job interrupted by server restart — please resubmit"
+        if stale_jobs:
+            db.commit()
+            logger.warning(f"Marked {len(stale_jobs)} stale ingestion job(s) as error")
+    except Exception as e:
+        db.rollback()
+        logger.error(f"Failed to recover stale jobs: {e}")
+    finally:
+        db.close()
+
+
 @asynccontextmanager
 async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
     """Startup and shutdown events."""
     # Startup
     logger.info("Starting Streaming Video-RAG API...")
     settings.ensure_dirs()
     init_db()
+    _recover_stale_jobs()
     logger.success(
         f"API ready — LLM={settings.llm_provider.value}/{settings.llm_model}, "
         f"Whisper={settings.whisper_mode.value}/{settings.whisper_model_size}, "
@@ -83,16 +110,35 @@ def health() -> dict[str, str]:
 
 
 if __name__ == "__main__":
+    from pathlib import Path
+
     import uvicorn
 
+    # Only enable SSL if both cert files exist
+    ssl_kwargs = {}
+    if settings.api_ssl_certfile and settings.api_ssl_keyfile:
+        cert_path = Path(settings.api_ssl_certfile)
+        key_path = Path(settings.api_ssl_keyfile)
+        if cert_path.exists() and key_path.exists():
+            ssl_kwargs["ssl_certfile"] = settings.api_ssl_certfile
+            ssl_kwargs["ssl_keyfile"] = settings.api_ssl_keyfile
+            logger.info(f"SSL enabled — cert={cert_path}, key={key_path}")
+        else:
+            logger.warning(
+                f"SSL cert/key files not found at {cert_path}, {key_path}. "
+                "Running without SSL. Generate certs with: "
+                "openssl req -x509 -newkey rsa:4096 -days 365 -nodes "
+                "-keyout certs/localhost-key.pem -out certs/localhost.pem "
+                "-subj '/CN=localhost' -addext 'subjectAltName=DNS:localhost'"
+            )
+
     uvicorn.run(
         "api.main:app",
         host=settings.api_host,
         port=settings.api_port,
         reload=settings.api_reload,
         http="h11",
-        ssl_certfile=settings.api_ssl_certfile,
-        ssl_keyfile=settings.api_ssl_keyfile,
         timeout_keep_alive=300,  # 5 minute timeout for long running tasks
         timeout_graceful_shutdown=300,
+        **ssl_kwargs,
     )

api/routes/ingest.py

@@ -173,6 +173,20 @@ def update_job(
         ingester = _get_ingester(source_type, platform, api_credentials)
         asset = ingester.ingest(source)
 
+        # Idempotency: skip videos that are already fully indexed to avoid
+        # re-transcribing and re-embedding (duplicate cost + duplicate vectors).
+        existing = db.query(Video).filter(Video.id == asset.video_id).first()
+        if existing is not None and str(existing.status) == "indexed":
+            update_job(
+                "done",
+                f"Video already indexed ({existing.chunk_count} chunks) — skipped",
+                video_id=asset.video_id,
+            )
+            logger.info(
+                f"[Ingest] Job {job_id} — video {asset.video_id} already indexed, skipping"
+            )
+            return
+
         update_job(
             "transcribing",
             "Transcribing audio with Whisper...",
@@ -262,8 +276,11 @@ def update_job(
                     video.status = "error"
                     video.error_message = str(e)
                     db.commit()
-        except Exception:
-            pass
+        except Exception as cleanup_error:
+            db.rollback()
+            logger.warning(
+                f"[Ingest] Job {job_id} — failed to mark video as error: {cleanup_error}"
+            )
     finally:
         db.close()
 

api/routes/query.py

@@ -22,7 +22,13 @@
 router = APIRouter(tags=["RAG"])
 
 
-@router.post("/query", response_model=QueryResponse)
+@router.post(
+    "/query",
+    response_model=QueryResponse,
+    responses={
+        503: {"description": "LLM service unavailable (e.g. Ollama not running)"},
+    },
+)
 def query_videos(request: QueryRequest) -> QueryResponse:
     """
     Ask a question and get an answer grounded in indexed video content.
@@ -31,11 +37,22 @@ def query_videos(request: QueryRequest) -> QueryResponse:
     logger.info(f"[API/query] '{request.question[:60]}'")
 
     chain = QAChain()
-    result = chain.ask(
-        question=request.question,
-        video_id=request.video_id,
-        top_k=request.top_k,
-    )
+    try:
+        result = chain.ask(
+            question=request.question,
+            video_id=request.video_id,
+            top_k=request.top_k,
+        )
+    except Exception as exc:
+        logger.error(f"[API/query] LLM invocation failed: {exc}")
+        raise HTTPException(
+            status_code=503,
+            detail=(
+                f"Cannot reach the LLM backend: {exc}. "
+                "Check that Ollama is running (`ollama serve`) "
+                "or verify your LLM_PROVIDER and API key configuration."
+            ),
+        ) from exc
 
     sources = [SourceCitation.model_validate(s) for s in result.source_citations]
 
@@ -49,7 +66,10 @@ def query_videos(request: QueryRequest) -> QueryResponse:
 @router.post(
     "/summarize",
     response_model=SummarizeResponse,
-    responses={404: {"description": "Video not found"}},
+    responses={
+        404: {"description": "Video not found"},
+        503: {"description": "LLM service unavailable (e.g. Ollama not running)"},
+    },
 )
 def summarize_video(
     request: SummarizeRequest, db: Annotated[Session, Depends(get_db)]
@@ -67,11 +87,24 @@ def summarize_video(
         )
 
     summarizer = Summarizer()
-    result = summarizer.summarize(
-        video_id=request.video_id,
-        title=str(video.title),
-        include_chapters=request.include_chapters,
-    )
+    try:
+        result = summarizer.summarize(
+            video_id=request.video_id,
+            title=str(video.title),
+            include_chapters=request.include_chapters,
+        )
+    except Exception as exc:
+        # Catch connection errors (e.g. Ollama not running),
+        # timeouts, and any other LLM invocation failures.
+        logger.error(f"[API/summarize] LLM invocation failed: {exc}")
+        raise HTTPException(
+            status_code=503,
+            detail=(
+                f"Cannot reach the LLM backend: {exc}. "
+                "Check that Ollama is running (`ollama serve`) "
+                "or verify your LLM_PROVIDER and API key configuration."
+            ),
+        ) from exc
 
     chapter_summaries = None
     if result.chapter_summaries:
@@ -86,4 +119,4 @@ def summarize_video(
         overall_summary=result.overall_summary,
         chapter_summaries=chapter_summaries,
         chunk_count=result.chunk_count,
-    )
+    )

config.py

@@ -43,7 +43,7 @@ class Settings(BaseSettings):
     # ── Whisper transcription ─────────────────────────────────────────────
     whisper_mode: WhisperMode = WhisperMode.LOCAL
     # Local model size: tiny | base | small | medium | large | large-v2
-    whisper_model_size: str = "large-v2"
+    whisper_model_size: str = "base"
 
     # ── Embeddings ────────────────────────────────────────────────────────
     embedding_mode: EmbeddingMode = EmbeddingMode.LOCAL
@@ -76,13 +76,27 @@ class Settings(BaseSettings):
     retrieval_top_k: int = 5
     chunk_duration_seconds: int = 60
     chunk_overlap_seconds: int = 15
-    use_hyde: bool = False  # Enable/disable HyDE query transformation
+    use_hyde: bool = False  # HyDE query transformation (adds 1 LLM call per query)
+
+    # ── Re-ranker ─────────────────────────────────────────────────────────
+    use_reranker: bool = True
+    reranker_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
+    reranker_top_k: int = 10  # Fetch this many from vector store before reranking
+
+    # ── MMR diversity ─────────────────────────────────────────────────────
+    use_mmr: bool = True
+    mmr_lambda: float = 0.5  # 0 = only diversity, 1 = only relevance
+
+    # ── Multi-query expansion ─────────────────────────────────────────────
+    use_multi_query: bool = False  # adds 1 LLM call + N extra searches per query
+    multi_query_count: int = 3
 
     # ── Live stream ───────────────────────────────────────────────────────
     live_stream_segment_seconds: int = 60  # capture window per segment
 
     # ── UI ────────────────────────────────────────────────────────────────
     ui_api_base_url: str = "http://localhost:8000"
+    ui_request_timeout: int = 120
 
     class Config:
         env_file = ".env"

processing/embedder.py

@@ -84,6 +84,24 @@ def embed_query(self, query: str) -> list[float]:
         else:
             return self._embed_openai([query])[0]
 
+    def embed_query_batch(self, texts: list[str]) -> list[list[float]]:
+        """
+        Embed a batch of query strings for retrieval.
+        Useful for MMR diversity scoring and multi-query expansion.
+
+        Args:
+            texts: List of text strings to embed
+
+        Returns:
+            List of float vectors, one per text (same order)
+        """
+        if not texts:
+            return []
+        if self.mode == EmbeddingMode.LOCAL:
+            return self._embed_local(texts)
+        else:
+            return self._embed_openai(texts)
+
     # ── Local (sentence-transformers) ────────────────────────────────────────
 
     def _embed_local(self, texts: list[str]) -> list[list[float]]:

rag/__init__.py

@@ -1,6 +1,14 @@
+from .pipeline import RAGPipeline
 from .qa_chain import QAChain
 from .retriever import Retriever, get_retriever
 from .search import SearchEngine
 from .summarizer import Summarizer
 
-__all__ = ["QAChain", "Retriever", "SearchEngine", "Summarizer", "get_retriever"]
+__all__ = [
+    "QAChain",
+    "RAGPipeline",
+    "Retriever",
+    "SearchEngine",
+    "Summarizer",
+    "get_retriever",
+]