fix(backend): harden bulk endpoints and redact pdf-queue email for non-admins

- W1: _parse_ids_param now returns 400 on non-integer input instead of 500
- W2: bulk_delete_scholars wraps commit in IntegrityError guard; router
  catches ScholarServiceError → 409
- W3: Move `from typing import Any` to top-level import in application.py
- W4: Redact requested_by_email in PDF queue response for non-admin users

app/api/routers/admin_dbops.py

@@ -57,7 +57,7 @@ def _requested_by_value(*, payload, admin_user: User) -> str:
     return from_payload or admin_user.email
 
 
-def _serialize_pdf_queue_item(item) -> dict[str, object]:
+def _serialize_pdf_queue_item(item, *, is_admin: bool = False) -> dict[str, object]:
     return {
         "publication_id": item.publication_id,
         "title": item.title,
@@ -69,7 +69,7 @@ def _serialize_pdf_queue_item(item) -> dict[str, object]:
         "last_failure_detail": item.last_failure_detail,
         "last_source": item.last_source,
         "requested_by_user_id": item.requested_by_user_id,
-        "requested_by_email": item.requested_by_email,
+        "requested_by_email": item.requested_by_email if is_admin else None,
         "queued_at": item.queued_at,
         "last_attempt_at": item.last_attempt_at,
         "resolved_at": item.resolved_at,
@@ -215,7 +215,7 @@ async def get_pdf_queue(
     return success_payload(
         request,
         data={
-            "items": [_serialize_pdf_queue_item(item) for item in queue_page.items],
+            "items": [_serialize_pdf_queue_item(item, is_admin=current_user.is_admin) for item in queue_page.items],
             **_pdf_queue_page_data(
                 total_count=queue_page.total_count,
                 page=resolved_page,

app/api/routers/scholars.py

@@ -51,7 +51,14 @@ def _parse_ids_param(ids: str | None) -> list[int] | None:
     parts = [p.strip() for p in ids.split(",") if p.strip()]
     if not parts:
         return None
-    return [int(p) for p in parts]
+    try:
+        return [int(p) for p in parts]
+    except ValueError:
+        raise ApiException(
+            status_code=400,
+            code="invalid_ids_param",
+            message="The 'ids' parameter must be a comma-separated list of integers.",
+        )
 
 
 @router.get(
@@ -104,12 +111,19 @@ async def bulk_delete_scholars(
     db_session: AsyncSession = Depends(get_db_session),
     current_user: User = Depends(get_api_current_user),
 ):
-    deleted_count = await scholar_service.bulk_delete_scholars(
+    try:
-        db_session,
+        deleted_count = await scholar_service.bulk_delete_scholars(
-        user_id=current_user.id,
+            db_session,
-        scholar_profile_ids=payload.scholar_profile_ids,
+            user_id=current_user.id,
-        upload_dir=settings.scholar_image_upload_dir,
+            scholar_profile_ids=payload.scholar_profile_ids,
-    )
+            upload_dir=settings.scholar_image_upload_dir,
+        )
+    except scholar_service.ScholarServiceError as exc:
+        raise ApiException(
+            status_code=409,
+            code="scholar_bulk_delete_failed",
+            message=str(exc),
+        ) from exc
     structured_log(
         logger,
         "info",

app/services/scholars/application.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+from typing import Any
 from uuid import uuid4
 
 from sqlalchemy.exc import IntegrityError
@@ -52,7 +53,11 @@ async def bulk_delete_scholars(
             _safe_remove_upload(upload_root, profile.profile_image_upload_path)
     for profile in profiles:
         await db_session.delete(profile)
-    await db_session.commit()
+    try:
+        await db_session.commit()
+    except IntegrityError as exc:
+        await db_session.rollback()
+        raise ScholarServiceError("Unable to bulk-delete scholars due to a database constraint.") from exc
     return len(profiles)
 
 
@@ -63,8 +68,6 @@ async def bulk_toggle_scholars(
     scholar_profile_ids: list[int],
     is_enabled: bool,
 ) -> int:
-    from typing import Any
-
     from sqlalchemy import CursorResult, update
 
     cursor: CursorResult[Any] = await db_session.execute(  # type: ignore[assignment]