Create-knowledge-find-decode/acquisition/runner.py

"""Formal acquisition runner backed by the repository boundary."""
from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Callable, Iterable
from uuid import UUID

from acquisition.classification.coarse import ClassificationResult, coarse_classify_item
from acquisition.content_mode import guard_content_for_processing, infer_content_mode
from acquisition.crawler import RateLimiter
from acquisition.domain import AcquisitionJob, Query
from acquisition.media.service import StabilizedMedia, stabilize_media_urls
from acquisition.platforms import PlatformAdapter, get_platform_adapter
from acquisition.platforms.base import PlatformCandidate, PlatformSearchPage
from acquisition.repositories.base import AcquisitionRepository
from acquisition.unique_key import build_unique_key
from core.config import Settings
from core.text_limits import (
    BODY_TEXT_MAX_CHARS,
    ERROR_MESSAGE_MAX_CHARS,
    RAW_SUMMARY_MAX_CHARS,
    clip_text,
)
from pipeline.tracing import NoopTraceWriter, TraceContext, TraceWriter

DEFAULT_PLATFORMS = ("xiaohongshu", "weixin", "douyin")
PAGINATION_STRATEGY = "creation_ratio_two_page_v1"
PAGINATION_CREATION_THRESHOLD = 0.4
PAGINATION_MAX_PAGES = 2
LOW_RESULT_RETRY_THRESHOLD = 5
LOW_RESULT_MAX_RETRIES = 1

AdapterFactory = Callable[[str], PlatformAdapter]
Classifier = Callable[..., ClassificationResult]
MediaStabilizer = Callable[..., list[StabilizedMedia]]
RateLimiterFactory = Callable[[str], Any]


@dataclass(frozen=True)
class RunBatchResult:
    run_id: UUID
    total_jobs: int
    done: int
    partial: int
    failed: int
    skipped: int


@dataclass(frozen=True)
class RecordCandidateResult:
    displayed: bool
    skipped_processing: bool = False
    classified: bool = False
    is_creation_knowledge: bool | None = None


class PlatformRateLimiter:
    """Share one rate bucket per platform across search and detail calls."""

    def __init__(self, platform: str, delegate: RateLimiter | None = None) -> None:
        self.platform = platform
        self.delegate = delegate or RateLimiter(
            min_interval_seconds=10.0,
            max_interval_seconds=12.0,
        )

    def wait(self, bucket: str) -> None:
        self.delegate.wait(self.platform)


def _query_id(query: Query) -> UUID:
    if query.id is None:
        raise RuntimeError("formal acquisition queries must have id before running")
    return query.id


def _job_id(job: AcquisitionJob) -> UUID:
    if job.id is None:
        raise RuntimeError("formal acquisition jobs must have id before running")
    return job.id


def _best_video_url(media: list[StabilizedMedia]) -> str:
    for row in media:
        if row.media_type == "video" and row.status == "ready" and row.cdn_url:
            return row.cdn_url
    return ""


def _image_urls(media: list[StabilizedMedia]) -> list[str]:
    return [row.cdn_url for row in media if row.media_type == "image" and row.cdn_url]


def _source_payload(candidate: Any, detail: Any) -> dict[str, Any]:
    return {
        "candidate": candidate.model_dump() if hasattr(candidate, "model_dump") else {},
        "detail": detail.raw if isinstance(getattr(detail, "raw", None), dict) else {},
    }


def _candidate_provider(candidate: Any) -> str:
    provider = getattr(candidate, "provider", "") or ""
    raw = getattr(candidate, "raw", None)
    if not provider and isinstance(raw, dict):
        provider = raw.get("search_provider") or ""
    return provider


def _detail_provider(detail: Any, candidate: Any) -> str:
    provider = getattr(detail, "provider", "") or ""
    if not provider:
        raw = getattr(detail, "raw", None)
        if isinstance(raw, dict):
            provider = raw.get("detail_provider") or ""
    return provider or _candidate_provider(candidate)


def _candidate_page_metadata(candidate: Any) -> dict[str, Any]:
    raw = getattr(candidate, "raw", None)
    if not isinstance(raw, dict):
        return {}
    out: dict[str, Any] = {}
    for key in ("page_index", "page_rank", "source_cursor"):
        value = raw.get(key)
        if value not in (None, ""):
            out[key] = value
    return out


def _content_mode(platform: str, detail: Any) -> str:
    explicit = getattr(detail, "content_mode", "") or ""
    if explicit:
        return explicit
    return infer_content_mode(
        platform=platform,
        content_type=getattr(detail, "content_type", "") or "",
        body_text=getattr(detail, "body_text", "") or "",
        image_urls=getattr(detail, "image_urls", []) or [],
        video_urls=getattr(detail, "video_urls", []) or [],
        raw=getattr(detail, "raw", {}) if isinstance(getattr(detail, "raw", None), dict) else {},
    )


def _skip_reason_message(reason: str) -> str:
    if reason == "video_url_missing":
        return "Video post has no usable video URL; skipped before classification."
    if reason == "video_oss_failed":
        return "Video OSS transfer failed after retries; skipped before classification."
    if reason == "unsupported_content_mode":
        return "Unsupported content mode; skipped before classification."
    return reason or "Skipped before classification."


def _requires_strict_video_oss(platform: str, content_mode: str) -> bool:
    return content_mode == "video_post"


def _video_media_errors(media_rows: list[StabilizedMedia]) -> list[dict[str, Any]]:
    errors: list[dict[str, Any]] = []
    for row in media_rows:
        if row.media_type != "video" or row.status == "ready":
            continue
        errors.append(
            {
                "source_url": row.source_url,
                "status": row.status,
                "error_message": row.error_message,
            }
        )
    return errors


def _record_candidate(
    repo: AcquisitionRepository,
    *,
    job: AcquisitionJob,
    query: Query,
    platform: str,
    candidate: Any,
    detail: Any,
    settings: Settings,
    media_stabilizer: MediaStabilizer,
    classifier: Classifier,
    classify: bool,
    attempt_index: int = 1,
    trace_writer: TraceWriter | None = None,
    trace_context: TraceContext | None = None,
) -> RecordCandidateResult:
    source_payload = _source_payload(candidate, detail)
    platform_item_id = detail.source_id or candidate.source_id or None
    canonical_url = detail.url or candidate.url or None
    content_mode = _content_mode(platform, detail)
    guard = guard_content_for_processing(
        content_mode=content_mode,
        video_urls=detail.video_urls,
    )
    metadata = {
        "candidate_rank": candidate.rank,
        "acquisition_match_status": "created",
        "content_mode": content_mode,
        "retry_attempt_index": attempt_index,
        **_candidate_page_metadata(candidate),
    }
    search_provider = _candidate_provider(candidate)
    detail_provider = _detail_provider(detail, candidate)
    if search_provider:
        metadata["search_provider"] = search_provider
    if detail_provider:
        metadata["detail_provider"] = detail_provider
    if guard.metadata:
        metadata.update(guard.metadata)
    body_text = clip_text(detail.body_text or "", BODY_TEXT_MAX_CHARS)
    item = repo.upsert_candidate_item(
        platform=platform,
        job_id=_job_id(job),
        query_id=_query_id(query),
        platform_item_id=platform_item_id,
        unique_key=build_unique_key(
            platform,
            platform_item_id=platform_item_id,
            url=canonical_url,
            raw_payload=source_payload,
        ),
        canonical_url=canonical_url,
        content_type=detail.content_type or None,
        content_mode=content_mode,
        title=detail.title or None,
        author_name=detail.author or None,
        body_text=body_text or None,
        raw_summary=clip_text(body_text, RAW_SUMMARY_MAX_CHARS) or None,
        status="candidate" if guard.can_process else "skipped",
        source_payload=source_payload,
        metadata=metadata,
        error_message=None if guard.can_process else guard.reason,
    )
    if item.id is None:
        raise RuntimeError("repository returned candidate without id")
    _record_candidate_hit(
        trace_writer,
        trace_context,
        item_id=item.id,
        platform=platform,
        unique_key=item.unique_key,
        platform_item_id=item.platform_item_id,
        search_provider=search_provider,
        detail_provider=detail_provider,
        candidate=candidate,
        attempt_index=attempt_index,
        is_duplicate_hit=False,
        metadata={"content_mode": content_mode, "status": item.status},
    )

    if not guard.can_process:
        _trace_event(
            trace_writer,
            trace_context.child(item_id=item.id, platform=platform) if trace_context else None,
            stage="classify",
            event_type="item_skipped_before_classification",
            status="skipped",
            target_table="candidate_items",
            target_id=item.id,
            payload={"reason": guard.reason, "content_mode": content_mode},
        )
        repo.add_item_classification(
            item_id=item.id,
            is_creation_knowledge=None,
            label=guard.label,
            confidence=None,
            reason=_skip_reason_message(guard.reason),
            model_name=None,
            prompt_version="content_mode_guard",
            result_payload={
                "content_mode": content_mode,
                "skip_reason": guard.reason,
                **(guard.metadata or {}),
            },
            status="skipped",
            error_message=guard.reason,
        )
        return RecordCandidateResult(displayed=True, skipped_processing=True)

    strict_video_oss = _requires_strict_video_oss(platform, content_mode)
    media_rows = media_stabilizer(
        image_urls=detail.image_urls,
        video_urls=detail.video_urls,
        settings=settings,
        video_fallback_to_source=not strict_video_oss,
        video_retry_delays_seconds=(
            settings.video_oss_retry_delays_seconds if strict_video_oss else ()
        ),
    )

    for row in media_rows:
        repo.add_media_asset(
            item_id=item.id,
            media_type=row.media_type,
            source_url=row.source_url,
            oss_url=row.cdn_url,
            cdn_url=row.cdn_url,
            position=row.position,
            status=row.status,
            source_payload={},
            metadata={"error_message": row.error_message} if row.error_message else {},
        )

    if strict_video_oss and not _best_video_url(media_rows):
        repo.add_item_classification(
            item_id=item.id,
            is_creation_knowledge=None,
            label="video_oss_failed",
            confidence=None,
            reason=_skip_reason_message("video_oss_failed"),
            model_name=None,
            prompt_version="media_stabilization_guard",
            result_payload={
                "content_mode": content_mode,
                "skip_reason": "video_oss_failed",
                "video_oss_retry_delays_seconds": list(
                    settings.video_oss_retry_delays_seconds
                ),
                "media_errors": _video_media_errors(media_rows),
            },
            status="skipped",
            error_message="video_oss_failed",
        )
        _trace_event(
            trace_writer,
            trace_context.child(item_id=item.id, platform=platform) if trace_context else None,
            stage="classify",
            event_type="item_skipped_before_classification",
            status="skipped",
            target_table="candidate_items",
            target_id=item.id,
            payload={"reason": "video_oss_failed", "content_mode": content_mode},
        )
        return RecordCandidateResult(displayed=True, skipped_processing=True)

    if classify:
        classifier_kwargs = {
            "platform": platform,
            "content_mode": content_mode,
            "title": detail.title,
            "body_text": detail.body_text,
            "image_urls": _image_urls(media_rows),
            "video_url": _best_video_url(media_rows),
            "settings": settings,
        }
        if classifier is coarse_classify_item:
            classifier_kwargs.update(
                {
                    "trace_writer": trace_writer,
                    "trace_context": trace_context.child(item_id=item.id, platform=platform) if trace_context else None,
                }
            )
        result = classifier(**classifier_kwargs)
        repo.add_item_classification(
            item_id=item.id,
            is_creation_knowledge=result.is_creation_knowledge,
            label=result.label,
            confidence=result.confidence,
            reason=result.reason,
            model_name=settings.video_model,
            prompt_version=result.prompt_version,
            result_payload=result.result_payload,
            status=result.status,
            error_message=result.error_message,
        )
        _trace_event(
            trace_writer,
            trace_context.child(item_id=item.id, platform=platform) if trace_context else None,
            stage="classify",
            event_type="classified",
            status=result.status,
            target_table="candidate_items",
            target_id=item.id,
            payload={
                "is_creation_knowledge": result.is_creation_knowledge,
                "label": result.label,
                "confidence": result.confidence,
            },
            error_message=result.error_message,
        )
        return RecordCandidateResult(
            displayed=True,
            classified=result.status == "classified" and result.is_creation_knowledge is not None,
            is_creation_knowledge=result.is_creation_knowledge,
        )
    return RecordCandidateResult(displayed=True)


def _candidate_unique_key(platform: str, candidate: PlatformCandidate) -> str | None:
    return build_unique_key(
        platform,
        platform_item_id=candidate.source_id or None,
        url=candidate.url or None,
        raw_payload={"candidate": candidate.model_dump()},
    )


def _candidate_dedupe_key(platform: str, candidate: PlatformCandidate) -> str | None:
    unique_key = _candidate_unique_key(platform, candidate)
    if unique_key:
        return f"unique:{unique_key}"
    if candidate.source_id:
        return f"source:{platform}:{candidate.source_id}"
    if candidate.url:
        return f"url:{platform}:{candidate.url}"
    return None


def _attach_existing_candidate(
    repo: AcquisitionRepository,
    *,
    job: AcquisitionJob,
    query: Query,
    candidate: PlatformCandidate,
    unique_key: str,
    attempt_index: int = 1,
    trace_writer: TraceWriter | None = None,
    trace_context: TraceContext | None = None,
) -> bool:
    get_by_key = getattr(repo, "get_candidate_item_by_unique_key", None)
    attach = getattr(repo, "attach_existing_candidate_item", None)
    if get_by_key is None or attach is None:
        return False
    existing = get_by_key(unique_key)
    if existing is None or existing.id is None:
        return False
    attach(
        existing.id,
        job_id=_job_id(job),
        query_id=_query_id(query),
        metadata={
            "acquisition_match_status": "existing",
            "matched_unique_key": unique_key,
            "matched_candidate": candidate.model_dump(),
            "candidate_rank": candidate.rank,
            "retry_attempt_index": attempt_index,
            **_candidate_page_metadata(candidate),
            **({"search_provider": _candidate_provider(candidate)} if _candidate_provider(candidate) else {}),
        },
    )
    _record_candidate_hit(
        trace_writer,
        trace_context,
        item_id=existing.id,
        platform=candidate.platform,
        unique_key=unique_key,
        platform_item_id=candidate.source_id or None,
        search_provider=_candidate_provider(candidate),
        detail_provider=None,
        candidate=candidate,
        attempt_index=attempt_index,
        is_duplicate_hit=True,
        metadata={"acquisition_match_status": "existing"},
    )
    _trace_event(
        trace_writer,
        trace_context.child(item_id=existing.id, platform=candidate.platform) if trace_context else None,
        stage="search",
        event_type="duplicate_candidate_reused",
        status="skipped",
        target_table="candidate_items",
        target_id=existing.id,
        payload={"unique_key": unique_key},
    )
    return True


def _trace_event(
    trace_writer: TraceWriter | None,
    context: TraceContext | None,
    *,
    stage: str,
    event_type: str,
    **kwargs: Any,
) -> None:
    if trace_writer is None or context is None:
        return
    trace_writer.event(context=context, stage=stage, event_type=event_type, **kwargs)


def _record_candidate_hit(
    trace_writer: TraceWriter | None,
    context: TraceContext | None,
    *,
    item_id: UUID | None,
    platform: str,
    unique_key: str | None,
    platform_item_id: str | None,
    search_provider: str | None,
    detail_provider: str | None,
    candidate: PlatformCandidate,
    attempt_index: int,
    is_duplicate_hit: bool,
    metadata: dict[str, Any] | None = None,
) -> None:
    if trace_writer is None or context is None:
        return
    raw = candidate.model_dump() if hasattr(candidate, "model_dump") else {}
    trace_writer.candidate_hit(
        context=context.child(item_id=item_id, platform=platform),
        item_id=item_id,
        platform=platform,
        unique_key=unique_key,
        platform_item_id=platform_item_id,
        search_provider=search_provider,
        detail_provider=detail_provider,
        attempt_index=attempt_index,
        page_index=raw.get("raw", {}).get("page_index") if isinstance(raw.get("raw"), dict) else None,
        page_rank=raw.get("raw", {}).get("page_rank") if isinstance(raw.get("raw"), dict) else None,
        candidate_rank=candidate.rank,
        source_cursor=raw.get("raw", {}).get("source_cursor") if isinstance(raw.get("raw"), dict) else None,
        is_duplicate_hit=is_duplicate_hit,
        raw_candidate=raw,
        metadata=metadata or {},
    )


def _pagination_config(threshold: float, max_pages: int) -> dict[str, Any]:
    return {
        "strategy": PAGINATION_STRATEGY,
        "creation_threshold": threshold,
        "max_pages": max_pages,
    }


def _ratio(numerator: int, denominator: int) -> float | None:
    if denominator <= 0:
        return None
    return numerator / denominator


def _search_pages(
    adapter: PlatformAdapter,
    query_text: str,
    *,
    settings: Settings,
    limit: int,
    rate_limiter: Any,
    max_pages: int,
) -> Iterable[PlatformSearchPage]:
    search_pages = getattr(adapter, "search_pages", None)
    if callable(search_pages):
        return search_pages(
            query_text,
            settings=settings,
            limit=limit,
            rate_limiter=rate_limiter,
            max_pages=max_pages,
        )
    candidates = adapter.search(
        query_text,
        settings=settings,
        limit=limit,
        rate_limiter=rate_limiter,
    )
    return [
        PlatformSearchPage(
            page_index=1,
            candidates=candidates,
            raw_count=len(candidates),
            has_more=False,
        )
    ]


def run_batch(
    repo: AcquisitionRepository,
    *,
    batch_id: UUID,
    settings: Settings,
    platforms: tuple[str, ...] | list[str] = DEFAULT_PLATFORMS,
    search_limit: int = 10,
    display_limit: int = 5,
    classify: bool = True,
    resume: bool = True,
    skip_done: bool = True,
    run_key: str | None = None,
    adapter_factory: AdapterFactory = get_platform_adapter,
    media_stabilizer: MediaStabilizer = stabilize_media_urls,
    classifier: Classifier = coarse_classify_item,
    rate_limiter_factory: RateLimiterFactory | None = None,
    pagination_creation_threshold: float = PAGINATION_CREATION_THRESHOLD,
    pagination_max_pages: int = PAGINATION_MAX_PAGES,
    low_result_retry_threshold: int = LOW_RESULT_RETRY_THRESHOLD,
    low_result_max_retries: int = LOW_RESULT_MAX_RETRIES,
    trace_writer: TraceWriter | None = None,
    trace_context: TraceContext | None = None,
) -> RunBatchResult:
    """Run query x platform acquisition and write formal cloud-state rows."""
    queries = repo.list_queries_for_batch(batch_id, keep=True)
    run = repo.create_acquisition_run(
        batch_id=batch_id,
        run_key=run_key or f"acquisition:{batch_id}",
        status="running",
        metadata={
            "platforms": list(platforms),
            "search_limit": search_limit,
            "display_limit": display_limit,
            "classify": classify,
            "resume": resume,
            "pagination": _pagination_config(
                pagination_creation_threshold,
                pagination_max_pages,
            ),
            "low_result_retry": {
                "threshold": low_result_retry_threshold,
                "max_retries": low_result_max_retries,
            },
        },
    )
    if run.id is None:
        raise RuntimeError("repository returned acquisition run without id")
    trace_writer = trace_writer or NoopTraceWriter()
    run_trace_context = (trace_context or TraceContext()).child(
        acquisition_run_id=run.id,
        stage="search",
    )
    _trace_event(
        trace_writer,
        run_trace_context,
        stage="search",
        event_type="acquisition_started",
        status="running",
        target_table="acquisition_runs",
        target_id=run.id,
        payload={"batch_id": str(batch_id), "platforms": list(platforms)},
    )

    total_jobs = len(queries) * len(platforms)
    done = partial = failed = skipped = 0
    gates: dict[str, Any] = {}

    for query in queries:
        query_id = _query_id(query)
        for platform in platforms:
            job = repo.ensure_acquisition_job(
                run_id=run.id,
                query_id=query_id,
                platform=platform,
                search_limit=search_limit,
                display_limit=display_limit,
                status="pending",
                metadata={"query_text": query.query_text},
            )
            if skip_done and job.status == "done":
                _trace_event(
                    trace_writer,
                    run_trace_context.child(
                        acquisition_job_id=_job_id(job),
                        query_id=query_id,
                        platform=platform,
                    ),
                    stage="search",
                    event_type="job_skipped_done",
                    status="skipped",
                    target_table="acquisition_jobs",
                    target_id=_job_id(job),
                )
                skipped += 1
                continue

            attempts = job.attempt_count + 1
            job = repo.update_acquisition_job(
                _job_id(job),
                status="running",
                attempt_count=attempts,
                error_message=None,
            )
            job_trace_context = run_trace_context.child(
                acquisition_job_id=_job_id(job),
                query_id=query_id,
                platform=platform,
            )
            _trace_event(
                trace_writer,
                job_trace_context,
                stage="search",
                event_type="job_started",
                status="running",
                target_table="acquisition_jobs",
                target_id=_job_id(job),
                payload={"query_text": query.query_text, "platform": platform},
            )
            errors: list[str] = []
            display_count = 0
            searched_count = 0
            raw_searched_count = 0
            skipped_item_count = 0
            search_page_count = 0
            page_summaries: list[dict[str, Any]] = []
            first_page_classified_count = 0
            first_page_creation_count = 0
            first_page_creation_ratio: float | None = None
            requested_second_page = False
            pagination_stop_reason = ""
            try:
                adapter = adapter_factory(platform)
                gate = gates.get(platform)
                if gate is None:
                    gate = (
                        rate_limiter_factory(platform)
                        if rate_limiter_factory
                        else PlatformRateLimiter(platform)
                    )
                    gates[platform] = gate
                retry_enabled = (
                    low_result_max_retries > 0
                    and low_result_retry_threshold >= 0
                    and search_limit > low_result_retry_threshold
                )
                max_attempts = 1 + (low_result_max_retries if retry_enabled else 0)
                seen_candidate_keys: set[str] = set()
                retry_attempts: list[dict[str, Any]] = []
                retry_triggered = False

                for attempt_index in range(1, max_attempts + 1):
                    attempt_display_start = display_count
                    attempt_searched_start = searched_count
                    attempt_raw_start = raw_searched_count
                    attempt_pages_start = search_page_count
                    attempt_page_summaries: list[dict[str, Any]] = []
                    attempt_first_page_classified_count = 0
                    attempt_first_page_creation_count = 0
                    attempt_first_page_creation_ratio: float | None = None
                    attempt_requested_second_page = False
                    attempt_stop_reason = ""

                    pages = _search_pages(
                        adapter,
                        query.query_text,
                        settings=settings,
                        limit=search_limit,
                        rate_limiter=gate,
                        max_pages=pagination_max_pages,
                    )
                    for page in pages:
                        search_page_count += 1
                        raw_searched_count += page.raw_count
                        page_classified_count = 0
                        page_creation_count = 0
                        unique_candidate_count = 0

                        for candidate in page.candidates[:search_limit]:
                            dedupe_key = _candidate_dedupe_key(platform, candidate)
                            if dedupe_key and dedupe_key in seen_candidate_keys:
                                continue
                            if dedupe_key:
                                seen_candidate_keys.add(dedupe_key)
                            unique_candidate_count += 1
                            searched_count += 1
                            try:
                                unique_key = _candidate_unique_key(platform, candidate)
                                if unique_key and _attach_existing_candidate(
                                    repo,
                                    job=job,
                                    query=query,
                                    candidate=candidate,
                                    unique_key=unique_key,
                                    attempt_index=attempt_index,
                                    trace_writer=trace_writer,
                                    trace_context=job_trace_context,
                                ):
                                    display_count += 1
                                    continue
                                detail = adapter.fetch_detail(
                                    candidate,
                                    settings=settings,
                                    rate_limiter=gate,
                                )
                                recorded = _record_candidate(
                                    repo,
                                    job=job,
                                    query=query,
                                    platform=platform,
                                    candidate=candidate,
                                    detail=detail,
                                    settings=settings,
                                    media_stabilizer=media_stabilizer,
                                    classifier=classifier,
                                    classify=classify,
                                    attempt_index=attempt_index,
                                    trace_writer=trace_writer,
                                    trace_context=job_trace_context,
                                )
                                if recorded.displayed:
                                    display_count += 1
                                    if recorded.skipped_processing:
                                        skipped_item_count += 1
                                if recorded.classified:
                                    page_classified_count += 1
                                    if recorded.is_creation_knowledge is True:
                                        page_creation_count += 1
                            except Exception as exc:
                                message = clip_text(exc, ERROR_MESSAGE_MAX_CHARS)
                                errors.append(message)
                                _trace_event(
                                    trace_writer,
                                    job_trace_context,
                                    stage="search",
                                    event_type="candidate_failed",
                                    status="failed",
                                    severity="warning",
                                    payload={
                                        "candidate": candidate.model_dump() if hasattr(candidate, "model_dump") else {},
                                        "attempt_index": attempt_index,
                                    },
                                    error_message=message,
                                )
                                continue

                        page_ratio = _ratio(page_creation_count, page_classified_count)
                        page_summary = {
                            "attempt_index": attempt_index,
                            "page_index": page.page_index,
                            "raw_count": page.raw_count,
                            "candidate_count": len(page.candidates),
                            "unique_candidate_count": unique_candidate_count,
                            "classified_count": page_classified_count,
                            "creation_count": page_creation_count,
                            "creation_ratio": page_ratio,
                            "has_more": page.has_more,
                            "next_cursor": page.next_cursor,
                            "provider": page.provider,
                        }
                        page_summaries.append(page_summary)
                        attempt_page_summaries.append(page_summary)
                        _trace_event(
                            trace_writer,
                            job_trace_context,
                            stage="search",
                            event_type="page_searched",
                            status="done",
                            payload=page_summary,
                            attempt_index=attempt_index,
                        )

                        if page.page_index == 1:
                            attempt_first_page_classified_count = page_classified_count
                            attempt_first_page_creation_count = page_creation_count
                            attempt_first_page_creation_ratio = page_ratio
                            if pagination_max_pages <= 1:
                                attempt_stop_reason = "max_pages_reached"
                                break
                            if not page.has_more or not page.next_cursor:
                                attempt_stop_reason = "no_more_pages"
                                break
                            if page_classified_count <= 0:
                                attempt_stop_reason = "no_classified_candidates"
                                break
                            if page_ratio is None or page_ratio < pagination_creation_threshold:
                                attempt_stop_reason = "below_threshold"
                                break
                            attempt_requested_second_page = True
                            continue

                        attempt_stop_reason = "max_pages_reached"
                        break

                    if not attempt_stop_reason:
                        attempt_stop_reason = "pages_exhausted"

                    first_page_classified_count = attempt_first_page_classified_count
                    first_page_creation_count = attempt_first_page_creation_count
                    first_page_creation_ratio = attempt_first_page_creation_ratio
                    requested_second_page = attempt_requested_second_page
                    pagination_stop_reason = attempt_stop_reason
                    attempt_display_count = display_count - attempt_display_start
                    retry_attempts.append(
                        {
                            "attempt_index": attempt_index,
                            "display_count": attempt_display_count,
                            "searched_count": searched_count - attempt_searched_start,
                            "raw_searched_count": raw_searched_count - attempt_raw_start,
                            "search_page_count": search_page_count - attempt_pages_start,
                            "stop_reason": attempt_stop_reason,
                            "requested_second_page": attempt_requested_second_page,
                            "first_page_classified_count": attempt_first_page_classified_count,
                            "first_page_creation_count": attempt_first_page_creation_count,
                            "first_page_creation_ratio": attempt_first_page_creation_ratio,
                            "pages": attempt_page_summaries,
                        }
                    )

                    if (
                        attempt_index <= low_result_max_retries
                        and display_count <= low_result_retry_threshold
                    ):
                        retry_triggered = True
                        continue
                    break

                status = "done" if display_count else "failed"
                if status == "done":
                    done += 1
                else:
                    failed += 1
                repo.update_acquisition_job(
                    _job_id(job),
                    status=status,
                    attempt_count=attempts,
                    error_message=None if status != "failed" else "; ".join(errors[-3:]),
                    metadata={
                        "query_text": query.query_text,
                        "searched_count": searched_count,
                        "raw_searched_count": raw_searched_count,
                        "display_count": display_count,
                        "skipped_item_count": skipped_item_count,
                        "search_page_count": search_page_count,
                        "pagination": {
                            **_pagination_config(
                                pagination_creation_threshold,
                                pagination_max_pages,
                            ),
                            "first_page_classified_count": first_page_classified_count,
                            "first_page_creation_count": first_page_creation_count,
                            "first_page_creation_ratio": first_page_creation_ratio,
                            "requested_second_page": requested_second_page,
                            "stop_reason": pagination_stop_reason,
                            "pages": page_summaries,
                        },
                        "low_result_retry": {
                            "threshold": low_result_retry_threshold,
                            "max_retries": low_result_max_retries if retry_enabled else 0,
                            "triggered": retry_triggered,
                            "attempt_count": len(retry_attempts),
                            "attempts": retry_attempts,
                        },
                        "errors": errors[-3:],
                    },
                )
                _trace_event(
                    trace_writer,
                    job_trace_context,
                    stage="search",
                    event_type="job_finished",
                    status=status,
                    target_table="acquisition_jobs",
                    target_id=_job_id(job),
                    payload={
                        "searched_count": searched_count,
                        "raw_searched_count": raw_searched_count,
                        "display_count": display_count,
                        "skipped_item_count": skipped_item_count,
                        "search_page_count": search_page_count,
                    },
                    error_message=None if status != "failed" else "; ".join(errors[-3:]),
                )
            except Exception as exc:
                message = clip_text(exc, ERROR_MESSAGE_MAX_CHARS)
                failed += 1
                repo.update_acquisition_job(
                    _job_id(job),
                    status="failed",
                    attempt_count=attempts,
                    error_message=message,
                    metadata={
                        "query_text": query.query_text,
                        "searched_count": searched_count,
                        "raw_searched_count": raw_searched_count,
                        "display_count": display_count,
                        "skipped_item_count": skipped_item_count,
                        "search_page_count": search_page_count,
                        "pagination": {
                            **_pagination_config(
                                pagination_creation_threshold,
                                pagination_max_pages,
                            ),
                            "first_page_classified_count": first_page_classified_count,
                            "first_page_creation_count": first_page_creation_count,
                            "first_page_creation_ratio": first_page_creation_ratio,
                            "requested_second_page": requested_second_page,
                            "stop_reason": pagination_stop_reason or "failed",
                            "pages": page_summaries,
                        },
                        "errors": errors[-3:],
                    },
                )
                _trace_event(
                    trace_writer,
                    job_trace_context,
                    stage="search",
                    event_type="job_failed",
                    status="failed",
                    severity="error",
                    target_table="acquisition_jobs",
                    target_id=_job_id(job),
                    error_message=message,
                )

    run_status = "done" if done > 0 else "failed"
    update_run = getattr(repo, "update_acquisition_run", None)
    if update_run:
        update_run(
            run.id,
            status=run_status,
            metadata={
                "total_jobs": total_jobs,
                "done": done,
                "partial": partial,
                "failed": failed,
                "skipped": skipped,
                "pagination": _pagination_config(
                    pagination_creation_threshold,
                    pagination_max_pages,
                ),
            },
        )
    _trace_event(
        trace_writer,
        run_trace_context,
        stage="search",
        event_type="acquisition_finished",
        status=run_status,
        target_table="acquisition_runs",
        target_id=run.id,
        payload={
            "total_jobs": total_jobs,
            "done": done,
            "partial": partial,
            "failed": failed,
            "skipped": skipped,
        },
    )

    return RunBatchResult(
        run_id=run.id,
        total_jobs=total_jobs,
        done=done,
        partial=partial,
        failed=failed,
        skipped=skipped,
    )