feat(phase-2): centralized exception handling for API layer

Replace scattered HTTPException(500) raises with typed OpenRAGError
subclasses that flow through the global handler in api.py:

- Add common.py: RayActorError, FileStorageError, ToolExecutionError,
  UnexpectedError
- All routers use `except OpenRAGError: raise` passthrough pattern
- RayTaskError/TaskCancelledError → RayActorError with `from e` chaining
- OSError → FileStorageError for disk I/O failures
- Generic Exception → UnexpectedError as catch-all
- Streaming errors use _make_sse_error() helper for consistent SSE format
- utils.py: add httpx.TimeoutException→504, httpx.HTTPError→503 for
  LLM health checks (appropriate as HTTPException for external services)

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

Author: paultranvan

openrag/api.py

@@ -167,6 +167,16 @@ async def openrag_exception_handler(request: Request, exc: OpenRAGError):
     return JSONResponse(status_code=exc.status_code, content=exc.to_dict())
 
 
+@app.exception_handler(Exception)
+async def unhandled_exception_handler(request: Request, exc: Exception):
+    logger = get_logger()
+    logger.exception("Unhandled exception", error=str(exc))
+    return JSONResponse(
+        status_code=500,
+        content={"detail": "[UNEXPECTED_ERROR]: An unexpected error occurred", "extra": {}},
+    )
+
+
 # Add CORS middleware
 allow_origins = [
     "http://localhost:3042",

openrag/components/pipeline.py

@@ -178,25 +178,17 @@ async def _prepare_for_completions(self, partition: list[str], payload: dict):
         return payload, docs
 
     async def completions(self, partition: list[str], payload: dict):
-        try:
-            if partition is None:
-                docs = []
-            else:
-                payload, docs = await self._prepare_for_completions(partition=partition, payload=payload)
-            llm_output = self.llm_client.completions(request=payload)
-            return llm_output, docs
-        except Exception as e:
-            logger.error(f"Error during chat completion: {e!s}")
-            raise e
+        if partition is None:
+            docs = []
+        else:
+            payload, docs = await self._prepare_for_completions(partition=partition, payload=payload)
+        llm_output = self.llm_client.completions(request=payload)
+        return llm_output, docs
 
     async def chat_completion(self, partition: list[str] | None, payload: dict):
-        try:
-            if partition is None:
-                docs = []
-            else:
-                payload, docs = await self._prepare_for_chat_completion(partition=partition, payload=payload)
-            llm_output = self.llm_client.chat_completion(request=payload)
-            return llm_output, docs
-        except Exception as e:
-            logger.error(f"Error during chat completion: {e!s}")
-            raise e
+        if partition is None:
+            docs = []
+        else:
+            payload, docs = await self._prepare_for_chat_completion(partition=partition, payload=payload)
+        llm_output = self.llm_client.chat_completion(request=payload)
+        return llm_output, docs

openrag/routers/actors.py

@@ -51,24 +51,17 @@
 )
 async def list_ray_actors():
     """List all known Ray actors and their status."""
-    try:
-        actors = [
-            {
-                "actor_id": a.actor_id,
-                "name": a.name,
-                "class_name": a.class_name,
-                "state": a.state,
-                "namespace": a.ray_namespace,
-            }
-            for a in list_actors()
-        ]
-        return JSONResponse(status_code=status.HTTP_200_OK, content={"actors": actors})
-    except Exception:
-        logger.exception("Error getting actor summaries")
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="Failed to retrieve actor summaries.",
-        )
+    actors = [
+        {
+            "actor_id": a.actor_id,
+            "name": a.name,
+            "class_name": a.class_name,
+            "state": a.state,
+            "namespace": a.ray_namespace,
+        }
+        for a in list_actors()
+    ]
+    return JSONResponse(status_code=status.HTTP_200_OK, content={"actors": actors})
 
 
 @router.post(
@@ -116,29 +109,16 @@ async def restart_actor(
         logger.info(f"Killed actor: {actor_name}")
     except ValueError:
         logger.warning("Actor not found. Creating new instance.", actor=actor_name)
-    except Exception as e:
-        logger.exception("Failed to kill actor", actor=actor_name)
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to kill actor {actor_name}: {e!s}",
-        )
 
-    try:
-        new_actor = actor_creation_map[actor_name]()
-        if "Semaphore" in actor_name:
-            new_actor = new_actor._actor
-        logger.info(f"Restarted actor: {actor_name}")
-        return JSONResponse(
-            status_code=status.HTTP_200_OK,
-            content={
-                "message": f"Actor {actor_name} restarted successfully.",
-                "actor_name": actor_name,
-                "actor_id": new_actor._actor_id.hex(),
-            },
-        )
-    except Exception as e:
-        logger.exception("Failed to restart actor", actor=actor_name)
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to restart actor {actor_name}: {e!s}",
-        )
+    new_actor = actor_creation_map[actor_name]()
+    if "Semaphore" in actor_name:
+        new_actor = new_actor._actor
+    logger.info(f"Restarted actor: {actor_name}")
+    return JSONResponse(
+        status_code=status.HTTP_200_OK,
+        content={
+            "message": f"Actor {actor_name} restarted successfully.",
+            "actor_name": actor_name,
+            "actor_id": new_actor._actor_id.hex(),
+        },
+    )

openrag/routers/extract.py

@@ -48,31 +48,23 @@ async def get_extract(
     user_partitions=Depends(current_user_or_admin_partitions_list),
 ):
     log = logger.bind(extract_id=extract_id)
-    try:
-        chunk = await vectordb.get_chunk_by_id.remote(extract_id)
-        if chunk is None:
-            log.warning("Extract not found.")
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"Extract '{extract_id}' not found.",
-            )
-        chunk_partition = chunk.metadata["partition"]
-        log.info(f"User partitions: {user_partitions}, Chunk partition: {chunk_partition}")
-        if chunk_partition not in user_partitions and user_partitions != ["all"]:
-            log.warning("User does not have access to this extract.")
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail=f"User does not have access to extract '{extract_id}'.",
-            )
-        log.info("Extract successfully retrieved.")
-    except HTTPException:
-        raise
-    except Exception as e:
-        log.exception("Failed to retrieve extract.", error=str(e))
+
+    chunk = await vectordb.get_chunk_by_id.remote(extract_id)
+    if chunk is None:
+        log.warning("Extract not found.")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Extract '{extract_id}' not found.",
+        )
+    chunk_partition = chunk.metadata["partition"]
+    log.info(f"User partitions: {user_partitions}, Chunk partition: {chunk_partition}")
+    if chunk_partition not in user_partitions and user_partitions != ["all"]:
+        log.warning("User does not have access to this extract.")
         raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=f"Failed to retrieve extract: {e!s}",
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=f"User does not have access to extract '{extract_id}'.",
         )
+    log.info("Extract successfully retrieved.")
 
     return JSONResponse(
         status_code=status.HTTP_200_OK,

openrag/routers/indexer.py

@@ -123,30 +123,16 @@ async def add_file(
     vectordb=Depends(get_vectordb),
     user=Depends(require_partition_editor),
 ):
-    log = logger.bind(
-        file_id=file_id,
-        partition=partition,
-        filename=file.filename,
-        user=user.get("display_name"),
-    )
-
     if await vectordb.file_exists.remote(file_id, partition):
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
             detail=f"File '{file_id}' already exists in partition {partition}",
         )
 
     save_dir = Path(DATA_DIR)
-    try:
-        original_filename = file.filename
-        file.filename = sanitize_filename(file.filename)
-        file_path = await save_file_to_disk(file, save_dir, with_random_prefix=True)
-    except Exception as e:
-        log.exception("Failed to save file to disk.", error=str(e))
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail=str(e),
-        )
+    original_filename = file.filename
+    file.filename = sanitize_filename(file.filename)
+    file_path = await save_file_to_disk(file, save_dir, with_random_prefix=True)
 
     metadata.update(
         {
@@ -240,8 +226,6 @@ async def put_file(
     vectordb=Depends(get_vectordb),
     user=Depends(require_partition_editor),
 ):
-    log = logger.bind(file_id=file_id, partition=partition, filename=file.filename)
-
     if not await vectordb.file_exists.remote(file_id, partition):
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -252,16 +236,9 @@ async def put_file(
     await indexer.delete_file.remote(file_id, partition)
 
     save_dir = Path(DATA_DIR)
-    try:
-        original_filename = file.filename
-        file.filename = sanitize_filename(file.filename)
-        file_path = await save_file_to_disk(file, save_dir, with_random_prefix=True)
-    except Exception:
-        log.exception("Failed to save file to disk.")
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="Failed to save uploaded file.",
-        )
+    original_filename = file.filename
+    file.filename = sanitize_filename(file.filename)
+    file_path = await save_file_to_disk(file, save_dir, with_random_prefix=True)
 
     metadata.update(
         {
@@ -441,21 +418,13 @@ async def get_task_error(
     task_state_manager=Depends(get_task_state_manager),
     task_details=Depends(require_task_owner),
 ):
-    try:
-        error = await task_state_manager.get_error.remote(task_id)
-        if error is None:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"No error found for task '{task_id}'.",
-            )
-        return {"task_id": task_id, "traceback": error.splitlines()}
-    except HTTPException:
-        raise
-    except Exception:
+    error = await task_state_manager.get_error.remote(task_id)
+    if error is None:
         raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="Failed to retrieve task error.",
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"No error found for task '{task_id}'.",
         )
+    return {"task_id": task_id, "traceback": error.splitlines()}
 
 
 @router.get(
@@ -475,32 +444,27 @@ async def get_task_error(
 """,
 )
 async def get_task_logs(task_id: str, max_lines: int = 100, task_details=Depends(require_task_owner)):
-    try:
-        if not LOG_FILE.exists():
-            raise HTTPException(status_code=500, detail="Log file not found.")
-
-        logs = []
-        with open(LOG_FILE, errors="replace") as f:
-            for line in reversed(list(f)):
-                try:
-                    record = json.loads(line).get("record", {})
-                    if record.get("extra", {}).get("task_id") == task_id:
-                        logs.append(
-                            f"{record['time']['repr']} - {record['level']['name']} - {record['message']} - {(record['extra'])}"
-                        )
-                        if len(logs) >= max_lines:
-                            break
-                except json.JSONDecodeError:
-                    continue
-
-        if not logs:
-            raise HTTPException(status_code=404, detail=f"No logs found for task '{task_id}'")
-
-        return JSONResponse(content={"task_id": task_id, "logs": logs[::-1]})  # restore order
-    except HTTPException:
-        raise
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Failed to fetch logs: {e!s}")
+    if not LOG_FILE.exists():
+        raise HTTPException(status_code=500, detail="Log file not found.")
+
+    logs = []
+    with open(LOG_FILE, errors="replace") as f:
+        for line in reversed(list(f)):
+            try:
+                record = json.loads(line).get("record", {})
+                if record.get("extra", {}).get("task_id") == task_id:
+                    logs.append(
+                        f"{record['time']['repr']} - {record['level']['name']} - {record['message']} - {(record['extra'])}"
+                    )
+                    if len(logs) >= max_lines:
+                        break
+            except json.JSONDecodeError:
+                continue
+
+    if not logs:
+        raise HTTPException(status_code=404, detail=f"No logs found for task '{task_id}'")
+
+    return JSONResponse(content={"task_id": task_id, "logs": logs[::-1]})  # restore order
 
 
 @router.delete(
@@ -525,13 +489,9 @@ async def cancel_task(
     task_state_manager=Depends(get_task_state_manager),
     task_details=Depends(require_task_owner),
 ):
-    try:
-        obj_ref = await task_state_manager.get_object_ref.remote(task_id)
-        if obj_ref is None:
-            raise HTTPException(404, f"No ObjectRef stored for task {task_id}")
-
-        ray.cancel(obj_ref["ref"], recursive=True)
-        return {"message": f"Cancellation signal sent for task {task_id}"}
-    except Exception as e:
-        logger.exception("Failed to cancel task.")
-        raise HTTPException(status_code=500, detail=str(e))
+    obj_ref = await task_state_manager.get_object_ref.remote(task_id)
+    if obj_ref is None:
+        raise HTTPException(404, f"No ObjectRef stored for task {task_id}")
+
+    ray.cancel(obj_ref["ref"], recursive=True)
+    return {"message": f"Cancellation signal sent for task {task_id}"}