refactor: extract helpers from oversized router and scheduler files

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

DECOMPOSITION_SLICES.md

@@ -29,7 +29,7 @@ This file contains self-contained decomposition prompts ("slices") to bring the
 | 2 | Ingestion: pagination + publication upsert | **DONE** |
 | 3 | Ingestion: enrichment + scholar processing + run completion | **DONE** |
 | 4 | Scholars service + PDF queue | **DONE** |
-| 5 | Routers + scheduler | pending |
+| 5 | Routers + scheduler | **DONE** |
 
 ---
 
@@ -177,9 +177,21 @@ Commit message: `refactor: decompose scholars service and pdf_queue into focused
 
 ---
 
-## Slice 5: Decompose router files and scheduler
+## Slice 5: Decompose router files and scheduler — DONE
 
-**Why:** Three files slightly over the 600-line ceiling. Each needs helper extraction.
+Completed. Extracted helpers from three oversized files:
+
+| File | Lines | Contents |
+|---|---|---|
+| `app/api/routers/run_serializers.py` | 238 | `serialize_run`, `serialize_queue_item`, `normalize_scholar_result`, `manual_run_payload_from_run`, `manual_run_success_payload`, value coercion helpers, idempotency key validation |
+| `app/api/routers/run_manual.py` | 237 | `load_safety_state`, `raise_manual_runs_disabled`, `reused_manual_run_payload`, `run_ingestion_for_manual`, `recover_integrity_error`, `execute_manual_run`, safety/failure raise helpers |
+| `app/api/routers/runs.py` | 440 | Route handlers only + imports |
+| `app/api/routers/scholar_helpers.py` | 234 | `serialize_scholar`, `hydrate_scholar_metadata_if_needed`, `enqueue_initial_scrape_job_for_scholar`, `search_kwargs`, `search_response_data`, `require_user_profile`, `read_uploaded_image` |
+| `app/api/routers/scholars.py` | 406 | Route handlers only + imports |
+| `app/services/ingestion/queue_runner.py` | 379 | `QueueJobRunner` class — drain continuation queue, job lifecycle (drop/retry/reschedule), ingestion dispatch, finalization |
+| `app/services/ingestion/scheduler.py` | 312 | `SchedulerService` — auto-run scheduling, `_run_loop`, `_tick_once`, candidate loading, PDF queue drain |
+
+Also updated one test file (`test_scholars_create_hydration.py`) to monkeypatch `scholar_helpers` instead of the `scholars` router for the 4 helper-related tests.
 
 **Files to create:** `app/api/routers/run_serializers.py`, `app/api/routers/run_manual.py`, `app/api/routers/scholar_helpers.py`, `app/services/ingestion/queue_runner.py`
 

app/api/routers/run_manual.py

@@ -0,0 +1,237 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from fastapi import Request
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.errors import ApiException
+from app.api.responses import success_payload
+from app.api.routers.run_serializers import manual_run_payload_from_run
+from app.db.models import RunStatus, RunTriggerType
+from app.logging_utils import structured_log
+from app.services.ingestion import application as ingestion_service
+from app.services.ingestion import safety as run_safety_service
+from app.services.runs import application as run_service
+from app.services.settings import application as user_settings_service
+from app.settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+async def load_safety_state(
+    db_session: AsyncSession,
+    *,
+    user_id: int,
+) -> dict[str, Any]:
+    user_settings = await user_settings_service.get_or_create_settings(
+        db_session,
+        user_id=user_id,
+    )
+    previous_safety_state = run_safety_service.get_safety_event_context(user_settings)
+    if run_safety_service.clear_expired_cooldown(user_settings):
+        await db_session.commit()
+        await db_session.refresh(user_settings)
+        structured_log(
+            logger,
+            "info",
+            "api.runs.safety_cooldown_cleared",
+            user_id=user_id,
+            reason=previous_safety_state.get("cooldown_reason"),
+            cooldown_until=previous_safety_state.get("cooldown_until"),
+        )
+    return run_safety_service.get_safety_state_payload(user_settings)
+
+
+def raise_manual_runs_disabled(*, user_id: int, safety_state: dict[str, Any]) -> None:
+    structured_log(
+        logger,
+        "warning",
+        "api.runs.manual_blocked_policy",
+        user_id=user_id,
+        policy={"manual_run_allowed": False},
+        safety_state=safety_state,
+    )
+    raise ApiException(
+        status_code=403,
+        code="manual_runs_disabled",
+        message="Manual checks are disabled by server policy.",
+        details={
+            "policy": {"manual_run_allowed": False},
+            "safety_state": safety_state,
+        },
+    )
+
+
+async def reused_manual_run_payload(
+    db_session: AsyncSession,
+    *,
+    request: Request,
+    user_id: int,
+    idempotency_key: str | None,
+    safety_state: dict[str, Any],
+) -> dict[str, Any] | None:
+    if idempotency_key is None:
+        return None
+    previous_run = await run_service.get_manual_run_by_idempotency_key(
+        db_session,
+        user_id=user_id,
+        idempotency_key=idempotency_key,
+    )
+    if previous_run is None:
+        return None
+    if previous_run.status in (RunStatus.RUNNING, RunStatus.RESOLVING):
+        raise ApiException(
+            status_code=409,
+            code="run_in_progress",
+            message="A run with this idempotency key is still in progress.",
+            details={"run_id": int(previous_run.id), "idempotency_key": idempotency_key},
+        )
+    return success_payload(
+        request,
+        data=manual_run_payload_from_run(
+            previous_run,
+            idempotency_key=idempotency_key,
+            reused_existing_run=True,
+            safety_state=safety_state,
+        ),
+    )
+
+
+async def run_ingestion_for_manual(
+    db_session: AsyncSession,
+    *,
+    ingest_service: ingestion_service.ScholarIngestionService,
+    user_id: int,
+    idempotency_key: str | None,
+):
+    user_settings = await user_settings_service.get_or_create_settings(
+        db_session,
+        user_id=user_id,
+    )
+    return await ingest_service.run_for_user(
+        db_session,
+        user_id=user_id,
+        trigger_type=RunTriggerType.MANUAL,
+        request_delay_seconds=user_settings.request_delay_seconds,
+        network_error_retries=settings.ingestion_network_error_retries,
+        retry_backoff_seconds=settings.ingestion_retry_backoff_seconds,
+        max_pages_per_scholar=settings.ingestion_max_pages_per_scholar,
+        page_size=settings.ingestion_page_size,
+        auto_queue_continuations=settings.ingestion_continuation_queue_enabled,
+        queue_delay_seconds=settings.ingestion_continuation_base_delay_seconds,
+        idempotency_key=idempotency_key,
+        alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold,
+        alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold,
+        alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold,
+    )
+
+
+async def recover_integrity_error(
+    db_session: AsyncSession,
+    *,
+    request: Request,
+    user_id: int,
+    idempotency_key: str | None,
+    original_exc: IntegrityError,
+) -> dict[str, Any]:
+    if idempotency_key is None:
+        logger.exception(
+            "api.runs.manual_integrity_error",
+            extra={"user_id": user_id},
+        )
+        raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc
+    existing_run = await run_service.get_manual_run_by_idempotency_key(
+        db_session,
+        user_id=user_id,
+        idempotency_key=idempotency_key,
+    )
+    if existing_run is None:
+        logger.exception(
+            "api.runs.manual_integrity_error",
+            extra={"user_id": user_id},
+        )
+        raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc
+    if existing_run.status in (RunStatus.RUNNING, RunStatus.RESOLVING):
+        raise ApiException(
+            status_code=409,
+            code="run_in_progress",
+            message="A run with this idempotency key is still in progress.",
+            details={"run_id": int(existing_run.id), "idempotency_key": idempotency_key},
+        ) from original_exc
+    return success_payload(
+        request,
+        data=manual_run_payload_from_run(
+            existing_run,
+            idempotency_key=idempotency_key,
+            reused_existing_run=True,
+            safety_state=await load_safety_state(db_session, user_id=user_id),
+        ),
+    )
+
+
+async def execute_manual_run(
+    db_session: AsyncSession,
+    *,
+    request: Request,
+    ingest_service: ingestion_service.ScholarIngestionService,
+    user_id: int,
+    idempotency_key: str | None,
+):
+    try:
+        return await run_ingestion_for_manual(
+            db_session,
+            ingest_service=ingest_service,
+            user_id=user_id,
+            idempotency_key=idempotency_key,
+        )
+    except ingestion_service.RunAlreadyInProgressError as exc:
+        await db_session.rollback()
+        raise ApiException(
+            status_code=409,
+            code="run_in_progress",
+            message="A run is already in progress for this account.",
+        ) from exc
+    except ingestion_service.RunBlockedBySafetyPolicyError as exc:
+        await db_session.rollback()
+        raise_manual_blocked_safety(exc=exc, user_id=user_id)
+    except IntegrityError as exc:
+        await db_session.rollback()
+        return await recover_integrity_error(
+            db_session,
+            request=request,
+            user_id=user_id,
+            idempotency_key=idempotency_key,
+            original_exc=exc,
+        )
+    except Exception as exc:
+        await db_session.rollback()
+        raise_manual_failed(exc=exc, user_id=user_id)
+
+
+def raise_manual_blocked_safety(*, exc, user_id: int) -> None:
+    structured_log(
+        logger,
+        "info",
+        "api.runs.manual_blocked_safety",
+        user_id=user_id,
+        reason=exc.safety_state.get("cooldown_reason"),
+        cooldown_until=exc.safety_state.get("cooldown_until"),
+        cooldown_remaining_seconds=exc.safety_state.get("cooldown_remaining_seconds"),
+    )
+    raise ApiException(
+        status_code=429,
+        code=exc.code,
+        message=exc.message,
+        details={"safety_state": exc.safety_state},
+    ) from exc
+
+
+def raise_manual_failed(*, exc: Exception, user_id: int) -> None:
+    logger.exception(
+        "api.runs.manual_failed",
+        extra={"user_id": user_id},
+    )
+    raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from exc

app/api/routers/run_serializers.py

@@ -0,0 +1,238 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from app.api.errors import ApiException
+from app.services.runs import application as run_service
+
+IDEMPOTENCY_HEADER = "Idempotency-Key"
+IDEMPOTENCY_MAX_LENGTH = 128
+IDEMPOTENCY_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{8,128}$")
+
+
+def _int_value(value: Any, default: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _str_value(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text if text else None
+
+
+def _bool_value(value: Any, default: bool = False) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        lowered = value.strip().lower()
+        if lowered in {"1", "true", "yes", "on"}:
+            return True
+        if lowered in {"0", "false", "no", "off"}:
+            return False
+    return default
+
+
+def normalize_idempotency_key(raw_value: str | None) -> str | None:
+    if raw_value is None:
+        return None
+    candidate = raw_value.strip()
+    if not candidate:
+        return None
+    if len(candidate) > IDEMPOTENCY_MAX_LENGTH or not IDEMPOTENCY_PATTERN.match(candidate):
+        raise ApiException(
+            status_code=400,
+            code="invalid_idempotency_key",
+            message=("Invalid Idempotency-Key. Use 8-128 characters from: A-Z a-z 0-9 . _ : -"),
+        )
+    return candidate
+
+
+def serialize_run(run) -> dict[str, Any]:
+    summary = run_service.extract_run_summary(run.error_log)
+    return {
+        "id": int(run.id),
+        "trigger_type": run.trigger_type.value,
+        "status": run.status.value,
+        "start_dt": run.start_dt,
+        "end_dt": run.end_dt,
+        "scholar_count": int(run.scholar_count or 0),
+        "new_publication_count": int(run.new_pub_count or 0),
+        "failed_count": int(summary["failed_count"]),
+        "partial_count": int(summary["partial_count"]),
+    }
+
+
+def serialize_queue_item(item) -> dict[str, Any]:
+    return {
+        "id": int(item.id),
+        "scholar_profile_id": int(item.scholar_profile_id),
+        "scholar_label": item.scholar_label,
+        "status": item.status,
+        "reason": item.reason,
+        "dropped_reason": item.dropped_reason,
+        "attempt_count": int(item.attempt_count),
+        "resume_cstart": int(item.resume_cstart),
+        "next_attempt_dt": item.next_attempt_dt,
+        "updated_at": item.updated_at,
+        "last_error": item.last_error,
+        "last_run_id": item.last_run_id,
+    }
+
+
+def _normalize_attempt_log(value: Any) -> list[dict[str, Any]]:
+    if not isinstance(value, list):
+        return []
+    normalized: list[dict[str, Any]] = []
+    for item in value:
+        if not isinstance(item, dict):
+            continue
+        normalized.append(
+            {
+                "attempt": _int_value(item.get("attempt"), 0),
+                "cstart": _int_value(item.get("cstart"), 0),
+                "state": _str_value(item.get("state")),
+                "state_reason": _str_value(item.get("state_reason")),
+                "status_code": (_int_value(item.get("status_code")) if item.get("status_code") is not None else None),
+                "fetch_error": _str_value(item.get("fetch_error")),
+            }
+        )
+    return normalized
+
+
+def _normalize_page_logs(value: Any) -> list[dict[str, Any]]:
+    if not isinstance(value, list):
+        return []
+    normalized: list[dict[str, Any]] = []
+    for item in value:
+        if not isinstance(item, dict):
+            continue
+        warning_codes = item.get("warning_codes")
+        normalized.append(
+            {
+                "page": _int_value(item.get("page"), 0),
+                "cstart": _int_value(item.get("cstart"), 0),
+                "state": _str_value(item.get("state")) or "unknown",
+                "state_reason": _str_value(item.get("state_reason")),
+                "status_code": (_int_value(item.get("status_code")) if item.get("status_code") is not None else None),
+                "publication_count": _int_value(item.get("publication_count"), 0),
+                "attempt_count": _int_value(item.get("attempt_count"), 0),
+                "has_show_more_button": _bool_value(item.get("has_show_more_button"), False),
+                "articles_range": _str_value(item.get("articles_range")),
+                "warning_codes": [str(code) for code in (warning_codes if isinstance(warning_codes, list) else [])],
+            }
+        )
+    return normalized
+
+
+def _normalize_debug(value: Any) -> dict[str, Any] | None:
+    if not isinstance(value, dict):
+        return None
+    marker_counts = value.get("marker_counts_nonzero")
+    warning_codes = value.get("warning_codes")
+    return {
+        "status_code": (_int_value(value.get("status_code")) if value.get("status_code") is not None else None),
+        "final_url": _str_value(value.get("final_url")),
+        "fetch_error": _str_value(value.get("fetch_error")),
+        "body_sha256": _str_value(value.get("body_sha256")),
+        "body_length": (_int_value(value.get("body_length")) if value.get("body_length") is not None else None),
+        "has_show_more_button": (
+            _bool_value(value.get("has_show_more_button"), False)
+            if value.get("has_show_more_button") is not None
+            else None
+        ),
+        "articles_range": _str_value(value.get("articles_range")),
+        "state_reason": _str_value(value.get("state_reason")),
+        "warning_codes": [str(code) for code in (warning_codes if isinstance(warning_codes, list) else [])],
+        "marker_counts_nonzero": {
+            str(key): _int_value(count, 0)
+            for key, count in (marker_counts.items() if isinstance(marker_counts, dict) else [])
+        },
+        "page_logs": _normalize_page_logs(value.get("page_logs")),
+        "attempt_log": _normalize_attempt_log(value.get("attempt_log")),
+    }
+
+
+def normalize_scholar_result(value: Any) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        return {
+            "scholar_profile_id": 0,
+            "scholar_id": "unknown",
+            "state": "unknown",
+            "state_reason": None,
+            "outcome": "failed",
+            "attempt_count": 0,
+            "publication_count": 0,
+            "start_cstart": 0,
+            "continuation_cstart": None,
+            "continuation_enqueued": False,
+            "continuation_cleared": False,
+            "warnings": [],
+            "error": None,
+            "debug": None,
+        }
+    warnings = value.get("warnings")
+    return {
+        "scholar_profile_id": _int_value(value.get("scholar_profile_id"), 0),
+        "scholar_id": _str_value(value.get("scholar_id")) or "unknown",
+        "state": _str_value(value.get("state")) or "unknown",
+        "state_reason": _str_value(value.get("state_reason")),
+        "outcome": _str_value(value.get("outcome")) or "failed",
+        "attempt_count": _int_value(value.get("attempt_count"), 0),
+        "publication_count": _int_value(value.get("publication_count"), 0),
+        "start_cstart": _int_value(value.get("start_cstart"), 0),
+        "continuation_cstart": (
+            _int_value(value.get("continuation_cstart")) if value.get("continuation_cstart") is not None else None
+        ),
+        "continuation_enqueued": _bool_value(value.get("continuation_enqueued"), False),
+        "continuation_cleared": _bool_value(value.get("continuation_cleared"), False),
+        "warnings": [str(item) for item in (warnings if isinstance(warnings, list) else [])],
+        "error": _str_value(value.get("error")),
+        "debug": _normalize_debug(value.get("debug")),
+    }
+
+
+def manual_run_payload_from_run(
+    run,
+    *,
+    idempotency_key: str | None,
+    reused_existing_run: bool,
+    safety_state: dict[str, Any],
+) -> dict[str, Any]:
+    summary = run_service.extract_run_summary(run.error_log)
+    return {
+        "run_id": int(run.id),
+        "status": run.status.value,
+        "scholar_count": int(run.scholar_count or 0),
+        "succeeded_count": int(summary["succeeded_count"]),
+        "failed_count": int(summary["failed_count"]),
+        "partial_count": int(summary["partial_count"]),
+        "new_publication_count": int(run.new_pub_count or 0),
+        "reused_existing_run": reused_existing_run,
+        "idempotency_key": idempotency_key,
+        "safety_state": safety_state,
+    }
+
+
+def manual_run_success_payload(
+    *,
+    run_summary,
+    idempotency_key: str | None,
+    safety_state: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "run_id": run_summary.crawl_run_id,
+        "status": run_summary.status.value,
+        "scholar_count": run_summary.scholar_count,
+        "succeeded_count": run_summary.succeeded_count,
+        "failed_count": run_summary.failed_count,
+        "partial_count": run_summary.partial_count,
+        "new_publication_count": run_summary.new_publication_count,
+        "reused_existing_run": False,
+        "idempotency_key": idempotency_key,
+        "safety_state": safety_state,
+    }

app/api/routers/runs.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import asyncio
 import logging
-import re
 from typing import Any
 
 from fastapi import APIRouter, Depends, Query, Request
@@ -13,6 +12,21 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.api.deps import get_api_current_user
 from app.api.errors import ApiException
 from app.api.responses import success_payload
+from app.api.routers.run_manual import (
+    load_safety_state,
+    raise_manual_blocked_safety,
+    raise_manual_failed,
+    raise_manual_runs_disabled,
+    recover_integrity_error,
+    reused_manual_run_payload,
+)
+from app.api.routers.run_serializers import (
+    IDEMPOTENCY_HEADER,
+    normalize_idempotency_key,
+    normalize_scholar_result,
+    serialize_queue_item,
+    serialize_run,
+)
 from app.api.runtime_deps import get_ingestion_service
 from app.api.schemas import (
     ManualRunEnvelope,
@@ -24,9 +38,7 @@ from app.api.schemas import (
 )
 from app.db.models import RunStatus, RunTriggerType, User
 from app.db.session import get_db_session
-from app.logging_utils import structured_log
 from app.services.ingestion import application as ingestion_service
-from app.services.ingestion import safety as run_safety_service
 from app.services.runs import application as run_service
 from app.services.runs.events import event_generator
 from app.services.settings import application as user_settings_service
@@ -38,453 +50,6 @@ _background_tasks: set[asyncio.Task[Any]] = set()
 router = APIRouter(prefix="/runs", tags=["api-runs"])
 ACTIVE_RUN_STATUSES = {RunStatus.RUNNING, RunStatus.RESOLVING}
 
-IDEMPOTENCY_HEADER = "Idempotency-Key"
-IDEMPOTENCY_MAX_LENGTH = 128
-IDEMPOTENCY_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{8,128}$")
-
-
-def _int_value(value: Any, default: int = 0) -> int:
-    try:
-        return int(value)
-    except (TypeError, ValueError):
-        return default
-
-
-def _str_value(value: Any) -> str | None:
-    if value is None:
-        return None
-    text = str(value).strip()
-    return text if text else None
-
-
-def _bool_value(value: Any, default: bool = False) -> bool:
-    if isinstance(value, bool):
-        return value
-    if isinstance(value, str):
-        lowered = value.strip().lower()
-        if lowered in {"1", "true", "yes", "on"}:
-            return True
-        if lowered in {"0", "false", "no", "off"}:
-            return False
-    return default
-
-
-def _normalize_idempotency_key(raw_value: str | None) -> str | None:
-    if raw_value is None:
-        return None
-    candidate = raw_value.strip()
-    if not candidate:
-        return None
-    if len(candidate) > IDEMPOTENCY_MAX_LENGTH or not IDEMPOTENCY_PATTERN.match(candidate):
-        raise ApiException(
-            status_code=400,
-            code="invalid_idempotency_key",
-            message=("Invalid Idempotency-Key. Use 8-128 characters from: A-Z a-z 0-9 . _ : -"),
-        )
-    return candidate
-
-
-def _serialize_run(run) -> dict[str, Any]:
-    summary = run_service.extract_run_summary(run.error_log)
-    return {
-        "id": int(run.id),
-        "trigger_type": run.trigger_type.value,
-        "status": run.status.value,
-        "start_dt": run.start_dt,
-        "end_dt": run.end_dt,
-        "scholar_count": int(run.scholar_count or 0),
-        "new_publication_count": int(run.new_pub_count or 0),
-        "failed_count": int(summary["failed_count"]),
-        "partial_count": int(summary["partial_count"]),
-    }
-
-
-def _serialize_queue_item(item) -> dict[str, Any]:
-    return {
-        "id": int(item.id),
-        "scholar_profile_id": int(item.scholar_profile_id),
-        "scholar_label": item.scholar_label,
-        "status": item.status,
-        "reason": item.reason,
-        "dropped_reason": item.dropped_reason,
-        "attempt_count": int(item.attempt_count),
-        "resume_cstart": int(item.resume_cstart),
-        "next_attempt_dt": item.next_attempt_dt,
-        "updated_at": item.updated_at,
-        "last_error": item.last_error,
-        "last_run_id": item.last_run_id,
-    }
-
-
-def _normalize_attempt_log(value: Any) -> list[dict[str, Any]]:
-    if not isinstance(value, list):
-        return []
-    normalized: list[dict[str, Any]] = []
-    for item in value:
-        if not isinstance(item, dict):
-            continue
-        normalized.append(
-            {
-                "attempt": _int_value(item.get("attempt"), 0),
-                "cstart": _int_value(item.get("cstart"), 0),
-                "state": _str_value(item.get("state")),
-                "state_reason": _str_value(item.get("state_reason")),
-                "status_code": (_int_value(item.get("status_code")) if item.get("status_code") is not None else None),
-                "fetch_error": _str_value(item.get("fetch_error")),
-            }
-        )
-    return normalized
-
-
-def _normalize_page_logs(value: Any) -> list[dict[str, Any]]:
-    if not isinstance(value, list):
-        return []
-    normalized: list[dict[str, Any]] = []
-    for item in value:
-        if not isinstance(item, dict):
-            continue
-        warning_codes = item.get("warning_codes")
-        normalized.append(
-            {
-                "page": _int_value(item.get("page"), 0),
-                "cstart": _int_value(item.get("cstart"), 0),
-                "state": _str_value(item.get("state")) or "unknown",
-                "state_reason": _str_value(item.get("state_reason")),
-                "status_code": (_int_value(item.get("status_code")) if item.get("status_code") is not None else None),
-                "publication_count": _int_value(item.get("publication_count"), 0),
-                "attempt_count": _int_value(item.get("attempt_count"), 0),
-                "has_show_more_button": _bool_value(item.get("has_show_more_button"), False),
-                "articles_range": _str_value(item.get("articles_range")),
-                "warning_codes": [str(code) for code in (warning_codes if isinstance(warning_codes, list) else [])],
-            }
-        )
-    return normalized
-
-
-def _normalize_debug(value: Any) -> dict[str, Any] | None:
-    if not isinstance(value, dict):
-        return None
-    marker_counts = value.get("marker_counts_nonzero")
-    warning_codes = value.get("warning_codes")
-    return {
-        "status_code": (_int_value(value.get("status_code")) if value.get("status_code") is not None else None),
-        "final_url": _str_value(value.get("final_url")),
-        "fetch_error": _str_value(value.get("fetch_error")),
-        "body_sha256": _str_value(value.get("body_sha256")),
-        "body_length": (_int_value(value.get("body_length")) if value.get("body_length") is not None else None),
-        "has_show_more_button": (
-            _bool_value(value.get("has_show_more_button"), False)
-            if value.get("has_show_more_button") is not None
-            else None
-        ),
-        "articles_range": _str_value(value.get("articles_range")),
-        "state_reason": _str_value(value.get("state_reason")),
-        "warning_codes": [str(code) for code in (warning_codes if isinstance(warning_codes, list) else [])],
-        "marker_counts_nonzero": {
-            str(key): _int_value(count, 0)
-            for key, count in (marker_counts.items() if isinstance(marker_counts, dict) else [])
-        },
-        "page_logs": _normalize_page_logs(value.get("page_logs")),
-        "attempt_log": _normalize_attempt_log(value.get("attempt_log")),
-    }
-
-
-def _normalize_scholar_result(value: Any) -> dict[str, Any]:
-    if not isinstance(value, dict):
-        return {
-            "scholar_profile_id": 0,
-            "scholar_id": "unknown",
-            "state": "unknown",
-            "state_reason": None,
-            "outcome": "failed",
-            "attempt_count": 0,
-            "publication_count": 0,
-            "start_cstart": 0,
-            "continuation_cstart": None,
-            "continuation_enqueued": False,
-            "continuation_cleared": False,
-            "warnings": [],
-            "error": None,
-            "debug": None,
-        }
-    warnings = value.get("warnings")
-    return {
-        "scholar_profile_id": _int_value(value.get("scholar_profile_id"), 0),
-        "scholar_id": _str_value(value.get("scholar_id")) or "unknown",
-        "state": _str_value(value.get("state")) or "unknown",
-        "state_reason": _str_value(value.get("state_reason")),
-        "outcome": _str_value(value.get("outcome")) or "failed",
-        "attempt_count": _int_value(value.get("attempt_count"), 0),
-        "publication_count": _int_value(value.get("publication_count"), 0),
-        "start_cstart": _int_value(value.get("start_cstart"), 0),
-        "continuation_cstart": (
-            _int_value(value.get("continuation_cstart")) if value.get("continuation_cstart") is not None else None
-        ),
-        "continuation_enqueued": _bool_value(value.get("continuation_enqueued"), False),
-        "continuation_cleared": _bool_value(value.get("continuation_cleared"), False),
-        "warnings": [str(item) for item in (warnings if isinstance(warnings, list) else [])],
-        "error": _str_value(value.get("error")),
-        "debug": _normalize_debug(value.get("debug")),
-    }
-
-
-def _manual_run_payload_from_run(
-    run,
-    *,
-    idempotency_key: str | None,
-    reused_existing_run: bool,
-    safety_state: dict[str, Any],
-) -> dict[str, Any]:
-    summary = run_service.extract_run_summary(run.error_log)
-    return {
-        "run_id": int(run.id),
-        "status": run.status.value,
-        "scholar_count": int(run.scholar_count or 0),
-        "succeeded_count": int(summary["succeeded_count"]),
-        "failed_count": int(summary["failed_count"]),
-        "partial_count": int(summary["partial_count"]),
-        "new_publication_count": int(run.new_pub_count or 0),
-        "reused_existing_run": reused_existing_run,
-        "idempotency_key": idempotency_key,
-        "safety_state": safety_state,
-    }
-
-
-async def _load_safety_state(
-    db_session: AsyncSession,
-    *,
-    user_id: int,
-) -> dict[str, Any]:
-    user_settings = await user_settings_service.get_or_create_settings(
-        db_session,
-        user_id=user_id,
-    )
-    previous_safety_state = run_safety_service.get_safety_event_context(user_settings)
-    if run_safety_service.clear_expired_cooldown(user_settings):
-        await db_session.commit()
-        await db_session.refresh(user_settings)
-        structured_log(
-            logger,
-            "info",
-            "api.runs.safety_cooldown_cleared",
-            user_id=user_id,
-            reason=previous_safety_state.get("cooldown_reason"),
-            cooldown_until=previous_safety_state.get("cooldown_until"),
-        )
-    return run_safety_service.get_safety_state_payload(user_settings)
-
-
-def _raise_manual_runs_disabled(*, user_id: int, safety_state: dict[str, Any]) -> None:
-    structured_log(
-        logger,
-        "warning",
-        "api.runs.manual_blocked_policy",
-        user_id=user_id,
-        policy={"manual_run_allowed": False},
-        safety_state=safety_state,
-    )
-    raise ApiException(
-        status_code=403,
-        code="manual_runs_disabled",
-        message="Manual checks are disabled by server policy.",
-        details={
-            "policy": {"manual_run_allowed": False},
-            "safety_state": safety_state,
-        },
-    )
-
-
-async def _reused_manual_run_payload(
-    db_session: AsyncSession,
-    *,
-    request: Request,
-    user_id: int,
-    idempotency_key: str | None,
-    safety_state: dict[str, Any],
-) -> dict[str, Any] | None:
-    if idempotency_key is None:
-        return None
-    previous_run = await run_service.get_manual_run_by_idempotency_key(
-        db_session,
-        user_id=user_id,
-        idempotency_key=idempotency_key,
-    )
-    if previous_run is None:
-        return None
-    if previous_run.status in (RunStatus.RUNNING, RunStatus.RESOLVING):
-        raise ApiException(
-            status_code=409,
-            code="run_in_progress",
-            message="A run with this idempotency key is still in progress.",
-            details={"run_id": int(previous_run.id), "idempotency_key": idempotency_key},
-        )
-    return success_payload(
-        request,
-        data=_manual_run_payload_from_run(
-            previous_run,
-            idempotency_key=idempotency_key,
-            reused_existing_run=True,
-            safety_state=safety_state,
-        ),
-    )
-
-
-async def _run_ingestion_for_manual(
-    db_session: AsyncSession,
-    *,
-    ingest_service: ingestion_service.ScholarIngestionService,
-    user_id: int,
-    idempotency_key: str | None,
-):
-    user_settings = await user_settings_service.get_or_create_settings(
-        db_session,
-        user_id=user_id,
-    )
-    return await ingest_service.run_for_user(
-        db_session,
-        user_id=user_id,
-        trigger_type=RunTriggerType.MANUAL,
-        request_delay_seconds=user_settings.request_delay_seconds,
-        network_error_retries=settings.ingestion_network_error_retries,
-        retry_backoff_seconds=settings.ingestion_retry_backoff_seconds,
-        max_pages_per_scholar=settings.ingestion_max_pages_per_scholar,
-        page_size=settings.ingestion_page_size,
-        auto_queue_continuations=settings.ingestion_continuation_queue_enabled,
-        queue_delay_seconds=settings.ingestion_continuation_base_delay_seconds,
-        idempotency_key=idempotency_key,
-        alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold,
-        alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold,
-        alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold,
-    )
-
-
-async def _recover_integrity_error(
-    db_session: AsyncSession,
-    *,
-    request: Request,
-    user_id: int,
-    idempotency_key: str | None,
-    original_exc: IntegrityError,
-) -> dict[str, Any]:
-    if idempotency_key is None:
-        logger.exception(
-            "api.runs.manual_integrity_error",
-            extra={"user_id": user_id},
-        )
-        raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc
-    existing_run = await run_service.get_manual_run_by_idempotency_key(
-        db_session,
-        user_id=user_id,
-        idempotency_key=idempotency_key,
-    )
-    if existing_run is None:
-        logger.exception(
-            "api.runs.manual_integrity_error",
-            extra={"user_id": user_id},
-        )
-        raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc
-    if existing_run.status in (RunStatus.RUNNING, RunStatus.RESOLVING):
-        raise ApiException(
-            status_code=409,
-            code="run_in_progress",
-            message="A run with this idempotency key is still in progress.",
-            details={"run_id": int(existing_run.id), "idempotency_key": idempotency_key},
-        ) from original_exc
-    return success_payload(
-        request,
-        data=_manual_run_payload_from_run(
-            existing_run,
-            idempotency_key=idempotency_key,
-            reused_existing_run=True,
-            safety_state=await _load_safety_state(db_session, user_id=user_id),
-        ),
-    )
-
-
-async def _execute_manual_run(
-    db_session: AsyncSession,
-    *,
-    request: Request,
-    ingest_service: ingestion_service.ScholarIngestionService,
-    user_id: int,
-    idempotency_key: str | None,
-):
-    try:
-        return await _run_ingestion_for_manual(
-            db_session,
-            ingest_service=ingest_service,
-            user_id=user_id,
-            idempotency_key=idempotency_key,
-        )
-    except ingestion_service.RunAlreadyInProgressError as exc:
-        await db_session.rollback()
-        raise ApiException(
-            status_code=409,
-            code="run_in_progress",
-            message="A run is already in progress for this account.",
-        ) from exc
-    except ingestion_service.RunBlockedBySafetyPolicyError as exc:
-        await db_session.rollback()
-        _raise_manual_blocked_safety(exc=exc, user_id=user_id)
-    except IntegrityError as exc:
-        await db_session.rollback()
-        return await _recover_integrity_error(
-            db_session,
-            request=request,
-            user_id=user_id,
-            idempotency_key=idempotency_key,
-            original_exc=exc,
-        )
-    except Exception as exc:
-        await db_session.rollback()
-        _raise_manual_failed(exc=exc, user_id=user_id)
-
-
-def _raise_manual_blocked_safety(*, exc, user_id: int) -> None:
-    structured_log(
-        logger,
-        "info",
-        "api.runs.manual_blocked_safety",
-        user_id=user_id,
-        reason=exc.safety_state.get("cooldown_reason"),
-        cooldown_until=exc.safety_state.get("cooldown_until"),
-        cooldown_remaining_seconds=exc.safety_state.get("cooldown_remaining_seconds"),
-    )
-    raise ApiException(
-        status_code=429,
-        code=exc.code,
-        message=exc.message,
-        details={"safety_state": exc.safety_state},
-    ) from exc
-
-
-def _raise_manual_failed(*, exc: Exception, user_id: int) -> None:
-    logger.exception(
-        "api.runs.manual_failed",
-        extra={"user_id": user_id},
-    )
-    raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from exc
-
-
-def _manual_run_success_payload(
-    *,
-    run_summary,
-    idempotency_key: str | None,
-    safety_state: dict[str, Any],
-) -> dict[str, Any]:
-    return {
-        "run_id": run_summary.crawl_run_id,
-        "status": run_summary.status.value,
-        "scholar_count": run_summary.scholar_count,
-        "succeeded_count": run_summary.succeeded_count,
-        "failed_count": run_summary.failed_count,
-        "partial_count": run_summary.partial_count,
-        "new_publication_count": run_summary.new_publication_count,
-        "reused_existing_run": False,
-        "idempotency_key": idempotency_key,
-        "safety_state": safety_state,
-    }
-
 
 @router.get(
     "",
@@ -503,14 +68,14 @@ async def list_runs(
         limit=limit,
         failed_only=failed_only,
     )
-    safety_state = await _load_safety_state(
+    safety_state = await load_safety_state(
         db_session,
         user_id=current_user.id,
     )
     return success_payload(
         request,
         data={
-            "runs": [_serialize_run(run) for run in runs],
+            "runs": [serialize_run(run) for run in runs],
             "safety_state": safety_state,
         },
     )
@@ -541,16 +106,16 @@ async def get_run(
     scholar_results = error_log.get("scholar_results")
     if not isinstance(scholar_results, list):
         scholar_results = []
-    safety_state = await _load_safety_state(
+    safety_state = await load_safety_state(
         db_session,
         user_id=current_user.id,
     )
     return success_payload(
         request,
         data={
-            "run": _serialize_run(run),
+            "run": serialize_run(run),
             "summary": run_service.extract_run_summary(error_log),
-            "scholar_results": [_normalize_scholar_result(item) for item in scholar_results],
+            "scholar_results": [normalize_scholar_result(item) for item in scholar_results],
             "safety_state": safety_state,
         },
     )
@@ -595,7 +160,7 @@ async def cancel_run(
     if not isinstance(scholar_results, list):
         scholar_results = []
 
-    safety_state = await _load_safety_state(
+    safety_state = await load_safety_state(
         db_session,
         user_id=current_user.id,
     )
@@ -603,9 +168,9 @@ async def cancel_run(
     return success_payload(
         request,
         data={
-            "run": _serialize_run(run),
+            "run": serialize_run(run),
             "summary": run_service.extract_run_summary(error_log),
-            "scholar_results": [_normalize_scholar_result(item) for item in scholar_results],
+            "scholar_results": [normalize_scholar_result(item) for item in scholar_results],
             "safety_state": safety_state,
         },
     )
@@ -621,12 +186,12 @@ async def run_manual(
     current_user: User = Depends(get_api_current_user),
     ingest_service: ingestion_service.ScholarIngestionService = Depends(get_ingestion_service),
 ):
-    safety_state = await _load_safety_state(db_session, user_id=current_user.id)
+    safety_state = await load_safety_state(db_session, user_id=current_user.id)
     if not settings.ingestion_manual_run_allowed:
-        _raise_manual_runs_disabled(user_id=current_user.id, safety_state=safety_state)
+        raise_manual_runs_disabled(user_id=current_user.id, safety_state=safety_state)
 
-    idempotency_key = _normalize_idempotency_key(request.headers.get(IDEMPOTENCY_HEADER))
+    idempotency_key = normalize_idempotency_key(request.headers.get(IDEMPOTENCY_HEADER))
-    reused_payload = await _reused_manual_run_payload(
+    reused_payload = await reused_manual_run_payload(
         db_session,
         request=request,
         user_id=current_user.id,
@@ -695,12 +260,12 @@ async def run_manual(
                 "new_publication_count": 0,
                 "reused_existing_run": False,
                 "idempotency_key": idempotency_key,
-                "safety_state": await _load_safety_state(db_session, user_id=current_user.id),
+                "safety_state": await load_safety_state(db_session, user_id=current_user.id),
             },
         )
     except ingestion_service.RunBlockedBySafetyPolicyError as exc:
         await db_session.rollback()
-        _raise_manual_blocked_safety(exc=exc, user_id=current_user.id)
+        raise_manual_blocked_safety(exc=exc, user_id=current_user.id)
     except ingestion_service.RunAlreadyInProgressError as exc:
         await db_session.rollback()
         raise ApiException(
@@ -710,7 +275,7 @@ async def run_manual(
         ) from exc
     except IntegrityError as exc:
         await db_session.rollback()
-        return await _recover_integrity_error(
+        return await recover_integrity_error(
             db_session,
             request=request,
             user_id=current_user.id,
@@ -719,7 +284,7 @@ async def run_manual(
         )
     except Exception as exc:
         await db_session.rollback()
-        _raise_manual_failed(exc=exc, user_id=current_user.id)
+        raise_manual_failed(exc=exc, user_id=current_user.id)
 
 
 @router.get(
@@ -740,7 +305,7 @@ async def list_queue_items(
     return success_payload(
         request,
         data={
-            "queue_items": [_serialize_queue_item(item) for item in items],
+            "queue_items": [serialize_queue_item(item) for item in items],
         },
     )
 
@@ -776,7 +341,7 @@ async def retry_queue_item(
         )
     return success_payload(
         request,
-        data=_serialize_queue_item(queue_item),
+        data=serialize_queue_item(queue_item),
     )
 
 
@@ -811,7 +376,7 @@ async def drop_queue_item(
         )
     return success_payload(
         request,
-        data=_serialize_queue_item(dropped),
+        data=serialize_queue_item(dropped),
     )
 
 

app/api/routers/scholar_helpers.py

@@ -0,0 +1,234 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from fastapi import UploadFile
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.errors import ApiException
+from app.logging_utils import structured_log
+from app.services.ingestion import queue as ingestion_queue_service
+from app.services.scholar import rate_limit as scholar_rate_limit
+from app.services.scholar.source import ScholarSource
+from app.services.scholars import application as scholar_service
+from app.services.scholars import search_hints as scholar_search_hints
+from app.settings import settings
+
+logger = logging.getLogger(__name__)
+
+CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS = 5.0
+INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS = 0
+INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON = "scholar_added_initial_scrape"
+
+
+def needs_metadata_hydration(profile) -> bool:
+    if not profile.profile_image_url:
+        return True
+    return not (profile.display_name or "").strip()
+
+
+def is_create_hydration_rate_limited() -> tuple[bool, float]:
+    remaining_seconds = scholar_rate_limit.remaining_scholar_slot_seconds(
+        min_interval_seconds=float(settings.ingestion_min_request_delay_seconds),
+    )
+    return remaining_seconds > 0, remaining_seconds
+
+
+def auto_enqueue_new_scholar_enabled() -> bool:
+    if not settings.scheduler_enabled:
+        return False
+    if not settings.ingestion_automation_allowed:
+        return False
+    return bool(settings.ingestion_continuation_queue_enabled)
+
+
+async def enqueue_initial_scrape_job_for_scholar(
+    db_session: AsyncSession,
+    *,
+    profile,
+    user_id: int,
+) -> bool:
+    if not auto_enqueue_new_scholar_enabled():
+        return False
+    try:
+        await ingestion_queue_service.upsert_job(
+            db_session,
+            user_id=user_id,
+            scholar_profile_id=int(profile.id),
+            resume_cstart=0,
+            reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON,
+            run_id=None,
+            delay_seconds=INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS,
+        )
+        await db_session.commit()
+    except Exception:
+        await db_session.rollback()
+        structured_log(
+            logger,
+            "warning",
+            "api.scholars.initial_scrape_enqueue_failed",
+            user_id=user_id,
+            scholar_profile_id=profile.id,
+        )
+        return False
+
+    structured_log(
+        logger,
+        "info",
+        "api.scholars.initial_scrape_enqueued",
+        user_id=user_id,
+        scholar_profile_id=profile.id,
+        reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON,
+    )
+    return True
+
+
+def uploaded_image_media_path(scholar_profile_id: int) -> str:
+    return f"/scholar-images/{scholar_profile_id}/upload"
+
+
+def serialize_scholar(profile) -> dict[str, object]:
+    uploaded_image_url = None
+    if profile.profile_image_upload_path:
+        uploaded_image_url = uploaded_image_media_path(int(profile.id))
+
+    profile_image_url, profile_image_source = scholar_search_hints.resolve_profile_image(
+        profile,
+        uploaded_image_url=uploaded_image_url,
+    )
+
+    return {
+        "id": int(profile.id),
+        "scholar_id": profile.scholar_id,
+        "display_name": profile.display_name,
+        "profile_image_url": profile_image_url,
+        "profile_image_source": profile_image_source,
+        "is_enabled": bool(profile.is_enabled),
+        "baseline_completed": bool(profile.baseline_completed),
+        "last_run_dt": profile.last_run_dt,
+        "last_run_status": (profile.last_run_status.value if profile.last_run_status is not None else None),
+    }
+
+
+async def hydrate_scholar_metadata_if_needed(
+    db_session: AsyncSession,
+    *,
+    profile,
+    source: ScholarSource,
+    user_id: int,
+):
+    if not needs_metadata_hydration(profile):
+        return profile
+
+    should_skip, remaining_seconds = is_create_hydration_rate_limited()
+    if should_skip:
+        structured_log(
+            logger,
+            "info",
+            "api.scholars.create_metadata_hydration_skipped",
+            reason="scholar_request_throttle_active",
+            user_id=user_id,
+            scholar_profile_id=profile.id,
+            retry_after_seconds=round(remaining_seconds, 3),
+        )
+        return profile
+
+    try:
+        return await asyncio.wait_for(
+            scholar_service.hydrate_profile_metadata(
+                db_session,
+                profile=profile,
+                source=source,
+            ),
+            timeout=CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS,
+        )
+    except TimeoutError:
+        structured_log(
+            logger,
+            "info",
+            "api.scholars.create_metadata_hydration_skipped",
+            reason="create_timeout",
+            user_id=user_id,
+            scholar_profile_id=profile.id,
+        )
+    except Exception:
+        structured_log(
+            logger,
+            "warning",
+            "api.scholars.create_metadata_hydration_failed",
+            user_id=user_id,
+            scholar_profile_id=profile.id,
+        )
+    return profile
+
+
+def search_kwargs() -> dict[str, Any]:
+    return {
+        "network_error_retries": settings.ingestion_network_error_retries,
+        "retry_backoff_seconds": settings.ingestion_retry_backoff_seconds,
+        "search_enabled": settings.scholar_name_search_enabled,
+        "cache_ttl_seconds": settings.scholar_name_search_cache_ttl_seconds,
+        "blocked_cache_ttl_seconds": settings.scholar_name_search_blocked_cache_ttl_seconds,
+        "cache_max_entries": settings.scholar_name_search_cache_max_entries,
+        "min_interval_seconds": settings.scholar_name_search_min_interval_seconds,
+        "interval_jitter_seconds": settings.scholar_name_search_interval_jitter_seconds,
+        "cooldown_block_threshold": settings.scholar_name_search_cooldown_block_threshold,
+        "cooldown_seconds": settings.scholar_name_search_cooldown_seconds,
+        "retry_alert_threshold": settings.scholar_name_search_alert_retry_count_threshold,
+        "cooldown_rejection_alert_threshold": (settings.scholar_name_search_alert_cooldown_rejections_threshold),
+    }
+
+
+def search_response_data(query: str, parsed) -> dict[str, object]:
+    return {
+        "query": query.strip(),
+        "state": parsed.state.value,
+        "state_reason": parsed.state_reason,
+        "action_hint": scholar_search_hints.scrape_state_hint(
+            state=parsed.state,
+            state_reason=parsed.state_reason,
+        ),
+        "candidates": [
+            {
+                "scholar_id": item.scholar_id,
+                "display_name": item.display_name,
+                "affiliation": item.affiliation,
+                "email_domain": item.email_domain,
+                "cited_by_count": item.cited_by_count,
+                "interests": item.interests,
+                "profile_url": item.profile_url,
+                "profile_image_url": item.profile_image_url,
+            }
+            for item in parsed.candidates
+        ],
+        "warnings": parsed.warnings,
+    }
+
+
+async def read_uploaded_image(image: UploadFile) -> bytes:
+    try:
+        return await image.read()
+    finally:
+        await image.close()
+
+
+async def require_user_profile(
+    db_session: AsyncSession,
+    *,
+    user_id: int,
+    scholar_profile_id: int,
+):
+    profile = await scholar_service.get_user_scholar_by_id(
+        db_session,
+        user_id=user_id,
+        scholar_profile_id=scholar_profile_id,
+    )
+    if profile is None:
+        raise ApiException(
+            status_code=404,
+            code="scholar_not_found",
+            message="Scholar not found.",
+        )
+    return profile

app/api/routers/scholars.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
-import asyncio
 import logging
-from typing import Any
 
 from fastapi import APIRouter, Depends, File, Query, Request, UploadFile
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -10,6 +8,15 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.api.deps import get_api_current_user
 from app.api.errors import ApiException
 from app.api.responses import success_payload
+from app.api.routers.scholar_helpers import (
+    enqueue_initial_scrape_job_for_scholar,
+    hydrate_scholar_metadata_if_needed,
+    read_uploaded_image,
+    require_user_profile,
+    search_kwargs,
+    search_response_data,
+    serialize_scholar,
+)
 from app.api.runtime_deps import get_scholar_source
 from app.api.schemas import (
     DataExportEnvelope,
@@ -25,231 +32,14 @@ from app.api.schemas import (
 from app.db.models import User
 from app.db.session import get_db_session
 from app.logging_utils import structured_log
-from app.services.ingestion import queue as ingestion_queue_service
 from app.services.portability import application as import_export_service
-from app.services.scholar import rate_limit as scholar_rate_limit
 from app.services.scholar.source import ScholarSource
 from app.services.scholars import application as scholar_service
-from app.services.scholars import search_hints as scholar_search_hints
 from app.settings import settings
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/scholars", tags=["api-scholars"])
-CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS = 5.0
-INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS = 0
-INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON = "scholar_added_initial_scrape"
-
-
-def _needs_metadata_hydration(profile) -> bool:
-    if not profile.profile_image_url:
-        return True
-    return not (profile.display_name or "").strip()
-
-
-def _is_create_hydration_rate_limited() -> tuple[bool, float]:
-    remaining_seconds = scholar_rate_limit.remaining_scholar_slot_seconds(
-        min_interval_seconds=float(settings.ingestion_min_request_delay_seconds),
-    )
-    return remaining_seconds > 0, remaining_seconds
-
-
-def _auto_enqueue_new_scholar_enabled() -> bool:
-    if not settings.scheduler_enabled:
-        return False
-    if not settings.ingestion_automation_allowed:
-        return False
-    return bool(settings.ingestion_continuation_queue_enabled)
-
-
-async def _enqueue_initial_scrape_job_for_scholar(
-    db_session: AsyncSession,
-    *,
-    profile,
-    user_id: int,
-) -> bool:
-    if not _auto_enqueue_new_scholar_enabled():
-        return False
-    try:
-        await ingestion_queue_service.upsert_job(
-            db_session,
-            user_id=user_id,
-            scholar_profile_id=int(profile.id),
-            resume_cstart=0,
-            reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON,
-            run_id=None,
-            delay_seconds=INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS,
-        )
-        await db_session.commit()
-    except Exception:
-        await db_session.rollback()
-        structured_log(
-            logger,
-            "warning",
-            "api.scholars.initial_scrape_enqueue_failed",
-            user_id=user_id,
-            scholar_profile_id=profile.id,
-        )
-        return False
-
-    structured_log(
-        logger,
-        "info",
-        "api.scholars.initial_scrape_enqueued",
-        user_id=user_id,
-        scholar_profile_id=profile.id,
-        reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON,
-    )
-    return True
-
-
-def _uploaded_image_media_path(scholar_profile_id: int) -> str:
-    return f"/scholar-images/{scholar_profile_id}/upload"
-
-
-def _serialize_scholar(profile) -> dict[str, object]:
-    uploaded_image_url = None
-    if profile.profile_image_upload_path:
-        uploaded_image_url = _uploaded_image_media_path(int(profile.id))
-
-    profile_image_url, profile_image_source = scholar_search_hints.resolve_profile_image(
-        profile,
-        uploaded_image_url=uploaded_image_url,
-    )
-
-    return {
-        "id": int(profile.id),
-        "scholar_id": profile.scholar_id,
-        "display_name": profile.display_name,
-        "profile_image_url": profile_image_url,
-        "profile_image_source": profile_image_source,
-        "is_enabled": bool(profile.is_enabled),
-        "baseline_completed": bool(profile.baseline_completed),
-        "last_run_dt": profile.last_run_dt,
-        "last_run_status": (profile.last_run_status.value if profile.last_run_status is not None else None),
-    }
-
-
-async def _hydrate_scholar_metadata_if_needed(
-    db_session: AsyncSession,
-    *,
-    profile,
-    source: ScholarSource,
-    user_id: int,
-):
-    if not _needs_metadata_hydration(profile):
-        return profile
-
-    should_skip, remaining_seconds = _is_create_hydration_rate_limited()
-    if should_skip:
-        structured_log(
-            logger,
-            "info",
-            "api.scholars.create_metadata_hydration_skipped",
-            reason="scholar_request_throttle_active",
-            user_id=user_id,
-            scholar_profile_id=profile.id,
-            retry_after_seconds=round(remaining_seconds, 3),
-        )
-        return profile
-
-    try:
-        return await asyncio.wait_for(
-            scholar_service.hydrate_profile_metadata(
-                db_session,
-                profile=profile,
-                source=source,
-            ),
-            timeout=CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS,
-        )
-    except TimeoutError:
-        structured_log(
-            logger,
-            "info",
-            "api.scholars.create_metadata_hydration_skipped",
-            reason="create_timeout",
-            user_id=user_id,
-            scholar_profile_id=profile.id,
-        )
-    except Exception:
-        structured_log(
-            logger,
-            "warning",
-            "api.scholars.create_metadata_hydration_failed",
-            user_id=user_id,
-            scholar_profile_id=profile.id,
-        )
-    return profile
-
-
-def _search_kwargs() -> dict[str, Any]:
-    return {
-        "network_error_retries": settings.ingestion_network_error_retries,
-        "retry_backoff_seconds": settings.ingestion_retry_backoff_seconds,
-        "search_enabled": settings.scholar_name_search_enabled,
-        "cache_ttl_seconds": settings.scholar_name_search_cache_ttl_seconds,
-        "blocked_cache_ttl_seconds": settings.scholar_name_search_blocked_cache_ttl_seconds,
-        "cache_max_entries": settings.scholar_name_search_cache_max_entries,
-        "min_interval_seconds": settings.scholar_name_search_min_interval_seconds,
-        "interval_jitter_seconds": settings.scholar_name_search_interval_jitter_seconds,
-        "cooldown_block_threshold": settings.scholar_name_search_cooldown_block_threshold,
-        "cooldown_seconds": settings.scholar_name_search_cooldown_seconds,
-        "retry_alert_threshold": settings.scholar_name_search_alert_retry_count_threshold,
-        "cooldown_rejection_alert_threshold": (settings.scholar_name_search_alert_cooldown_rejections_threshold),
-    }
-
-
-def _search_response_data(query: str, parsed) -> dict[str, object]:
-    return {
-        "query": query.strip(),
-        "state": parsed.state.value,
-        "state_reason": parsed.state_reason,
-        "action_hint": scholar_search_hints.scrape_state_hint(
-            state=parsed.state,
-            state_reason=parsed.state_reason,
-        ),
-        "candidates": [
-            {
-                "scholar_id": item.scholar_id,
-                "display_name": item.display_name,
-                "affiliation": item.affiliation,
-                "email_domain": item.email_domain,
-                "cited_by_count": item.cited_by_count,
-                "interests": item.interests,
-                "profile_url": item.profile_url,
-                "profile_image_url": item.profile_image_url,
-            }
-            for item in parsed.candidates
-        ],
-        "warnings": parsed.warnings,
-    }
-
-
-async def _read_uploaded_image(image: UploadFile) -> bytes:
-    try:
-        return await image.read()
-    finally:
-        await image.close()
-
-
-async def _require_user_profile(
-    db_session: AsyncSession,
-    *,
-    user_id: int,
-    scholar_profile_id: int,
-):
-    profile = await scholar_service.get_user_scholar_by_id(
-        db_session,
-        user_id=user_id,
-        scholar_profile_id=scholar_profile_id,
-    )
-    if profile is None:
-        raise ApiException(
-            status_code=404,
-            code="scholar_not_found",
-            message="Scholar not found.",
-        )
-    return profile
 
 
 @router.get(
@@ -268,7 +58,7 @@ async def list_scholars(
     return success_payload(
         request,
         data={
-            "scholars": [_serialize_scholar(profile) for profile in scholars],
+            "scholars": [serialize_scholar(profile) for profile in scholars],
         },
     )
 
@@ -354,13 +144,13 @@ async def create_scholar(
             message=str(exc),
         ) from exc
     structured_log(logger, "info", "api.scholars.created", user_id=current_user.id, scholar_profile_id=created.id)
-    did_queue_initial_scrape = await _enqueue_initial_scrape_job_for_scholar(
+    did_queue_initial_scrape = await enqueue_initial_scrape_job_for_scholar(
         db_session,
         profile=created,
         user_id=current_user.id,
     )
     if not did_queue_initial_scrape:
-        created = await _hydrate_scholar_metadata_if_needed(
+        created = await hydrate_scholar_metadata_if_needed(
             db_session,
             profile=created,
             source=source,
@@ -369,7 +159,7 @@ async def create_scholar(
 
     return success_payload(
         request,
-        data=_serialize_scholar(created),
+        data=serialize_scholar(created),
     )
 
 
@@ -391,7 +181,7 @@ async def search_scholars(
             db_session=db_session,
             query=query,
             limit=limit,
-            **_search_kwargs(),
+            **search_kwargs(),
         )
     except scholar_service.ScholarServiceError as exc:
         raise ApiException(
@@ -411,7 +201,7 @@ async def search_scholars(
     )
     return success_payload(
         request,
-        data=_search_response_data(query, parsed),
+        data=search_response_data(query, parsed),
     )
 
 
@@ -447,7 +237,7 @@ async def toggle_scholar(
     )
     return success_payload(
         request,
-        data=_serialize_scholar(updated),
+        data=serialize_scholar(updated),
     )
 
 
@@ -528,7 +318,7 @@ async def update_scholar_image_url(
     )
     return success_payload(
         request,
-        data=_serialize_scholar(updated),
+        data=serialize_scholar(updated),
     )
 
 
@@ -543,13 +333,13 @@ async def upload_scholar_image(
     db_session: AsyncSession = Depends(get_db_session),
     current_user: User = Depends(get_api_current_user),
 ):
-    profile = await _require_user_profile(
+    profile = await require_user_profile(
         db_session,
         user_id=current_user.id,
         scholar_profile_id=scholar_profile_id,
     )
 
-    image_bytes = await _read_uploaded_image(image)
+    image_bytes = await read_uploaded_image(image)
     try:
         updated = await scholar_service.set_profile_image_upload(
             db_session,
@@ -578,7 +368,7 @@ async def upload_scholar_image(
     )
     return success_payload(
         request,
-        data=_serialize_scholar(updated),
+        data=serialize_scholar(updated),
     )
 
 
@@ -612,5 +402,5 @@ async def clear_scholar_image_customization(
     structured_log(logger, "info", "api.scholars.image_cleared", user_id=current_user.id, scholar_profile_id=updated.id)
     return success_payload(
         request,
-        data=_serialize_scholar(updated),
+        data=serialize_scholar(updated),
     )

app/services/ingestion/queue_runner.py

@@ -0,0 +1,379 @@
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+
+from app.db.models import QueueItemStatus, RunTriggerType, ScholarProfile, UserSetting
+from app.db.session import get_session_factory
+from app.logging_utils import structured_log
+from app.services.ingestion import queue as queue_service
+from app.services.ingestion.application import (
+    RunAlreadyInProgressError,
+    RunBlockedBySafetyPolicyError,
+    ScholarIngestionService,
+)
+from app.services.scholar.source import LiveScholarSource
+from app.settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+def _effective_request_delay_seconds(value: int | None, *, floor: int) -> int:
+    try:
+        parsed = int(value) if value is not None else floor
+    except (TypeError, ValueError):
+        parsed = floor
+    return max(floor, parsed)
+
+
+class QueueJobRunner:
+    def __init__(
+        self,
+        *,
+        tick_seconds: int,
+        network_error_retries: int,
+        retry_backoff_seconds: float,
+        max_pages_per_scholar: int,
+        page_size: int,
+        continuation_queue_enabled: bool,
+        continuation_base_delay_seconds: int,
+        continuation_max_delay_seconds: int,
+        continuation_max_attempts: int,
+        queue_batch_size: int,
+    ) -> None:
+        self._tick_seconds = tick_seconds
+        self._network_error_retries = network_error_retries
+        self._retry_backoff_seconds = retry_backoff_seconds
+        self._max_pages_per_scholar = max_pages_per_scholar
+        self._page_size = page_size
+        self._continuation_queue_enabled = continuation_queue_enabled
+        self._continuation_base_delay_seconds = continuation_base_delay_seconds
+        self._continuation_max_delay_seconds = continuation_max_delay_seconds
+        self._continuation_max_attempts = continuation_max_attempts
+        self._queue_batch_size = queue_batch_size
+        self._source = LiveScholarSource()
+
+    async def drain_continuation_queue(self) -> None:
+        now = datetime.now(UTC)
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            jobs = await queue_service.list_due_jobs(
+                session,
+                now=now,
+                limit=self._queue_batch_size,
+            )
+        for job in jobs:
+            await self._run_queue_job(job)
+
+    async def _drop_queue_job_if_max_attempts(
+        self,
+        job: queue_service.ContinuationQueueJob,
+    ) -> bool:
+        if job.attempt_count < self._continuation_max_attempts:
+            return False
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            dropped = await queue_service.mark_dropped(
+                session,
+                job_id=job.id,
+                reason="max_attempts_reached",
+            )
+            await session.commit()
+        if dropped is not None:
+            structured_log(
+                logger,
+                "warning",
+                "scheduler.queue_item_dropped_max_attempts",
+                queue_item_id=job.id,
+                user_id=job.user_id,
+                scholar_profile_id=job.scholar_profile_id,
+                attempt_count=job.attempt_count,
+                max_attempts=self._continuation_max_attempts,
+            )
+        return True
+
+    async def _mark_queue_job_retrying(
+        self,
+        job: queue_service.ContinuationQueueJob,
+    ) -> bool:
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            queue_item = await queue_service.mark_retrying(session, job_id=job.id)
+            await session.commit()
+        if queue_item is None:
+            return False
+        return queue_item.status != QueueItemStatus.DROPPED.value
+
+    async def _queue_job_has_available_scholar(
+        self,
+        job: queue_service.ContinuationQueueJob,
+    ) -> bool:
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            scholar_result = await session.execute(
+                select(ScholarProfile.id).where(
+                    ScholarProfile.user_id == job.user_id,
+                    ScholarProfile.id == job.scholar_profile_id,
+                    ScholarProfile.is_enabled.is_(True),
+                )
+            )
+            scholar_id = scholar_result.scalar_one_or_none()
+            if scholar_id is not None:
+                return True
+            dropped = await queue_service.mark_dropped(
+                session,
+                job_id=job.id,
+                reason="scholar_unavailable",
+            )
+            await session.commit()
+        if dropped is not None:
+            structured_log(
+                logger,
+                "info",
+                "scheduler.queue_item_dropped_scholar_unavailable",
+                queue_item_id=job.id,
+                user_id=job.user_id,
+                scholar_profile_id=job.scholar_profile_id,
+            )
+        return False
+
+    async def _reschedule_queue_job_lock_active(self, job: queue_service.ContinuationQueueJob) -> None:
+        session_factory = get_session_factory()
+        async with session_factory() as recovery_session:
+            await queue_service.reschedule_job(
+                recovery_session,
+                job_id=job.id,
+                delay_seconds=max(self._tick_seconds, 15),
+                reason="user_run_lock_active",
+                error="run_already_in_progress",
+            )
+            await recovery_session.commit()
+        structured_log(
+            logger,
+            "info",
+            "scheduler.queue_item_deferred_lock",
+            queue_item_id=job.id,
+            user_id=job.user_id,
+        )
+
+    async def _reschedule_queue_job_safety_cooldown(
+        self,
+        job: queue_service.ContinuationQueueJob,
+        exc: RunBlockedBySafetyPolicyError,
+    ) -> None:
+        cooldown_remaining_seconds = max(
+            self._tick_seconds,
+            int(exc.safety_state.get("cooldown_remaining_seconds") or 0),
+        )
+        session_factory = get_session_factory()
+        async with session_factory() as recovery_session:
+            await queue_service.reschedule_job(
+                recovery_session,
+                job_id=job.id,
+                delay_seconds=max(self._tick_seconds, cooldown_remaining_seconds),
+                reason="scrape_safety_cooldown",
+                error=str(exc.message),
+            )
+            await recovery_session.commit()
+        structured_log(
+            logger,
+            "info",
+            "scheduler.queue_item_deferred_safety_cooldown",
+            queue_item_id=job.id,
+            user_id=job.user_id,
+            reason=exc.safety_state.get("cooldown_reason"),
+            cooldown_remaining_seconds=cooldown_remaining_seconds,
+        )
+
+    async def _reschedule_queue_job_after_exception(
+        self,
+        job: queue_service.ContinuationQueueJob,
+        *,
+        exc: Exception,
+    ) -> None:
+        session_factory = get_session_factory()
+        async with session_factory() as recovery_session:
+            queue_item = await queue_service.increment_attempt_count(recovery_session, job_id=job.id)
+            if queue_item is None:
+                await recovery_session.commit()
+                return
+            if int(queue_item.attempt_count) >= self._continuation_max_attempts:
+                await queue_service.mark_dropped(
+                    recovery_session,
+                    job_id=job.id,
+                    reason="scheduler_exception_max_attempts",
+                    error=str(exc),
+                )
+                await recovery_session.commit()
+                structured_log(
+                    logger,
+                    "warning",
+                    "scheduler.queue_item_dropped_after_exception",
+                    queue_item_id=job.id,
+                    user_id=job.user_id,
+                    attempt_count=queue_item.attempt_count,
+                )
+                return
+            delay_seconds = queue_service.compute_backoff_seconds(
+                base_seconds=self._continuation_base_delay_seconds,
+                attempt_count=int(queue_item.attempt_count),
+                max_seconds=self._continuation_max_delay_seconds,
+            )
+            await queue_service.reschedule_job(
+                recovery_session,
+                job_id=job.id,
+                delay_seconds=delay_seconds,
+                reason="scheduler_exception",
+                error=str(exc),
+            )
+            await recovery_session.commit()
+        logger.exception("scheduler.queue_item_run_failed", extra={"queue_item_id": job.id, "user_id": job.user_id})
+
+    async def _load_request_delay_for_user(self, user_id: int, *, floor: int) -> int:
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            result = await session.execute(
+                select(UserSetting.request_delay_seconds).where(UserSetting.user_id == user_id)
+            )
+            delay = result.scalar_one_or_none()
+        return _effective_request_delay_seconds(delay, floor=floor)
+
+    async def _run_ingestion_for_queue_job(
+        self,
+        job: queue_service.ContinuationQueueJob,
+        *,
+        request_delay_floor: int,
+    ):
+        request_delay_seconds = await self._load_request_delay_for_user(job.user_id, floor=request_delay_floor)
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            ingestion = ScholarIngestionService(source=self._source)
+            try:
+                return await ingestion.run_for_user(
+                    session,
+                    user_id=job.user_id,
+                    trigger_type=RunTriggerType.SCHEDULED,
+                    request_delay_seconds=request_delay_seconds,
+                    network_error_retries=self._network_error_retries,
+                    retry_backoff_seconds=self._retry_backoff_seconds,
+                    rate_limit_retries=settings.ingestion_rate_limit_retries,
+                    rate_limit_backoff_seconds=settings.ingestion_rate_limit_backoff_seconds,
+                    max_pages_per_scholar=self._max_pages_per_scholar,
+                    page_size=self._page_size,
+                    scholar_profile_ids={job.scholar_profile_id},
+                    start_cstart_by_scholar_id={job.scholar_profile_id: job.resume_cstart},
+                    auto_queue_continuations=self._continuation_queue_enabled,
+                    queue_delay_seconds=self._continuation_base_delay_seconds,
+                    alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold,
+                    alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold,
+                    alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold,
+                )
+            except RunAlreadyInProgressError:
+                await session.rollback()
+                await self._reschedule_queue_job_lock_active(job)
+            except RunBlockedBySafetyPolicyError as exc:
+                await session.rollback()
+                await self._reschedule_queue_job_safety_cooldown(job, exc)
+            except Exception as exc:
+                await session.rollback()
+                await self._reschedule_queue_job_after_exception(job, exc=exc)
+        return None
+
+    async def _finalize_queue_job_after_run(self, job: queue_service.ContinuationQueueJob, run_summary) -> None:
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            if int(run_summary.failed_count) <= 0:
+                queue_item = await queue_service.reset_attempt_count(session, job_id=job.id)
+                await session.commit()
+                if queue_item is None:
+                    structured_log(
+                        logger,
+                        "info",
+                        "scheduler.queue_item_resolved",
+                        queue_item_id=job.id,
+                        user_id=job.user_id,
+                        run_id=run_summary.crawl_run_id,
+                        status=run_summary.status.value,
+                    )
+                    return
+                structured_log(
+                    logger,
+                    "info",
+                    "scheduler.queue_item_progressed",
+                    queue_item_id=job.id,
+                    user_id=job.user_id,
+                    run_id=run_summary.crawl_run_id,
+                    status=run_summary.status.value,
+                    attempt_count=int(queue_item.attempt_count),
+                )
+                return
+            queue_item = await queue_service.increment_attempt_count(session, job_id=job.id)
+            if queue_item is None:
+                await session.commit()
+                structured_log(
+                    logger,
+                    "info",
+                    "scheduler.queue_item_resolved",
+                    queue_item_id=job.id,
+                    user_id=job.user_id,
+                    run_id=run_summary.crawl_run_id,
+                    status=run_summary.status.value,
+                )
+                return
+            if int(queue_item.attempt_count) >= self._continuation_max_attempts:
+                await queue_service.mark_dropped(session, job_id=job.id, reason="max_attempts_after_run")
+                await session.commit()
+                structured_log(
+                    logger,
+                    "warning",
+                    "scheduler.queue_item_dropped_max_attempts_after_run",
+                    queue_item_id=job.id,
+                    user_id=job.user_id,
+                    attempt_count=queue_item.attempt_count,
+                    run_id=run_summary.crawl_run_id,
+                    status=run_summary.status.value,
+                )
+                return
+            delay_seconds = queue_service.compute_backoff_seconds(
+                base_seconds=self._continuation_base_delay_seconds,
+                attempt_count=int(queue_item.attempt_count),
+                max_seconds=self._continuation_max_delay_seconds,
+            )
+            await queue_service.reschedule_job(
+                session,
+                job_id=job.id,
+                delay_seconds=delay_seconds,
+                reason=queue_item.reason,
+                error=queue_item.last_error,
+            )
+            await session.commit()
+        structured_log(
+            logger,
+            "info",
+            "scheduler.queue_item_rescheduled_failed",
+            queue_item_id=job.id,
+            user_id=job.user_id,
+            run_id=run_summary.crawl_run_id,
+            status=run_summary.status.value,
+            attempt_count=int(queue_item.attempt_count),
+            delay_seconds=delay_seconds,
+        )
+
+    async def _run_queue_job(self, job: queue_service.ContinuationQueueJob) -> None:
+        if await self._drop_queue_job_if_max_attempts(job):
+            return
+        if not await self._mark_queue_job_retrying(job):
+            return
+        if not await self._queue_job_has_available_scholar(job):
+            return
+        from app.services.settings import application as user_settings_service
+
+        request_delay_floor = user_settings_service.resolve_request_delay_minimum(
+            settings.ingestion_min_request_delay_seconds,
+        )
+        run_summary = await self._run_ingestion_for_queue_job(job, request_delay_floor=request_delay_floor)
+        if run_summary is None:
+            return
+        await self._finalize_queue_job_after_run(job, run_summary)

app/services/ingestion/scheduler.py

@@ -6,24 +6,21 @@ from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 
 from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db.models import (
     CrawlRun,
-    QueueItemStatus,
     RunTriggerType,
-    ScholarProfile,
     User,
     UserSetting,
 )
 from app.db.session import get_session_factory
 from app.logging_utils import structured_log
-from app.services.ingestion import queue as queue_service
 from app.services.ingestion.application import (
     RunAlreadyInProgressError,
     RunBlockedBySafetyPolicyError,
     ScholarIngestionService,
 )
+from app.services.ingestion.queue_runner import QueueJobRunner
 from app.services.scholar.source import LiveScholarSource
 from app.services.settings import application as user_settings_service
 from app.settings import settings
@@ -85,6 +82,18 @@ class SchedulerService:
         self._queue_batch_size = max(1, int(queue_batch_size))
         self._task: asyncio.Task[None] | None = None
         self._source = LiveScholarSource()
+        self._queue_runner = QueueJobRunner(
+            tick_seconds=self._tick_seconds,
+            network_error_retries=self._network_error_retries,
+            retry_backoff_seconds=self._retry_backoff_seconds,
+            max_pages_per_scholar=self._max_pages_per_scholar,
+            page_size=self._page_size,
+            continuation_queue_enabled=self._continuation_queue_enabled,
+            continuation_base_delay_seconds=self._continuation_base_delay_seconds,
+            continuation_max_delay_seconds=self._continuation_max_delay_seconds,
+            continuation_max_attempts=self._continuation_max_attempts,
+            queue_batch_size=self._queue_batch_size,
+        )
 
     async def start(self) -> None:
         if not self._enabled:
@@ -136,7 +145,7 @@ class SchedulerService:
 
     async def _tick_once(self) -> None:
         if self._continuation_queue_enabled:
-            await self._drain_continuation_queue()
+            await self._queue_runner.drain_continuation_queue()
 
         await self._drain_pdf_queue()
 
@@ -281,313 +290,6 @@ class SchedulerService:
             new_publication_count=run_summary.new_publication_count,
         )
 
-    async def _drain_continuation_queue(self) -> None:
-        now = datetime.now(UTC)
-        session_factory = get_session_factory()
-        async with session_factory() as session:
-            jobs = await queue_service.list_due_jobs(
-                session,
-                now=now,
-                limit=self._queue_batch_size,
-            )
-        for job in jobs:
-            await self._run_queue_job(job)
-
-    async def _drop_queue_job_if_max_attempts(
-        self,
-        job: queue_service.ContinuationQueueJob,
-    ) -> bool:
-        if job.attempt_count < self._continuation_max_attempts:
-            return False
-        session_factory = get_session_factory()
-        async with session_factory() as session:
-            dropped = await queue_service.mark_dropped(
-                session,
-                job_id=job.id,
-                reason="max_attempts_reached",
-            )
-            await session.commit()
-        if dropped is not None:
-            structured_log(
-                logger,
-                "warning",
-                "scheduler.queue_item_dropped_max_attempts",
-                queue_item_id=job.id,
-                user_id=job.user_id,
-                scholar_profile_id=job.scholar_profile_id,
-                attempt_count=job.attempt_count,
-                max_attempts=self._continuation_max_attempts,
-            )
-        return True
-
-    async def _mark_queue_job_retrying(
-        self,
-        job: queue_service.ContinuationQueueJob,
-    ) -> bool:
-        session_factory = get_session_factory()
-        async with session_factory() as session:
-            queue_item = await queue_service.mark_retrying(session, job_id=job.id)
-            await session.commit()
-        if queue_item is None:
-            return False
-        return queue_item.status != QueueItemStatus.DROPPED.value
-
-    async def _queue_job_has_available_scholar(
-        self,
-        job: queue_service.ContinuationQueueJob,
-    ) -> bool:
-        session_factory = get_session_factory()
-        async with session_factory() as session:
-            scholar_result = await session.execute(
-                select(ScholarProfile.id).where(
-                    ScholarProfile.user_id == job.user_id,
-                    ScholarProfile.id == job.scholar_profile_id,
-                    ScholarProfile.is_enabled.is_(True),
-                )
-            )
-            scholar_id = scholar_result.scalar_one_or_none()
-            if scholar_id is not None:
-                return True
-            dropped = await queue_service.mark_dropped(
-                session,
-                job_id=job.id,
-                reason="scholar_unavailable",
-            )
-            await session.commit()
-        if dropped is not None:
-            structured_log(
-                logger,
-                "info",
-                "scheduler.queue_item_dropped_scholar_unavailable",
-                queue_item_id=job.id,
-                user_id=job.user_id,
-                scholar_profile_id=job.scholar_profile_id,
-            )
-        return False
-
-    async def _reschedule_queue_job_lock_active(self, job: queue_service.ContinuationQueueJob) -> None:
-        session_factory = get_session_factory()
-        async with session_factory() as recovery_session:
-            await queue_service.reschedule_job(
-                recovery_session,
-                job_id=job.id,
-                delay_seconds=max(self._tick_seconds, 15),
-                reason="user_run_lock_active",
-                error="run_already_in_progress",
-            )
-            await recovery_session.commit()
-        structured_log(
-            logger,
-            "info",
-            "scheduler.queue_item_deferred_lock",
-            queue_item_id=job.id,
-            user_id=job.user_id,
-        )
-
-    async def _reschedule_queue_job_safety_cooldown(
-        self,
-        job: queue_service.ContinuationQueueJob,
-        exc: RunBlockedBySafetyPolicyError,
-    ) -> None:
-        cooldown_remaining_seconds = max(
-            self._tick_seconds,
-            int(exc.safety_state.get("cooldown_remaining_seconds") or 0),
-        )
-        session_factory = get_session_factory()
-        async with session_factory() as recovery_session:
-            await queue_service.reschedule_job(
-                recovery_session,
-                job_id=job.id,
-                delay_seconds=max(self._tick_seconds, cooldown_remaining_seconds),
-                reason="scrape_safety_cooldown",
-                error=str(exc.message),
-            )
-            await recovery_session.commit()
-        structured_log(
-            logger,
-            "info",
-            "scheduler.queue_item_deferred_safety_cooldown",
-            queue_item_id=job.id,
-            user_id=job.user_id,
-            reason=exc.safety_state.get("cooldown_reason"),
-            cooldown_remaining_seconds=cooldown_remaining_seconds,
-        )
-
-    async def _reschedule_queue_job_after_exception(
-        self,
-        job: queue_service.ContinuationQueueJob,
-        *,
-        exc: Exception,
-    ) -> None:
-        session_factory = get_session_factory()
-        async with session_factory() as recovery_session:
-            queue_item = await queue_service.increment_attempt_count(recovery_session, job_id=job.id)
-            if queue_item is None:
-                await recovery_session.commit()
-                return
-            if int(queue_item.attempt_count) >= self._continuation_max_attempts:
-                await queue_service.mark_dropped(
-                    recovery_session,
-                    job_id=job.id,
-                    reason="scheduler_exception_max_attempts",
-                    error=str(exc),
-                )
-                await recovery_session.commit()
-                structured_log(
-                    logger,
-                    "warning",
-                    "scheduler.queue_item_dropped_after_exception",
-                    queue_item_id=job.id,
-                    user_id=job.user_id,
-                    attempt_count=queue_item.attempt_count,
-                )
-                return
-            delay_seconds = queue_service.compute_backoff_seconds(
-                base_seconds=self._continuation_base_delay_seconds,
-                attempt_count=int(queue_item.attempt_count),
-                max_seconds=self._continuation_max_delay_seconds,
-            )
-            await queue_service.reschedule_job(
-                recovery_session,
-                job_id=job.id,
-                delay_seconds=delay_seconds,
-                reason="scheduler_exception",
-                error=str(exc),
-            )
-            await recovery_session.commit()
-        logger.exception("scheduler.queue_item_run_failed", extra={"queue_item_id": job.id, "user_id": job.user_id})
-
-    async def _run_ingestion_for_queue_job(
-        self,
-        job: queue_service.ContinuationQueueJob,
-    ):
-        session_factory = get_session_factory()
-        async with session_factory() as session:
-            request_delay_seconds = await self._load_request_delay_for_user(session, user_id=job.user_id)
-            ingestion = ScholarIngestionService(source=self._source)
-            try:
-                return await ingestion.run_for_user(
-                    session,
-                    user_id=job.user_id,
-                    trigger_type=RunTriggerType.SCHEDULED,
-                    request_delay_seconds=request_delay_seconds,
-                    network_error_retries=self._network_error_retries,
-                    retry_backoff_seconds=self._retry_backoff_seconds,
-                    rate_limit_retries=settings.ingestion_rate_limit_retries,
-                    rate_limit_backoff_seconds=settings.ingestion_rate_limit_backoff_seconds,
-                    max_pages_per_scholar=self._max_pages_per_scholar,
-                    page_size=self._page_size,
-                    scholar_profile_ids={job.scholar_profile_id},
-                    start_cstart_by_scholar_id={job.scholar_profile_id: job.resume_cstart},
-                    auto_queue_continuations=self._continuation_queue_enabled,
-                    queue_delay_seconds=self._continuation_base_delay_seconds,
-                    alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold,
-                    alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold,
-                    alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold,
-                )
-            except RunAlreadyInProgressError:
-                await session.rollback()
-                await self._reschedule_queue_job_lock_active(job)
-            except RunBlockedBySafetyPolicyError as exc:
-                await session.rollback()
-                await self._reschedule_queue_job_safety_cooldown(job, exc)
-            except Exception as exc:
-                await session.rollback()
-                await self._reschedule_queue_job_after_exception(job, exc=exc)
-        return None
-
-    async def _finalize_queue_job_after_run(self, job: queue_service.ContinuationQueueJob, run_summary) -> None:
-        session_factory = get_session_factory()
-        async with session_factory() as session:
-            if int(run_summary.failed_count) <= 0:
-                queue_item = await queue_service.reset_attempt_count(session, job_id=job.id)
-                await session.commit()
-                if queue_item is None:
-                    structured_log(
-                        logger,
-                        "info",
-                        "scheduler.queue_item_resolved",
-                        queue_item_id=job.id,
-                        user_id=job.user_id,
-                        run_id=run_summary.crawl_run_id,
-                        status=run_summary.status.value,
-                    )
-                    return
-                structured_log(
-                    logger,
-                    "info",
-                    "scheduler.queue_item_progressed",
-                    queue_item_id=job.id,
-                    user_id=job.user_id,
-                    run_id=run_summary.crawl_run_id,
-                    status=run_summary.status.value,
-                    attempt_count=int(queue_item.attempt_count),
-                )
-                return
-            queue_item = await queue_service.increment_attempt_count(session, job_id=job.id)
-            if queue_item is None:
-                await session.commit()
-                structured_log(
-                    logger,
-                    "info",
-                    "scheduler.queue_item_resolved",
-                    queue_item_id=job.id,
-                    user_id=job.user_id,
-                    run_id=run_summary.crawl_run_id,
-                    status=run_summary.status.value,
-                )
-                return
-            if int(queue_item.attempt_count) >= self._continuation_max_attempts:
-                await queue_service.mark_dropped(session, job_id=job.id, reason="max_attempts_after_run")
-                await session.commit()
-                structured_log(
-                    logger,
-                    "warning",
-                    "scheduler.queue_item_dropped_max_attempts_after_run",
-                    queue_item_id=job.id,
-                    user_id=job.user_id,
-                    attempt_count=queue_item.attempt_count,
-                    run_id=run_summary.crawl_run_id,
-                    status=run_summary.status.value,
-                )
-                return
-            delay_seconds = queue_service.compute_backoff_seconds(
-                base_seconds=self._continuation_base_delay_seconds,
-                attempt_count=int(queue_item.attempt_count),
-                max_seconds=self._continuation_max_delay_seconds,
-            )
-            await queue_service.reschedule_job(
-                session,
-                job_id=job.id,
-                delay_seconds=delay_seconds,
-                reason=queue_item.reason,
-                error=queue_item.last_error,
-            )
-            await session.commit()
-        structured_log(
-            logger,
-            "info",
-            "scheduler.queue_item_rescheduled_failed",
-            queue_item_id=job.id,
-            user_id=job.user_id,
-            run_id=run_summary.crawl_run_id,
-            status=run_summary.status.value,
-            attempt_count=int(queue_item.attempt_count),
-            delay_seconds=delay_seconds,
-        )
-
-    async def _run_queue_job(self, job: queue_service.ContinuationQueueJob) -> None:
-        if await self._drop_queue_job_if_max_attempts(job):
-            return
-        if not await self._mark_queue_job_retrying(job):
-            return
-        if not await self._queue_job_has_available_scholar(job):
-            return
-        run_summary = await self._run_ingestion_for_queue_job(job)
-        if run_summary is None:
-            return
-        await self._finalize_queue_job_after_run(job, run_summary)
-
     async def _drain_pdf_queue(self) -> None:
         from app.services.publications.pdf_queue import drain_ready_jobs
 
@@ -608,15 +310,3 @@ class SchedulerService:
                     )
             except Exception:
                 logger.exception("scheduler.pdf_queue_drain_failed", extra={})
-
-    async def _load_request_delay_for_user(
-        self,
-        db_session: AsyncSession,
-        *,
-        user_id: int,
-    ) -> int:
-        result = await db_session.execute(
-            select(UserSetting.request_delay_seconds).where(UserSetting.user_id == user_id)
-        )
-        delay = result.scalar_one_or_none()
-        return _effective_request_delay_seconds(delay)

tests/unit/test_scholars_create_hydration.py

@@ -2,7 +2,7 @@ from types import SimpleNamespace
 
 import pytest
 
-from app.api.routers import scholars as scholars_router
+from app.api.routers import scholar_helpers
 
 
 class _FakeSession:
@@ -31,17 +31,17 @@ async def test_create_hydration_skips_when_global_throttle_active(
         raise AssertionError("hydrate_profile_metadata should not run when throttle is active")
 
     monkeypatch.setattr(
-        scholars_router.scholar_rate_limit,
+        scholar_helpers.scholar_rate_limit,
         "remaining_scholar_slot_seconds",
         lambda **_kwargs: 3.5,
     )
     monkeypatch.setattr(
-        scholars_router.scholar_service,
+        scholar_helpers.scholar_service,
         "hydrate_profile_metadata",
         _fail_hydration,
     )
 
-    result = await scholars_router._hydrate_scholar_metadata_if_needed(
+    result = await scholar_helpers.hydrate_scholar_metadata_if_needed(
         db_session=None,
         profile=profile,
         source=object(),
@@ -66,17 +66,17 @@ async def test_create_hydration_runs_when_throttle_is_clear(
         return profile
 
     monkeypatch.setattr(
-        scholars_router.scholar_rate_limit,
+        scholar_helpers.scholar_rate_limit,
         "remaining_scholar_slot_seconds",
         lambda **_kwargs: 0.0,
     )
     monkeypatch.setattr(
-        scholars_router.scholar_service,
+        scholar_helpers.scholar_service,
         "hydrate_profile_metadata",
         _hydrate_profile_metadata,
     )
 
-    result = await scholars_router._hydrate_scholar_metadata_if_needed(
+    result = await scholar_helpers.hydrate_scholar_metadata_if_needed(
         db_session=None,
         profile=profile,
         source=object(),
@@ -98,17 +98,17 @@ async def test_initial_scrape_job_enqueued_on_create(
         upsert_calls.append(kwargs)
 
     monkeypatch.setattr(
-        scholars_router,
+        scholar_helpers,
-        "_auto_enqueue_new_scholar_enabled",
+        "auto_enqueue_new_scholar_enabled",
         lambda: True,
     )
     monkeypatch.setattr(
-        scholars_router.ingestion_queue_service,
+        scholar_helpers.ingestion_queue_service,
         "upsert_job",
         _fake_upsert_job,
     )
 
-    queued = await scholars_router._enqueue_initial_scrape_job_for_scholar(
+    queued = await scholar_helpers.enqueue_initial_scrape_job_for_scholar(
         session,
         profile=profile,
         user_id=9,
@@ -117,7 +117,7 @@ async def test_initial_scrape_job_enqueued_on_create(
     assert queued is True
     assert session.commits == 1
     assert session.rollbacks == 0
-    assert upsert_calls[0]["reason"] == scholars_router.INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON
+    assert upsert_calls[0]["reason"] == scholar_helpers.INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON
 
 
 @pytest.mark.asyncio
@@ -128,12 +128,12 @@ async def test_initial_scrape_job_not_enqueued_when_disabled(
     session = _FakeSession()
 
     monkeypatch.setattr(
-        scholars_router,
+        scholar_helpers,
-        "_auto_enqueue_new_scholar_enabled",
+        "auto_enqueue_new_scholar_enabled",
         lambda: False,
     )
 
-    queued = await scholars_router._enqueue_initial_scrape_job_for_scholar(
+    queued = await scholar_helpers.enqueue_initial_scrape_job_for_scholar(
         session,
         profile=profile,
         user_id=11,