优化数据存储

agents/find_agent/PRD.md

@@ -213,7 +213,9 @@ TikHub 翻页需要原样复用上一页的 `next_cursor`、`search_id`、`backt
 - 播放、点赞、评论、收藏和分享数据；
 - 发布时间和视频时长。
 
-详情工具返回证据，不直接修改候选数据库。模型需要再次调用候选评估保存工具才能写入。
+详情工具的新流程使用 `run_id + candidate_ids`：完整上游响应先写入
+`video_discovery_evidence`，再由程序抽取详情字段、更新同一运行内的候选并重算门禁。
+旧 `content_ids` 调用保留兼容，但不会自动关联候选。
 
 ### 7.2 年龄画像
 
@@ -227,6 +229,9 @@ TikHub 翻页需要原样复用上一页的 `next_cursor`、`search_id`、`backt
 任一上游画像请求可能单独失败；保存时未获取侧保持 `missing`，只要视频侧或作者侧任一侧
 达到对应 50+ 门槛，画像门禁即通过。
 
+批量画像的新流程使用 `run_id + candidate_ids`，内容侧和账号侧原始响应分别追加保存；
+标准化和候选更新由工具完成，不再要求模型搬运画像 JSON。
+
 单条画像工具只返回原始画像，需要额外调用 `normalize_age_portraits`。
 
 ### 7.3 年龄标准化

agents/find_agent/prompt/system_prompt.md

@@ -239,11 +239,11 @@ P0 门槛是 `primary` 的程序准入底线。判断时**必须严格参考**
   相关性、老年倾向和分享价值，不能因作者优秀就直接推荐。工具会自动保存作者搜索页
   和候选。
 - `douyin_detail`：用于核验候选的真实发布时间、时长、最新分享数与其他互动数据，
-  以及作者、页面链接和标题/描述等文本证据。最终候选更新必须把
-  `publish_at / duration_seconds / share_count` 一并写回。
+  以及作者、页面链接和标题/描述等文本证据。使用 `run_id + candidate_ids` 调用；工具会
+  保存完整原始响应、直接更新候选结构化字段并重算门禁，无需把详情数据再次传给更新工具。
 - `batch_fetch_portraits`：用于批量取得视频点赞画像；对正式候选应设置
-  `fetch_account_portrait=true`，同时取得作者粉丝画像。批量结果会自动附带
-  `age_normalization`，无需为同一候选再单独调用标准化工具。上游可能只返回视频侧或只返回
+  `fetch_account_portrait=true`，同时取得作者粉丝画像。使用 `run_id + candidate_ids` 调用；
+  工具会保存双侧原始响应、标准化年龄画像、更新候选并重算门禁。上游可能只返回视频侧或只返回
   作者侧画像；必须保留已成功的一侧，另一侧记为 `missing`，不得因为未同时获得双侧画像而
   拒绝候选。任一已获取侧达到对应 50+ 门槛即可通过画像门禁。
 - `get_content_fans_portrait` / `get_account_fans_portrait`：用于补充或复核单条画像。
@@ -253,11 +253,9 @@ P0 门槛是 `primary` 的程序准入底线。判断时**必须严格参考**
   不要重复调用。标准化结果中的 `elder_score_cap` 是当前证据条件下 `E` 的上限，
   `elder_score` 不得超过该值。
 - `batch_update_video_discovery_candidates`：严格按搜索结果返回的 `candidate_id`
-  更新详情、证据、**0~1 的 R/E/S/V 评分**和 `decision_bucket`。更新正式候选时必须
-  提供最新的 `publish_at / duration_seconds / share_count`、已经获取到的画像及
-  `age_normalization`。不要求双侧同时存在；工具会从标准化结果拆出视频侧和账号侧 50+
-  指标，并在保存
-  `primary` 前执行程序硬门槛；未通过的 `primary` 会自动保存为 `rejected` 并返回原因码，
+  更新 **0~1 的 R/E/S/V 评分**、`decision_reason` 和 `decision_bucket`；不要重复提交
+  详情、互动量或画像。工具在保存 `primary` 前读取数据库证据执行程序硬门槛；明确未通过的
+  `primary` 会自动保存为 `rejected`，证据尚未就绪的会保留为 `pending_evaluation`，
   同一批中其他合格候选仍会成功保存。工具不会
   新增候选、修改搜索记录或修改运行状态。`decision_reason` 必须分别覆盖相关性、视频
   侧与作者侧 50+（说明采纳哪一侧或双侧均达标）、分享价值、时间有效性和主要限制。

agents/find_agent/support/douyin_detail.py

@@ -15,6 +15,11 @@ from typing import Any, Optional
 import httpx
 
 from supply_infra.video_discovery_gates import parse_datetime_value
+from supply_infra.services.video_discovery_service import (
+    RunNotFoundError,
+    build_evidence_values,
+    get_video_discovery_service,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -237,7 +242,7 @@ async def _wait_rate_limit() -> None:
 async def _fetch_one_detail(
     client: httpx.AsyncClient,
     content_id: str,
-) -> dict[str, Any]:
+) -> tuple[dict[str, Any], dict[str, Any]]:
     """拉取单条详情。成功返回 detail 字典；失败抛出 Exception。"""
     await _wait_rate_limit()
     response = await client.post(
@@ -256,12 +261,40 @@ async def _fetch_one_detail(
     if not detail_raw:
         raise RuntimeError(f"未查到视频详情: content_id={content_id}")
 
-    return _build_detail_result(detail_raw, content_id)
+    return _build_detail_result(detail_raw, content_id), body
+
+
+def _detail_candidate_values(detail: dict[str, Any]) -> dict[str, Any]:
+    topics = detail.get("topic_list")
+    tags = topics if isinstance(topics, list) else []
+    publish_at = parse_datetime_value(detail.get("publish_at"))
+    return {
+        "title": detail.get("title") or detail.get("body_text"),
+        "content_link": detail.get("content_link"),
+        "author_name": detail.get("channel_account_name"),
+        "author_sec_uid": detail.get("channel_account_id"),
+        "tags_json": json.dumps(tags, ensure_ascii=False) if tags else None,
+        "publish_at": (
+            publish_at.replace(tzinfo=None) if publish_at is not None else None
+        ),
+        "duration_seconds": detail.get("duration_seconds"),
+        "play_count": (
+            detail.get("play_count")
+            if detail.get("play_count") is not None
+            else detail.get("view_count")
+        ),
+        "like_count": detail.get("like_count"),
+        "comment_count": detail.get("comment_count"),
+        "collect_count": detail.get("collect_count"),
+        "share_count": detail.get("share_count"),
+    }
 
 
 async def douyin_detail(
-    content_ids: list[str],
+    content_ids: list[str] | None = None,
     timeout: Optional[float] = None,
+    run_id: str | None = None,
+    candidate_ids: list[int] | None = None,
 ) -> str:
     """
     抖音视频详情（支持批量）
@@ -270,9 +303,10 @@ async def douyin_detail(
     用于在 douyin_search 选中目标视频后，再拉取可播放的 video_url。
 
     Args:
-        content_ids: 视频 ID 列表，对应搜索结果中的 aweme_id。
-            单个传 ["123"]，多个传 ["123", "456"]
+        content_ids: 兼容旧调用的视频 ID 列表；新流程不要使用。
         timeout: 单次请求超时时间（秒），默认 60
+        run_id: 视频发现运行 id；与 candidate_ids 一起使用时自动落库并重算门禁。
+        candidate_ids: 搜索工具返回的数据库候选 id。工具会从数据库读取 aweme_id。
 
     Returns:
         JSON 字符串，包含：
@@ -283,7 +317,25 @@ async def douyin_detail(
     """
     start_time = time.time()
     request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
-    ids = _normalize_content_ids(content_ids)
+    candidate_by_aweme: dict[str, int] = {}
+    if candidate_ids:
+        if not run_id:
+            return _error_result("candidate_ids 模式必须提供 run_id", input_error=True)
+        try:
+            candidates = await asyncio.to_thread(
+                get_video_discovery_service().get_candidates_by_ids,
+                str(run_id),
+                [int(value) for value in candidate_ids],
+            )
+        except (RunNotFoundError, ValueError) as exc:
+            return _error_result(str(exc), input_error=True)
+        for candidate in candidates:
+            aweme_id = str(candidate.get("aweme_id") or "")
+            if aweme_id:
+                candidate_by_aweme.setdefault(aweme_id, int(candidate["candidate_id"]))
+        ids = _normalize_content_ids(list(candidate_by_aweme))
+    else:
+        ids = _normalize_content_ids(content_ids or [])
 
     if not ids:
         return _error_result("content_ids 不能为空")
@@ -304,8 +356,29 @@ async def douyin_detail(
         ) as client:
             for content_id in ids:
                 try:
-                    detail = await _fetch_one_detail(client, content_id)
+                    detail, raw_response = await _fetch_one_detail(client, content_id)
                     details.append(detail)
+                    if run_id and content_id in candidate_by_aweme:
+                        normalized = _detail_candidate_values(detail)
+                        evidence_values = build_evidence_values(
+                            run_id=str(run_id),
+                            evidence_type="detail",
+                            provider="crawler",
+                            subject_key=content_id,
+                            request={"content_id": content_id},
+                            raw_response=raw_response,
+                            normalized=normalized,
+                            fetch_status="success",
+                            trigger_candidate_id=candidate_by_aweme[content_id],
+                            aweme_id=content_id,
+                        )
+                        await asyncio.to_thread(
+                            get_video_discovery_service().record_candidate_evidence,
+                            run_id=str(run_id),
+                            evidence_values=evidence_values,
+                            aweme_id=content_id,
+                            normalized=normalized,
+                        )
                 except httpx.HTTPStatusError as e:
                     msg = f"HTTP {e.response.status_code}: {e.response.text}"
                     logger.error("douyin_detail HTTP error: content_id=%s status=%d", content_id, e.response.status_code)
@@ -323,6 +396,28 @@ async def douyin_detail(
                     logger.warning("douyin_detail item failed: content_id=%s error=%s", content_id, e)
                     errors.append({"content_id": content_id, "error": msg})
 
+                if errors and errors[-1].get("content_id") == content_id and run_id and content_id in candidate_by_aweme:
+                    error_message = str(errors[-1].get("error") or "详情获取失败")
+                    evidence_values = build_evidence_values(
+                        run_id=str(run_id),
+                        evidence_type="detail",
+                        provider="crawler",
+                        subject_key=content_id,
+                        request={"content_id": content_id},
+                        raw_response={"error": error_message},
+                        fetch_status="failed",
+                        trigger_candidate_id=candidate_by_aweme[content_id],
+                        aweme_id=content_id,
+                        error_message=error_message,
+                    )
+                    await asyncio.to_thread(
+                        get_video_discovery_service().record_candidate_evidence,
+                        run_id=str(run_id),
+                        evidence_values=evidence_values,
+                        aweme_id=content_id,
+                        normalized={},
+                    )
+
         duration_ms = int((time.time() - start_time) * 1000)
         logger.info(
             "douyin_detail completed: requested=%d success=%d failed=%d duration_ms=%d",
@@ -332,7 +427,7 @@ async def douyin_detail(
             duration_ms,
         )
 
-        if not details and errors:
+        if not details and errors and not (run_id and candidate_ids):
             return _error_result(
                 f"全部失败（{len(errors)} 条）: {errors[0].get('error', '')}"
             )
@@ -347,8 +442,34 @@ async def douyin_detail(
             "errors": errors,
             "duration_ms": duration_ms,
         }
+        if run_id and candidate_ids:
+            refreshed = await asyncio.to_thread(
+                get_video_discovery_service().get_candidates_by_ids,
+                str(run_id),
+                [int(value) for value in candidate_ids],
+            )
+            payload["candidates"] = [
+                {
+                    "candidate_id": item["candidate_id"],
+                    "aweme_id": item["aweme_id"],
+                    "gate_status": item["gate_status"],
+                    "failed_reason_codes": (
+                        item.get("gate_results", {}).get("failed_reason_codes") or []
+                    ),
+                    "evidence_version": item.get("evidence_version"),
+                }
+                for item in refreshed
+            ]
+            payload.pop("details", None)
+            payload["errors"] = [
+                {
+                    "content_id": item.get("content_id"),
+                    "error": str(item.get("error") or "")[:500],
+                }
+                for item in errors
+            ]
         # 单条时额外提供 detail，方便旧逻辑取值
-        if len(details) == 1:
+        if len(details) == 1 and not (run_id and candidate_ids):
             payload["detail"] = details[0]
         return json.dumps(payload, ensure_ascii=False)
 

agents/find_agent/support/douyin_search.py

@@ -142,13 +142,19 @@ def _success_result(
         "next_cursor": cursor_value,
         "search_results": search_results,
         "duration_ms": duration_ms,
+        "_raw_response": data,
     }
     return json.dumps(payload, ensure_ascii=False)
 
 
-def _error_result(error: str, *, title: str = "抖音搜索失败") -> str:
+def _error_result(
+    error: str, *, title: str = "抖音搜索失败", raw_response: Any = None
+) -> str:
     """构建失败时的 JSON 字符串返回值。"""
-    return json.dumps({"error": error, "title": title}, ensure_ascii=False)
+    payload = {"error": error, "title": title}
+    if raw_response is not None:
+        payload["_raw_response"] = raw_response
+    return json.dumps(payload, ensure_ascii=False)
 
 
 async def _douyin_search_raw(
@@ -254,7 +260,10 @@ async def _douyin_search_raw(
             keyword,
             e.response.status_code,
         )
-        return _error_result(f"HTTP {e.response.status_code}: {e.response.text}")
+        return _error_result(
+            f"HTTP {e.response.status_code}: {e.response.text}",
+            raw_response=e.response.text,
+        )
     except httpx.TimeoutException:
         logger.error("douyin_search timeout: keyword=%s timeout=%s", keyword, request_timeout)
         return _error_result(f"请求超时（{request_timeout}秒）")

agents/find_agent/support/douyin_search_tikhub.py

@@ -217,11 +217,11 @@ def _summary(
     return "\n".join(lines).rstrip()
 
 
-def _error_result(error: str) -> str:
-    return json.dumps(
-        {"error": error, "title": "TikHub 抖音搜索失败"},
-        ensure_ascii=False,
-    )
+def _error_result(error: str, *, raw_response: Any = None) -> str:
+    payload = {"error": error, "title": "TikHub 抖音搜索失败"}
+    if raw_response is not None:
+        payload["_raw_response"] = raw_response
+    return json.dumps(payload, ensure_ascii=False)
 
 
 async def _wait_rate_limit() -> None:
@@ -382,6 +382,7 @@ async def _douyin_search_tikhub_raw(
             "backtrace": next_backtrace,
             "search_results": results,
             "duration_ms": duration_ms,
+            "_raw_response": body,
         }
         logger.info(
             "douyin_search_tikhub completed: keyword=%s results=%d has_more=%s duration_ms=%d",
@@ -398,7 +399,10 @@ async def _douyin_search_tikhub_raw(
             keyword_text,
             exc.response.status_code,
         )
-        return _error_result(f"HTTP {exc.response.status_code}: {text}")
+        return _error_result(
+            f"HTTP {exc.response.status_code}: {text}",
+            raw_response=exc.response.text,
+        )
     except httpx.TimeoutException:
         return _error_result(f"请求超时（{request_timeout}秒）")
     except httpx.RequestError as exc:

agents/find_agent/support/douyin_user_videos.py

@@ -128,11 +128,11 @@ def _summary(
     return "\n".join(lines).rstrip()
 
 
-def _error_result(error: str) -> str:
-    return json.dumps(
-        {"error": error, "title": "抖音作者作品获取失败"},
-        ensure_ascii=False,
-    )
+def _error_result(error: str, *, raw_response: Any = None) -> str:
+    payload = {"error": error, "title": "抖音作者作品获取失败"}
+    if raw_response is not None:
+        payload["_raw_response"] = raw_response
+    return json.dumps(payload, ensure_ascii=False)
 
 
 async def _wait_rate_limit() -> None:
@@ -241,6 +241,7 @@ async def _douyin_user_videos_raw(
             "user_videos": results,
             "search_results": results,
             "duration_ms": duration_ms,
+            "_raw_response": body,
         }
         logger.info(
             "douyin_user_videos completed: account_id=%s results=%d has_more=%s duration_ms=%d",
@@ -257,7 +258,10 @@ async def _douyin_user_videos_raw(
             account_text,
             exc.response.status_code,
         )
-        return _error_result(f"HTTP {exc.response.status_code}: {text}")
+        return _error_result(
+            f"HTTP {exc.response.status_code}: {text}",
+            raw_response=exc.response.text,
+        )
     except httpx.TimeoutException:
         return _error_result(f"请求超时（{request_timeout}秒）")
     except httpx.RequestError as exc:

agents/find_agent/support/portrait.py

@@ -15,6 +15,11 @@ from typing import Any, Optional
 import httpx
 
 from agents.find_agent.support.age_portrait import normalize_age_portrait_pair
+from supply_infra.services.video_discovery_service import (
+    RunNotFoundError,
+    build_evidence_values,
+    get_video_discovery_service,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -186,6 +191,39 @@ def _error_result(
     )
 
 
+def _portrait_candidate_values(
+    *,
+    content_portrait: dict[str, Any],
+    account_portrait: dict[str, Any],
+    age_normalization: dict[str, Any],
+) -> dict[str, Any]:
+    content = age_normalization.get("content") or {}
+    account = age_normalization.get("account") or {}
+    return {
+        "content_age_evidence_json": json.dumps(
+            content_portrait, ensure_ascii=False
+        ),
+        "account_age_evidence_json": json.dumps(
+            account_portrait, ensure_ascii=False
+        ),
+        "age_normalization_json": json.dumps(
+            age_normalization, ensure_ascii=False
+        ),
+        "content_50_plus_ratio": (
+            content.get("older_ratio") if content.get("has_age_portrait") else None
+        ),
+        "content_50_plus_tgi": (
+            content.get("older_tgi") if content.get("has_age_portrait") else None
+        ),
+        "account_50_plus_ratio": (
+            account.get("older_ratio") if account.get("has_age_portrait") else None
+        ),
+        "account_50_plus_tgi": (
+            account.get("older_tgi") if account.get("has_age_portrait") else None
+        ),
+    }
+
+
 async def get_account_fans_portrait(
     account_id: str,
     need_province: bool = False,
@@ -375,7 +413,7 @@ async def get_content_fans_portrait(
 
 
 async def batch_fetch_portraits(
-    candidates_json: str,
+    candidates_json: str = "",
     fetch_account_portrait: bool = False,
     need_province: bool = False,
     need_city: bool = False,
@@ -385,6 +423,8 @@ async def batch_fetch_portraits(
     need_phone_brand: bool = False,
     need_phone_price: bool = False,
     timeout: Optional[float] = None,
+    run_id: str | None = None,
+    candidate_ids: list[int] | None = None,
 ) -> str:
     """
     批量获取多条候选视频的画像
@@ -394,13 +434,15 @@ async def batch_fetch_portraits(
     一次调用返回所有条目，便于比较同一候选的两侧年龄证据。
 
     Args:
-        candidates_json: JSON 数组字符串。每项为对象，字段：
+        candidates_json: 兼容旧调用的 JSON 数组字符串。每项为对象，字段：
             - aweme_id (必填): 视频 id
             - author_sec_uid (可选): 作者 sec_uid，作者画像或兜底时需要
             - try_account_fallback (可选，默认 true): 为 false 时不请求账号画像
         fetch_account_portrait: 是否为每个候选同时获取作者粉丝画像，默认 False。
             老年受众判断建议设为 True；缺少 author_sec_uid 的条目会跳过作者画像。
         need_* / timeout: 与各单条画像工具一致
+        run_id / candidate_ids: 新流程使用数据库候选 id；工具自动读取视频和作者 id、
+            保存原始响应及标准化画像并重算门禁。
 
     Returns:
         JSON 字符串，包含 output（人类可读摘要）和 results（结构化列表）。
@@ -408,7 +450,38 @@ async def batch_fetch_portraits(
     """
     start_time = time.time()
     request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
-    raw = (candidates_json or "").strip()
+    candidate_id_by_aweme: dict[str, int] = {}
+    if candidate_ids:
+        if not run_id:
+            return _error_result(
+                "candidate_ids 模式必须提供 run_id",
+                title="批量画像失败",
+                input_error=True,
+            )
+        try:
+            db_candidates = await asyncio.to_thread(
+                get_video_discovery_service().get_candidates_by_ids,
+                str(run_id),
+                [int(value) for value in candidate_ids],
+            )
+        except (RunNotFoundError, ValueError) as exc:
+            return _error_result(str(exc), title="批量画像失败", input_error=True)
+        parsed_from_db: list[dict[str, Any]] = []
+        for candidate in db_candidates:
+            aweme_id = str(candidate.get("aweme_id") or "")
+            if not aweme_id or aweme_id in candidate_id_by_aweme:
+                continue
+            candidate_id_by_aweme[aweme_id] = int(candidate["candidate_id"])
+            parsed_from_db.append(
+                {
+                    "aweme_id": aweme_id,
+                    "author_sec_uid": candidate.get("author_sec_uid"),
+                    "try_account_fallback": True,
+                }
+            )
+        raw = json.dumps(parsed_from_db, ensure_ascii=False)
+    else:
+        raw = (candidates_json or "").strip()
 
     if not raw:
         return _error_result(
@@ -523,6 +596,7 @@ async def batch_fetch_portraits(
                 need_account = fetch_account_portrait or (
                     bool(try_fallback) and not content_has
                 )
+                aok: dict[str, Any] | None = None
 
                 if need_account:
                     if not author_sec or not isinstance(author_sec, str):
@@ -573,6 +647,84 @@ async def batch_fetch_portraits(
                     content_block.get("portrait_data"),
                     account_block.get("portrait_data"),
                 )
+                if run_id and aweme_id in candidate_id_by_aweme:
+                    candidate_id = candidate_id_by_aweme[aweme_id]
+                    normalized = _portrait_candidate_values(
+                        content_portrait=content_block.get("portrait_data") or {},
+                        account_portrait=account_block.get("portrait_data") or {},
+                        age_normalization=item_result["age_normalization"],
+                    )
+                    content_status = (
+                        "failed"
+                        if content_block.get("error")
+                        else (
+                            "success"
+                            if content_block.get("has_portrait")
+                            else "unavailable"
+                        )
+                    )
+                    content_evidence = build_evidence_values(
+                        run_id=str(run_id),
+                        evidence_type="content_portrait",
+                        provider="douhot",
+                        subject_key=aweme_id,
+                        request={"content_id": aweme_id, **flags},
+                        raw_response=(
+                            cok.get("raw_data")
+                            if cok is not None
+                            else {"error": content_block.get("error")}
+                        ),
+                        normalized=normalized,
+                        fetch_status=content_status,
+                        trigger_candidate_id=candidate_id,
+                        aweme_id=aweme_id,
+                        error_message=content_block.get("error"),
+                    )
+                    await asyncio.to_thread(
+                        get_video_discovery_service().record_candidate_evidence,
+                        run_id=str(run_id),
+                        evidence_values=content_evidence,
+                        aweme_id=aweme_id,
+                        normalized=normalized,
+                    )
+
+                    if account_block.get("attempted"):
+                        account_status = (
+                            "failed"
+                            if account_block.get("error")
+                            else (
+                                "success"
+                                if account_block.get("has_portrait")
+                                else "unavailable"
+                            )
+                        )
+                        account_evidence = build_evidence_values(
+                            run_id=str(run_id),
+                            evidence_type="account_portrait",
+                            provider="douhot",
+                            subject_key=str(author_sec or aweme_id),
+                            request={"account_id": author_sec, **flags},
+                            raw_response=(
+                                aok.get("raw_data")
+                                if aok is not None
+                                else {"error": account_block.get("error")}
+                            ),
+                            normalized=normalized,
+                            fetch_status=account_status,
+                            trigger_candidate_id=candidate_id,
+                            aweme_id=aweme_id,
+                            author_sec_uid=(
+                                str(author_sec) if author_sec is not None else None
+                            ),
+                            error_message=account_block.get("error"),
+                        )
+                        await asyncio.to_thread(
+                            get_video_discovery_service().record_candidate_evidence,
+                            run_id=str(run_id),
+                            evidence_values=account_evidence,
+                            aweme_id=aweme_id,
+                            normalized=normalized,
+                        )
                 results.append(item_result)
                 c_part = item_result["content"] or {}
                 a_part = item_result["account"] or {}
@@ -590,15 +742,33 @@ async def batch_fetch_portraits(
             len(parsed),
             duration_ms,
         )
-        return _success_result(
-            {
+        payload = {
                 "title": f"批量画像完成 ({len(results)} 条)",
                 "output": "\n".join(output_chunks),
                 "results": results,
                 "count": len(results),
                 "duration_ms": duration_ms,
             }
-        )
+        if run_id and candidate_ids:
+            refreshed = await asyncio.to_thread(
+                get_video_discovery_service().get_candidates_by_ids,
+                str(run_id),
+                [int(value) for value in candidate_ids],
+            )
+            payload["candidates"] = [
+                {
+                    "candidate_id": item["candidate_id"],
+                    "aweme_id": item["aweme_id"],
+                    "gate_status": item["gate_status"],
+                    "failed_reason_codes": (
+                        item.get("gate_results", {}).get("failed_reason_codes") or []
+                    ),
+                    "evidence_version": item.get("evidence_version"),
+                }
+                for item in refreshed
+            ]
+            payload.pop("results", None)
+        return _success_result(payload)
 
     except Exception as e:
         logger.error("batch_fetch_portraits unexpected error: error=%s", e, exc_info=True)

agents/find_agent/support/search_persistence.py

@@ -12,6 +12,7 @@ from supply_infra.video_discovery_gates import (
 )
 from supply_infra.services.video_discovery_service import (
     RunNotFoundError,
+    build_evidence_values,
     format_db_error,
     get_video_discovery_service,
 )
@@ -58,6 +59,13 @@ def _nonnegative_int(value: Any) -> int | None:
         return None
 
 
+def _first_present(*values: Any) -> Any:
+    for value in values:
+        if value is not None and value != "":
+            return value
+    return None
+
+
 def _search_key(values: dict[str, Any]) -> str:
     identity = {
         key: values.get(key)
@@ -120,19 +128,19 @@ def _candidate_from_search_result(
             max_length=256,
         ),
         "like_count": _nonnegative_int(
-            stats.get("digg_count") or item.get("like_count")
+            _first_present(stats.get("digg_count"), item.get("like_count"))
         ),
         "comment_count": _nonnegative_int(
-            stats.get("comment_count") or item.get("comment_count")
+            _first_present(stats.get("comment_count"), item.get("comment_count"))
         ),
         "share_count": _nonnegative_int(
-            stats.get("share_count") or item.get("share_count")
+            _first_present(stats.get("share_count"), item.get("share_count"))
         ),
         "collect_count": _nonnegative_int(
-            stats.get("collect_count") or item.get("collect_count")
+            _first_present(stats.get("collect_count"), item.get("collect_count"))
         ),
         "play_count": _nonnegative_int(
-            stats.get("play_count") or item.get("play_count")
+            _first_present(stats.get("play_count"), item.get("play_count"))
         ),
         "duration_seconds": duration,
         "publish_at": (
@@ -164,6 +172,7 @@ def persist_search_payload(
 ) -> str:
     """新增搜索记录和本页全部候选，并把数据库 ID 拼回搜索结果。"""
     payload = _load_payload(payload_json)
+    provider_raw_response = payload.get("_raw_response", payload)
     run_text = _clean_text(run_id, max_length=64)
     keyword_text = _clean_text(keyword, max_length=256)
     reason_text = _clean_text(query_reason)
@@ -231,12 +240,31 @@ def persist_search_payload(
         "error_message": _clean_text(payload.get("error")),
     }
     search_values["search_key"] = _search_key(search_values)
+    evidence_values = build_evidence_values(
+        run_id=run_text,
+        evidence_type="search_page",
+        provider=str(search_values["provider"]),
+        subject_key=f"search:{search_values['search_key']}:{normalized_page_no}",
+        request={
+            "keyword": keyword_text,
+            "content_type": search_values["content_type"],
+            "sort_type": search_values["sort_type"],
+            "publish_time": search_values["publish_time"],
+            "cursor": search_values["cursor"],
+            "page_no": normalized_page_no,
+        },
+        raw_response=provider_raw_response,
+        normalized={"candidate_count": len(candidate_rows)},
+        fetch_status="failed" if payload.get("error") else "success",
+        error_message=_clean_text(payload.get("error")),
+    )
 
     try:
         saved = get_video_discovery_service().save_search_page(
             run_text,
             search_values,
             candidate_rows,
+            evidence_values,
         )
     except RunNotFoundError as exc:
         payload["error"] = str(exc)
@@ -249,8 +277,25 @@ def persist_search_payload(
 
     payload.pop("search_results", None)
     payload.pop("user_videos", None)
+    payload.pop("_raw_response", None)
     if provider_search_id not in (None, ""):
         payload["provider_search_id"] = provider_search_id
+    saved_candidates = saved.get("candidates") or []
+    saved["candidates"] = [
+        {
+            "candidate_id": item.get("candidate_id"),
+            "search_id": item.get("search_id"),
+            "aweme_id": item.get("aweme_id"),
+            "title": item.get("title"),
+            "author_name": item.get("author_name"),
+            "author_sec_uid": item.get("author_sec_uid"),
+            "publish_at": item.get("publish_at"),
+            "duration_seconds": item.get("duration_seconds"),
+            "share_count": item.get("share_count"),
+            "gate_status": item.get("gate_status"),
+        }
+        for item in saved_candidates
+    ]
     payload.update(saved)
     payload["persisted"] = True
     return json.dumps(payload, ensure_ascii=False, default=str)

agents/find_agent/support/video_discovery.py

@@ -319,6 +319,24 @@ def batch_update_video_discovery_candidates(
             rows,
         )
         reclassified = list(updated.get("reclassified") or [])
+        candidate_summaries = []
+        for candidate in updated["candidates"]:
+            gate_results = candidate.get("gate_results") or {}
+            candidate_summaries.append(
+                {
+                    "candidate_id": candidate.get("candidate_id"),
+                    "aweme_id": candidate.get("aweme_id"),
+                    "decision_bucket": candidate.get("decision_bucket"),
+                    "gate_status": candidate.get("gate_status"),
+                    "failed_reason_codes": gate_results.get("failed_reason_codes") or [],
+                    "pending_reason_codes": gate_results.get("pending_reason_codes") or [],
+                    "evidence_version": candidate.get("evidence_version"),
+                    "gate_evidence_version": candidate.get("gate_evidence_version"),
+                    "latest_gate_evaluation_id": candidate.get(
+                        "latest_gate_evaluation_id"
+                    ),
+                }
+            )
         payload = {
             "title": "候选已更新",
             "run_id": run_text,
@@ -327,10 +345,10 @@ def batch_update_video_discovery_candidates(
                 updated.get("reclassified_count", len(reclassified))
             ),
             "reclassified": reclassified,
-            "candidates": updated["candidates"],
+            "candidates": candidate_summaries,
             "output": (
                 f"更新 {updated['updated_count']} 条候选；"
-                f"P0 自动降级 {int(updated.get('reclassified_count', len(reclassified)))} 条"
+                f"门禁调整 {int(updated.get('reclassified_count', len(reclassified)))} 条"
             ),
         }
         return _json(payload)

alembic/versions/20260803_11_add_find_agent_evidence.py

@@ -0,0 +1,175 @@
+"""add find_agent raw evidence and versioned gate evaluations
+
+Revision ID: 20260803_11
+Revises: 20260803_10
+Create Date: 2026-08-03
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import mysql
+
+revision: str = "20260803_11"
+down_revision: str | None = "20260803_10"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "video_discovery_evidence",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("run_id", sa.String(length=64), nullable=False),
+        sa.Column("search_id", sa.BigInteger(), nullable=True),
+        sa.Column("trigger_candidate_id", sa.BigInteger(), nullable=True),
+        sa.Column("aweme_id", sa.String(length=64), nullable=True),
+        sa.Column("author_sec_uid", sa.String(length=256), nullable=True),
+        sa.Column("subject_key", sa.String(length=320), nullable=False),
+        sa.Column("evidence_type", sa.String(length=32), nullable=False),
+        sa.Column("provider", sa.String(length=32), nullable=False),
+        sa.Column("tool_call_id", sa.String(length=64), nullable=False),
+        sa.Column("request_json", mysql.LONGTEXT(), nullable=True),
+        sa.Column("raw_response_json", mysql.LONGTEXT(), nullable=True),
+        sa.Column("normalized_json", mysql.LONGTEXT(), nullable=True),
+        sa.Column("response_sha256", sa.String(length=64), nullable=True),
+        sa.Column("fetch_status", sa.String(length=16), nullable=False),
+        sa.Column(
+            "processing_status",
+            sa.String(length=16),
+            nullable=False,
+            server_default="pending",
+        ),
+        sa.Column("error_code", sa.String(length=64), nullable=True),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.Column(
+            "schema_version",
+            sa.String(length=32),
+            nullable=False,
+            server_default="find-agent-evidence-v1",
+        ),
+        sa.Column(
+            "fetched_at",
+            sa.DateTime(),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.Column("latency_ms", sa.Integer(), nullable=True),
+        sa.Column(
+            "create_time",
+            sa.DateTime(),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "tool_call_id",
+            "evidence_type",
+            "subject_key",
+            name="uk_video_discovery_evidence_call_subject",
+        ),
+    )
+    op.create_index(
+        "idx_video_discovery_evidence_run_type",
+        "video_discovery_evidence",
+        ["run_id", "evidence_type", "id"],
+    )
+    op.create_index(
+        "idx_video_discovery_evidence_subject",
+        "video_discovery_evidence",
+        ["run_id", "aweme_id", "evidence_type", "id"],
+    )
+    op.create_index(
+        "idx_video_discovery_evidence_processing",
+        "video_discovery_evidence",
+        ["processing_status", "id"],
+    )
+
+    op.create_table(
+        "video_discovery_gate_evaluation",
+        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
+        sa.Column("run_id", sa.String(length=64), nullable=False),
+        sa.Column("candidate_id", sa.BigInteger(), nullable=False),
+        sa.Column("rule_version", sa.String(length=64), nullable=False),
+        sa.Column("evidence_version", sa.Integer(), nullable=False),
+        sa.Column("gate_input_hash", sa.String(length=64), nullable=False),
+        sa.Column("gate_input_json", mysql.LONGTEXT(), nullable=False),
+        sa.Column("gate_result_json", mysql.LONGTEXT(), nullable=False),
+        sa.Column("gate_status", sa.String(length=16), nullable=False),
+        sa.Column("failed_reason_codes_json", sa.Text(), nullable=True),
+        sa.Column(
+            "evaluated_at",
+            sa.DateTime(),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "candidate_id",
+            "rule_version",
+            "evidence_version",
+            "gate_input_hash",
+            name="uk_video_discovery_gate_input",
+        ),
+    )
+    op.create_index(
+        "idx_video_discovery_gate_candidate",
+        "video_discovery_gate_evaluation",
+        ["candidate_id", "id"],
+    )
+    op.create_index(
+        "idx_video_discovery_gate_run_status",
+        "video_discovery_gate_evaluation",
+        ["run_id", "gate_status", "id"],
+    )
+
+    op.add_column(
+        "video_discovery_search",
+        sa.Column("raw_evidence_id", sa.BigInteger(), nullable=True),
+    )
+    candidate_columns = (
+        sa.Column("latest_search_evidence_id", sa.BigInteger(), nullable=True),
+        sa.Column("latest_detail_evidence_id", sa.BigInteger(), nullable=True),
+        sa.Column("latest_content_portrait_evidence_id", sa.BigInteger(), nullable=True),
+        sa.Column("latest_account_portrait_evidence_id", sa.BigInteger(), nullable=True),
+        sa.Column("detail_fetch_status", sa.String(length=16), nullable=True),
+        sa.Column("content_portrait_fetch_status", sa.String(length=16), nullable=True),
+        sa.Column("account_portrait_fetch_status", sa.String(length=16), nullable=True),
+        sa.Column(
+            "evidence_version", sa.Integer(), nullable=False, server_default="0"
+        ),
+        sa.Column("gate_evidence_version", sa.Integer(), nullable=True),
+        sa.Column("latest_gate_evaluation_id", sa.BigInteger(), nullable=True),
+        sa.Column("gate_evaluated_at", sa.DateTime(), nullable=True),
+    )
+    for column in candidate_columns:
+        op.add_column("video_discovery_candidate", column)
+    op.execute(
+        sa.text(
+            "UPDATE video_discovery_candidate "
+            "SET gate_evidence_version = evidence_version "
+            "WHERE gate_status IS NOT NULL"
+        )
+    )
+
+
+def downgrade() -> None:
+    for name in (
+        "gate_evaluated_at",
+        "latest_gate_evaluation_id",
+        "gate_evidence_version",
+        "evidence_version",
+        "account_portrait_fetch_status",
+        "content_portrait_fetch_status",
+        "detail_fetch_status",
+        "latest_account_portrait_evidence_id",
+        "latest_content_portrait_evidence_id",
+        "latest_detail_evidence_id",
+        "latest_search_evidence_id",
+    ):
+        op.drop_column("video_discovery_candidate", name)
+    op.drop_column("video_discovery_search", "raw_evidence_id")
+    op.drop_table("video_discovery_gate_evaluation")
+    op.drop_table("video_discovery_evidence")

sql/video_discovery_tables.sql

@@ -60,6 +60,7 @@ CREATE TABLE IF NOT EXISTS `video_discovery_search` (
   `result_ids_json` TEXT NULL COMMENT '本页 aweme_id 列表 JSON',
   `status` VARCHAR(16) NOT NULL DEFAULT 'success' COMMENT 'success / failed',
   `error_message` TEXT NULL COMMENT '搜索失败信息',
+  `raw_evidence_id` BIGINT NULL COMMENT '本页完整原始响应证据 id',
   `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
   `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
     ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
@@ -106,6 +107,17 @@ CREATE TABLE IF NOT EXISTS `video_discovery_candidate` (
   `gate_results_json` TEXT NULL COMMENT '逐项硬门槛结果 JSON',
   `reject_reason_code` VARCHAR(64) NULL COMMENT '主要淘汰原因码',
   `rule_version` VARCHAR(64) NULL COMMENT '候选评估使用的规则版本',
+  `latest_search_evidence_id` BIGINT NULL COMMENT '最新搜索原始证据 id',
+  `latest_detail_evidence_id` BIGINT NULL COMMENT '最新详情原始证据 id',
+  `latest_content_portrait_evidence_id` BIGINT NULL COMMENT '最新内容画像原始证据 id',
+  `latest_account_portrait_evidence_id` BIGINT NULL COMMENT '最新账号画像原始证据 id',
+  `detail_fetch_status` VARCHAR(16) NULL COMMENT '详情获取状态',
+  `content_portrait_fetch_status` VARCHAR(16) NULL COMMENT '内容画像获取状态',
+  `account_portrait_fetch_status` VARCHAR(16) NULL COMMENT '账号画像获取状态',
+  `evidence_version` INT NOT NULL DEFAULT 0 COMMENT '门禁输入数据版本',
+  `gate_evidence_version` INT NULL COMMENT '最近门禁使用的证据版本',
+  `latest_gate_evaluation_id` BIGINT NULL COMMENT '最近门禁评估记录 id',
+  `gate_evaluated_at` DATETIME NULL COMMENT '最近门禁评估时间',
   `relevance_score` DECIMAL(8,6) NULL COMMENT 'R，范围 0~1',
   `elder_score` DECIMAL(8,6) NULL COMMENT 'E，范围 0~1',
   `share_score` DECIMAL(8,6) NULL COMMENT 'S，范围 0~1',
@@ -125,3 +137,56 @@ CREATE TABLE IF NOT EXISTS `video_discovery_candidate` (
     ON DELETE RESTRICT
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
   COMMENT='本次运行发现的视频及评估快照';
+
+CREATE TABLE IF NOT EXISTS `video_discovery_evidence` (
+  `id` BIGINT NOT NULL AUTO_INCREMENT,
+  `run_id` VARCHAR(64) NOT NULL,
+  `search_id` BIGINT NULL,
+  `trigger_candidate_id` BIGINT NULL,
+  `aweme_id` VARCHAR(64) NULL,
+  `author_sec_uid` VARCHAR(256) NULL,
+  `subject_key` VARCHAR(320) NOT NULL,
+  `evidence_type` VARCHAR(32) NOT NULL,
+  `provider` VARCHAR(32) NOT NULL,
+  `tool_call_id` VARCHAR(64) NOT NULL,
+  `request_json` LONGTEXT NULL,
+  `raw_response_json` LONGTEXT NULL,
+  `normalized_json` LONGTEXT NULL,
+  `response_sha256` VARCHAR(64) NULL,
+  `fetch_status` VARCHAR(16) NOT NULL,
+  `processing_status` VARCHAR(16) NOT NULL DEFAULT 'pending',
+  `error_code` VARCHAR(64) NULL,
+  `error_message` TEXT NULL,
+  `schema_version` VARCHAR(32) NOT NULL DEFAULT 'find-agent-evidence-v1',
+  `fetched_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  `latency_ms` INT NULL,
+  `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_video_discovery_evidence_call_subject`
+    (`tool_call_id`, `evidence_type`, `subject_key`),
+  KEY `idx_video_discovery_evidence_run_type` (`run_id`, `evidence_type`, `id`),
+  KEY `idx_video_discovery_evidence_subject`
+    (`run_id`, `aweme_id`, `evidence_type`, `id`),
+  KEY `idx_video_discovery_evidence_processing` (`processing_status`, `id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+  COMMENT='find_agent 工具原始响应和规范化证据';
+
+CREATE TABLE IF NOT EXISTS `video_discovery_gate_evaluation` (
+  `id` BIGINT NOT NULL AUTO_INCREMENT,
+  `run_id` VARCHAR(64) NOT NULL,
+  `candidate_id` BIGINT NOT NULL,
+  `rule_version` VARCHAR(64) NOT NULL,
+  `evidence_version` INT NOT NULL,
+  `gate_input_hash` VARCHAR(64) NOT NULL,
+  `gate_input_json` LONGTEXT NOT NULL,
+  `gate_result_json` LONGTEXT NOT NULL,
+  `gate_status` VARCHAR(16) NOT NULL,
+  `failed_reason_codes_json` TEXT NULL,
+  `evaluated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_video_discovery_gate_input`
+    (`candidate_id`, `rule_version`, `evidence_version`, `gate_input_hash`),
+  KEY `idx_video_discovery_gate_candidate` (`candidate_id`, `id`),
+  KEY `idx_video_discovery_gate_run_status` (`run_id`, `gate_status`, `id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+  COMMENT='find_agent 候选门禁评估历史';

supply_infra/db/models/__init__.py

@@ -33,6 +33,8 @@ from supply_infra.db.models.pipeline_run import PipelineRun
 from supply_infra.db.models.pipeline_step_run import PipelineStepRun
 from supply_infra.db.models.video_discovery import (
     VideoDiscoveryCandidate,
+    VideoDiscoveryEvidence,
+    VideoDiscoveryGateEvaluation,
     VideoDiscoveryRun,
     VideoDiscoverySearch,
 )
@@ -66,6 +68,8 @@ __all__ = [
     "PipelineRun",
     "PipelineStepRun",
     "VideoDiscoveryCandidate",
+    "VideoDiscoveryEvidence",
+    "VideoDiscoveryGateEvaluation",
     "VideoDiscoveryRun",
     "VideoDiscoverySearch",
 ]

supply_infra/db/models/video_discovery.py

@@ -15,10 +15,15 @@ from sqlalchemy import (
     func,
 )
 from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.dialects.mysql import LONGTEXT
 
 from supply_infra.db.base import Base
 
 
+_LONG_TEXT = Text().with_variant(LONGTEXT(), "mysql")
+_BIGINT_PK = BigInteger().with_variant(Integer, "sqlite")
+
+
 class VideoDiscoveryRun(Base):
     """一次需求找视频运行，保存输入、意图解释和完成状态。"""
 
@@ -186,6 +191,9 @@ class VideoDiscoverySearch(Base):
     error_message: Mapped[str | None] = mapped_column(
         Text, nullable=True, comment="搜索失败信息"
     )
+    raw_evidence_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="本页完整原始响应证据 id"
+    )
     create_time: Mapped[datetime] = mapped_column(
         nullable=False, server_default=func.now(), comment="创建时间"
     )
@@ -301,6 +309,39 @@ class VideoDiscoveryCandidate(Base):
     rule_version: Mapped[str | None] = mapped_column(
         String(64), nullable=True, comment="候选评估使用的规则版本"
     )
+    latest_search_evidence_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="最新搜索原始证据 id"
+    )
+    latest_detail_evidence_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="最新详情原始证据 id"
+    )
+    latest_content_portrait_evidence_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="最新内容画像原始证据 id"
+    )
+    latest_account_portrait_evidence_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="最新账号画像原始证据 id"
+    )
+    detail_fetch_status: Mapped[str | None] = mapped_column(
+        String(16), nullable=True, comment="pending / success / failed / unavailable"
+    )
+    content_portrait_fetch_status: Mapped[str | None] = mapped_column(
+        String(16), nullable=True, comment="内容画像获取状态"
+    )
+    account_portrait_fetch_status: Mapped[str | None] = mapped_column(
+        String(16), nullable=True, comment="账号画像获取状态"
+    )
+    evidence_version: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="门禁输入数据版本"
+    )
+    gate_evidence_version: Mapped[int | None] = mapped_column(
+        Integer, nullable=True, comment="最近门禁使用的证据版本"
+    )
+    latest_gate_evaluation_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="最近门禁评估记录 id"
+    )
+    gate_evaluated_at: Mapped[datetime | None] = mapped_column(
+        nullable=True, comment="最近门禁评估时间"
+    )
     relevance_score: Mapped[Decimal | None] = mapped_column(
         Numeric(8, 6), nullable=True, comment="R，范围 0~1"
     )
@@ -341,3 +382,84 @@ class VideoDiscoveryCandidate(Base):
         onupdate=func.now(),
         comment="更新时间",
     )
+
+
+class VideoDiscoveryEvidence(Base):
+    """工具原始响应及规范化结果的不可变证据快照。"""
+
+    __tablename__ = "video_discovery_evidence"
+    __table_args__ = (
+        Index("idx_video_discovery_evidence_run_type", "run_id", "evidence_type", "id"),
+        Index(
+            "idx_video_discovery_evidence_subject",
+            "run_id",
+            "aweme_id",
+            "evidence_type",
+            "id",
+        ),
+        Index("idx_video_discovery_evidence_processing", "processing_status", "id"),
+        UniqueConstraint(
+            "tool_call_id",
+            "evidence_type",
+            "subject_key",
+            name="uk_video_discovery_evidence_call_subject",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(_BIGINT_PK, primary_key=True, autoincrement=True)
+    run_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    search_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    trigger_candidate_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    aweme_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    author_sec_uid: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    subject_key: Mapped[str] = mapped_column(String(320), nullable=False)
+    evidence_type: Mapped[str] = mapped_column(String(32), nullable=False)
+    provider: Mapped[str] = mapped_column(String(32), nullable=False)
+    tool_call_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    request_json: Mapped[str | None] = mapped_column(_LONG_TEXT, nullable=True)
+    raw_response_json: Mapped[str | None] = mapped_column(_LONG_TEXT, nullable=True)
+    normalized_json: Mapped[str | None] = mapped_column(_LONG_TEXT, nullable=True)
+    response_sha256: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    fetch_status: Mapped[str] = mapped_column(String(16), nullable=False)
+    processing_status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="pending"
+    )
+    error_code: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    schema_version: Mapped[str] = mapped_column(
+        String(32), nullable=False, default="find-agent-evidence-v1"
+    )
+    fetched_at: Mapped[datetime] = mapped_column(nullable=False, server_default=func.now())
+    latency_ms: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    create_time: Mapped[datetime] = mapped_column(nullable=False, server_default=func.now())
+
+
+class VideoDiscoveryGateEvaluation(Base):
+    """候选每个输入版本的门禁评估历史。"""
+
+    __tablename__ = "video_discovery_gate_evaluation"
+    __table_args__ = (
+        Index("idx_video_discovery_gate_candidate", "candidate_id", "id"),
+        Index("idx_video_discovery_gate_run_status", "run_id", "gate_status", "id"),
+        UniqueConstraint(
+            "candidate_id",
+            "rule_version",
+            "evidence_version",
+            "gate_input_hash",
+            name="uk_video_discovery_gate_input",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(_BIGINT_PK, primary_key=True, autoincrement=True)
+    run_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    candidate_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    rule_version: Mapped[str] = mapped_column(String(64), nullable=False)
+    evidence_version: Mapped[int] = mapped_column(Integer, nullable=False)
+    gate_input_hash: Mapped[str] = mapped_column(String(64), nullable=False)
+    gate_input_json: Mapped[str] = mapped_column(_LONG_TEXT, nullable=False)
+    gate_result_json: Mapped[str] = mapped_column(_LONG_TEXT, nullable=False)
+    gate_status: Mapped[str] = mapped_column(String(16), nullable=False)
+    failed_reason_codes_json: Mapped[str | None] = mapped_column(Text, nullable=True)
+    evaluated_at: Mapped[datetime] = mapped_column(
+        nullable=False, server_default=func.now()
+    )

supply_infra/db/repositories/video_discovery_repo.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import json
+import hashlib
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any
 
 from sqlalchemy import func, select, update
@@ -13,6 +15,8 @@ from supply_infra.video_discovery_gates import (
 )
 from supply_infra.db.models.video_discovery import (
     VideoDiscoveryCandidate,
+    VideoDiscoveryEvidence,
+    VideoDiscoveryGateEvaluation,
     VideoDiscoveryRun,
     VideoDiscoverySearch,
 )
@@ -45,6 +49,60 @@ def _merge_json_list(raw: str | None, values: list[Any]) -> str | None:
 _PUBLISHABLE_BUCKETS = ("primary",)
 _BUSINESS_GOAL_PRIMARY_COUNT = 5
 _PENDING_FINAL_REJECT_REASON = "NOT_SELECTED_AFTER_EVALUATION"
+_EVIDENCE_INCOMPLETE_REJECT_REASON = "EVIDENCE_INCOMPLETE"
+
+
+def _gate_input(entity: VideoDiscoveryCandidate) -> dict[str, Any]:
+    """Build the single authoritative gate input from persisted candidate facts."""
+    return {
+        "title": entity.title,
+        "tags_json": entity.tags_json,
+        "publish_at": entity.publish_at,
+        "duration_seconds": entity.duration_seconds,
+        "share_count": entity.share_count,
+        "like_count": entity.like_count,
+        "play_count": entity.play_count,
+        "content_50_plus_ratio": entity.content_50_plus_ratio,
+        "account_50_plus_ratio": entity.account_50_plus_ratio,
+        "relevance_score": entity.relevance_score,
+        "elder_score": entity.elder_score,
+        "share_score": entity.share_score,
+        "value_score": entity.value_score,
+        "temporal_type": entity.temporal_type,
+        "temporal_status": entity.temporal_status,
+        "temporal_evidence_json": entity.temporal_evidence_json,
+        "detail_fetch_status": entity.detail_fetch_status,
+        "content_portrait_fetch_status": entity.content_portrait_fetch_status,
+        "account_portrait_fetch_status": entity.account_portrait_fetch_status,
+        "latest_search_evidence_id": entity.latest_search_evidence_id,
+        "latest_detail_evidence_id": entity.latest_detail_evidence_id,
+        "latest_content_portrait_evidence_id": entity.latest_content_portrait_evidence_id,
+        "latest_account_portrait_evidence_id": entity.latest_account_portrait_evidence_id,
+        "evidence_version": int(entity.evidence_version or 0),
+    }
+
+
+def _pending_gate_codes(
+    gate: dict[str, Any], candidate: VideoDiscoveryCandidate
+) -> list[str]:
+    """Return missing-evidence failures that are still retryable."""
+    pending: list[str] = []
+    detail_terminal = candidate.detail_fetch_status in {"success", "unavailable"}
+    portrait_terminal = (
+        candidate.content_portrait_fetch_status in {"success", "unavailable"}
+        or candidate.account_portrait_fetch_status in {"success", "unavailable"}
+    )
+    for code in gate.get("failed_reason_codes") or []:
+        code_text = str(code)
+        if code_text in {
+            "TEMPORAL_UNKNOWN",
+            "DURATION_UNKNOWN",
+            "SHARE_COUNT_UNKNOWN",
+        } and not detail_terminal:
+            pending.append(code_text)
+        elif code_text == "CONTENT_PORTRAIT_MISSING" and not portrait_terminal:
+            pending.append(code_text)
+    return pending
 
 
 @dataclass(frozen=True)
@@ -134,16 +192,167 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         self.session.flush()
         return entity
 
+    def create_evidence(self, values: dict[str, Any]) -> VideoDiscoveryEvidence:
+        entity = VideoDiscoveryEvidence(**values)
+        self.session.add(entity)
+        self.session.flush()
+        return entity
+
+    def get_evidence(self, evidence_id: int) -> VideoDiscoveryEvidence | None:
+        return self.session.get(VideoDiscoveryEvidence, int(evidence_id))
+
+    def get_candidates_by_ids(
+        self, run_id: str, candidate_ids: list[int]
+    ) -> list[VideoDiscoveryCandidate]:
+        if not candidate_ids:
+            return []
+        stmt = select(VideoDiscoveryCandidate).where(
+            VideoDiscoveryCandidate.run_id == run_id,
+            VideoDiscoveryCandidate.id.in_([int(value) for value in candidate_ids]),
+        )
+        rows = list(self.session.scalars(stmt).all())
+        by_id = {int(item.id): item for item in rows}
+        missing = [int(value) for value in candidate_ids if int(value) not in by_id]
+        if missing:
+            raise ValueError(
+                f"candidate_id 不存在或不属于 run_id={run_id}: {missing}"
+            )
+        return [by_id[int(value)] for value in candidate_ids]
+
+    def _evaluate_candidate(
+        self,
+        entity: VideoDiscoveryCandidate,
+        rule_snapshot: dict[str, Any],
+    ) -> dict[str, Any]:
+        input_payload = _gate_input(entity)
+        gate = evaluate_candidate_gate(input_payload, rule_snapshot)
+        pending_codes = _pending_gate_codes(gate, entity)
+        if pending_codes:
+            gate["status"] = "pending"
+            gate["primary_eligible"] = False
+            gate["pending_reason_codes"] = pending_codes
+
+        entity.content_portrait_status = gate["content_portrait_status"]
+        entity.account_portrait_status = gate["account_portrait_status"]
+        entity.portrait_conflict = int(bool(gate["portrait_conflict"]))
+        entity.temporal_type = gate["temporal"]["temporal_type"]
+        entity.temporal_status = gate["temporal"]["status"]
+        entity.temporal_evidence_json = json.dumps(
+            gate["temporal"], ensure_ascii=False, default=str
+        )
+        entity.gate_status = str(gate["status"])
+        entity.gate_results_json = json.dumps(gate, ensure_ascii=False, default=str)
+        entity.rule_version = str(gate["rule_version"])
+        entity.gate_evidence_version = int(entity.evidence_version or 0)
+        entity.gate_evaluated_at = datetime.now()
+
+        input_json = json.dumps(
+            input_payload, ensure_ascii=False, sort_keys=True, default=str
+        )
+        input_hash = hashlib.sha256(input_json.encode("utf-8")).hexdigest()
+        evaluation = VideoDiscoveryGateEvaluation(
+            run_id=entity.run_id,
+            candidate_id=int(entity.id),
+            rule_version=str(gate["rule_version"]),
+            evidence_version=int(entity.evidence_version or 0),
+            gate_input_hash=input_hash,
+            gate_input_json=input_json,
+            gate_result_json=json.dumps(gate, ensure_ascii=False, default=str),
+            gate_status=str(gate["status"]),
+            failed_reason_codes_json=json.dumps(
+                gate.get("failed_reason_codes") or [], ensure_ascii=False
+            ),
+        )
+        self.session.add(evaluation)
+        self.session.flush()
+        entity.latest_gate_evaluation_id = int(evaluation.id)
+        return gate
+
+    def apply_evidence(
+        self,
+        run_id: str,
+        *,
+        aweme_id: str,
+        evidence: VideoDiscoveryEvidence,
+        normalized: dict[str, Any],
+    ) -> list[VideoDiscoveryCandidate]:
+        """Apply one immutable observation to every occurrence of a run video."""
+        run = self.get_run(run_id)
+        if run is None:
+            raise ValueError(f"run_id 不存在: {run_id}")
+        rule_snapshot = load_rule_snapshot(run.rule_config_json)
+        stmt = select(VideoDiscoveryCandidate).where(
+            VideoDiscoveryCandidate.run_id == run_id,
+            VideoDiscoveryCandidate.aweme_id == aweme_id,
+        )
+        candidates = list(self.session.scalars(stmt).all())
+        if not candidates:
+            raise ValueError(f"run_id={run_id} 中不存在 aweme_id={aweme_id}")
+
+        pointer_name = {
+            "detail": "latest_detail_evidence_id",
+            "content_portrait": "latest_content_portrait_evidence_id",
+            "account_portrait": "latest_account_portrait_evidence_id",
+        }.get(evidence.evidence_type)
+        status_name = {
+            "detail": "detail_fetch_status",
+            "content_portrait": "content_portrait_fetch_status",
+            "account_portrait": "account_portrait_fetch_status",
+        }.get(evidence.evidence_type)
+        for candidate in candidates:
+            if pointer_name:
+                setattr(candidate, pointer_name, int(evidence.id))
+            if status_name:
+                setattr(candidate, status_name, evidence.fetch_status)
+            for key, value in normalized.items():
+                if key in {"id", "candidate_id", "run_id", "search_id", "aweme_id"}:
+                    continue
+                if value is not None and hasattr(candidate, key):
+                    setattr(candidate, key, value)
+            candidate.evidence_version = int(candidate.evidence_version or 0) + 1
+            self.session.flush()
+            gate = self._evaluate_candidate(candidate, rule_snapshot)
+            if candidate.decision_bucket == "primary" and gate["status"] != "pass":
+                if gate["status"] == "pending":
+                    candidate.decision_bucket = "pending_evaluation"
+                    candidate.reject_reason_code = None
+                else:
+                    candidate.decision_bucket = "rejected"
+                    failed_codes = gate.get("failed_reason_codes") or []
+                    candidate.reject_reason_code = str(
+                        failed_codes[0] if failed_codes else "P0_GATE_FAILED"
+                    )
+        evidence.processing_status = "applied"
+        evidence.normalized_json = json.dumps(
+            normalized, ensure_ascii=False, default=str
+        )
+        self.session.flush()
+        self._refresh_run_counts(run_id)
+        return candidates
+
     def save_search_page(
         self,
         search_values: dict[str, Any],
         candidate_rows: list[dict[str, Any]],
+        evidence_values: dict[str, Any] | None = None,
+        evidence_id: int | None = None,
     ) -> tuple[VideoDiscoverySearch, list[VideoDiscoveryCandidate]]:
         """新增一个搜索页，并为本页每条结果新增独立候选记录。"""
         run_id = str(search_values["run_id"])
+        evidence = (
+            self.get_evidence(evidence_id)
+            if evidence_id is not None
+            else (self.create_evidence(evidence_values) if evidence_values else None)
+        )
+        if evidence_id is not None and evidence is None:
+            raise ValueError(f"evidence_id 不存在: {evidence_id}")
         search = VideoDiscoverySearch(**search_values)
+        if evidence is not None:
+            search.raw_evidence_id = int(evidence.id)
         self.session.add(search)
         self.session.flush()
+        if evidence is not None:
+            evidence.search_id = int(search.id)
 
         candidates: list[VideoDiscoveryCandidate] = []
         aweme_ids: list[str] = []
@@ -164,6 +373,8 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
                 ),
                 source_search_ids_json=json.dumps([int(search.id)]),
                 decision_bucket="pending_evaluation",
+                latest_search_evidence_id=(int(evidence.id) if evidence else None),
+                evidence_version=(1 if evidence else 0),
             )
             for key, value in row.items():
                 if value is None:
@@ -179,6 +390,8 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
             json.dumps(aweme_ids, ensure_ascii=False) if aweme_ids else None
         )
         self.session.flush()
+        if evidence is not None:
+            evidence.processing_status = "applied"
         self._refresh_run_counts(run_id)
         return search, candidates
 
@@ -214,6 +427,7 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         for row in rows:
             candidate_id = int(row["candidate_id"])
             entity = existing[candidate_id]
+            requested_bucket = str(row.get("decision_bucket") or entity.decision_bucket)
             for key, value in row.items():
                 if key in {"candidate_id", "id", "run_id", "search_id", "aweme_id"}:
                     continue
@@ -228,42 +442,25 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
                     setattr(entity, key, _merge_json_list(getattr(entity, key), values))
                 elif hasattr(entity, key):
                     setattr(entity, key, value)
+            entity.evidence_version = int(entity.evidence_version or 0) + 1
+            self.session.flush()
+            gate = self._evaluate_candidate(entity, rule_snapshot)
 
-            gate = evaluate_candidate_gate(
-                {
-                    "title": entity.title,
-                    "tags_json": entity.tags_json,
-                    "publish_at": entity.publish_at,
-                    "duration_seconds": entity.duration_seconds,
-                    "share_count": entity.share_count,
-                    "content_50_plus_ratio": entity.content_50_plus_ratio,
-                    "account_50_plus_ratio": entity.account_50_plus_ratio,
-                    "temporal_type": entity.temporal_type,
-                    "temporal_status": entity.temporal_status,
-                    "temporal_evidence_json": entity.temporal_evidence_json,
-                },
-                rule_snapshot,
-            )
-            entity.content_portrait_status = gate["content_portrait_status"]
-            entity.account_portrait_status = gate["account_portrait_status"]
-            entity.portrait_conflict = int(bool(gate["portrait_conflict"]))
-            entity.temporal_type = gate["temporal"]["temporal_type"]
-            entity.temporal_status = gate["temporal"]["status"]
-            entity.temporal_evidence_json = json.dumps(
-                gate["temporal"],
-                ensure_ascii=False,
-                default=str,
-            )
-            entity.gate_status = gate["status"]
-            entity.gate_results_json = json.dumps(
-                gate,
-                ensure_ascii=False,
-                default=str,
-            )
-            entity.rule_version = str(gate["rule_version"])
-
-            if entity.decision_bucket == "primary":
-                if not gate["primary_eligible"]:
+            if requested_bucket == "primary":
+                if gate["status"] == "pending":
+                    entity.decision_bucket = "pending_evaluation"
+                    entity.reject_reason_code = None
+                    reclassified.append(
+                        {
+                            "candidate_id": candidate_id,
+                            "requested_bucket": "primary",
+                            "saved_bucket": "pending_evaluation",
+                            "failed_reason_codes": list(
+                                gate.get("pending_reason_codes") or []
+                            ),
+                        }
+                    )
+                elif not gate["primary_eligible"]:
                     failed_codes = gate["failed_reason_codes"]
                     entity.decision_bucket = "rejected"
                     entity.reject_reason_code = str(
@@ -278,8 +475,10 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
                         }
                     )
                 else:
+                    entity.decision_bucket = "primary"
                     entity.reject_reason_code = None
             else:
+                entity.decision_bucket = requested_bucket
                 failed_codes = gate["failed_reason_codes"]
                 if failed_codes:
                     entity.reject_reason_code = str(failed_codes[0])
@@ -328,13 +527,24 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         return "no_match"
 
     def _finalize_pending_candidates(self, run_id: str) -> None:
+        pending_buckets = ("pending_evaluation", "unreviewed")
         self.session.execute(
             update(VideoDiscoveryCandidate)
             .where(
                 VideoDiscoveryCandidate.run_id == run_id,
-                VideoDiscoveryCandidate.decision_bucket.in_(
-                    ("pending_evaluation", "unreviewed")
-                ),
+                VideoDiscoveryCandidate.decision_bucket.in_(pending_buckets),
+                VideoDiscoveryCandidate.gate_status == "pending",
+            )
+            .values(
+                decision_bucket="rejected",
+                reject_reason_code=_EVIDENCE_INCOMPLETE_REJECT_REASON,
+            )
+        )
+        self.session.execute(
+            update(VideoDiscoveryCandidate)
+            .where(
+                VideoDiscoveryCandidate.run_id == run_id,
+                VideoDiscoveryCandidate.decision_bucket.in_(pending_buckets),
             )
             .values(
                 decision_bucket="rejected",
@@ -395,6 +605,10 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
             )
             .where(VideoDiscoveryCandidate.decision_bucket.in_(_PUBLISHABLE_BUCKETS))
             .where(VideoDiscoveryCandidate.gate_status == "pass")
+            .where(
+                VideoDiscoveryCandidate.gate_evidence_version
+                == VideoDiscoveryCandidate.evidence_version
+            )
             .where(VideoDiscoveryCandidate.aweme_id.is_not(None))
             .where(func.trim(VideoDiscoveryCandidate.aweme_id) != "")
             .order_by(
@@ -424,6 +638,10 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
             .where(VideoDiscoveryRun.biz_dt == biz_dt)
             .where(VideoDiscoveryCandidate.decision_bucket.in_(_PUBLISHABLE_BUCKETS))
             .where(VideoDiscoveryCandidate.gate_status == "pass")
+            .where(
+                VideoDiscoveryCandidate.gate_evidence_version
+                == VideoDiscoveryCandidate.evidence_version
+            )
             .where(VideoDiscoveryCandidate.aweme_id.is_not(None))
             .where(func.trim(VideoDiscoveryCandidate.aweme_id) != "")
         )
@@ -479,6 +697,8 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
             VideoDiscoveryCandidate.run_id == run_id,
             VideoDiscoveryCandidate.decision_bucket == "primary",
             VideoDiscoveryCandidate.gate_status == "pass",
+            VideoDiscoveryCandidate.gate_evidence_version
+            == VideoDiscoveryCandidate.evidence_version,
             VideoDiscoveryCandidate.aweme_id.is_not(None),
             func.trim(VideoDiscoveryCandidate.aweme_id) != "",
         )

supply_infra/services/video_discovery_service.py

@@ -2,6 +2,8 @@
 from __future__ import annotations
 
 import json
+import hashlib
+import uuid
 from dataclasses import dataclass
 from typing import Any
 
@@ -78,6 +80,7 @@ def _serialize_search(item: Any) -> dict[str, Any]:
         "next_cursor": item.next_cursor,
         "status": item.status,
         "error_message": item.error_message,
+        "raw_evidence_id": item.raw_evidence_id,
     }
 
 
@@ -148,6 +151,80 @@ def _serialize_candidate(item: Any) -> dict[str, Any]:
         "reject_reason_code": item.reject_reason_code,
         "rule_version": item.rule_version,
         "decision_reason": item.decision_reason,
+        "latest_search_evidence_id": item.latest_search_evidence_id,
+        "latest_detail_evidence_id": item.latest_detail_evidence_id,
+        "latest_content_portrait_evidence_id": item.latest_content_portrait_evidence_id,
+        "latest_account_portrait_evidence_id": item.latest_account_portrait_evidence_id,
+        "detail_fetch_status": item.detail_fetch_status,
+        "content_portrait_fetch_status": item.content_portrait_fetch_status,
+        "account_portrait_fetch_status": item.account_portrait_fetch_status,
+        "evidence_version": int(item.evidence_version or 0),
+        "gate_evidence_version": item.gate_evidence_version,
+        "latest_gate_evaluation_id": item.latest_gate_evaluation_id,
+        "gate_evaluated_at": (
+            item.gate_evaluated_at.isoformat() if item.gate_evaluated_at else None
+        ),
+    }
+
+
+def build_evidence_values(
+    *,
+    run_id: str,
+    evidence_type: str,
+    provider: str,
+    subject_key: str,
+    request: Any = None,
+    raw_response: Any = None,
+    normalized: Any = None,
+    fetch_status: str = "success",
+    processing_status: str = "pending",
+    tool_call_id: str | None = None,
+    search_id: int | None = None,
+    trigger_candidate_id: int | None = None,
+    aweme_id: str | None = None,
+    author_sec_uid: str | None = None,
+    error_code: str | None = None,
+    error_message: str | None = None,
+    latency_ms: int | None = None,
+) -> dict[str, Any]:
+    """Create a sanitized, append-only evidence row payload."""
+    request_json = (
+        json.dumps(request, ensure_ascii=False, default=str)
+        if request is not None
+        else None
+    )
+    raw_json = (
+        json.dumps(raw_response, ensure_ascii=False, default=str)
+        if raw_response is not None
+        else None
+    )
+    normalized_json = (
+        json.dumps(normalized, ensure_ascii=False, default=str)
+        if normalized is not None
+        else None
+    )
+    return {
+        "run_id": str(run_id),
+        "search_id": search_id,
+        "trigger_candidate_id": trigger_candidate_id,
+        "aweme_id": aweme_id,
+        "author_sec_uid": author_sec_uid,
+        "subject_key": str(subject_key),
+        "evidence_type": str(evidence_type),
+        "provider": str(provider),
+        "tool_call_id": tool_call_id or uuid.uuid4().hex,
+        "request_json": request_json,
+        "raw_response_json": raw_json,
+        "normalized_json": normalized_json,
+        "response_sha256": (
+            hashlib.sha256(raw_json.encode("utf-8")).hexdigest() if raw_json else None
+        ),
+        "fetch_status": str(fetch_status),
+        "processing_status": str(processing_status),
+        "error_code": error_code,
+        "error_message": error_message,
+        "schema_version": "find-agent-evidence-v1",
+        "latency_ms": latency_ms,
     }
 
 
@@ -192,12 +269,24 @@ class VideoDiscoveryService:
         run_id: str,
         search_values: dict[str, Any],
         candidate_rows: list[dict[str, Any]],
+        evidence_values: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
+        evidence_id: int | None = None
+        if evidence_values is not None:
+            with get_session() as session:
+                repo = VideoDiscoveryRepository(session)
+                if repo.get_run(run_id) is None:
+                    raise RunNotFoundError(f"run_id 不存在: {run_id}")
+                evidence_id = int(repo.create_evidence(evidence_values).id)
         with get_session() as session:
             repo = VideoDiscoveryRepository(session)
             if repo.get_run(run_id) is None:
                 raise RunNotFoundError(f"run_id 不存在: {run_id}")
-            search, candidates = repo.save_search_page(search_values, candidate_rows)
+            search, candidates = repo.save_search_page(
+                search_values,
+                candidate_rows,
+                evidence_id=evidence_id,
+            )
             return {
                 **_serialize_search(search),
                 "run_id": run_id,
@@ -207,6 +296,58 @@ class VideoDiscoveryService:
                 ],
             }
 
+    def get_candidates_by_ids(
+        self, run_id: str, candidate_ids: list[int]
+    ) -> list[dict[str, Any]]:
+        with get_session() as session:
+            repo = VideoDiscoveryRepository(session)
+            if repo.get_run(run_id) is None:
+                raise RunNotFoundError(f"run_id 不存在: {run_id}")
+            return [
+                _serialize_candidate(item)
+                for item in repo.get_candidates_by_ids(run_id, candidate_ids)
+            ]
+
+    def record_candidate_evidence(
+        self,
+        *,
+        run_id: str,
+        evidence_values: dict[str, Any],
+        aweme_id: str,
+        normalized: dict[str, Any],
+    ) -> dict[str, Any]:
+        with get_session() as session:
+            repo = VideoDiscoveryRepository(session)
+            if repo.get_run(run_id) is None:
+                raise RunNotFoundError(f"run_id 不存在: {run_id}")
+            evidence_id = int(repo.create_evidence(evidence_values).id)
+        with get_session() as session:
+            repo = VideoDiscoveryRepository(session)
+            evidence = repo.get_evidence(evidence_id)
+            if evidence is None:
+                raise ValueError(f"evidence_id 不存在: {evidence_id}")
+            candidates = repo.apply_evidence(
+                run_id,
+                aweme_id=aweme_id,
+                evidence=evidence,
+                normalized=normalized,
+            )
+            return {
+                "evidence_id": evidence_id,
+                "updated_count": len(candidates),
+                "candidates": [_serialize_candidate(item) for item in candidates],
+            }
+
+    def record_failed_evidence(self, evidence_values: dict[str, Any]) -> int:
+        with get_session() as session:
+            repo = VideoDiscoveryRepository(session)
+            run_id = str(evidence_values.get("run_id") or "")
+            if repo.get_run(run_id) is None:
+                raise RunNotFoundError(f"run_id 不存在: {run_id}")
+            evidence_values = {**evidence_values, "processing_status": "failed"}
+            evidence = repo.create_evidence(evidence_values)
+            return int(evidence.id)
+
     def update_candidates(
         self,
         run_id: str,

tests/supply_infra/scheduler/test_discover_videos_from_demands.py

@@ -36,6 +36,8 @@ from agents.find_agent.support.douyin_user_videos import (
 from supply_infra.video_discovery_gates import evaluate_candidate_gate
 from supply_infra.db.models.video_discovery import (
     VideoDiscoveryCandidate,
+    VideoDiscoveryEvidence,
+    VideoDiscoveryGateEvaluation,
     VideoDiscoveryRun,
     VideoDiscoverySearch,
 )
@@ -54,6 +56,8 @@ def _expire_on_commit_session_factory() -> sessionmaker[Session]:
     VideoDiscoveryRun.__table__.create(engine)
     VideoDiscoverySearch.__table__.create(engine)
     VideoDiscoveryCandidate.__table__.create(engine)
+    VideoDiscoveryEvidence.__table__.create(engine)
+    VideoDiscoveryGateEvaluation.__table__.create(engine)
     return sessionmaker(bind=engine, autoflush=False, autocommit=False)
 
 
@@ -527,6 +531,184 @@ def test_repository_reclassifies_primary_that_fails_p0_gate() -> None:
         ]
 
 
+def test_repository_gate_uses_persisted_like_and_play_compensation() -> None:
+    factory = _expire_on_commit_session_factory()
+    with factory() as session:
+        session.add(
+            VideoDiscoveryRun(
+                id=1,
+                run_id="complete-gate-input",
+                demand_word="生活技巧",
+                relevant_points_json="[]",
+                status="running",
+                rule_version="test-p0",
+                rule_config_json=json.dumps(_P0_RULES, ensure_ascii=False),
+            )
+        )
+        session.add(
+            VideoDiscoveryCandidate(
+                id=2,
+                run_id="complete-gate-input",
+                aweme_id="video-like-compensation",
+                title="适合家庭分享的生活技巧",
+                publish_at=datetime(2026, 7, 31, 9, 0),
+                duration_seconds=Decimal("30.000"),
+                share_count=None,
+                like_count=8000,
+                content_50_plus_ratio=Decimal("0.280000"),
+                detail_fetch_status="success",
+                content_portrait_fetch_status="success",
+                decision_bucket="pending_evaluation",
+            )
+        )
+        session.commit()
+
+    with factory() as session:
+        batch = VideoDiscoveryRepository(session).update_candidates(
+            "complete-gate-input",
+            [{"candidate_id": 2, "decision_bucket": "primary"}],
+        )
+        session.commit()
+        assert batch.candidates[0].decision_bucket == "primary"
+        assert batch.candidates[0].gate_status == "pass"
+        evaluation = session.get(
+            VideoDiscoveryGateEvaluation,
+            batch.candidates[0].latest_gate_evaluation_id,
+        )
+        assert evaluation is not None
+        assert json.loads(evaluation.gate_input_json)["like_count"] == 8000
+
+
+def test_publishable_candidates_require_current_gate_version() -> None:
+    factory = _expire_on_commit_session_factory()
+    with factory() as session:
+        session.add(
+            VideoDiscoveryRun(
+                id=1,
+                run_id="stale-gate-run",
+                demand_word="生活技巧",
+                relevant_points_json="[]",
+                status="finished",
+            )
+        )
+        session.add_all(
+            [
+                VideoDiscoveryCandidate(
+                    id=2,
+                    run_id="stale-gate-run",
+                    aweme_id="current",
+                    decision_bucket="primary",
+                    gate_status="pass",
+                    evidence_version=2,
+                    gate_evidence_version=2,
+                ),
+                VideoDiscoveryCandidate(
+                    id=3,
+                    run_id="stale-gate-run",
+                    aweme_id="stale",
+                    decision_bucket="primary",
+                    gate_status="pass",
+                    evidence_version=2,
+                    gate_evidence_version=1,
+                ),
+            ]
+        )
+        session.commit()
+
+    with factory() as session:
+        rows = VideoDiscoveryRepository(session).list_publishable_candidates(
+            run_id="stale-gate-run"
+        )
+        assert [item.aweme_id for item in rows] == ["current"]
+
+
+def test_candidate_evidence_is_applied_and_gate_is_recomputed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from supply_infra.services.video_discovery_service import (
+        build_evidence_values,
+        get_video_discovery_service,
+    )
+
+    factory = _expire_on_commit_session_factory()
+    _patch_service_session(monkeypatch, factory)
+    with factory() as session:
+        session.add(
+            VideoDiscoveryRun(
+                id=1,
+                run_id="evidence-apply-run",
+                demand_word="生活技巧",
+                relevant_points_json="[]",
+                status="running",
+                rule_version="test-p0",
+                rule_config_json=json.dumps(_P0_RULES, ensure_ascii=False),
+            )
+        )
+        session.add(
+            VideoDiscoveryCandidate(
+                id=2,
+                run_id="evidence-apply-run",
+                aweme_id="evidence-video",
+                title="适合家庭分享的生活技巧",
+                decision_bucket="pending_evaluation",
+            )
+        )
+        session.commit()
+
+    detail = {
+        "publish_at": datetime(2026, 7, 31, 9, 0),
+        "duration_seconds": Decimal("30.000"),
+        "share_count": 1000,
+        "like_count": 8000,
+    }
+    detail_evidence = build_evidence_values(
+        run_id="evidence-apply-run",
+        evidence_type="detail",
+        provider="crawler",
+        subject_key="evidence-video",
+        raw_response={"data": {"share_count": 1000}},
+        normalized=detail,
+        fetch_status="success",
+        aweme_id="evidence-video",
+    )
+    first = get_video_discovery_service().record_candidate_evidence(
+        run_id="evidence-apply-run",
+        evidence_values=detail_evidence,
+        aweme_id="evidence-video",
+        normalized=detail,
+    )
+    assert first["candidates"][0]["gate_status"] == "pending"
+
+    portrait = {
+        "content_50_plus_ratio": Decimal("0.280000"),
+        "content_50_plus_tgi": Decimal("130.0000"),
+    }
+    portrait_evidence = build_evidence_values(
+        run_id="evidence-apply-run",
+        evidence_type="content_portrait",
+        provider="douhot",
+        subject_key="evidence-video",
+        raw_response={"data": {"年龄": {"50-": {"percentage": 0.28}}}},
+        normalized=portrait,
+        fetch_status="success",
+        aweme_id="evidence-video",
+    )
+    second = get_video_discovery_service().record_candidate_evidence(
+        run_id="evidence-apply-run",
+        evidence_values=portrait_evidence,
+        aweme_id="evidence-video",
+        normalized=portrait,
+    )
+    candidate = second["candidates"][0]
+    assert candidate["gate_status"] == "pass"
+    assert candidate["evidence_version"] == 2
+    assert candidate["gate_evidence_version"] == 2
+
+    with factory() as session:
+        assert len(session.scalars(select(VideoDiscoveryEvidence)).all()) == 2
+        assert len(session.scalars(select(VideoDiscoveryGateEvaluation)).all()) == 2
+
+
 def test_repository_keeps_valid_items_when_same_batch_contains_gate_failure() -> None:
     factory = _expire_on_commit_session_factory()
     with factory() as session:
@@ -603,6 +785,8 @@ def test_finish_run_computes_partial_from_distinct_valid_primary_and_archives_pe
                     aweme_id="same-video",
                     decision_bucket="primary",
                     gate_status="pass",
+                    evidence_version=1,
+                    gate_evidence_version=1,
                 ),
                 VideoDiscoveryCandidate(
                     id=3,
@@ -610,6 +794,8 @@ def test_finish_run_computes_partial_from_distinct_valid_primary_and_archives_pe
                     aweme_id="same-video",
                     decision_bucket="primary",
                     gate_status="pass",
+                    evidence_version=1,
+                    gate_evidence_version=1,
                 ),
                 VideoDiscoveryCandidate(
                     id=4,
@@ -707,8 +893,11 @@ def test_new_run_lifecycle_accepts_one_portrait_side_and_finishes_partial(
     assert updated["reclassified_count"] == 0
     assert updated["candidates"][0]["decision_bucket"] == "primary"
     assert updated["candidates"][0]["gate_status"] == "pass"
-    assert updated["candidates"][0]["content_50_plus_ratio"] is None
-    assert updated["candidates"][0]["account_50_plus_ratio"] == 0.55
+    with factory() as session:
+        saved = session.get(VideoDiscoveryCandidate, 2)
+        assert saved is not None
+        assert saved.content_50_plus_ratio is None
+        assert saved.account_50_plus_ratio == Decimal("0.550000")
 
     finished = json.loads(
         video_discovery_store.update_video_discovery_run_status(
@@ -798,6 +987,76 @@ async def test_douyin_search_automatically_persists_page(
     assert persisted["provider"] == "internal_keyword"
 
 
+def test_search_persistence_keeps_complete_raw_response(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    persistence = importlib.import_module(
+        "agents.find_agent.support.search_persistence"
+    )
+    captured: dict[str, object] = {}
+
+    class FakeService:
+        def save_search_page(
+            self, run_id, search_values, candidate_rows, evidence_values
+        ):
+            captured["run_id"] = run_id
+            captured["evidence_values"] = evidence_values
+            return {
+                "search_id": 11,
+                "new_candidate_count": 1,
+                "candidates": [
+                    {
+                        "candidate_id": 21,
+                        "search_id": 11,
+                        "aweme_id": "raw-video",
+                        "title": "原始响应测试",
+                    }
+                ],
+            }
+
+    monkeypatch.setattr(
+        persistence,
+        "get_video_discovery_service",
+        lambda: FakeService(),
+    )
+    raw_payload = {
+        "results_count": 1,
+        "has_more": False,
+        "provider_extension": {"must_keep": [1, 2, 3]},
+        "_raw_response": {
+            "provider_extension": {"must_keep": [1, 2, 3]},
+            "unparsed_provider_field": "kept",
+        },
+        "search_results": [
+            {
+                "aweme_id": "raw-video",
+                "desc": "原始响应测试",
+                "statistics": {"share_count": 0},
+            }
+        ],
+    }
+    result = json.loads(
+        persistence.persist_search_payload(
+            json.dumps(raw_payload, ensure_ascii=False),
+            run_id="raw-run",
+            keyword="原始响应",
+            query_reason="验证完整落库",
+            source_type="demand",
+            cursor="0",
+            page_no=1,
+            provider="internal_keyword",
+        )
+    )
+
+    evidence_values = captured["evidence_values"]
+    assert isinstance(evidence_values, dict)
+    saved_raw = json.loads(str(evidence_values["raw_response_json"]))
+    assert saved_raw["provider_extension"] == {"must_keep": [1, 2, 3]}
+    assert saved_raw["unparsed_provider_field"] == "kept"
+    assert "_raw_response" not in result
+    assert result["candidates"][0]["candidate_id"] == 21
+
+
 @pytest.mark.asyncio
 async def test_batch_search_records_each_page_and_carries_parent(
     monkeypatch: pytest.MonkeyPatch,
@@ -1241,3 +1500,124 @@ async def test_find_agent_timeout_closes_async_client() -> None:
         )
 
     assert agent.llm._async_client.closed is True
+
+
+@pytest.mark.asyncio
+async def test_detail_candidate_mode_persists_raw_and_returns_gate_summary(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    detail_module = importlib.import_module(
+        "agents.find_agent.support.douyin_detail"
+    )
+    recorded: list[dict[str, object]] = []
+
+    class FakeService:
+        def get_candidates_by_ids(self, _run_id, _candidate_ids):
+            return [
+                {
+                    "candidate_id": 21,
+                    "aweme_id": "7641118685977614586",
+                    "gate_status": "pending" if not recorded else "pass",
+                    "gate_results": {"failed_reason_codes": []},
+                    "evidence_version": len(recorded),
+                }
+            ]
+
+        def record_candidate_evidence(self, **kwargs):
+            recorded.append(kwargs)
+            return {"evidence_id": 1, "updated_count": 1, "candidates": []}
+
+    async def fake_fetch(_client, content_id):
+        return (
+            {
+                "content_id": content_id,
+                "title": "详情自动落库",
+                "publish_at": "2026-07-31T09:00:00",
+                "duration_seconds": 60,
+                "share_count": 2000,
+            },
+            {"code": 0, "data": {"unparsed": "raw-detail"}},
+        )
+
+    service = FakeService()
+    monkeypatch.setattr(detail_module, "get_video_discovery_service", lambda: service)
+    monkeypatch.setattr(detail_module, "_fetch_one_detail", fake_fetch)
+
+    result = json.loads(
+        await detail_module.douyin_detail(
+            run_id="detail-run",
+            candidate_ids=[21],
+        )
+    )
+
+    assert len(recorded) == 1
+    evidence = recorded[0]["evidence_values"]
+    assert isinstance(evidence, dict)
+    assert "raw-detail" in str(evidence["raw_response_json"])
+    assert "details" not in result
+    assert "detail" not in result
+    assert result["candidates"][0]["candidate_id"] == 21
+
+
+@pytest.mark.asyncio
+async def test_portrait_candidate_mode_persists_both_sides_and_returns_summary(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    portrait_module = importlib.import_module(
+        "agents.find_agent.support.portrait"
+    )
+    recorded: list[dict[str, object]] = []
+
+    class FakeService:
+        def get_candidates_by_ids(self, _run_id, _candidate_ids):
+            return [
+                {
+                    "candidate_id": 31,
+                    "aweme_id": "7641118685977614586",
+                    "author_sec_uid": "MS4wLjABAAAA-test",
+                    "gate_status": "pass",
+                    "gate_results": {"failed_reason_codes": []},
+                    "evidence_version": len(recorded),
+                }
+            ]
+
+        def record_candidate_evidence(self, **kwargs):
+            recorded.append(kwargs)
+            return {"evidence_id": len(recorded), "updated_count": 1, "candidates": []}
+
+    async def fake_content(_client, _content_id, _flags):
+        portrait = {"年龄": {"50-": {"percentage": 0.3, "preference": 130}}}
+        return None, {
+            "has_portrait": True,
+            "portrait_data": portrait,
+            "output": "content",
+            "raw_data": {"raw": "content-portrait"},
+        }
+
+    async def fake_account(_client, _account_id, _flags):
+        portrait = {"年龄": {"50-": {"percentage": 0.4, "preference": 140}}}
+        return None, {
+            "has_portrait": True,
+            "portrait_data": portrait,
+            "output": "account",
+            "raw_data": {"raw": "account-portrait"},
+        }
+
+    service = FakeService()
+    monkeypatch.setattr(portrait_module, "get_video_discovery_service", lambda: service)
+    monkeypatch.setattr(portrait_module, "_fetch_content_portrait", fake_content)
+    monkeypatch.setattr(portrait_module, "_fetch_account_portrait", fake_account)
+
+    result = json.loads(
+        await portrait_module.batch_fetch_portraits(
+            run_id="portrait-run",
+            candidate_ids=[31],
+            fetch_account_portrait=True,
+        )
+    )
+
+    assert [
+        item["evidence_values"]["evidence_type"] for item in recorded
+    ] == ["content_portrait", "account_portrait"]
+    assert "results" not in result
+    assert result["candidates"][0]["candidate_id"] == 31