refactor(phase-3): replace RuntimeError with typed OpenRAGError subclasses

Replace all 12 RuntimeError raises introduced by phase-3 with proper
OpenRAGError subclasses that flow through the global handler:

- indexer.py: RuntimeError → UnexpectedError, VDBError/EmbeddingError
  catches widened to OpenRAGError
- llm.py: RuntimeError → UnexpectedError for streaming failures
- reranker.py: RuntimeError → UnexpectedError
- serializer.py: RuntimeError → FileStorageError (I/O) / UnexpectedError
- marker.py: RuntimeError → FileStorageError (I/O) / UnexpectedError
- ray_utils.py: RuntimeError → RayActorError (wraps RayTaskError)
- Fix pre-existing lint issues (unused vars, unsorted imports)

All exception raises now use `from e` chaining for proper tracebacks.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: paultranvan

openrag/components/indexer/chunker/chunker.py

@@ -2,6 +2,7 @@
 
 import openai
 from components.indexer.utils.text_sanitizer import sanitize_text
+from utils.exceptions.base import OpenRAGError
 from components.prompts import CHUNK_CONTEXTUALIZER_PROMPT
 from components.utils import detect_language, get_vlm_semaphore, load_config
 from langchain_core.documents.base import Document
@@ -79,7 +80,7 @@ async def _generate_context(
                 logger.error("VLM context generation failed", error=str(e))
                 return ""
 
-            except Exception as e:
+            except Exception:
                 # Unexpected errors - log but still gracefully degrade
                 logger.exception("Unexpected error during context generation")
                 return ""
@@ -132,7 +133,9 @@ async def contextualize_chunks(
                 for chunk, context in zip(chunks, contexts, strict=True)
             ]
 
-        except Exception as e:
+        except OpenRAGError:
+            raise
+        except Exception:
             logger.exception("Error contextualizing chunks", filename=filename)
             return chunks
 

openrag/components/indexer/embeddings/openai.py

@@ -32,7 +32,7 @@ def embedding_dimension(self) -> int:
             logger.error("Invalid embedding response format", error=str(e))
             raise EmbeddingResponseError("Unexpected response format", error=str(e))
 
-        except Exception as e:
+        except Exception:
             logger.exception("Unexpected error getting embedding dimension")
             raise UnexpectedEmbeddingError("An unexpected error occurred")
 

openrag/components/indexer/indexer.py

@@ -10,8 +10,8 @@
 import torch
 from config import load_config
 from langchain_core.documents.base import Document
-from utils.exceptions.embeddings import EmbeddingError
-from utils.exceptions.vectordb import VDBError
+from utils.exceptions.base import OpenRAGError
+from utils.exceptions.common import UnexpectedError
 
 from .chunker import BaseChunker, ChunkerFactory
 from .utils import serialize_file
@@ -121,25 +121,8 @@ async def add_file(
             # Mark task as completed
             await task_state_manager.set_state.remote(task_id, "COMPLETED")
 
-        except OSError as e:
-            # File I/O errors (FileNotFoundError, PermissionError, etc.)
-            log.error("File operation failed", path=path, error=str(e))
-            tb = "".join(traceback.format_exception(type(e), e, e.__traceback__))
-            await task_state_manager.set_state.remote(task_id, "FAILED")
-            await task_state_manager.set_error.remote(task_id, tb)
-            raise
-
-        except VDBError as e:
-            # Database errors (already typed from vectordb)
-            log.error("Database operation failed", error=str(e))
-            tb = "".join(traceback.format_exception(type(e), e, e.__traceback__))
-            await task_state_manager.set_state.remote(task_id, "FAILED")
-            await task_state_manager.set_error.remote(task_id, tb)
-            raise
-
-        except EmbeddingError as e:
-            # Embedding errors (already typed from embeddings)
-            log.error("Embedding generation failed", error=str(e))
+        except OpenRAGError as e:
+            log.error("Operation failed during file ingestion", code=e.code, error=e.message)
             tb = "".join(traceback.format_exception(type(e), e, e.__traceback__))
             await task_state_manager.set_state.remote(task_id, "FAILED")
             await task_state_manager.set_error.remote(task_id, tb)
@@ -151,7 +134,7 @@ async def add_file(
             tb = "".join(traceback.format_exception(type(e), e, e.__traceback__))
             await task_state_manager.set_state.remote(task_id, "FAILED")
             await task_state_manager.set_error.remote(task_id, tb)
-            raise RuntimeError("An unexpected error occurred during file processing")
+            raise UnexpectedError("An unexpected error occurred during file processing") from e
 
         finally:
             # GPU cleanup
@@ -185,15 +168,12 @@ async def delete_file(self, file_id: str, partition: str) -> bool:
             await vectordb.delete_file.remote(file_id, partition)
             log.info("Deleted file from partition.", file_id=file_id, partition=partition)
 
-        except VDBError as e:
-            # Database errors (already typed from vectordb)
-            log.error("Database operation failed in delete_file", error=str(e))
+        except OpenRAGError:
             raise
 
         except Exception as e:
-            # Unexpected errors
             log.exception("Unexpected error in delete_file")
-            raise RuntimeError("An unexpected error occurred during file deletion")
+            raise UnexpectedError("An unexpected error occurred during file deletion") from e
 
     @ray.method(concurrency_group="update")
     async def update_file_metadata(
@@ -219,15 +199,12 @@ async def update_file_metadata(
 
             log.info("Metadata updated for file.")
 
-        except VDBError as e:
-            # Database errors (already typed from vectordb)
-            log.error("Database operation failed in update_file_metadata", error=str(e))
+        except OpenRAGError:
             raise
 
         except Exception as e:
-            # Unexpected errors
             log.exception("Unexpected error in update_file_metadata")
-            raise RuntimeError("An unexpected error occurred during metadata update")
+            raise UnexpectedError("An unexpected error occurred during metadata update") from e
 
     @ray.method(concurrency_group="update")
     async def copy_file(
@@ -258,15 +235,12 @@ async def copy_file(
                 new_partition=metadata.get("partition"),
             )
 
-        except VDBError as e:
-            # Database errors (already typed from vectordb)
-            log.error("Database operation failed in copy_file", error=str(e))
+        except OpenRAGError:
             raise
 
         except Exception as e:
-            # Unexpected errors
             log.exception("Unexpected error in copy_file")
-            raise RuntimeError("An unexpected error occurred during file copy")
+            raise UnexpectedError("An unexpected error occurred during file copy") from e
 
     @ray.method(concurrency_group="search")
     async def asearch(

openrag/components/indexer/loaders/base.py

@@ -1,6 +1,5 @@
 import asyncio
 import base64
-import binascii
 import re
 from abc import ABC, abstractmethod
 from io import BytesIO

openrag/components/indexer/loaders/eml_loader.py

@@ -9,6 +9,7 @@
 
 from langchain_core.documents.base import Document
 from PIL import Image, UnidentifiedImageError
+from utils.exceptions.common import FileStorageError, UnexpectedError
 
 from . import get_loader_classes
 from .base import BaseLoader
@@ -327,13 +328,13 @@ async def aload_document(self, file_path, metadata: dict | None = None, save_mar
                 metadata["markdown_path"] = str(markdown_path)
         except OSError as e:
             # File I/O error reading email file
-            raise ValueError(f"Cannot read email file: {e}")
+            raise FileStorageError(f"Cannot read email file: {e}") from e
         except email.errors.MessageError as e:
             # Email parsing error
-            raise ValueError(f"Invalid email format: {e}")
+            raise UnexpectedError(f"Invalid email format: {e}") from e
         except Exception as e:
             # Unexpected error
-            raise ValueError(f"Failed to parse the EML file {file_path}: {e}")
+            raise UnexpectedError(f"Failed to parse the EML file {file_path}: {e}") from e
 
         document = Document(page_content=content_body, metadata=metadata)
         return document

openrag/components/indexer/loaders/media_loader.py

@@ -7,7 +7,7 @@
 import numpy as np
 from components.utils import get_audio_semaphore
 from langchain_core.documents.base import Document
-from openai import AsyncOpenAI, APIError
+from openai import APIError, AsyncOpenAI
 from pydub import AudioSegment, silence
 from tqdm.asyncio import tqdm
 from utils.logger import get_logger

openrag/components/indexer/loaders/pdf_loaders/marker.py

@@ -9,6 +9,7 @@
 from config import load_config
 from langchain_core.documents.base import Document
 from marker.converters.pdf import PdfConverter
+from utils.exceptions.common import FileStorageError, UnexpectedError
 from utils.logger import get_logger
 
 from ..base import BaseLoader
@@ -107,13 +108,11 @@ def _process_pdf(file_path, config):
             logger.info("PDF processing cancelled", path=file_path)
             raise
         except OSError as e:
-            # File I/O error
             logger.error("Cannot read PDF file", path=file_path, error=str(e))
-            raise RuntimeError(f"Cannot read PDF file: {e}")
+            raise FileStorageError(f"Cannot read PDF file: {e}") from e
         except Exception as e:
-            # Marker library errors or unexpected failures
             logger.exception("Error processing PDF", path=file_path, error=str(e))
-            raise RuntimeError("Failed to process PDF document")
+            raise UnexpectedError("Failed to process PDF document") from e
         finally:
             gc.collect()
             if torch.cuda.is_available():
@@ -133,7 +132,7 @@ def run_with_timeout():
                 return result
             except MPTimeoutError:
                 self.logger.exception("MarkerWorker child process timed out", path=file_path)
-                raise RuntimeError(f"PDF processing timed out")
+                raise UnexpectedError("PDF processing timed out")
             except asyncio.CancelledError:
                 # Cancellation - propagate
                 self.logger.info("PDF processing cancelled", path=file_path)
@@ -242,7 +241,7 @@ async def aload_document(
             )
 
             if not markdown:
-                raise RuntimeError(f"Conversion failed for {file_path_str}")
+                raise UnexpectedError(f"Conversion failed for {file_path_str}")
 
             if self.image_captioning:
                 keys = list(images.keys())
@@ -271,9 +270,8 @@ async def aload_document(
             logger.info("PDF loading cancelled", path=file_path_str)
             raise
         except OSError as e:
-            # File I/O error
             logger.error("Cannot read PDF file", path=file_path_str, error=str(e))
-            raise RuntimeError(f"Cannot read PDF file: {e}")
+            raise FileStorageError(f"Cannot read PDF file: {e}") from e
         except Exception:
             # Ray actor errors or PDF processing failures
             logger.exception("Error in aload_document", path=file_path_str)

openrag/components/indexer/loaders/serializer.py

@@ -5,6 +5,7 @@
 import torch
 from config import load_config
 from langchain_core.documents.base import Document
+from utils.exceptions.common import FileStorageError, UnexpectedError
 
 from . import get_loader_classes
 
@@ -85,10 +86,8 @@ async def serialize_document(
             log.info("Document serialized successfully")
             return doc
         except OSError as e:
-            # File operation failed (file not found, permission denied, etc.)
             log.error("File operation failed during serialization", path=str(path), error=str(e))
-            raise RuntimeError(f"Cannot read file: {e}")
+            raise FileStorageError(f"Cannot read file: {e}") from e
         except Exception as e:
-            # Loader-specific errors or unexpected failures
             log.exception("Failed to serialize document", path=str(path), file_type=file_ext, error=str(e))
-            raise RuntimeError("Failed to serialize document: unsupported format or corrupted file")
+            raise UnexpectedError("Failed to serialize document: unsupported format or corrupted file") from e

openrag/components/indexer/vectordb/utils.py

@@ -152,7 +152,7 @@ def __init__(self, database_url: str, logger=logger):
             AUTH_TOKEN = os.getenv("AUTH_TOKEN")
             self._ensure_admin_user(AUTH_TOKEN)
 
-        except Exception as e:
+        except Exception:
             raise VDBConnectionError(
                 "An unexpected database error occurred",
                 db_url=str(database_url),

openrag/components/llm.py

@@ -3,6 +3,7 @@
 import json
 
 import httpx
+from utils.exceptions.common import UnexpectedError
 from utils.logger import get_logger
 
 logger = get_logger()
@@ -38,10 +39,10 @@ async def completions(self, request: dict):
                 data = response.json()
                 yield data
             except httpx.HTTPStatusError as e:
-                error_detail = e.response.text
-                raise ValueError(f"LLM API error ({e.response.status_code}): {error_detail}")
+                logger.error("LLM API returned error", status_code=e.response.status_code)
+                raise UnexpectedError(f"LLM API error ({e.response.status_code})") from e
             except json.JSONDecodeError as e:
-                raise ValueError(f"Invalid JSON in API response: {str(e)}")
+                raise UnexpectedError("Invalid JSON in LLM API response") from e
 
     async def chat_completion(self, request: dict):
         request.pop("model")
@@ -71,17 +72,16 @@ async def chat_completion(self, request: dict):
                 except httpx.HTTPStatusError as e:
                     # 4xx/5xx responses
                     logger.error("LLM API returned error", status_code=e.response.status_code)
-                    raise
+                    raise UnexpectedError(f"LLM API error ({e.response.status_code})") from e
 
                 except httpx.RequestError as e:
                     # Network/connection failures
                     logger.error("Network error during LLM streaming", error=str(e))
-                    raise
+                    raise UnexpectedError("Network error during LLM streaming") from e
 
                 except Exception as e:
-                    # Truly unexpected errors
                     logger.exception("Unexpected error during LLM streaming")
-                    raise RuntimeError("An unexpected error occurred during streaming")
+                    raise UnexpectedError("An unexpected error occurred during streaming") from e
 
             else:  # Handle non-streaming response
                 try:
@@ -94,7 +94,7 @@ async def chat_completion(self, request: dict):
                     data = response.json()
                     yield data
                 except httpx.HTTPStatusError as e:
-                    error_detail = e.response.text
-                    raise ValueError(f"LLM API error ({e.response.status_code}): {error_detail}")
+                    logger.error("LLM API returned error", status_code=e.response.status_code)
+                    raise UnexpectedError(f"LLM API error ({e.response.status_code})") from e
                 except json.JSONDecodeError as e:
-                    raise ValueError(f"Invalid JSON in API response: {str(e)}")
+                    raise UnexpectedError("Invalid JSON in LLM API response") from e