refactor: extract pagination engine and publication upsert from ingestion service

Split app/services/ingestion/application.py (2,955 lines) into three
focused modules:

- page_fetch.py (219 lines): PageFetcher class for single-page fetch,
  parse-or-layout-error handling, and retry with backoff
- pagination.py (486 lines): PaginationEngine class for multi-page
  orchestration, loop state management, and short-circuit detection
- publication_upsert.py (239 lines): module-level functions for
  resolve_publication, upsert_profile_publications, and all
  find/create/update helpers

Also fixes two external import paths in portability/ that incorrectly
sourced normalize_title and build_publication_url from application.py
instead of fingerprints.py.

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

app/services/ingestion/application.py

@@ -21,7 +21,6 @@ from app.db.models import (
 )
 from app.logging_utils import structured_log
 from app.services.arxiv.errors import ArxivRateLimitError
-from app.services.doi.normalize import first_doi_from_texts
 from app.services.ingestion import queue as queue_service
 from app.services.ingestion import safety as run_safety_service
 from app.services.ingestion.constants import (
@@ -37,16 +36,10 @@ from app.services.ingestion.constants import (
 )
 from app.services.ingestion.fingerprints import (
     _build_body_excerpt,
-    _dedupe_publication_candidates,
-    _next_cstart_value,
-    build_initial_page_fingerprint,
-    build_publication_fingerprint,
-    build_publication_url,
-    canonical_title_for_dedup,
-    normalize_title,
 )
+from app.services.ingestion.pagination import PaginationEngine
+from app.services.ingestion.publication_upsert import upsert_profile_publications
 from app.services.ingestion.types import (
-    PagedLoopState,
     PagedParseResult,
     RunAlertSummary,
     RunAlreadyInProgressError,
@@ -62,8 +55,6 @@ from app.services.scholar.parser import (
     ParsedProfilePage,
     ParseState,
     PublicationCandidate,
-    ScholarParserError,
-    parse_profile_page,
 )
 from app.services.scholar.source import FetchResult, ScholarSource
 from app.services.settings import application as user_settings_service
@@ -115,6 +106,7 @@ _background_tasks: set[asyncio.Task[Any]] = set()
 class ScholarIngestionService:
     def __init__(self, *, source: ScholarSource) -> None:
         self._source = source
+        self._pagination = PaginationEngine(source=source)
 
     @staticmethod
     def _effective_request_delay_seconds(value: int) -> int:
@@ -290,7 +282,9 @@ class ScholarIngestionService:
         paged_parse_result: PagedParseResult,
     ) -> None:
         parsed_page = paged_parse_result.parsed_page
-        if parsed_page.state in {ParseState.OK, ParseState.NO_RESULTS} and any(code.startswith("layout_") for code in parsed_page.warnings):
+        if parsed_page.state in {ParseState.OK, ParseState.NO_RESULTS} and any(
+            code.startswith("layout_") for code in parsed_page.warnings
+        ):
             raise RuntimeError(f"Layout warning marked as terminal for scholar_id={scholar_id}.")
         for publication in paged_parse_result.publications:
             if not publication.title.strip():
@@ -445,7 +439,7 @@ class ScholarIngestionService:
         has_partial_publication_set: bool,
     ) -> ScholarProcessingOutcome:
         # We no longer aggregate and upsert the publications here.
-        # Eager DB insertions have already saved and committed them inside `_fetch_and_parse_all_pages_with_retry` and `_paginate_loop`.
+        # Eager DB insertions have already saved and committed them inside `fetch_and_parse_all_pages` and `_paginate_loop`.
         discovered_count = paged_parse_result.discovered_publication_count
         is_partial = (
             paged_parse_result.has_more_remaining
@@ -683,7 +677,7 @@ class ScholarIngestionService:
         page_size: int,
     ) -> tuple[datetime, PagedParseResult, dict[str, Any]]:
         run_dt = datetime.now(UTC)
-        paged_parse_result = await self._fetch_and_parse_all_pages_with_retry(
+        paged_parse_result = await self._pagination.fetch_and_parse_all_pages(
             scholar=scholar,
             run=run,
             db_session=db_session,
@@ -696,6 +690,7 @@ class ScholarIngestionService:
             max_pages=max_pages_per_scholar,
             page_size=page_size,
             previous_initial_page_fingerprint_sha256=scholar.last_initial_page_fingerprint_sha256,
+            upsert_publications_fn=upsert_profile_publications,
         )
         self._assert_valid_paged_parse_result(scholar_id=scholar.scholar_id, paged_parse_result=paged_parse_result)
         self._apply_first_page_profile_metadata(scholar=scholar, paged_parse_result=paged_parse_result, run_dt=run_dt)
@@ -1701,650 +1696,6 @@ class ScholarIngestionService:
         )
         return self._run_execution_summary(run=run, scholars=scholars, progress=progress)
 
-    async def _fetch_profile_page(
-        self,
-        *,
-        scholar_id: str,
-        cstart: int,
-        page_size: int,
-    ) -> FetchResult:
-        try:
-            page_fetcher = getattr(self._source, "fetch_profile_page_html", None)
-            if callable(page_fetcher):
-                return await page_fetcher(
-                    scholar_id,
-                    cstart=cstart,
-                    pagesize=page_size,
-                )
-            if cstart <= 0:
-                return await self._source.fetch_profile_html(scholar_id)
-            return FetchResult(
-                requested_url=(
-                    f"https://scholar.google.com/citations?hl=en&user={scholar_id}&cstart={cstart}&pagesize={page_size}"
-                ),
-                status_code=None,
-                final_url=None,
-                body="",
-                error="source_does_not_support_pagination",
-            )
-        except Exception as exc:
-            logger.exception(
-                "ingestion.fetch_unexpected_error",
-                extra={
-                    "scholar_id": scholar_id,
-                    "cstart": cstart,
-                    "page_size": page_size,
-                },
-            )
-            return FetchResult(
-                requested_url=(
-                    f"https://scholar.google.com/citations?hl=en&user={scholar_id}&cstart={cstart}&pagesize={page_size}"
-                ),
-                status_code=None,
-                final_url=None,
-                body="",
-                error=str(exc),
-            )
-
-    @staticmethod
-    def _should_retry_page_fetch(
-        *,
-        parsed_page: ParsedProfilePage,
-        network_attempt_count: int,
-        rate_limit_attempt_count: int,
-        network_error_retries: int,
-        rate_limit_retries: int,
-    ) -> bool:
-        if parsed_page.state == ParseState.NETWORK_ERROR:
-            return network_attempt_count <= network_error_retries
-        if (
-            parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA
-            and parsed_page.state_reason == "blocked_http_429_rate_limited"
-        ):
-            return rate_limit_attempt_count <= rate_limit_retries
-        return False
-
-    @staticmethod
-    async def _sleep_retry_backoff(
-        *,
-        scholar_id: str,
-        cstart: int,
-        network_attempt_count: int,
-        rate_limit_attempt_count: int,
-        retry_backoff_seconds: float,
-        rate_limit_backoff_seconds: float,
-        state_reason: str,
-    ) -> None:
-        if state_reason == "blocked_http_429_rate_limited":
-            sleep_seconds = rate_limit_backoff_seconds * rate_limit_attempt_count
-            attempt_label = rate_limit_attempt_count
-        else:
-            sleep_seconds = retry_backoff_seconds * (2 ** (network_attempt_count - 1))
-            attempt_label = network_attempt_count
-
-        structured_log(
-            logger,
-            "warning",
-            "ingestion.scholar_retry_scheduled",
-            scholar_id=scholar_id,
-            cstart=cstart,
-            attempt_count=attempt_label,
-            sleep_seconds=sleep_seconds,
-            state_reason=state_reason,
-        )
-        if sleep_seconds > 0:
-            await asyncio.sleep(sleep_seconds)
-
-    async def _fetch_and_parse_page_with_retry(
-        self,
-        *,
-        scholar_id: str,
-        cstart: int,
-        page_size: int,
-        network_error_retries: int,
-        retry_backoff_seconds: float,
-        rate_limit_retries: int,
-        rate_limit_backoff_seconds: float,
-    ) -> tuple[FetchResult, ParsedProfilePage, list[dict[str, Any]]]:
-        network_attempts = 0
-        rate_limit_attempts = 0
-        attempt_log: list[dict[str, Any]] = []
-        fetch_result: FetchResult | None = None
-        parsed_page: ParsedProfilePage | None = None
-
-        while True:
-            fetch_result = await self._fetch_profile_page(
-                scholar_id=scholar_id,
-                cstart=cstart,
-                page_size=page_size,
-            )
-            parsed_page = self._parse_profile_page_or_layout_error(fetch_result=fetch_result)
-
-            if parsed_page.state == ParseState.NETWORK_ERROR:
-                network_attempts += 1
-                total_attempts = network_attempts
-            elif (
-                parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA
-                and parsed_page.state_reason == "blocked_http_429_rate_limited"
-            ):
-                rate_limit_attempts += 1
-                total_attempts = rate_limit_attempts
-            else:
-                total_attempts = network_attempts + rate_limit_attempts + 1
-
-            attempt_log.append(
-                {
-                    "attempt": total_attempts,
-                    "cstart": cstart,
-                    "state": parsed_page.state.value,
-                    "state_reason": parsed_page.state_reason,
-                    "status_code": fetch_result.status_code,
-                    "fetch_error": fetch_result.error,
-                }
-            )
-
-            if not self._should_retry_page_fetch(
-                parsed_page=parsed_page,
-                network_attempt_count=network_attempts,
-                rate_limit_attempt_count=rate_limit_attempts,
-                network_error_retries=network_error_retries,
-                rate_limit_retries=rate_limit_retries,
-            ):
-                break
-
-            await self._sleep_retry_backoff(
-                scholar_id=scholar_id,
-                cstart=cstart,
-                network_attempt_count=network_attempts,
-                rate_limit_attempt_count=rate_limit_attempts,
-                retry_backoff_seconds=max(float(retry_backoff_seconds), 0.0),
-                rate_limit_backoff_seconds=max(float(rate_limit_backoff_seconds), 0.0),
-                state_reason=parsed_page.state_reason,
-            )
-
-        if fetch_result is None or parsed_page is None:
-            raise RuntimeError("Fetch-and-parse retry loop produced no result.")
-        return fetch_result, parsed_page, attempt_log
-
-    def _parse_profile_page_or_layout_error(
-        self,
-        *,
-        fetch_result: FetchResult,
-    ) -> ParsedProfilePage:
-        try:
-            return parse_profile_page(fetch_result)
-        except ScholarParserError as exc:
-            return self._parsed_page_from_parser_error(
-                fetch_result=fetch_result,
-                code=exc.code,
-            )
-
-    @staticmethod
-    def _parsed_page_from_parser_error(
-        *,
-        fetch_result: FetchResult,
-        code: str,
-    ) -> ParsedProfilePage:
-        return ParsedProfilePage(
-            state=ParseState.LAYOUT_CHANGED,
-            state_reason=code,
-            profile_name=None,
-            profile_image_url=None,
-            publications=[],
-            marker_counts={},
-            warnings=[code],
-            has_show_more_button=False,
-            has_operation_error_banner=False,
-            articles_range=None,
-        )
-
-    @staticmethod
-    def _should_skip_no_change(
-        *,
-        start_cstart: int,
-        first_page_fingerprint_sha256: str | None,
-        previous_initial_page_fingerprint_sha256: str | None,
-        parsed_page: ParsedProfilePage,
-    ) -> bool:
-        return (
-            start_cstart <= 0
-            and first_page_fingerprint_sha256 is not None
-            and previous_initial_page_fingerprint_sha256 is not None
-            and first_page_fingerprint_sha256 == previous_initial_page_fingerprint_sha256
-            and parsed_page.state in {ParseState.OK, ParseState.NO_RESULTS}
-        )
-
-    @staticmethod
-    def _skip_no_change_result(
-        *,
-        fetch_result: FetchResult,
-        parsed_page: ParsedProfilePage,
-        first_page_fingerprint_sha256: str | None,
-        attempt_log: list[dict[str, Any]],
-        page_logs: list[dict[str, Any]],
-    ) -> PagedParseResult:
-        return PagedParseResult(
-            fetch_result=fetch_result,
-            parsed_page=parsed_page,
-            first_page_fetch_result=fetch_result,
-            first_page_parsed_page=parsed_page,
-            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
-            publications=[],
-            attempt_log=attempt_log,
-            page_logs=page_logs,
-            pages_fetched=1,
-            pages_attempted=1,
-            has_more_remaining=False,
-            pagination_truncated_reason=None,
-            continuation_cstart=None,
-            skipped_no_change=True,
-            discovered_publication_count=0,
-        )
-
-    @staticmethod
-    def _initial_failure_result(
-        *,
-        fetch_result: FetchResult,
-        parsed_page: ParsedProfilePage,
-        first_page_fingerprint_sha256: str | None,
-        start_cstart: int,
-        attempt_log: list[dict[str, Any]],
-        page_logs: list[dict[str, Any]],
-    ) -> PagedParseResult:
-        continuation_cstart = start_cstart if parsed_page.state == ParseState.NETWORK_ERROR else None
-        return PagedParseResult(
-            fetch_result=fetch_result,
-            parsed_page=parsed_page,
-            first_page_fetch_result=fetch_result,
-            first_page_parsed_page=parsed_page,
-            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
-            publications=[],
-            attempt_log=attempt_log,
-            page_logs=page_logs,
-            pages_fetched=0,
-            pages_attempted=1,
-            has_more_remaining=False,
-            pagination_truncated_reason=None,
-            continuation_cstart=continuation_cstart,
-            skipped_no_change=False,
-            discovered_publication_count=0,
-        )
-
-    @staticmethod
-    def _build_loop_state(
-        *,
-        start_cstart: int,
-        fetch_result: FetchResult,
-        parsed_page: ParsedProfilePage,
-        attempt_log: list[dict[str, Any]],
-        page_logs: list[dict[str, Any]],
-    ) -> PagedLoopState:
-        next_cstart = _next_cstart_value(
-            articles_range=parsed_page.articles_range,
-            fallback=start_cstart + len(parsed_page.publications),
-        )
-        return PagedLoopState(
-            fetch_result=fetch_result,
-            parsed_page=parsed_page,
-            attempt_log=attempt_log,
-            page_logs=page_logs,
-            publications=list(parsed_page.publications),
-            pages_fetched=1,
-            pages_attempted=1,
-            current_cstart=start_cstart,
-            next_cstart=next_cstart,
-        )
-
-    @staticmethod
-    def _set_truncated_state(
-        *,
-        state: PagedLoopState,
-        reason: str,
-        continuation_cstart: int,
-    ) -> None:
-        state.has_more_remaining = True
-        state.pagination_truncated_reason = reason
-        state.continuation_cstart = continuation_cstart
-
-    def _should_stop_pagination(self, *, state: PagedLoopState, bounded_max_pages: int) -> bool:
-        if state.pages_fetched >= bounded_max_pages:
-            self._set_truncated_state(
-                state=state,
-                reason="max_pages_reached",
-                continuation_cstart=(
-                    state.next_cstart if state.next_cstart > state.current_cstart else state.current_cstart
-                ),
-            )
-            return True
-        if state.next_cstart <= state.current_cstart:
-            self._set_truncated_state(
-                state=state,
-                reason="pagination_cursor_stalled",
-                continuation_cstart=state.current_cstart,
-            )
-            return True
-        return False
-
-    async def _fetch_next_page(
-        self,
-        *,
-        scholar_id: str,
-        state: PagedLoopState,
-        request_delay_seconds: int,
-        bounded_page_size: int,
-        network_error_retries: int,
-        retry_backoff_seconds: float,
-        rate_limit_retries: int,
-        rate_limit_backoff_seconds: float,
-    ) -> tuple[FetchResult, ParsedProfilePage, list[dict[str, Any]]]:
-        if request_delay_seconds > 0:
-            await asyncio.sleep(float(request_delay_seconds))
-        state.current_cstart = state.next_cstart
-        return await self._fetch_and_parse_page_with_retry(
-            scholar_id=scholar_id,
-            cstart=state.current_cstart,
-            page_size=bounded_page_size,
-            network_error_retries=network_error_retries,
-            retry_backoff_seconds=retry_backoff_seconds,
-            rate_limit_retries=rate_limit_retries,
-            rate_limit_backoff_seconds=rate_limit_backoff_seconds,
-        )
-
-    @staticmethod
-    def _record_next_page(
-        *,
-        state: PagedLoopState,
-        fetch_result: FetchResult,
-        parsed_page: ParsedProfilePage,
-        page_attempt_log: list[dict[str, Any]],
-    ) -> None:
-        state.pages_attempted += 1
-        state.attempt_log.extend(page_attempt_log)
-        state.page_logs.append(
-            {
-                "page": state.pages_attempted,
-                "cstart": state.current_cstart,
-                "state": parsed_page.state.value,
-                "state_reason": parsed_page.state_reason,
-                "status_code": fetch_result.status_code,
-                "publication_count": len(parsed_page.publications),
-                "articles_range": parsed_page.articles_range,
-                "has_show_more_button": parsed_page.has_show_more_button,
-                "warning_codes": parsed_page.warnings,
-                "attempt_count": len(page_attempt_log),
-            }
-        )
-        state.fetch_result = fetch_result
-        state.parsed_page = parsed_page
-
-    @staticmethod
-    def _handle_page_state_transition(*, state: PagedLoopState) -> bool:
-        if state.parsed_page.state not in {ParseState.OK, ParseState.NO_RESULTS}:
-            ScholarIngestionService._set_truncated_state(
-                state=state,
-                reason=f"page_state_{state.parsed_page.state.value}",
-                continuation_cstart=state.current_cstart,
-            )
-            return True
-        if state.parsed_page.state == ParseState.NO_RESULTS and len(state.parsed_page.publications) == 0:
-            state.pages_fetched += 1
-            return True
-        state.pages_fetched += 1
-        state.publications.extend(state.parsed_page.publications)
-        state.next_cstart = _next_cstart_value(
-            articles_range=state.parsed_page.articles_range,
-            fallback=state.current_cstart + len(state.parsed_page.publications),
-        )
-        return False
-
-    async def _fetch_initial_page_context(
-        self,
-        *,
-        scholar_id: str,
-        start_cstart: int,
-        bounded_page_size: int,
-        network_error_retries: int,
-        retry_backoff_seconds: float,
-        rate_limit_retries: int,
-        rate_limit_backoff_seconds: float,
-    ) -> tuple[FetchResult, ParsedProfilePage, str | None, list[dict[str, Any]], list[dict[str, Any]]]:
-        fetch_result, parsed_page, first_attempt_log = await self._fetch_and_parse_page_with_retry(
-            scholar_id=scholar_id,
-            cstart=start_cstart,
-            page_size=bounded_page_size,
-            network_error_retries=network_error_retries,
-            retry_backoff_seconds=retry_backoff_seconds,
-            rate_limit_retries=rate_limit_retries,
-            rate_limit_backoff_seconds=rate_limit_backoff_seconds,
-        )
-        first_page_fingerprint_sha256 = build_initial_page_fingerprint(parsed_page)
-        attempt_log = list(first_attempt_log)
-        page_logs = [
-            {
-                "page": 1,
-                "cstart": start_cstart,
-                "state": parsed_page.state.value,
-                "state_reason": parsed_page.state_reason,
-                "status_code": fetch_result.status_code,
-                "publication_count": len(parsed_page.publications),
-                "articles_range": parsed_page.articles_range,
-                "has_show_more_button": parsed_page.has_show_more_button,
-                "warning_codes": parsed_page.warnings,
-                "attempt_count": len(first_attempt_log),
-            }
-        ]
-        return fetch_result, parsed_page, first_page_fingerprint_sha256, attempt_log, page_logs
-
-    async def _paginate_loop(
-        self,
-        *,
-        scholar: ScholarProfile,
-        run: CrawlRun,
-        db_session: AsyncSession,
-        state: PagedLoopState,
-        bounded_max_pages: int,
-        request_delay_seconds: int,
-        bounded_page_size: int,
-        network_error_retries: int,
-        retry_backoff_seconds: float,
-        rate_limit_retries: int,
-        rate_limit_backoff_seconds: float,
-    ) -> None:
-        # Cross-page canonical dedup state; grows across all pages of this scholar.
-        seen_canonical: set[str] = set()
-
-        if state.parsed_page.publications:
-            deduped_first = _dedupe_publication_candidates(
-                list(state.parsed_page.publications), seen_canonical=seen_canonical
-            )
-            if deduped_first:
-                discovered_count = await self._upsert_profile_publications(
-                    db_session, run=run, scholar=scholar, publications=deduped_first
-                )
-                state.discovered_publication_count += discovered_count
-
-        while state.parsed_page.has_show_more_button:
-            await db_session.refresh(run)
-            if run.status == RunStatus.CANCELED:
-                structured_log(
-                    logger,
-                    "info",
-                    "ingestion.pagination_canceled",
-                    run_id=run.id,
-                )
-                self._set_truncated_state(
-                    state=state,
-                    reason="run_canceled",
-                    continuation_cstart=state.current_cstart,
-                )
-                return
-
-            if self._should_stop_pagination(state=state, bounded_max_pages=bounded_max_pages):
-                return
-            next_fetch_result, next_parsed_page, next_attempt_log = await self._fetch_next_page(
-                scholar_id=scholar.scholar_id,
-                state=state,
-                request_delay_seconds=request_delay_seconds,
-                bounded_page_size=bounded_page_size,
-                network_error_retries=network_error_retries,
-                retry_backoff_seconds=retry_backoff_seconds,
-                rate_limit_retries=rate_limit_retries,
-                rate_limit_backoff_seconds=rate_limit_backoff_seconds,
-            )
-            self._record_next_page(
-                state=state,
-                fetch_result=next_fetch_result,
-                parsed_page=next_parsed_page,
-                page_attempt_log=next_attempt_log,
-            )
-
-            # Deduplicate against all publications already committed this run,
-            # then immediately commit the surviving candidates to the DB.
-            if next_parsed_page.publications:
-                deduped_next = _dedupe_publication_candidates(
-                    list(next_parsed_page.publications), seen_canonical=seen_canonical
-                )
-                if deduped_next:
-                    discovered_count = await self._upsert_profile_publications(
-                        db_session, run=run, scholar=scholar, publications=deduped_next
-                    )
-                    state.discovered_publication_count += discovered_count
-
-            if self._handle_page_state_transition(state=state):
-                return
-
-    @staticmethod
-    def _result_from_pagination_state(
-        *,
-        state: PagedLoopState,
-        first_page_fetch_result: FetchResult,
-        first_page_parsed_page: ParsedProfilePage,
-        first_page_fingerprint_sha256: str | None,
-    ) -> PagedParseResult:
-        return PagedParseResult(
-            fetch_result=state.fetch_result,
-            parsed_page=state.parsed_page,
-            first_page_fetch_result=first_page_fetch_result,
-            first_page_parsed_page=first_page_parsed_page,
-            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
-            publications=_dedupe_publication_candidates(state.publications),
-            attempt_log=state.attempt_log,
-            page_logs=state.page_logs,
-            pages_fetched=state.pages_fetched,
-            pages_attempted=state.pages_attempted,
-            has_more_remaining=state.has_more_remaining,
-            pagination_truncated_reason=state.pagination_truncated_reason,
-            continuation_cstart=state.continuation_cstart,
-            skipped_no_change=False,
-            discovered_publication_count=state.discovered_publication_count,
-        )
-
-    def _short_circuit_initial_page(
-        self,
-        *,
-        start_cstart: int,
-        previous_initial_page_fingerprint_sha256: str | None,
-        fetch_result: FetchResult,
-        parsed_page: ParsedProfilePage,
-        first_page_fingerprint_sha256: str | None,
-        attempt_log: list[dict[str, Any]],
-        page_logs: list[dict[str, Any]],
-    ) -> PagedParseResult | None:
-        if self._should_skip_no_change(
-            start_cstart=start_cstart,
-            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
-            previous_initial_page_fingerprint_sha256=previous_initial_page_fingerprint_sha256,
-            parsed_page=parsed_page,
-        ):
-            return self._skip_no_change_result(
-                fetch_result=fetch_result,
-                parsed_page=parsed_page,
-                first_page_fingerprint_sha256=first_page_fingerprint_sha256,
-                attempt_log=attempt_log,
-                page_logs=page_logs,
-            )
-        if parsed_page.state not in {ParseState.OK, ParseState.NO_RESULTS}:
-            return self._initial_failure_result(
-                fetch_result=fetch_result,
-                parsed_page=parsed_page,
-                first_page_fingerprint_sha256=first_page_fingerprint_sha256,
-                start_cstart=start_cstart,
-                attempt_log=attempt_log,
-                page_logs=page_logs,
-            )
-        return None
-
-    async def _fetch_and_parse_all_pages_with_retry(
-        self,
-        *,
-        scholar: ScholarProfile,
-        run: CrawlRun,
-        db_session: AsyncSession,
-        start_cstart: int,
-        request_delay_seconds: int,
-        network_error_retries: int,
-        retry_backoff_seconds: float,
-        rate_limit_retries: int,
-        rate_limit_backoff_seconds: float,
-        max_pages: int,
-        page_size: int,
-        previous_initial_page_fingerprint_sha256: str | None = None,
-    ) -> PagedParseResult:
-        bounded_max_pages = max(1, int(max_pages))
-        bounded_page_size = max(1, int(page_size))
-        (
-            fetch_result,
-            parsed_page,
-            first_page_fingerprint_sha256,
-            attempt_log,
-            page_logs,
-        ) = await self._fetch_initial_page_context(
-            scholar_id=scholar.scholar_id,
-            start_cstart=start_cstart,
-            bounded_page_size=bounded_page_size,
-            network_error_retries=network_error_retries,
-            retry_backoff_seconds=retry_backoff_seconds,
-            rate_limit_retries=rate_limit_retries,
-            rate_limit_backoff_seconds=rate_limit_backoff_seconds,
-        )
-        shortcut_result = self._short_circuit_initial_page(
-            start_cstart=start_cstart,
-            previous_initial_page_fingerprint_sha256=previous_initial_page_fingerprint_sha256,
-            fetch_result=fetch_result,
-            parsed_page=parsed_page,
-            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
-            attempt_log=attempt_log,
-            page_logs=page_logs,
-        )
-        if shortcut_result is not None:
-            return shortcut_result
-        state = self._build_loop_state(
-            start_cstart=start_cstart,
-            fetch_result=fetch_result,
-            parsed_page=parsed_page,
-            attempt_log=attempt_log,
-            page_logs=page_logs,
-        )
-        await self._paginate_loop(
-            scholar=scholar,
-            run=run,
-            db_session=db_session,
-            state=state,
-            bounded_max_pages=bounded_max_pages,
-            request_delay_seconds=request_delay_seconds,
-            bounded_page_size=bounded_page_size,
-            network_error_retries=network_error_retries,
-            retry_backoff_seconds=retry_backoff_seconds,
-            rate_limit_retries=rate_limit_retries,
-            rate_limit_backoff_seconds=rate_limit_backoff_seconds,
-        )
-        return self._result_from_pagination_state(
-            state=state,
-            first_page_fetch_result=fetch_result,
-            first_page_parsed_page=parsed_page,
-            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
-        )
-
     async def _run_is_canceled(
         self,
         db_session: AsyncSession,
@@ -2641,228 +1992,6 @@ class ScholarIngestionService:
                     enriched.append(p)
         return enriched
 
-    async def _upsert_profile_publications(
-        self,
-        db_session: AsyncSession,
-        *,
-        run: CrawlRun,
-        scholar: ScholarProfile,
-        publications: list[PublicationCandidate],
-    ) -> int:
-        # We no longer enrich inline here. Enrichment is now deferred to the end of the run.
-        seen_publication_ids: set[int] = set()
-        discovered_count = 0
-
-        for candidate in publications:
-            publication = await self._resolve_publication(db_session, candidate)
-            if publication.id in seen_publication_ids:
-                continue
-            seen_publication_ids.add(publication.id)
-
-            link_result = await db_session.execute(
-                select(ScholarPublication).where(
-                    ScholarPublication.scholar_profile_id == scholar.id,
-                    ScholarPublication.publication_id == publication.id,
-                )
-            )
-            link = link_result.scalar_one_or_none()
-            if link is not None:
-                continue
-
-            link = ScholarPublication(
-                scholar_profile_id=scholar.id,
-                publication_id=publication.id,
-                is_read=False,
-                first_seen_run_id=run.id,
-            )
-            db_session.add(link)
-            discovered_count += 1
-
-            await self._commit_discovered_publication(
-                db_session,
-                run=run,
-                scholar=scholar,
-                publication=publication,
-            )
-
-        if not scholar.baseline_completed:
-            scholar.baseline_completed = True
-
-        return discovered_count
-
-    async def _commit_discovered_publication(
-        self,
-        db_session: AsyncSession,
-        *,
-        run: CrawlRun,
-        scholar: ScholarProfile,
-        publication: Publication,
-    ) -> None:
-        run.new_pub_count = int(run.new_pub_count or 0) + 1
-        await db_session.commit()
-        await run_events.publish(
-            run_id=run.id,
-            event_type="publication_discovered",
-            data={
-                "publication_id": publication.id,
-                "title": publication.title_raw,
-                "pub_url": publication.pub_url,
-                "scholar_profile_id": scholar.id,
-                "scholar_label": scholar.display_name or scholar.scholar_id,
-                "first_seen_at": datetime.now(UTC).isoformat(),
-                "new_publication_count": int(run.new_pub_count or 0),
-            },
-        )
-
-    @staticmethod
-    def _validate_publication_candidate(candidate: PublicationCandidate) -> None:
-        if not candidate.title.strip():
-            raise RuntimeError("Publication candidate is missing title.")
-        if candidate.citation_count is not None and int(candidate.citation_count) < 0:
-            raise RuntimeError("Publication candidate has negative citation_count.")
-
-    async def _find_publication_by_cluster(
-        self,
-        db_session: AsyncSession,
-        *,
-        cluster_id: str | None,
-    ) -> Publication | None:
-        if not cluster_id:
-            return None
-        result = await db_session.execute(select(Publication).where(Publication.cluster_id == cluster_id))
-        return result.scalar_one_or_none()
-
-    async def _find_publication_by_fingerprint(
-        self,
-        db_session: AsyncSession,
-        *,
-        fingerprint: str,
-    ) -> Publication | None:
-        result = await db_session.execute(select(Publication).where(Publication.fingerprint_sha256 == fingerprint))
-        return result.scalar_one_or_none()
-
-    @staticmethod
-    def _select_existing_publication(
-        *,
-        cluster_publication: Publication | None,
-        fingerprint_publication: Publication | None,
-    ) -> Publication | None:
-        if cluster_publication is not None:
-            return cluster_publication
-        return fingerprint_publication
-
-    @staticmethod
-    def _compute_canonical_title_hash(title: str) -> str:
-        canonical = canonical_title_for_dedup(title)
-        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
-
-    async def _find_publication_by_canonical_title_hash(
-        self,
-        db_session: AsyncSession,
-        *,
-        canonical_title_hash: str,
-    ) -> Publication | None:
-        result = await db_session.execute(
-            select(Publication).where(Publication.canonical_title_hash == canonical_title_hash)
-        )
-        return result.scalar_one_or_none()
-
-    async def _create_publication(
-        self,
-        db_session: AsyncSession,
-        *,
-        candidate: PublicationCandidate,
-        fingerprint: str,
-    ) -> Publication:
-        publication = Publication(
-            cluster_id=candidate.cluster_id,
-            fingerprint_sha256=fingerprint,
-            title_raw=candidate.title,
-            title_normalized=normalize_title(candidate.title),
-            canonical_title_hash=self._compute_canonical_title_hash(candidate.title),
-            year=candidate.year,
-            citation_count=int(candidate.citation_count or 0),
-            author_text=candidate.authors_text,
-            venue_text=candidate.venue_text,
-            pub_url=build_publication_url(candidate.title_url),
-            pdf_url=None,
-        )
-        db_session.add(publication)
-        await db_session.flush()
-        return publication
-
-    @staticmethod
-    def _update_existing_publication(
-        *,
-        publication: Publication,
-        candidate: PublicationCandidate,
-    ) -> None:
-        if candidate.cluster_id and publication.cluster_id is None:
-            publication.cluster_id = candidate.cluster_id
-        publication.title_raw = candidate.title
-        publication.title_normalized = normalize_title(candidate.title)
-        if candidate.year is not None:
-            publication.year = candidate.year
-        if candidate.citation_count is not None:
-            publication.citation_count = int(candidate.citation_count)
-        if candidate.authors_text:
-            publication.author_text = candidate.authors_text
-        if candidate.venue_text:
-            publication.venue_text = candidate.venue_text
-        if candidate.title_url:
-            publication.pub_url = build_publication_url(candidate.title_url)
-        first_doi_from_texts(candidate.title_url, candidate.venue_text, candidate.title)
-
-    async def _resolve_publication(
-        self,
-        db_session: AsyncSession,
-        candidate: PublicationCandidate,
-    ) -> Publication:
-        self._validate_publication_candidate(candidate)
-        fingerprint = build_publication_fingerprint(candidate)
-        cluster_publication = await self._find_publication_by_cluster(
-            db_session,
-            cluster_id=candidate.cluster_id,
-        )
-        fingerprint_publication = await self._find_publication_by_fingerprint(
-            db_session,
-            fingerprint=fingerprint,
-        )
-        publication = self._select_existing_publication(
-            cluster_publication=cluster_publication,
-            fingerprint_publication=fingerprint_publication,
-        )
-        if publication is None:
-            # Fallback: canonical title hash — catches cross-scholar noise variants
-            # (e.g. "Adam preprint (2014)" vs "Adam arXiv 2017" → same normalized form)
-            canonical_hash = self._compute_canonical_title_hash(candidate.title)
-            publication = await self._find_publication_by_canonical_title_hash(
-                db_session,
-                canonical_title_hash=canonical_hash,
-            )
-        if publication is None:
-            created = await self._create_publication(
-                db_session,
-                candidate=candidate,
-                fingerprint=fingerprint,
-            )
-            # Sync identifiers from local fields only for fast UI response
-            await identifier_service.sync_identifiers_for_publication_fields(
-                db_session,
-                publication=created,
-            )
-            return created
-        self._update_existing_publication(
-            publication=publication,
-            candidate=candidate,
-        )
-        # Sync identifiers from local fields only for fast UI response
-        await identifier_service.sync_identifiers_for_publication_fields(
-            db_session,
-            publication=publication,
-        )
-        return publication
-
     def _resolve_run_status(
         self,
         *,

app/services/ingestion/page_fetch.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from app.logging_utils import structured_log
+from app.services.scholar.parser import (
+    ParsedProfilePage,
+    ParseState,
+    ScholarParserError,
+    parse_profile_page,
+)
+from app.services.scholar.source import FetchResult, ScholarSource
+
+logger = logging.getLogger(__name__)
+
+
+class PageFetcher:
+    """Fetches and parses a single Google Scholar profile page with retry logic."""
+
+    def __init__(self, *, source: ScholarSource) -> None:
+        self._source = source
+
+    async def fetch_profile_page(
+        self,
+        *,
+        scholar_id: str,
+        cstart: int,
+        page_size: int,
+    ) -> FetchResult:
+        try:
+            page_fetcher = getattr(self._source, "fetch_profile_page_html", None)
+            if callable(page_fetcher):
+                return await page_fetcher(
+                    scholar_id,
+                    cstart=cstart,
+                    pagesize=page_size,
+                )
+            if cstart <= 0:
+                return await self._source.fetch_profile_html(scholar_id)
+            return FetchResult(
+                requested_url=(
+                    f"https://scholar.google.com/citations?hl=en&user={scholar_id}&cstart={cstart}&pagesize={page_size}"
+                ),
+                status_code=None,
+                final_url=None,
+                body="",
+                error="source_does_not_support_pagination",
+            )
+        except Exception as exc:
+            logger.exception(
+                "ingestion.fetch_unexpected_error",
+                extra={
+                    "scholar_id": scholar_id,
+                    "cstart": cstart,
+                    "page_size": page_size,
+                },
+            )
+            return FetchResult(
+                requested_url=(
+                    f"https://scholar.google.com/citations?hl=en&user={scholar_id}&cstart={cstart}&pagesize={page_size}"
+                ),
+                status_code=None,
+                final_url=None,
+                body="",
+                error=str(exc),
+            )
+
+    # ── Parse helpers ────────────────────────────────────────────────
+
+    def parse_page_or_layout_error(
+        self,
+        *,
+        fetch_result: FetchResult,
+    ) -> ParsedProfilePage:
+        try:
+            return parse_profile_page(fetch_result)
+        except ScholarParserError as exc:
+            return self._parsed_page_from_parser_error(code=exc.code)
+
+    @staticmethod
+    def _parsed_page_from_parser_error(*, code: str) -> ParsedProfilePage:
+        return ParsedProfilePage(
+            state=ParseState.LAYOUT_CHANGED,
+            state_reason=code,
+            profile_name=None,
+            profile_image_url=None,
+            publications=[],
+            marker_counts={},
+            warnings=[code],
+            has_show_more_button=False,
+            has_operation_error_banner=False,
+            articles_range=None,
+        )
+
+    # ── Retry logic ──────────────────────────────────────────────────
+
+    @staticmethod
+    def _should_retry(
+        *,
+        parsed_page: ParsedProfilePage,
+        network_attempt_count: int,
+        rate_limit_attempt_count: int,
+        network_error_retries: int,
+        rate_limit_retries: int,
+    ) -> bool:
+        if parsed_page.state == ParseState.NETWORK_ERROR:
+            return network_attempt_count <= network_error_retries
+        if (
+            parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA
+            and parsed_page.state_reason == "blocked_http_429_rate_limited"
+        ):
+            return rate_limit_attempt_count <= rate_limit_retries
+        return False
+
+    @staticmethod
+    async def _sleep_backoff(
+        *,
+        scholar_id: str,
+        cstart: int,
+        network_attempt_count: int,
+        rate_limit_attempt_count: int,
+        retry_backoff_seconds: float,
+        rate_limit_backoff_seconds: float,
+        state_reason: str,
+    ) -> None:
+        if state_reason == "blocked_http_429_rate_limited":
+            sleep_seconds = rate_limit_backoff_seconds * rate_limit_attempt_count
+            attempt_label = rate_limit_attempt_count
+        else:
+            sleep_seconds = retry_backoff_seconds * (2 ** (network_attempt_count - 1))
+            attempt_label = network_attempt_count
+
+        structured_log(
+            logger,
+            "warning",
+            "ingestion.scholar_retry_scheduled",
+            scholar_id=scholar_id,
+            cstart=cstart,
+            attempt_count=attempt_label,
+            sleep_seconds=sleep_seconds,
+            state_reason=state_reason,
+        )
+        if sleep_seconds > 0:
+            await asyncio.sleep(sleep_seconds)
+
+    async def fetch_and_parse_with_retry(
+        self,
+        *,
+        scholar_id: str,
+        cstart: int,
+        page_size: int,
+        network_error_retries: int,
+        retry_backoff_seconds: float,
+        rate_limit_retries: int,
+        rate_limit_backoff_seconds: float,
+    ) -> tuple[FetchResult, ParsedProfilePage, list[dict[str, Any]]]:
+        network_attempts = 0
+        rate_limit_attempts = 0
+        attempt_log: list[dict[str, Any]] = []
+        fetch_result: FetchResult | None = None
+        parsed_page: ParsedProfilePage | None = None
+
+        while True:
+            fetch_result = await self.fetch_profile_page(
+                scholar_id=scholar_id,
+                cstart=cstart,
+                page_size=page_size,
+            )
+            parsed_page = self.parse_page_or_layout_error(fetch_result=fetch_result)
+
+            if parsed_page.state == ParseState.NETWORK_ERROR:
+                network_attempts += 1
+                total_attempts = network_attempts
+            elif (
+                parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA
+                and parsed_page.state_reason == "blocked_http_429_rate_limited"
+            ):
+                rate_limit_attempts += 1
+                total_attempts = rate_limit_attempts
+            else:
+                total_attempts = network_attempts + rate_limit_attempts + 1
+
+            attempt_log.append(
+                {
+                    "attempt": total_attempts,
+                    "cstart": cstart,
+                    "state": parsed_page.state.value,
+                    "state_reason": parsed_page.state_reason,
+                    "status_code": fetch_result.status_code,
+                    "fetch_error": fetch_result.error,
+                }
+            )
+
+            if not self._should_retry(
+                parsed_page=parsed_page,
+                network_attempt_count=network_attempts,
+                rate_limit_attempt_count=rate_limit_attempts,
+                network_error_retries=network_error_retries,
+                rate_limit_retries=rate_limit_retries,
+            ):
+                break
+
+            await self._sleep_backoff(
+                scholar_id=scholar_id,
+                cstart=cstart,
+                network_attempt_count=network_attempts,
+                rate_limit_attempt_count=rate_limit_attempts,
+                retry_backoff_seconds=max(float(retry_backoff_seconds), 0.0),
+                rate_limit_backoff_seconds=max(float(rate_limit_backoff_seconds), 0.0),
+                state_reason=parsed_page.state_reason,
+            )
+
+        if fetch_result is None or parsed_page is None:
+            raise RuntimeError("Fetch-and-parse retry loop produced no result.")
+        return fetch_result, parsed_page, attempt_log

app/services/ingestion/pagination.py

@@ -0,0 +1,486 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from app.db.models import CrawlRun, RunStatus, ScholarProfile
+from app.logging_utils import structured_log
+from app.services.ingestion.fingerprints import (
+    _dedupe_publication_candidates,
+    _next_cstart_value,
+    build_initial_page_fingerprint,
+)
+from app.services.ingestion.page_fetch import PageFetcher
+from app.services.ingestion.types import PagedLoopState, PagedParseResult
+from app.services.scholar.parser import ParsedProfilePage, ParseState
+from app.services.scholar.source import FetchResult, ScholarSource
+
+logger = logging.getLogger(__name__)
+
+
+class PaginationEngine:
+    """Fetches and paginates Google Scholar profile pages.
+
+    Pure HTTP + parsing — no DB writes except via the provided upsert callback.
+    """
+
+    def __init__(self, *, source: ScholarSource) -> None:
+        self._source = source
+        self._fetcher = PageFetcher(source=source)
+
+    # ── No-change / initial failure short-circuits ───────────────────
+
+    @staticmethod
+    def _should_skip_no_change(
+        *,
+        start_cstart: int,
+        first_page_fingerprint_sha256: str | None,
+        previous_initial_page_fingerprint_sha256: str | None,
+        parsed_page: ParsedProfilePage,
+    ) -> bool:
+        return (
+            start_cstart <= 0
+            and first_page_fingerprint_sha256 is not None
+            and previous_initial_page_fingerprint_sha256 is not None
+            and first_page_fingerprint_sha256 == previous_initial_page_fingerprint_sha256
+            and parsed_page.state in {ParseState.OK, ParseState.NO_RESULTS}
+        )
+
+    @staticmethod
+    def _skip_no_change_result(
+        *,
+        fetch_result: FetchResult,
+        parsed_page: ParsedProfilePage,
+        first_page_fingerprint_sha256: str | None,
+        attempt_log: list[dict[str, Any]],
+        page_logs: list[dict[str, Any]],
+    ) -> PagedParseResult:
+        return PagedParseResult(
+            fetch_result=fetch_result,
+            parsed_page=parsed_page,
+            first_page_fetch_result=fetch_result,
+            first_page_parsed_page=parsed_page,
+            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
+            publications=[],
+            attempt_log=attempt_log,
+            page_logs=page_logs,
+            pages_fetched=1,
+            pages_attempted=1,
+            has_more_remaining=False,
+            pagination_truncated_reason=None,
+            continuation_cstart=None,
+            skipped_no_change=True,
+            discovered_publication_count=0,
+        )
+
+    @staticmethod
+    def _initial_failure_result(
+        *,
+        fetch_result: FetchResult,
+        parsed_page: ParsedProfilePage,
+        first_page_fingerprint_sha256: str | None,
+        start_cstart: int,
+        attempt_log: list[dict[str, Any]],
+        page_logs: list[dict[str, Any]],
+    ) -> PagedParseResult:
+        continuation_cstart = start_cstart if parsed_page.state == ParseState.NETWORK_ERROR else None
+        return PagedParseResult(
+            fetch_result=fetch_result,
+            parsed_page=parsed_page,
+            first_page_fetch_result=fetch_result,
+            first_page_parsed_page=parsed_page,
+            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
+            publications=[],
+            attempt_log=attempt_log,
+            page_logs=page_logs,
+            pages_fetched=0,
+            pages_attempted=1,
+            has_more_remaining=False,
+            pagination_truncated_reason=None,
+            continuation_cstart=continuation_cstart,
+            skipped_no_change=False,
+            discovered_publication_count=0,
+        )
+
+    # ── Loop state management ────────────────────────────────────────
+
+    @staticmethod
+    def _build_loop_state(
+        *,
+        start_cstart: int,
+        fetch_result: FetchResult,
+        parsed_page: ParsedProfilePage,
+        attempt_log: list[dict[str, Any]],
+        page_logs: list[dict[str, Any]],
+    ) -> PagedLoopState:
+        next_cstart = _next_cstart_value(
+            articles_range=parsed_page.articles_range,
+            fallback=start_cstart + len(parsed_page.publications),
+        )
+        return PagedLoopState(
+            fetch_result=fetch_result,
+            parsed_page=parsed_page,
+            attempt_log=attempt_log,
+            page_logs=page_logs,
+            publications=list(parsed_page.publications),
+            pages_fetched=1,
+            pages_attempted=1,
+            current_cstart=start_cstart,
+            next_cstart=next_cstart,
+        )
+
+    @staticmethod
+    def _set_truncated_state(
+        *,
+        state: PagedLoopState,
+        reason: str,
+        continuation_cstart: int,
+    ) -> None:
+        state.has_more_remaining = True
+        state.pagination_truncated_reason = reason
+        state.continuation_cstart = continuation_cstart
+
+    def _should_stop_pagination(self, *, state: PagedLoopState, bounded_max_pages: int) -> bool:
+        if state.pages_fetched >= bounded_max_pages:
+            self._set_truncated_state(
+                state=state,
+                reason="max_pages_reached",
+                continuation_cstart=(
+                    state.next_cstart if state.next_cstart > state.current_cstart else state.current_cstart
+                ),
+            )
+            return True
+        if state.next_cstart <= state.current_cstart:
+            self._set_truncated_state(
+                state=state,
+                reason="pagination_cursor_stalled",
+                continuation_cstart=state.current_cstart,
+            )
+            return True
+        return False
+
+    # ── Multi-page fetch helpers ─────────────────────────────────────
+
+    async def _fetch_next_page(
+        self,
+        *,
+        scholar_id: str,
+        state: PagedLoopState,
+        request_delay_seconds: int,
+        bounded_page_size: int,
+        network_error_retries: int,
+        retry_backoff_seconds: float,
+        rate_limit_retries: int,
+        rate_limit_backoff_seconds: float,
+    ) -> tuple[FetchResult, ParsedProfilePage, list[dict[str, Any]]]:
+        if request_delay_seconds > 0:
+            await asyncio.sleep(float(request_delay_seconds))
+        state.current_cstart = state.next_cstart
+        return await self._fetcher.fetch_and_parse_with_retry(
+            scholar_id=scholar_id,
+            cstart=state.current_cstart,
+            page_size=bounded_page_size,
+            network_error_retries=network_error_retries,
+            retry_backoff_seconds=retry_backoff_seconds,
+            rate_limit_retries=rate_limit_retries,
+            rate_limit_backoff_seconds=rate_limit_backoff_seconds,
+        )
+
+    @staticmethod
+    def _record_next_page(
+        *,
+        state: PagedLoopState,
+        fetch_result: FetchResult,
+        parsed_page: ParsedProfilePage,
+        page_attempt_log: list[dict[str, Any]],
+    ) -> None:
+        state.pages_attempted += 1
+        state.attempt_log.extend(page_attempt_log)
+        state.page_logs.append(
+            {
+                "page": state.pages_attempted,
+                "cstart": state.current_cstart,
+                "state": parsed_page.state.value,
+                "state_reason": parsed_page.state_reason,
+                "status_code": fetch_result.status_code,
+                "publication_count": len(parsed_page.publications),
+                "articles_range": parsed_page.articles_range,
+                "has_show_more_button": parsed_page.has_show_more_button,
+                "warning_codes": parsed_page.warnings,
+                "attempt_count": len(page_attempt_log),
+            }
+        )
+        state.fetch_result = fetch_result
+        state.parsed_page = parsed_page
+
+    def _handle_page_state_transition(self, *, state: PagedLoopState) -> bool:
+        if state.parsed_page.state not in {ParseState.OK, ParseState.NO_RESULTS}:
+            self._set_truncated_state(
+                state=state,
+                reason=f"page_state_{state.parsed_page.state.value}",
+                continuation_cstart=state.current_cstart,
+            )
+            return True
+        if state.parsed_page.state == ParseState.NO_RESULTS and len(state.parsed_page.publications) == 0:
+            state.pages_fetched += 1
+            return True
+        state.pages_fetched += 1
+        state.publications.extend(state.parsed_page.publications)
+        state.next_cstart = _next_cstart_value(
+            articles_range=state.parsed_page.articles_range,
+            fallback=state.current_cstart + len(state.parsed_page.publications),
+        )
+        return False
+
+    async def _fetch_initial_page_context(
+        self,
+        *,
+        scholar_id: str,
+        start_cstart: int,
+        bounded_page_size: int,
+        network_error_retries: int,
+        retry_backoff_seconds: float,
+        rate_limit_retries: int,
+        rate_limit_backoff_seconds: float,
+    ) -> tuple[FetchResult, ParsedProfilePage, str | None, list[dict[str, Any]], list[dict[str, Any]]]:
+        fetch_result, parsed_page, first_attempt_log = await self._fetcher.fetch_and_parse_with_retry(
+            scholar_id=scholar_id,
+            cstart=start_cstart,
+            page_size=bounded_page_size,
+            network_error_retries=network_error_retries,
+            retry_backoff_seconds=retry_backoff_seconds,
+            rate_limit_retries=rate_limit_retries,
+            rate_limit_backoff_seconds=rate_limit_backoff_seconds,
+        )
+        first_page_fingerprint_sha256 = build_initial_page_fingerprint(parsed_page)
+        attempt_log = list(first_attempt_log)
+        page_logs = [
+            {
+                "page": 1,
+                "cstart": start_cstart,
+                "state": parsed_page.state.value,
+                "state_reason": parsed_page.state_reason,
+                "status_code": fetch_result.status_code,
+                "publication_count": len(parsed_page.publications),
+                "articles_range": parsed_page.articles_range,
+                "has_show_more_button": parsed_page.has_show_more_button,
+                "warning_codes": parsed_page.warnings,
+                "attempt_count": len(first_attempt_log),
+            }
+        ]
+        return fetch_result, parsed_page, first_page_fingerprint_sha256, attempt_log, page_logs
+
+    # ── Pagination loop ──────────────────────────────────────────────
+
+    async def _paginate_loop(
+        self,
+        *,
+        scholar: ScholarProfile,
+        run: CrawlRun,
+        db_session: Any,
+        state: PagedLoopState,
+        bounded_max_pages: int,
+        request_delay_seconds: int,
+        bounded_page_size: int,
+        network_error_retries: int,
+        retry_backoff_seconds: float,
+        rate_limit_retries: int,
+        rate_limit_backoff_seconds: float,
+        upsert_publications_fn: Any,
+    ) -> None:
+        seen_canonical: set[str] = set()
+
+        if state.parsed_page.publications:
+            deduped_first = _dedupe_publication_candidates(
+                list(state.parsed_page.publications), seen_canonical=seen_canonical
+            )
+            if deduped_first:
+                discovered_count = await upsert_publications_fn(
+                    db_session, run=run, scholar=scholar, publications=deduped_first
+                )
+                state.discovered_publication_count += discovered_count
+
+        while state.parsed_page.has_show_more_button:
+            await db_session.refresh(run)
+            if run.status == RunStatus.CANCELED:
+                structured_log(
+                    logger,
+                    "info",
+                    "ingestion.pagination_canceled",
+                    run_id=run.id,
+                )
+                self._set_truncated_state(
+                    state=state,
+                    reason="run_canceled",
+                    continuation_cstart=state.current_cstart,
+                )
+                return
+
+            if self._should_stop_pagination(state=state, bounded_max_pages=bounded_max_pages):
+                return
+            next_fetch_result, next_parsed_page, next_attempt_log = await self._fetch_next_page(
+                scholar_id=scholar.scholar_id,
+                state=state,
+                request_delay_seconds=request_delay_seconds,
+                bounded_page_size=bounded_page_size,
+                network_error_retries=network_error_retries,
+                retry_backoff_seconds=retry_backoff_seconds,
+                rate_limit_retries=rate_limit_retries,
+                rate_limit_backoff_seconds=rate_limit_backoff_seconds,
+            )
+            self._record_next_page(
+                state=state,
+                fetch_result=next_fetch_result,
+                parsed_page=next_parsed_page,
+                page_attempt_log=next_attempt_log,
+            )
+
+            if next_parsed_page.publications:
+                deduped_next = _dedupe_publication_candidates(
+                    list(next_parsed_page.publications), seen_canonical=seen_canonical
+                )
+                if deduped_next:
+                    discovered_count = await upsert_publications_fn(
+                        db_session, run=run, scholar=scholar, publications=deduped_next
+                    )
+                    state.discovered_publication_count += discovered_count
+
+            if self._handle_page_state_transition(state=state):
+                return
+
+    @staticmethod
+    def _result_from_pagination_state(
+        *,
+        state: PagedLoopState,
+        first_page_fetch_result: FetchResult,
+        first_page_parsed_page: ParsedProfilePage,
+        first_page_fingerprint_sha256: str | None,
+    ) -> PagedParseResult:
+        return PagedParseResult(
+            fetch_result=state.fetch_result,
+            parsed_page=state.parsed_page,
+            first_page_fetch_result=first_page_fetch_result,
+            first_page_parsed_page=first_page_parsed_page,
+            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
+            publications=_dedupe_publication_candidates(state.publications),
+            attempt_log=state.attempt_log,
+            page_logs=state.page_logs,
+            pages_fetched=state.pages_fetched,
+            pages_attempted=state.pages_attempted,
+            has_more_remaining=state.has_more_remaining,
+            pagination_truncated_reason=state.pagination_truncated_reason,
+            continuation_cstart=state.continuation_cstart,
+            skipped_no_change=False,
+            discovered_publication_count=state.discovered_publication_count,
+        )
+
+    def _short_circuit_initial_page(
+        self,
+        *,
+        start_cstart: int,
+        previous_initial_page_fingerprint_sha256: str | None,
+        fetch_result: FetchResult,
+        parsed_page: ParsedProfilePage,
+        first_page_fingerprint_sha256: str | None,
+        attempt_log: list[dict[str, Any]],
+        page_logs: list[dict[str, Any]],
+    ) -> PagedParseResult | None:
+        if self._should_skip_no_change(
+            start_cstart=start_cstart,
+            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
+            previous_initial_page_fingerprint_sha256=previous_initial_page_fingerprint_sha256,
+            parsed_page=parsed_page,
+        ):
+            return self._skip_no_change_result(
+                fetch_result=fetch_result,
+                parsed_page=parsed_page,
+                first_page_fingerprint_sha256=first_page_fingerprint_sha256,
+                attempt_log=attempt_log,
+                page_logs=page_logs,
+            )
+        if parsed_page.state not in {ParseState.OK, ParseState.NO_RESULTS}:
+            return self._initial_failure_result(
+                fetch_result=fetch_result,
+                parsed_page=parsed_page,
+                first_page_fingerprint_sha256=first_page_fingerprint_sha256,
+                start_cstart=start_cstart,
+                attempt_log=attempt_log,
+                page_logs=page_logs,
+            )
+        return None
+
+    # ── Public entry point ───────────────────────────────────────────
+
+    async def fetch_and_parse_all_pages(
+        self,
+        *,
+        scholar: ScholarProfile,
+        run: CrawlRun,
+        db_session: Any,
+        start_cstart: int,
+        request_delay_seconds: int,
+        network_error_retries: int,
+        retry_backoff_seconds: float,
+        rate_limit_retries: int,
+        rate_limit_backoff_seconds: float,
+        max_pages: int,
+        page_size: int,
+        previous_initial_page_fingerprint_sha256: str | None = None,
+        upsert_publications_fn: Any = None,
+    ) -> PagedParseResult:
+        bounded_max_pages = max(1, int(max_pages))
+        bounded_page_size = max(1, int(page_size))
+        (
+            fetch_result,
+            parsed_page,
+            first_page_fingerprint_sha256,
+            attempt_log,
+            page_logs,
+        ) = await self._fetch_initial_page_context(
+            scholar_id=scholar.scholar_id,
+            start_cstart=start_cstart,
+            bounded_page_size=bounded_page_size,
+            network_error_retries=network_error_retries,
+            retry_backoff_seconds=retry_backoff_seconds,
+            rate_limit_retries=rate_limit_retries,
+            rate_limit_backoff_seconds=rate_limit_backoff_seconds,
+        )
+        shortcut_result = self._short_circuit_initial_page(
+            start_cstart=start_cstart,
+            previous_initial_page_fingerprint_sha256=previous_initial_page_fingerprint_sha256,
+            fetch_result=fetch_result,
+            parsed_page=parsed_page,
+            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
+            attempt_log=attempt_log,
+            page_logs=page_logs,
+        )
+        if shortcut_result is not None:
+            return shortcut_result
+        state = self._build_loop_state(
+            start_cstart=start_cstart,
+            fetch_result=fetch_result,
+            parsed_page=parsed_page,
+            attempt_log=attempt_log,
+            page_logs=page_logs,
+        )
+        await self._paginate_loop(
+            scholar=scholar,
+            run=run,
+            db_session=db_session,
+            state=state,
+            bounded_max_pages=bounded_max_pages,
+            request_delay_seconds=request_delay_seconds,
+            bounded_page_size=bounded_page_size,
+            network_error_retries=network_error_retries,
+            retry_backoff_seconds=retry_backoff_seconds,
+            rate_limit_retries=rate_limit_retries,
+            rate_limit_backoff_seconds=rate_limit_backoff_seconds,
+            upsert_publications_fn=upsert_publications_fn,
+        )
+        return self._result_from_pagination_state(
+            state=state,
+            first_page_fetch_result=fetch_result,
+            first_page_parsed_page=parsed_page,
+            first_page_fingerprint_sha256=first_page_fingerprint_sha256,
+        )

app/services/ingestion/publication_upsert.py

@@ -0,0 +1,239 @@
+from __future__ import annotations
+
+import hashlib
+import logging
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import CrawlRun, Publication, ScholarProfile, ScholarPublication
+from app.services.doi.normalize import first_doi_from_texts
+from app.services.ingestion.fingerprints import (
+    build_publication_fingerprint,
+    build_publication_url,
+    canonical_title_for_dedup,
+    normalize_title,
+)
+from app.services.publication_identifiers import application as identifier_service
+from app.services.runs.events import run_events
+from app.services.scholar.parser import PublicationCandidate
+
+logger = logging.getLogger(__name__)
+
+
+def validate_publication_candidate(candidate: PublicationCandidate) -> None:
+    if not candidate.title.strip():
+        raise RuntimeError("Publication candidate is missing title.")
+    if candidate.citation_count is not None and int(candidate.citation_count) < 0:
+        raise RuntimeError("Publication candidate has negative citation_count.")
+
+
+async def find_publication_by_cluster(
+    db_session: AsyncSession,
+    *,
+    cluster_id: str | None,
+) -> Publication | None:
+    if not cluster_id:
+        return None
+    result = await db_session.execute(select(Publication).where(Publication.cluster_id == cluster_id))
+    return result.scalar_one_or_none()
+
+
+async def find_publication_by_fingerprint(
+    db_session: AsyncSession,
+    *,
+    fingerprint: str,
+) -> Publication | None:
+    result = await db_session.execute(select(Publication).where(Publication.fingerprint_sha256 == fingerprint))
+    return result.scalar_one_or_none()
+
+
+def select_existing_publication(
+    *,
+    cluster_publication: Publication | None,
+    fingerprint_publication: Publication | None,
+) -> Publication | None:
+    if cluster_publication is not None:
+        return cluster_publication
+    return fingerprint_publication
+
+
+def compute_canonical_title_hash(title: str) -> str:
+    canonical = canonical_title_for_dedup(title)
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+async def find_publication_by_canonical_title_hash(
+    db_session: AsyncSession,
+    *,
+    canonical_title_hash: str,
+) -> Publication | None:
+    result = await db_session.execute(
+        select(Publication).where(Publication.canonical_title_hash == canonical_title_hash)
+    )
+    return result.scalar_one_or_none()
+
+
+async def create_publication(
+    db_session: AsyncSession,
+    *,
+    candidate: PublicationCandidate,
+    fingerprint: str,
+) -> Publication:
+    publication = Publication(
+        cluster_id=candidate.cluster_id,
+        fingerprint_sha256=fingerprint,
+        title_raw=candidate.title,
+        title_normalized=normalize_title(candidate.title),
+        canonical_title_hash=compute_canonical_title_hash(candidate.title),
+        year=candidate.year,
+        citation_count=int(candidate.citation_count or 0),
+        author_text=candidate.authors_text,
+        venue_text=candidate.venue_text,
+        pub_url=build_publication_url(candidate.title_url),
+        pdf_url=None,
+    )
+    db_session.add(publication)
+    await db_session.flush()
+    return publication
+
+
+def update_existing_publication(
+    *,
+    publication: Publication,
+    candidate: PublicationCandidate,
+) -> None:
+    if candidate.cluster_id and publication.cluster_id is None:
+        publication.cluster_id = candidate.cluster_id
+    publication.title_raw = candidate.title
+    publication.title_normalized = normalize_title(candidate.title)
+    if candidate.year is not None:
+        publication.year = candidate.year
+    if candidate.citation_count is not None:
+        publication.citation_count = int(candidate.citation_count)
+    if candidate.authors_text:
+        publication.author_text = candidate.authors_text
+    if candidate.venue_text:
+        publication.venue_text = candidate.venue_text
+    if candidate.title_url:
+        publication.pub_url = build_publication_url(candidate.title_url)
+    first_doi_from_texts(candidate.title_url, candidate.venue_text, candidate.title)
+
+
+async def resolve_publication(
+    db_session: AsyncSession,
+    candidate: PublicationCandidate,
+) -> Publication:
+    validate_publication_candidate(candidate)
+    fingerprint = build_publication_fingerprint(candidate)
+    cluster_publication = await find_publication_by_cluster(
+        db_session,
+        cluster_id=candidate.cluster_id,
+    )
+    fingerprint_publication = await find_publication_by_fingerprint(
+        db_session,
+        fingerprint=fingerprint,
+    )
+    publication = select_existing_publication(
+        cluster_publication=cluster_publication,
+        fingerprint_publication=fingerprint_publication,
+    )
+    if publication is None:
+        canonical_hash = compute_canonical_title_hash(candidate.title)
+        publication = await find_publication_by_canonical_title_hash(
+            db_session,
+            canonical_title_hash=canonical_hash,
+        )
+    if publication is None:
+        created = await create_publication(
+            db_session,
+            candidate=candidate,
+            fingerprint=fingerprint,
+        )
+        await identifier_service.sync_identifiers_for_publication_fields(
+            db_session,
+            publication=created,
+        )
+        return created
+    update_existing_publication(
+        publication=publication,
+        candidate=candidate,
+    )
+    await identifier_service.sync_identifiers_for_publication_fields(
+        db_session,
+        publication=publication,
+    )
+    return publication
+
+
+async def upsert_profile_publications(
+    db_session: AsyncSession,
+    *,
+    run: CrawlRun,
+    scholar: ScholarProfile,
+    publications: list[PublicationCandidate],
+) -> int:
+    seen_publication_ids: set[int] = set()
+    discovered_count = 0
+
+    for candidate in publications:
+        publication = await resolve_publication(db_session, candidate)
+        if publication.id in seen_publication_ids:
+            continue
+        seen_publication_ids.add(publication.id)
+
+        link_result = await db_session.execute(
+            select(ScholarPublication).where(
+                ScholarPublication.scholar_profile_id == scholar.id,
+                ScholarPublication.publication_id == publication.id,
+            )
+        )
+        link = link_result.scalar_one_or_none()
+        if link is not None:
+            continue
+
+        link = ScholarPublication(
+            scholar_profile_id=scholar.id,
+            publication_id=publication.id,
+            is_read=False,
+            first_seen_run_id=run.id,
+        )
+        db_session.add(link)
+        discovered_count += 1
+
+        await commit_discovered_publication(
+            db_session,
+            run=run,
+            scholar=scholar,
+            publication=publication,
+        )
+
+    if not scholar.baseline_completed:
+        scholar.baseline_completed = True
+
+    return discovered_count
+
+
+async def commit_discovered_publication(
+    db_session: AsyncSession,
+    *,
+    run: CrawlRun,
+    scholar: ScholarProfile,
+    publication: Publication,
+) -> None:
+    run.new_pub_count = int(run.new_pub_count or 0) + 1
+    await db_session.commit()
+    await run_events.publish(
+        run_id=run.id,
+        event_type="publication_discovered",
+        data={
+            "publication_id": publication.id,
+            "title": publication.title_raw,
+            "pub_url": publication.pub_url,
+            "scholar_profile_id": scholar.id,
+            "scholar_label": scholar.display_name or scholar.scholar_id,
+            "first_seen_at": datetime.now(UTC).isoformat(),
+            "new_publication_count": int(run.new_pub_count or 0),
+        },
+    )

app/services/portability/normalize.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import hashlib
 from typing import Any
 
-from app.services.ingestion.application import normalize_title
+from app.services.ingestion.fingerprints import normalize_title
 from app.services.portability.constants import (
     MAX_IMPORT_PUBLICATIONS,
     MAX_IMPORT_SCHOLARS,

app/services/portability/publication_import.py

@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db.models import Publication, ScholarProfile, ScholarPublication
 from app.services.doi.normalize import normalize_doi
-from app.services.ingestion.application import build_publication_url, normalize_title
+from app.services.ingestion.fingerprints import build_publication_url, normalize_title
 from app.services.portability.normalize import (
     _normalize_citation_count,
     _normalize_optional_text,