SupplyAgent/find_agent_v2/service.py

"""Transactional service for the isolated ``find_agent_v2_*`` tables."""

from __future__ import annotations

import json
import uuid
from dataclasses import asdict
from decimal import Decimal
from typing import Any

from sqlalchemy import case, func, select

from find_agent_v2.models import (
    FindAgentV2Candidate,
    FindAgentV2Evidence,
    FindAgentV2Round,
    FindAgentV2Run,
    FindAgentV2Search,
)
from find_agent_v2.state import DiscoverySnapshot, ExecutionPlan
from supply_infra.db.session import get_session
from find_agent_v2.gates import (
    build_rule_snapshot,
    evaluate_candidate_gate,
    parse_datetime_value,
)


RUN_TIMEOUT_MINUTES = 60
RUN_TIMEOUT_SECONDS = RUN_TIMEOUT_MINUTES * 60
RUN_TIMEOUT_REASON = "运行时间超过 60 分钟，系统自动标记失败"


class FindAgentV2RunNotFound(LookupError):
    pass


def _json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, default=str)


def _loads(value: str | None, default: Any) -> Any:
    try:
        return json.loads(value) if value else default
    except (TypeError, ValueError):
        return default


def _ratio(value: Any) -> Decimal | None:
    if value in (None, "") or isinstance(value, bool):
        return None
    number = Decimal(str(value))
    if number > 1 and number <= 100:
        number /= 100
    if number < 0 or number > 1:
        raise ValueError("ratio 必须在 0~1")
    return number.quantize(Decimal("0.000001"))


def _fails_search_share_gate(
    provider: str, share_count: Any, min_share_count: int,
) -> bool:
    """Reject any provider's explicit share metric below the configured threshold."""
    del provider
    return share_count is not None and int(share_count) < int(min_share_count)


def _candidate_dict(row: FindAgentV2Candidate) -> dict[str, Any]:
    detail = _loads(row.detail_json, {})
    return {
        "candidate_id": int(row.id),
        "aweme_id": row.aweme_id,
        "title": row.title,
        "content_link": row.content_link,
        "video_url": detail.get("video_url") if isinstance(detail, dict) else None,
        "author_name": row.author_name,
        "author_sec_uid": row.author_sec_uid,
        "source_keywords": _loads(row.source_keywords_json, []),
        "tags": _loads(row.tags_json, []),
        "publish_at": row.publish_at.isoformat() if row.publish_at else None,
        "duration_seconds": float(row.duration_seconds) if row.duration_seconds is not None else None,
        "play_count": row.play_count,
        "like_count": row.like_count,
        "comment_count": row.comment_count,
        "collect_count": row.collect_count,
        "share_count": row.share_count,
        "detail_status": row.detail_status,
        "portrait_status": row.portrait_status,
        "content_50_plus_ratio": float(row.content_50_plus_ratio) if row.content_50_plus_ratio is not None else None,
        "account_50_plus_ratio": float(row.account_50_plus_ratio) if row.account_50_plus_ratio is not None else None,
        "relevance_score": float(row.relevance_score) if row.relevance_score is not None else None,
        "elder_score": float(row.elder_score) if row.elder_score is not None else None,
        "share_score": float(row.share_score) if row.share_score is not None else None,
        "value_score": float(row.value_score) if row.value_score is not None else None,
        "gate_status": row.gate_status,
        "gate_result": _loads(row.gate_result_json, {}),
        "decision_bucket": row.decision_bucket,
        "decision_reason": row.decision_reason,
        "reject_reason_code": row.reject_reason_code,
    }


class FindAgentV2Service:
    def create_run(
        self,
        *,
        user_input: str,
        demand_word: str,
        demand_grade_id: int | None = None,
        run_id: str | None = None,
        rule_config: dict[str, Any] | None = None,
        triggered_by_user_id: int | None = None,
        triggered_by_username: str | None = None,
        trigger_source: str = "cli",
    ) -> str:
        run_key = str(run_id or uuid.uuid4().hex)[:64]
        with get_session() as session:
            exists = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_key))
            if exists is not None:
                raise ValueError(f"find_agent_v2 run_id 已存在: {run_key}")
            rules = rule_config or build_rule_snapshot()
            session.add(FindAgentV2Run(
                run_id=run_key,
                demand_grade_id=demand_grade_id,
                demand_word=str(demand_word)[:256],
                triggered_by_user_id=triggered_by_user_id,
                triggered_by_username=(str(triggered_by_username)[:64] if triggered_by_username else None),
                trigger_source=str(trigger_source or "cli")[:32],
                input_json=_json({"user_input": user_input}),
                rule_config_json=_json(rules),
                status="running",
            ))
        return run_key

    def lookup_run(self, run_id: str) -> dict[str, Any] | None:
        with get_session() as session:
            row = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if row is None:
                return None
            return {
                "run_id": row.run_id,
                "demand_grade_id": row.demand_grade_id,
                "demand_word": row.demand_word,
                "triggered_by_user_id": row.triggered_by_user_id,
                "triggered_by_username": row.triggered_by_username,
                "trigger_source": row.trigger_source,
                "status": row.status,
                "outcome_status": row.outcome_status,
                "current_round": row.current_round,
                "search_count": row.search_count,
                "candidate_count": row.candidate_count,
                "valid_primary_count": row.valid_primary_count,
                "input_tokens": row.input_tokens,
                "output_tokens": row.output_tokens,
                "total_tokens": row.total_tokens,
                "cost_usd": float(row.cost_usd or 0),
                "intent_summary": row.intent_summary,
                "stop_reason": row.stop_reason,
                "obagent_run_uid": row.obagent_run_uid,
                "rule_config": _loads(row.rule_config_json, {}),
                "create_time": row.create_time.isoformat() if row.create_time else None,
                "update_time": row.update_time.isoformat() if row.update_time else None,
            }

    def fail_run(self, run_id: str, reason: str) -> None:
        """Persist failures that happen outside the normal Agent finalization path."""
        with get_session() as session:
            row = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if row is None:
                raise FindAgentV2RunNotFound(run_id)
            row.status = "failed"
            row.outcome_status = "failed"
            row.stop_reason = str(reason)[:2000]

    def set_obagent_run_uid(self, run_id: str, run_uid: str | None) -> None:
        if not run_uid:
            return
        with get_session() as session:
            row = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if row is None:
                raise FindAgentV2RunNotFound(run_id)
            row.obagent_run_uid = str(run_uid)[:64]

    def add_usage(self, run_id: str, usage: dict[str, Any]) -> None:
        """Accumulate one execution attempt; resume never erases earlier usage."""
        with get_session() as session:
            row = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if row is None:
                raise FindAgentV2RunNotFound(run_id)
            row.input_tokens += int(usage.get("input_tokens") or 0)
            row.output_tokens += int(usage.get("output_tokens") or 0)
            row.total_tokens += int(usage.get("total_tokens") or 0)
            row.cost_usd = (
                Decimal(str(row.cost_usd or 0)) + Decimal(str(usage.get("cost") or 0))
            ).quantize(Decimal("0.00000001"))

    def prepare_resume(self, run_id: str) -> dict[str, Any]:
        """Reset a terminal run for a recovery round without deleting audit rows."""
        with get_session() as session:
            row = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if row is None:
                raise FindAgentV2RunNotFound(run_id)
            if row.status == "running":
                raise ValueError(f"run_id={run_id} 已处于 running，不能重复 resume")
            row.status = "running"
            row.outcome_status = None
            row.stop_reason = None
        return self.require_run(run_id)

    def require_run(self, run_id: str) -> dict[str, Any]:
        run = self.lookup_run(run_id)
        if run is None:
            raise FindAgentV2RunNotFound(f"find_agent_v2 run_id 不存在: {run_id}")
        return run

    def get_run_user_input(self, run_id: str) -> str:
        """Return the immutable task input stored when a v2 run was prepared."""
        with get_session() as session:
            row = session.scalar(select(FindAgentV2Run).where(
                FindAgentV2Run.run_id == run_id
            ))
            if row is None:
                raise FindAgentV2RunNotFound(run_id)
            user_input = _loads(row.input_json, {}).get("user_input")
            if not isinstance(user_input, str) or not user_input.strip():
                raise ValueError(f"run_id={run_id} 缺少有效 user_input")
            return user_input

    def get_latest_execution_plan(self, run_id: str) -> ExecutionPlan | None:
        """Restore the newest validated plan for continuation/resume without replanning raw input."""
        with get_session() as session:
            raw = session.scalar(
                select(FindAgentV2Round.plan_json)
                .where(
                    FindAgentV2Round.run_id == run_id,
                    FindAgentV2Round.plan_json.is_not(None),
                )
                .order_by(FindAgentV2Round.round_index.desc(), FindAgentV2Round.id.desc())
                .limit(1)
            )
        payload = _loads(raw, None)
        if not isinstance(payload, dict):
            return None
        try:
            return ExecutionPlan.model_validate(payload)
        except ValueError:
            # Old rounds may contain the pre-contract free-form plan shape.
            return None

    def begin_round(self, run_id: str, round_index: int, snapshot: DiscoverySnapshot) -> None:
        with get_session() as session:
            run = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if run is None:
                raise FindAgentV2RunNotFound(run_id)
            run.current_round = int(round_index)
            session.add(FindAgentV2Round(
                run_id=run_id,
                round_index=int(round_index),
                phase="planning",
                status="open",
                start_snapshot_json=_json(asdict(snapshot)),
            ))

    def update_round(
        self,
        run_id: str,
        round_index: int,
        *,
        phase: str | None = None,
        plan: str | None = None,
        status: str | None = None,
        snapshot: DiscoverySnapshot | None = None,
        error: str | None = None,
    ) -> None:
        with get_session() as session:
            row = session.scalar(select(FindAgentV2Round).where(
                FindAgentV2Round.run_id == run_id,
                FindAgentV2Round.round_index == int(round_index),
            ))
            if row is None:
                raise FindAgentV2RunNotFound(f"round 不存在: {run_id}/{round_index}")
            if phase is not None:
                row.phase = phase
            if plan is not None:
                row.plan_json = plan
            if status is not None:
                row.status = status
            if snapshot is not None:
                row.end_snapshot_json = _json(asdict(snapshot))
            if error is not None:
                row.error_message = error[:2000]

    def save_search(
        self,
        *,
        run_id: str,
        round_index: int,
        keyword: str,
        query_reason: str,
        source_type: str,
        provider: str,
        cursor: str,
        page_no: int,
        payload: dict[str, Any],
    ) -> dict[str, Any]:
        results = list(payload.get("search_results") or [])
        with get_session() as session:
            run = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if run is None:
                raise FindAgentV2RunNotFound(run_id)
            search = FindAgentV2Search(
                run_id=run_id,
                round_index=round_index,
                keyword=keyword[:256],
                query_reason=query_reason,
                source_type=source_type[:32],
                provider=provider[:32],
                cursor=str(cursor)[:128],
                page_no=page_no,
                has_more=int(bool(payload.get("has_more"))),
                next_cursor=str(payload.get("next_cursor") or "")[:128] or None,
                provider_state_json=_json({
                    "search_id": payload.get("search_id"),
                    "backtrace": payload.get("backtrace"),
                }),
                result_count=len(results),
                status="failed" if payload.get("error") else "success",
                error_message=str(payload.get("error") or "") or None,
                raw_response_json=_json(payload),
            )
            session.add(search)
            session.flush()
            new_count = 0
            share_gate_rejected_count = 0
            min_share_count = int(_loads(run.rule_config_json, {}).get("min_share_count", 1000))
            for item in results:
                aweme_id = str(item.get("aweme_id") or "").strip()
                if not aweme_id:
                    continue
                candidate = session.scalar(select(FindAgentV2Candidate).where(
                    FindAgentV2Candidate.run_id == run_id,
                    FindAgentV2Candidate.aweme_id == aweme_id,
                ))
                author = item.get("author") if isinstance(item.get("author"), dict) else {}
                stats = item.get("statistics") if isinstance(item.get("statistics"), dict) else {}
                if candidate is None:
                    candidate = FindAgentV2Candidate(
                        run_id=run_id,
                        first_search_id=search.id,
                        aweme_id=aweme_id,
                        decision_bucket="pending_evaluation",
                    )
                    session.add(candidate)
                    new_count += 1
                keywords = _loads(candidate.source_keywords_json, [])
                if keyword not in keywords:
                    keywords.append(keyword)
                candidate.source_keywords_json = _json(keywords)
                candidate.title = str(item.get("desc") or item.get("title") or candidate.title or "")[:512] or None
                candidate.content_link = str(item.get("url") or candidate.content_link or "")[:1024] or None
                candidate.author_name = str(author.get("nickname") or candidate.author_name or "")[:256] or None
                candidate.author_sec_uid = str(author.get("sec_uid") or candidate.author_sec_uid or "")[:256] or None
                candidate.tags_json = _json(item.get("topics") or item.get("tags") or [])
                duration_ms = item.get("duration_ms")
                if duration_ms:
                    candidate.duration_seconds = Decimal(str(duration_ms)) / 1000
                metric_fields = {
                    "play_count": stats.get("play_count"),
                    "like_count": (
                        stats.get("digg_count")
                        if stats.get("digg_count") is not None
                        else stats.get("like_count")
                    ),
                    "comment_count": stats.get("comment_count"),
                    "collect_count": stats.get("collect_count"),
                    "share_count": stats.get("share_count"),
                }
                for field, value in metric_fields.items():
                    if value is not None:
                        setattr(candidate, field, int(value))

                search_share_count = stats.get("share_count")
                if _fails_search_share_gate(provider, search_share_count, min_share_count):
                    gate = {
                        "stage": "search",
                        "status": "fail",
                        "primary_eligible": False,
                        "failed_reason_codes": ["SHARE_COUNT_TOO_LOW"],
                        "checks": [{
                            "name": "share_count",
                            "status": "fail",
                            "reason_code": "SHARE_COUNT_TOO_LOW",
                            "actual": int(search_share_count),
                            "threshold": min_share_count,
                            "compensated": False,
                        }],
                    }
                    candidate.gate_status = "fail"
                    candidate.gate_result_json = _json(gate)
                    candidate.decision_bucket = "rejected"
                    candidate.decision_reason = (
                        f"搜索结果分享数 {int(search_share_count)} 低于门槛 {min_share_count}"
                    )
                    candidate.reject_reason_code = "SHARE_COUNT_TOO_LOW"
                    share_gate_rejected_count += 1
                elif (
                    search_share_count is not None
                    and candidate.reject_reason_code == "SHARE_COUNT_TOO_LOW"
                ):
                    # A later search response may contain a refreshed metric.
                    candidate.gate_status = None
                    candidate.gate_result_json = None
                    candidate.decision_bucket = "pending_evaluation"
                    candidate.decision_reason = None
                    candidate.reject_reason_code = None
            run.search_count = int(session.scalar(select(func.count()).select_from(FindAgentV2Search).where(FindAgentV2Search.run_id == run_id)) or 0)
            session.flush()
            run.candidate_count = int(session.scalar(select(func.count()).select_from(FindAgentV2Candidate).where(FindAgentV2Candidate.run_id == run_id)) or 0)
            return {
                "search_id": int(search.id),
                "new_candidate_count": new_count,
                "result_count": len(results),
                "share_gate_rejected_count": share_gate_rejected_count,
            }

    def candidate_inputs(self, run_id: str, candidate_ids: list[int]) -> list[dict[str, Any]]:
        with get_session() as session:
            rows = list(session.scalars(select(FindAgentV2Candidate).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.id.in_([int(v) for v in candidate_ids]),
            )))
            return [_candidate_dict(row) for row in rows]

    def candidate_input(self, run_id: str, candidate_id: int) -> dict[str, Any]:
        """Return one candidate only when it belongs to the requested v2 run."""
        items = self.candidate_inputs(run_id, [candidate_id])
        if not items or int(items[0]["candidate_id"]) != int(candidate_id):
            raise ValueError(f"candidate_id 不属于 run: {candidate_id}")
        return items[0]

    def get_search_summaries(self, run_id: str) -> list[dict[str, Any]]:
        """Return compact search execution metadata without loading candidates."""
        self.require_run(run_id)
        with get_session() as session:
            rows = list(session.scalars(select(FindAgentV2Search).where(
                FindAgentV2Search.run_id == run_id,
            ).order_by(FindAgentV2Search.id)))
            return [{
                "search_id": int(row.id),
                "round_index": row.round_index,
                "keyword": row.keyword,
                "query_reason": row.query_reason,
                "provider": row.provider,
                "page_no": row.page_no,
                "has_more": bool(row.has_more),
                "next_cursor": row.next_cursor,
                "status": row.status,
                "result_count": row.result_count,
            } for row in rows]

    def get_candidate_progress(self, run_id: str) -> dict[str, int]:
        """Compute full-run candidate progress with SQL aggregates, never a row limit."""
        self.require_run(run_id)
        pending = FindAgentV2Candidate.decision_bucket == "pending_evaluation"
        detail_pending = FindAgentV2Candidate.detail_status == "pending"
        portrait_pending = FindAgentV2Candidate.portrait_status == "pending"
        with get_session() as session:
            row = session.execute(select(
                func.count(FindAgentV2Candidate.id),
                func.sum(case((pending, 1), else_=0)),
                func.sum(case((FindAgentV2Candidate.decision_bucket == "primary", 1), else_=0)),
                func.sum(case((FindAgentV2Candidate.decision_bucket == "rejected", 1), else_=0)),
                func.sum(case((pending & detail_pending, 1), else_=0)),
                func.sum(case((pending & (FindAgentV2Candidate.detail_status == "success"), 1), else_=0)),
                func.sum(case((pending & (FindAgentV2Candidate.detail_status == "failed"), 1), else_=0)),
                func.sum(case((pending & portrait_pending, 1), else_=0)),
                func.sum(case((pending & (FindAgentV2Candidate.portrait_status == "success"), 1), else_=0)),
                func.sum(case((pending & (FindAgentV2Candidate.portrait_status == "failed"), 1), else_=0)),
                func.sum(case((pending & ~detail_pending & ~portrait_pending, 1), else_=0)),
                func.sum(case((
                    pending
                    & (FindAgentV2Candidate.detail_status == "success")
                    & (FindAgentV2Candidate.portrait_status == "success"),
                    1,
                ), else_=0)),
            ).where(FindAgentV2Candidate.run_id == run_id)).one()
        values = [int(value or 0) for value in row]
        keys = (
            "total_count", "pending_count", "primary_count", "rejected_count",
            "detail_pending_count", "detail_success_count", "detail_failed_count",
            "portrait_pending_count", "portrait_success_count", "portrait_failed_count",
            "evidence_completed_count", "evidence_success_count",
        )
        return dict(zip(keys, values, strict=True))

    def list_pending_evidence_ids(
        self, run_id: str, evidence_type: str, *, limit: int,
    ) -> list[int]:
        """Claimable evidence work projection; callers repeatedly drain this queue."""
        if evidence_type not in {"detail", "portrait"}:
            raise ValueError("evidence_type 只能是 detail/portrait")
        status_column = (
            FindAgentV2Candidate.detail_status
            if evidence_type == "detail"
            else FindAgentV2Candidate.portrait_status
        )
        with get_session() as session:
            return [int(value) for value in session.scalars(select(
                FindAgentV2Candidate.id,
            ).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.decision_bucket == "pending_evaluation",
                status_column == "pending",
            ).order_by(FindAgentV2Candidate.id).limit(max(1, int(limit))))]

    def list_ready_evaluation_ids(self, run_id: str, *, limit: int) -> list[int]:
        """Return pending candidates whose detail and portrait attempts have completed."""
        with get_session() as session:
            return [int(value) for value in session.scalars(select(
                FindAgentV2Candidate.id,
            ).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.decision_bucket == "pending_evaluation",
                FindAgentV2Candidate.detail_status != "pending",
                FindAgentV2Candidate.portrait_status != "pending",
            ).order_by(FindAgentV2Candidate.id).limit(max(1, int(limit))))]

    def count_pending_candidates(self, run_id: str) -> int:
        with get_session() as session:
            return int(session.scalar(select(func.count()).select_from(
                FindAgentV2Candidate,
            ).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.decision_bucket == "pending_evaluation",
            )) or 0)

    def get_report_state(self, run_id: str) -> dict[str, Any]:
        """Report-only projection: aggregate summary, primaries and rejection distribution."""
        run = self.require_run(run_id)
        progress = self.get_candidate_progress(run_id)
        with get_session() as session:
            primaries = list(session.scalars(select(FindAgentV2Candidate).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.decision_bucket == "primary",
            ).order_by(
                FindAgentV2Candidate.value_score.desc(),
                FindAgentV2Candidate.id,
            )))
            rejection_rows = session.execute(select(
                FindAgentV2Candidate.reject_reason_code,
                func.count(FindAgentV2Candidate.id),
            ).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.decision_bucket == "rejected",
            ).group_by(FindAgentV2Candidate.reject_reason_code)).all()
            primary_candidates = [_candidate_dict(row) for row in primaries]
        return {
            "run": run,
            "summary": progress,
            "primary_candidates": primary_candidates,
            "rejection_reason_distribution": {
                str(reason or "UNSPECIFIED"): int(count)
                for reason, count in rejection_rows
            },
        }

    def save_details(
        self,
        run_id: str,
        details: list[dict[str, Any]],
        errors: list[dict[str, Any]],
        *,
        requested_aweme_ids: list[str] | None = None,
    ) -> None:
        by_id = {str(item.get("content_id") or ""): item for item in details}
        error_by_id = {str(item.get("content_id") or ""): item for item in errors}
        requested = {str(value) for value in (requested_aweme_ids or []) if str(value)}
        target_ids = set(by_id) | set(error_by_id) | requested
        with get_session() as session:
            rows = list(session.scalars(select(FindAgentV2Candidate).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.aweme_id.in_(target_ids),
            )))
            for row in rows:
                detail = by_id.get(row.aweme_id)
                error = error_by_id.get(row.aweme_id)
                if detail:
                    row.detail_status = "success"
                    row.detail_json = _json(detail)
                    row.title = str(detail.get("title") or detail.get("body_text") or row.title or "")[:512] or None
                    row.content_link = str(detail.get("content_link") or row.content_link or "")[:1024] or None
                    row.author_name = str(detail.get("channel_account_name") or row.author_name or "")[:256] or None
                    row.author_sec_uid = str(detail.get("channel_account_id") or row.author_sec_uid or "")[:256] or None
                    row.tags_json = _json(detail.get("topic_list") or [])
                    parsed = parse_datetime_value(detail.get("publish_at"))
                    row.publish_at = parsed.replace(tzinfo=None) if parsed else None
                    row.duration_seconds = detail.get("duration_seconds") or None
                    for key in ("play_count", "like_count", "comment_count", "collect_count", "share_count"):
                        value = detail.get(key)
                        if value is not None:
                            setattr(row, key, value)
                    status, raw = "success", detail
                else:
                    error = error or {
                        "error": "上游详情响应未包含已请求候选",
                        "error_code": "UPSTREAM_RESULT_MISSING",
                    }
                    row.detail_status = "failed"
                    status, raw = "failed", error
                session.add(FindAgentV2Evidence(
                    run_id=run_id,
                    candidate_id=row.id,
                    evidence_type="detail",
                    provider="crawler",
                    status=status,
                    raw_json=_json(raw),
                    error_message=str((error or {}).get("error") or "") or None,
                ))

    def save_portraits(
        self,
        run_id: str,
        results: list[dict[str, Any]],
        *,
        requested_aweme_ids: list[str] | None = None,
    ) -> None:
        by_id = {str(item.get("aweme_id") or ""): item for item in results}
        requested = {str(value) for value in (requested_aweme_ids or []) if str(value)}
        target_ids = set(by_id) | requested
        with get_session() as session:
            rows = list(session.scalars(select(FindAgentV2Candidate).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.aweme_id.in_(target_ids),
            )))
            for row in rows:
                item = by_id.get(row.aweme_id) or {
                    "aweme_id": row.aweme_id,
                    "error": "上游画像响应未包含已请求候选",
                    "error_code": "UPSTREAM_RESULT_MISSING",
                }
                normalization = item.get("age_normalization") or {}
                content = normalization.get("content") or {}
                account = normalization.get("account") or {}
                row.portrait_json = _json(item)
                row.portrait_status = "failed" if item.get("error") else "success"
                row.content_50_plus_ratio = _ratio(
                    content.get("older_ratio") if content.get("has_age_portrait") else None
                )
                row.account_50_plus_ratio = _ratio(
                    account.get("older_ratio") if account.get("has_age_portrait") else None
                )
                session.add(FindAgentV2Evidence(
                    run_id=run_id,
                    candidate_id=row.id,
                    evidence_type="portrait",
                    provider="douhot",
                    status=row.portrait_status,
                    raw_json=_json(item),
                    normalized_json=_json(normalization),
                    error_message=str(item.get("error") or "") or None,
                ))

    def evaluate(self, run_id: str, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
        run_data = self.require_run(run_id)
        output: list[dict[str, Any]] = []
        with get_session() as session:
            run = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            assert run is not None
            for item in items:
                row = session.scalar(select(FindAgentV2Candidate).where(
                    FindAgentV2Candidate.run_id == run_id,
                    FindAgentV2Candidate.id == int(item.get("candidate_id") or 0),
                ))
                if row is None:
                    raise ValueError(f"candidate_id 不属于 run: {item.get('candidate_id')}")
                for key in ("relevance_score", "elder_score", "share_score", "value_score"):
                    setattr(row, key, _ratio(item.get(key)))
                requested = str(item.get("decision_bucket") or "rejected")
                if requested not in {"primary", "rejected"}:
                    raise ValueError("decision_bucket 只能是 primary/rejected")
                gate_input = _candidate_dict(row)
                gate = evaluate_candidate_gate(gate_input, run_data["rule_config"])
                row.gate_status = gate["status"]
                row.gate_result_json = _json(gate)
                row.decision_bucket = "primary" if requested == "primary" and gate["status"] == "pass" else "rejected"
                row.decision_reason = str(item.get("decision_reason") or "")
                failed = list(gate.get("failed_reason_codes") or [])
                row.reject_reason_code = (str(item.get("reject_reason_code") or "") or (failed[0] if failed else None))
                output.append({"candidate_id": int(row.id), "decision_bucket": row.decision_bucket, "gate_status": row.gate_status})
            session.flush()
            primaries = list(session.scalars(select(FindAgentV2Candidate).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.decision_bucket == "primary",
            )))
            run.valid_primary_count = len({row.aweme_id for row in primaries if row.gate_status == "pass"})
        return output

    def reject_failed_gates(
        self, run_id: str, failures: list[tuple[int, dict[str, Any]]],
    ) -> list[int]:
        """Reject hard-gate failures before any evaluator model invocation."""
        rejected: list[int] = []
        if not failures:
            return rejected
        by_id = {int(candidate_id): gate for candidate_id, gate in failures}
        with get_session() as session:
            rows = list(session.scalars(select(FindAgentV2Candidate).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.id.in_(list(by_id)),
                FindAgentV2Candidate.decision_bucket == "pending_evaluation",
            )))
            for row in rows:
                gate = by_id[int(row.id)]
                failed = list(gate.get("failed_reason_codes") or [])
                row.gate_status = "fail"
                row.gate_result_json = _json(gate)
                row.decision_bucket = "rejected"
                row.decision_reason = "硬门禁未通过：" + ", ".join(failed)
                row.reject_reason_code = failed[0] if failed else "HARD_GATE_FAILED"
                rejected.append(int(row.id))
        return rejected

    def recount_valid_primary(self, run_id: str) -> int:
        """Recompute the denormalized primary counter after parallel worker writes."""
        with get_session() as session:
            run = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if run is None:
                raise FindAgentV2RunNotFound(run_id)
            count = int(session.scalar(select(func.count(func.distinct(
                FindAgentV2Candidate.aweme_id,
            ))).where(
                FindAgentV2Candidate.run_id == run_id,
                FindAgentV2Candidate.decision_bucket == "primary",
                FindAgentV2Candidate.gate_status == "pass",
            )) or 0)
            run.valid_primary_count = count
            return count

    def get_full_state(
        self, run_id: str, *, limit: int = 100, pending_only: bool = False,
    ) -> dict[str, Any]:
        run = self.require_run(run_id)
        with get_session() as session:
            searches = list(session.scalars(select(FindAgentV2Search).where(
                FindAgentV2Search.run_id == run_id,
            ).order_by(FindAgentV2Search.id)))
            candidate_query = select(FindAgentV2Candidate).where(
                FindAgentV2Candidate.run_id == run_id,
            )
            if pending_only:
                candidate_query = candidate_query.where(
                    FindAgentV2Candidate.decision_bucket == "pending_evaluation",
                )
            candidates = list(session.scalars(candidate_query.order_by(
                FindAgentV2Candidate.value_score.desc(), FindAgentV2Candidate.id,
            ).limit(max(1, min(limit, 500)))))
            return {
                "run": run,
                "searches": [{
                    "search_id": int(row.id), "round_index": row.round_index,
                    "keyword": row.keyword, "query_reason": row.query_reason,
                    "provider": row.provider, "page_no": row.page_no,
                    "has_more": bool(row.has_more), "next_cursor": row.next_cursor,
                    "status": row.status, "result_count": row.result_count,
                } for row in searches],
                "candidates": [_candidate_dict(row) for row in candidates],
            }

    def snapshot(self, run_id: str) -> DiscoverySnapshot:
        run = self.require_run(run_id)
        progress = self.get_candidate_progress(run_id)
        return DiscoverySnapshot(
            status=run["status"],
            search_count=run["search_count"],
            candidate_count=progress["total_count"],
            pending_count=progress["pending_count"],
            primary_count=progress["primary_count"],
            valid_primary_count=run["valid_primary_count"],
            rejected_count=progress["rejected_count"],
            outcome_status=run.get("outcome_status") or "",
        )

    def finalize(
        self,
        run_id: str,
        *,
        failed: bool = False,
        reason: str = "",
    ) -> dict[str, Any]:
        snapshot = self.snapshot(run_id)
        if failed:
            outcome, status = "failed", "failed"
        elif snapshot.valid_primary_count > 0:
            outcome, status = "goal_met", "finished"
        else:
            outcome, status = "no_match", "finished"
        with get_session() as session:
            run = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
            if run is None:
                raise FindAgentV2RunNotFound(run_id)
            if not (run.status == "failed" and run.outcome_status == "failed"):
                run.status = status
                run.outcome_status = outcome
                run.stop_reason = reason[:2000] or None
        return self.require_run(run_id)


_SERVICE = FindAgentV2Service()


def get_find_agent_v2_service() -> FindAgentV2Service:
    return _SERVICE