寻找agent优化

.env.example

@@ -63,6 +63,16 @@ SCHEDULER_TIMEZONE=Asia/Shanghai
 SCHEDULER_CRON_HOUR=15
 SCHEDULER_CRON_MINUTE=0
 
+# find_agent P0 质量门禁（每次运行会保存完整配置快照）
+FIND_AGENT_RULE_VERSION=find-agent-gate-v1
+FIND_AGENT_MIN_DURATION_SECONDS=30
+FIND_AGENT_MIN_SHARE_COUNT=1000
+FIND_AGENT_MIN_CONTENT_50_PLUS_RATIO=0.20
+FIND_AGENT_MIN_ACCOUNT_50_PLUS_RATIO=0.20
+FIND_AGENT_FESTIVAL_LEAD_DAYS=7
+FIND_AGENT_EVENT_MAX_AGE_DAYS=7
+FIND_AGENT_SEASONAL_MAX_AGE_DAYS=180
+
 # Pipeline control plane
 PROCESS_ROLE=api
 DATABASE_AUTO_CREATE=false

agents/find_agent/demand_run.py

@@ -10,6 +10,7 @@ from datetime import datetime
 from typing import Any
 from zoneinfo import ZoneInfo
 
+from supply_infra.video_discovery_gates import build_rule_snapshot
 from supply_agent.types import AgentResult
 from supply_infra.config import get_infra_settings
 from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
@@ -273,6 +274,7 @@ def prepare_video_discovery_run(
 ) -> tuple[str | None, str | None]:
     """执行 Agent 前预创建 video_discovery_run，返回 (run_id, skip_reason)。"""
     payload = build_run_input_payload(ctx)
+    rule_snapshot = build_rule_snapshot()
     primary = ctx.primary_video
     values = {
         "run_id": uuid.uuid4().hex,
@@ -285,6 +287,8 @@ def prepare_video_discovery_run(
         "intent_summary": None,
         "status": "running",
         "stop_reason": None,
+        "rule_version": rule_snapshot["rule_version"],
+        "rule_config_json": json.dumps(rule_snapshot, ensure_ascii=False),
     }
     run_id, skip_reason = get_video_discovery_service().prepare_scheduled_run(
         biz_dt=ctx.biz_dt,
@@ -338,14 +342,24 @@ def _flatten_relevant_points(ctx: FindDemandContext) -> list[dict[str, Any]]:
     return relevant_points
 
 
-def build_find_agent_user_input(ctx: FindDemandContext, run_id: str) -> str:
+def build_find_agent_user_input(
+    ctx: FindDemandContext,
+    run_id: str,
+    *,
+    rule_snapshot: dict[str, Any] | None = None,
+) -> str:
     """构建传给 find_agent 的用户消息。"""
     videos_payload = [_serialize_video(video) for video in ctx.videos]
+    active_rules = rule_snapshot or build_rule_snapshot()
 
     return (
         f"run_id：{run_id}\n"
         f"demand_grade_id：{ctx.demand_grade_id}\n"
         f"demand_word：{ctx.demand_name}\n"
+        f"current_datetime：{active_rules['current_datetime']}\n"
+        f"current_date：{active_rules['current_date']}\n"
+        f"timezone：{active_rules['timezone']}\n"
+        f"quality_gate_rules：{json.dumps(active_rules, ensure_ascii=False)}\n"
         f"reference_videos：{json.dumps(videos_payload, ensure_ascii=False)}\n"
         "说明：video_discovery_run 已由系统预创建，无需也不得由模型再次创建。\n"
         "请直接依据 reference_videos 中各视频的 title 和 points 理解需求并开始搜索；"
@@ -374,7 +388,12 @@ def discover_videos_for_demand(
     if not run_id:
         raise RuntimeError("prepare_video_discovery_run 未返回 run_id")
 
-    user_input = build_find_agent_user_input(ctx, run_id)
+    run_snapshot = get_video_discovery_service().lookup_run(run_id) or {}
+    user_input = build_find_agent_user_input(
+        ctx,
+        run_id,
+        rule_snapshot=run_snapshot.get("rule_config"),
+    )
     try:
         from agents.find_agent.run_outcome import evaluate_find_agent_run
 

agents/find_agent/prompt/system_prompt.md

@@ -4,6 +4,8 @@
 
 - `run_id`：系统预创建的视频发现运行 ID，所有存储工具必须复用它；
 - `demand_grade_id` 和 `demand_word`：需求记录 ID 与需求词，定义这次寻找的真实意图边界；
+- `current_datetime / current_date / timezone`：本次判断使用的当前时间上下文；
+- `quality_gate_rules`：本次运行固化的时长、分享数、50+ 占比和时效规则快照；
 - `reference_videos`：全部参考视频。每项包含 `video_id`、`title` 和该视频对应的
   `points`。
 
@@ -25,9 +27,12 @@
 对候选视频 `v`，定义三个彼此独立的命题：
 
 - `R(v)`（需求相关性）：视频是否满足 `demand_word` 在标题和相关点所限定的具体意图；
-- `E(v)`（老年受众倾向）：视频点赞用户画像与作者粉丝画像是否支持受众偏向较高年龄段；
+- `E(v)`（内容侧老年受众倾向）：视频点赞用户画像中的 50+ 占比是否支持受众偏老；
 - `S(v)`（分享价值）：视频是否已经表现出值得转发的行为信号和内容理由。
 
+作者粉丝画像单独记为账号先验 `A(v)`，不得与 `E(v)` 相加或平均。`A(v)` 可以增强或
+降低结论置信度，但不能替代视频侧画像，也不能使视频侧 50+ 门槛失败的候选通过。
+
 最终寻找的是联合事件：
 
 `G(v) = R(v) ∩ E(v) ∩ S(v)`
@@ -40,6 +45,24 @@
 `pending_evaluation` 只是搜索召回后的过程状态，不是最终等级。不存在 `backup`、
 补充推荐或人工备选等级。
 
+# P0 程序硬门槛
+
+用户消息中的 `quality_gate_rules` 是本次运行唯一有效的阈值快照。候选进入 `primary`
+前必须同时满足：
+
+1. 真实发布时间可用，且基础日期、节日、时段和内容新鲜度判断为有效；
+2. `duration_seconds >= min_duration_seconds`；
+3. 最新详情中的 `share_count >= min_share_count`；
+4. 视频点赞用户画像中的
+   `content_50_plus_ratio >= min_content_50_plus_ratio`。
+
+搜索工具会过滤已知时长不足 30 秒的结果；未知时长仍可能进入待补证候选，但必须通过
+`douyin_detail` 补齐后才能保存为 `primary`。分享数、发布时间或视频侧 50+ 占比缺失
+均属于证据不足，不能进入 `primary`。
+
+`batch_update_video_discovery_candidates` 会在保存层重新执行硬门槛。不得通过提高
+`R/E/S/V` 分数、使用账号画像或在理由中声称“综合表现较好”绕过门槛。
+
 # 决策公理与定理
 
 ## 1. 需求闸门公理
@@ -94,13 +117,17 @@
    适配特征；
 4. 题材或作者形象带来的直觉。
 
-第 4 层不得单独形成老年倾向结论。视频画像代表“这条内容吸引了谁”，作者画像代表“这个账号通常触达谁”；前者是直接内容证据，后者是账号先验。两者一致时增强置信度；冲突时优先视频画像并显式降置信度，不得静默平均。
+第 4 层不得单独形成老年倾向结论。视频画像代表“这条内容吸引了谁”，作者画像代表
+“这个账号通常触达谁”；两侧必须分别给出 `content_portrait_status` 和
+`account_portrait_status`。两者一致时增强置信度；冲突时保留 `portrait_conflict`，
+但始终以视频侧门槛为准，不得静默平均。
 
 年龄桶中，明确覆盖 `50岁及以上` 的桶才是直接老年信号；接口实际可能用 `50-`
 表示“50岁以上”，必须用 `normalize_age_portraits` 标准化后再判断。只提供
 `40岁及以上` 时只能称为“成熟人群代理信号”。同时观察占比和偏好度/TGI：占比回答
 “人多不多”，偏好度回答“相对平台基线是否更偏爱”。若接口没有给出中性基线，不得
-臆造阈值。
+臆造阈值。`40～49` 或 `41～50` 只能作为成熟人群辅助信息，不得计入
+`content_50_plus_ratio`。
 
 ## 6. 相对传播定理
 
@@ -120,10 +147,10 @@
 
 这意味着任一维度接近零，整体价值都会被明显压低。评分用于保持排序一致，不得制造虚假精确性；**数据库保存与最终报告中的 `R/E/S/V` 均使用 `0~1` 小数，不做百分制换算**。
 
-`R/E/S/V` 是帮助你保持判断一致的参考量，**不是程序校验线**。候选最终进入
-`primary / rejected` 完全由你根据全部证据判断。`batch_update_video_discovery_candidates`
-不会重算、换算、校验分数范围或改写 `decision_bucket`；存储时 `R/E/S` 最多保留
-6 位小数，`V` 最多保留 2 位小数。因此必须自行确保所有分数都在 `0~1` 内。
+`R/E/S/V` 用于语义判断和排序，不能替代程序硬门槛。候选是否语义相关仍由你判断；
+当你请求保存为 `primary` 时，保存层会校验时效、时长、分享数和视频侧 50+ 占比。
+`R/E/S` 最多保留 6 位小数，`V` 最多保留 2 位小数，必须自行确保所有分数在 `0~1`
+内。
 
 低相关候选即使原始分享规模、分享效率或老年倾向很强，也不能进入 `primary`。
 
@@ -150,7 +177,8 @@
   搜索 1～2 页；工具在每页返回后创建搜索记录并把本页结果逐条写入候选表。优先用它完成
   2～3 个根搜索。每次搜索和每条候选都会生成新的数据库记录，返回结果中的
   `search_id / candidate_id` 是后续更新依据；同一 `aweme_id` 在不同搜索中对应不同
-  `candidate_id`。
+  `candidate_id`。所有搜索源的客户端最短时长固定不低于 30 秒；对于强时效需求，
+  必须依据 `current_datetime` 选择“一天内/一周内/半年内”等发布时间参数。
 - `douyin_search`：用于单次内部关键词搜索和特殊场景回退。必须传入 `run_id`、搜索
   原因和来源；工具返回前自动保存本页搜索轨迹及候选。
 - `douyin_search_tikhub`：独立的 TikHub 搜索来源，返回标签、更多互动字段和
@@ -164,7 +192,9 @@
   潜力时，按最热或最新扩展作者作品。作者作品属于 `author` 搜索分支，仍需逐条判断
   相关性、老年倾向和分享价值，不能因作者优秀就直接推荐。工具会自动保存作者搜索页
   和候选。
-- `douyin_detail`：用于核验候选的最新互动数据、作者、页面链接和标题/描述等文本证据。
+- `douyin_detail`：用于核验候选的真实发布时间、时长、最新分享数与其他互动数据，
+  以及作者、页面链接和标题/描述等文本证据。最终候选更新必须把
+  `publish_at / duration_seconds / share_count` 一并写回。
 - `batch_fetch_portraits`：用于批量取得视频点赞画像；对正式候选应设置
   `fetch_account_portrait=true`，同时取得作者粉丝画像。批量结果会自动附带
   `age_normalization`，无需为同一候选再单独调用标准化工具。
@@ -175,9 +205,12 @@
   不要重复调用。标准化结果中的 `elder_score_cap` 是当前证据条件下 `E` 的上限，
   `elder_score` 不得超过该值。
 - `batch_update_video_discovery_candidates`：严格按搜索结果返回的 `candidate_id`
-  更新详情、证据、**0~1 的 R/E/S/V 评分**和 `decision_bucket`。工具只接受
-  `primary / rejected`，不会新增候选、重算分数、修改搜索记录或修改运行状态。
-  `decision_reason` 必须用一段综合理由覆盖相关性、年龄证据、分享价值和主要限制。
+  更新详情、证据、**0~1 的 R/E/S/V 评分**和 `decision_bucket`。更新正式候选时必须
+  提供最新的 `publish_at / duration_seconds / share_count`、双侧画像及
+  `age_normalization`。工具会从标准化结果拆出视频侧和账号侧 50+ 指标，并在保存
+  `primary` 前执行程序硬门槛；失败时根据返回的原因码补证或改为 `rejected`。工具不会
+  新增候选、修改搜索记录或修改运行状态。`decision_reason` 必须分别覆盖相关性、视频
+  侧 50+、账号侧画像、分享价值、时间有效性和主要限制。
   同一 `aweme_id` 对应多个 `candidate_id` 时必须分别判断和更新。
 - `update_video_discovery_run_status`：本轮搜索和评估流程结束后，单独把运行状态更新为
   `finished`，并保存意图摘要和停止原因。不得用它代替候选更新。
@@ -210,7 +243,8 @@
 
 - 根搜索默认形成 2～3 个语义不同的词；每个生产性首页最多继续 1 页，除非第二页仍能
   显著提升判断质量；
-- 搜索结果先按需求相关性、分享规模/效率和主推荐潜力做廉价预筛，然后进入 `douyin_detail` 和双画像阶段；
+- 搜索结果先按需求相关性、已知时长、分享规模/效率、时间有效性和主推荐潜力做廉价
+  预筛，然后进入 `douyin_detail` 和双画像阶段；
 - 内容相关性与分享动机主要依据标题、描述、`topic_list`、话题标签和详情字段判断
 - 执行新搜索、翻页、详情或画像后，应重新保存受影响候选；
 
@@ -234,7 +268,8 @@
 简要说明对需求意图的理解，再报告本次形成的主推荐及其主要证据。按需概括主要淘汰原因、
 缺失数据、画像冲突或接口失败，不要求逐条列出 `rejected` 候选。
 
-决定均由 Agent 作出。程序不会根据阈值重新解释或修改你的最终推荐。
+语义相关性、分享动机和排序由 Agent 判断；时效、时长、分享数和视频侧 50+ 占比由
+程序按本次规则快照做不可绕过的准入校验。
 
 禁止输出没有证据支撑的年龄结论，禁止把“内容讲老人”写成“观看者是老人”，禁止推荐
 低相关视频。

agents/find_agent/support/douyin_detail.py

@@ -14,6 +14,7 @@ from typing import Any, Optional
 
 import httpx
 
+from supply_infra.video_discovery_gates import parse_datetime_value
 
 logger = logging.getLogger(__name__)
 
@@ -53,6 +54,11 @@ _KEEP_FIELDS = (
     "comment_count",
     "share_count",
     "looking_count",
+    "create_time",
+    "create_timestamp",
+    "publish_at",
+    "publish_time",
+    "publish_timestamp",
     "modify_timestamp",
     "update_timestamp",
 )
@@ -136,9 +142,9 @@ def _build_detail_result(detail: dict[str, Any], content_id: str) -> dict[str, A
         "content_id": content_id,
         "video_url": _extract_video_url(detail),
         "video_duration": _extract_video_duration(detail),
+        "duration_seconds": _extract_video_duration(detail),
         "cover_url": _extract_cover_url(detail),
     }
-
     for key in _KEEP_FIELDS:
         if key not in detail:
             continue
@@ -152,6 +158,16 @@ def _build_detail_result(detail: dict[str, Any], content_id: str) -> dict[str, A
         else:
             result[key] = value
 
+    publish_at = parse_datetime_value(
+        detail.get("publish_at")
+        or detail.get("publish_time")
+        or detail.get("publish_timestamp")
+        or detail.get("create_time")
+        or detail.get("create_timestamp")
+    )
+    result["publish_at"] = (
+        publish_at.isoformat(timespec="seconds") if publish_at is not None else None
+    )
     return result
 
 

agents/find_agent/support/douyin_search.py

@@ -25,6 +25,34 @@ _last_request_monotonic: float = 0.0
 DOUYIN_SEARCH_API = "http://crawapi.piaoquantv.com/crawler/dou_yin/keyword"
 DEFAULT_TIMEOUT = 60.0
 DOUYIN_ACCOUNT_ID = "771431222"
+DEFAULT_MIN_DURATION_SECONDS = 30
+
+
+def _safe_int(value: Any, default: int = 0) -> int:
+    if isinstance(value, bool) or value is None:
+        return default
+    try:
+        return int(float(str(value).strip()))
+    except (TypeError, ValueError):
+        return default
+
+
+def _extract_duration_ms(item: dict[str, Any]) -> int:
+    video = item.get("video") if isinstance(item.get("video"), dict) else {}
+    return _safe_int(
+        item.get("duration_ms")
+        or video.get("duration")
+        or item.get("duration")
+    )
+
+
+def _extract_publish_at(item: dict[str, Any]) -> Any:
+    return (
+        item.get("publish_at")
+        or item.get("create_time")
+        or item.get("create_timestamp")
+        or item.get("publish_timestamp")
+    )
 
 
 def _build_search_results(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
@@ -48,6 +76,8 @@ def _build_search_results(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
                     "comment_count": stats.get("comment_count", 0),
                     "share_count": stats.get("share_count", 0),
                 },
+                "duration_ms": _extract_duration_ms(item),
+                "publish_at": _extract_publish_at(item),
             }
         )
     return results
@@ -98,6 +128,7 @@ def _success_result(
     has_more: bool,
     cursor_value: str,
     duration_ms: int,
+    filtered_count: int = 0,
 ) -> str:
     """构建成功时的 JSON 字符串返回值。"""
     search_results = _build_search_results(items)
@@ -106,6 +137,7 @@ def _success_result(
         "output": _build_output_summary(keyword, items, has_more, cursor_value),
         "keyword": keyword,
         "results_count": len(items),
+        "filtered_count": filtered_count,
         "has_more": has_more,
         "next_cursor": cursor_value,
         "search_results": search_results,
@@ -126,6 +158,7 @@ async def _douyin_search_raw(
     publish_time: str = "不限",
     cursor: str = "0",
     account_id: str = DOUYIN_ACCOUNT_ID,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: Optional[float] = None,
 ) -> str:
     """
@@ -140,6 +173,7 @@ async def _douyin_search_raw(
         publish_time: 发布时间范围（可选：不限/一天内/一周内/半年内, 默认 "不限"）
         cursor: 分页游标，用于获取下一页结果，默认 "0"
         account_id: 账号ID（可选）
+        min_duration_seconds: 客户端最短时长过滤，固定不低于 30 秒。
         timeout: 超时时间（秒），默认 60
 
     Returns:
@@ -179,6 +213,19 @@ async def _douyin_search_raw(
 
         data_block = data.get("data", {}) if isinstance(data.get("data"), dict) else {}
         items = data_block.get("data", []) if isinstance(data_block.get("data"), list) else []
+        minimum_ms = max(
+            DEFAULT_MIN_DURATION_SECONDS,
+            int(min_duration_seconds or DEFAULT_MIN_DURATION_SECONDS),
+        ) * 1000
+        filtered_count = 0
+        filtered_items: list[dict[str, Any]] = []
+        for item in items:
+            duration = _extract_duration_ms(item)
+            if duration and duration < minimum_ms:
+                filtered_count += 1
+                continue
+            filtered_items.append(item)
+        items = filtered_items
         has_more = bool(data_block.get("has_more", False))
         cursor_value = str(data_block.get("next_cursor", ""))
 
@@ -191,7 +238,15 @@ async def _douyin_search_raw(
             duration_ms,
         )
 
-        return _success_result(keyword, data, items, has_more, cursor_value, duration_ms)
+        return _success_result(
+            keyword,
+            data,
+            items,
+            has_more,
+            cursor_value,
+            duration_ms,
+            filtered_count,
+        )
 
     except httpx.HTTPStatusError as e:
         logger.error(
@@ -224,6 +279,7 @@ async def douyin_search(
     publish_time: str = "不限",
     cursor: str = "0",
     account_id: str = DOUYIN_ACCOUNT_ID,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: Optional[float] = None,
 ) -> str:
     """
@@ -236,6 +292,7 @@ async def douyin_search(
         publish_time=publish_time,
         cursor=cursor,
         account_id=account_id,
+        min_duration_seconds=min_duration_seconds,
         timeout=timeout,
     )
     return await asyncio.to_thread(

agents/find_agent/support/douyin_search_tikhub.py

@@ -20,6 +20,7 @@ DOUYIN_SEARCH_TIKHUB_API = (
     "https://api.tikhub.io/api/v1/douyin/search/fetch_video_search_v2"
 )
 DEFAULT_TIMEOUT = 60.0
+DEFAULT_MIN_DURATION_SECONDS = 30
 _MIN_REQUEST_INTERVAL_SECONDS = 1.0
 _rate_limit_lock = asyncio.Lock()
 _last_request_monotonic = 0.0
@@ -161,6 +162,11 @@ def _normalize_aweme(aweme: dict[str, Any]) -> dict[str, Any] | None:
             "play_count": _safe_int(stats.get("play_count")),
         },
         "duration_ms": _safe_int(aweme.get("duration")),
+        "publish_at": (
+            aweme.get("create_time")
+            or aweme.get("create_timestamp")
+            or aweme.get("publish_timestamp")
+        ),
         "topics": _extract_topics(aweme),
     }
 
@@ -236,7 +242,7 @@ async def _douyin_search_tikhub_raw(
     filter_duration: str = "不限",
     search_id: str = "",
     backtrace: str = "",
-    min_duration_seconds: int = 0,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: float | None = None,
 ) -> str:
     """
@@ -254,7 +260,7 @@ async def _douyin_search_tikhub_raw(
         filter_duration: 不限 / 一分钟内 / 1-5分钟 / 5分钟以上。
         search_id: 翻页状态，必须使用同一搜索返回值。
         backtrace: 翻页回溯状态，必须使用同一搜索返回值。
-        min_duration_seconds: 客户端最短时长过滤，默认 0 表示不过滤。
+        min_duration_seconds: 客户端最短时长过滤，固定不低于 30 秒。
         timeout: 请求超时秒数，默认 60。
 
     Returns:
@@ -330,7 +336,10 @@ async def _douyin_search_tikhub_raw(
         results: list[dict[str, Any]] = []
         seen: set[str] = set()
         filtered_count = 0
-        minimum_ms = max(0, int(min_duration_seconds)) * 1000
+        minimum_ms = max(
+            DEFAULT_MIN_DURATION_SECONDS,
+            int(min_duration_seconds or DEFAULT_MIN_DURATION_SECONDS),
+        ) * 1000
         for raw_item in business_data:
             normalized = _normalize_aweme(_get_aweme_info(raw_item))
             if normalized is None or normalized["aweme_id"] in seen:
@@ -419,7 +428,7 @@ async def douyin_search_tikhub(
     filter_duration: str = "不限",
     search_id: str = "",
     backtrace: str = "",
-    min_duration_seconds: int = 0,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: float | None = None,
 ) -> str:
     """

agents/find_agent/support/douyin_user_videos.py

@@ -15,6 +15,7 @@ logger = logging.getLogger(__name__)
 
 DOUYIN_USER_VIDEOS_API = "http://crawapi.piaoquantv.com/crawler/dou_yin/blogger"
 DEFAULT_TIMEOUT = 60.0
+DEFAULT_MIN_DURATION_SECONDS = 30
 _MIN_REQUEST_INTERVAL_SECONDS = 10.1
 _rate_limit_lock = asyncio.Lock()
 _last_request_monotonic = 0.0
@@ -81,6 +82,11 @@ def _normalize_video(item: dict[str, Any]) -> dict[str, Any] | None:
             "play_count": _safe_int(stats.get("play_count")),
         },
         "duration_ms": duration_ms,
+        "publish_at": (
+            item.get("create_time")
+            or item.get("create_timestamp")
+            or item.get("publish_timestamp")
+        ),
         "topics": _extract_topics(item),
     }
 
@@ -142,7 +148,7 @@ async def _douyin_user_videos_raw(
     account_id: str,
     sort_type: str = "最热",
     cursor: str = "",
-    min_duration_seconds: int = 0,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: float | None = None,
 ) -> str:
     """
@@ -155,7 +161,7 @@ async def _douyin_user_videos_raw(
         account_id: author.sec_uid，必须使用完整值。
         sort_type: 最热 / 最新，默认最热。
         cursor: 首次为空；翻页使用上次返回的 next_cursor。
-        min_duration_seconds: 最短时长过滤，默认 0 表示不过滤。
+        min_duration_seconds: 最短时长过滤，固定不低于 30 秒。
         timeout: 请求超时秒数，默认 60。
 
     Returns:
@@ -189,7 +195,10 @@ async def _douyin_user_videos_raw(
 
         data = body.get("data") if isinstance(body.get("data"), dict) else {}
         items = data.get("data") if isinstance(data.get("data"), list) else []
-        minimum_ms = max(0, int(min_duration_seconds)) * 1000
+        minimum_ms = max(
+            DEFAULT_MIN_DURATION_SECONDS,
+            int(min_duration_seconds or DEFAULT_MIN_DURATION_SECONDS),
+        ) * 1000
         filtered_count = 0
         seen: set[str] = set()
         results: list[dict[str, Any]] = []
@@ -272,7 +281,7 @@ async def douyin_user_videos(
     page_no: int = 1,
     sort_type: str = "最热",
     cursor: str = "",
-    min_duration_seconds: int = 0,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: float | None = None,
 ) -> str:
     """

agents/find_agent/support/search_persistence.py

@@ -6,6 +6,10 @@ import json
 import logging
 from typing import Any
 
+from supply_infra.video_discovery_gates import (
+    normalize_duration_seconds,
+    parse_datetime_value,
+)
 from supply_infra.services.video_discovery_service import (
     RunNotFoundError,
     format_db_error,
@@ -84,6 +88,22 @@ def _candidate_from_search_result(
     author = item.get("author") if isinstance(item.get("author"), dict) else {}
     stats = item.get("statistics") if isinstance(item.get("statistics"), dict) else {}
     topics = item.get("topics") if isinstance(item.get("topics"), list) else []
+    duration = normalize_duration_seconds(
+        item.get("duration_seconds")
+        if item.get("duration_seconds") not in (None, "")
+        else item.get("duration_ms"),
+        unit=(
+            "seconds"
+            if item.get("duration_seconds") not in (None, "")
+            else "milliseconds"
+        ),
+    )
+    publish_at = parse_datetime_value(
+        item.get("publish_at")
+        or item.get("create_time")
+        or item.get("create_timestamp")
+        or item.get("publish_timestamp")
+    )
     return {
         "aweme_id": aweme_id,
         "title": _clean_text(item.get("desc") or item.get("title"), max_length=512),
@@ -114,6 +134,10 @@ def _candidate_from_search_result(
         "play_count": _nonnegative_int(
             stats.get("play_count") or item.get("play_count")
         ),
+        "duration_seconds": duration,
+        "publish_at": (
+            publish_at.replace(tzinfo=None) if publish_at is not None else None
+        ),
         "tags_json": (
             json.dumps(topics, ensure_ascii=False) if topics else None
         ),

agents/find_agent/support/video_discovery.py

@@ -7,6 +7,11 @@ import uuid
 from decimal import Decimal
 from typing import Any
 
+from supply_infra.video_discovery_gates import (
+    build_rule_snapshot,
+    normalize_duration_seconds,
+    parse_datetime_value,
+)
 from supply_infra.services.video_discovery_service import (
     RunNotFoundError,
     format_db_error,
@@ -56,6 +61,23 @@ def _optional_decimal(value: Any, places: int) -> Decimal | None:
     return number.quantize(quantum)
 
 
+def _optional_ratio_decimal(value: Any, places: int = 6) -> Decimal | None:
+    number = _optional_decimal(value, places)
+    if number is not None and not Decimal("0") <= number <= Decimal("1"):
+        raise ValueError("比例字段必须在 0~1 范围内")
+    return number
+
+
+def _age_side(
+    age_normalization: Any,
+    side: str,
+) -> dict[str, Any]:
+    if not isinstance(age_normalization, dict):
+        return {}
+    value = age_normalization.get(side)
+    return value if isinstance(value, dict) else {}
+
+
 def create_video_discovery_run(
     demand_word: str,
     relevant_points: list[dict[str, Any]],
@@ -107,6 +129,7 @@ def create_video_discovery_run(
         return _input_error("demand_word 不能为空")
 
     new_run_id = cleaned_run_id or uuid.uuid4().hex
+    rule_snapshot = build_rule_snapshot()
     values = {
         "run_id": new_run_id,
         "demand_grade_id": demand_grade_id,
@@ -116,6 +139,8 @@ def create_video_discovery_run(
         "relevant_points_json": _json(relevant_points or []),
         "intent_summary": _clean_text(intent_summary),
         "status": "running",
+        "rule_version": rule_snapshot["rule_version"],
+        "rule_config_json": _json(rule_snapshot),
     }
     try:
         created = service.create_run(values)
@@ -143,6 +168,8 @@ def _normalize_candidate_update(item: dict[str, Any]) -> dict[str, Any]:
     content_age = item.get("content_age_evidence")
     account_age = item.get("account_age_evidence")
     age_normalization = item.get("age_normalization")
+    content_normalization = _age_side(age_normalization, "content")
+    account_normalization = _age_side(age_normalization, "account")
     decision_bucket = (
         _clean_text(item.get("decision_bucket"), max_length=24)
         or ""
@@ -152,6 +179,29 @@ def _normalize_candidate_update(item: dict[str, Any]) -> dict[str, Any]:
             "decision_bucket 必须是 primary 或 rejected"
         )
 
+    publish_at = parse_datetime_value(item.get("publish_at"))
+    duration = normalize_duration_seconds(
+        item.get("duration_seconds")
+        if item.get("duration_seconds") not in (None, "")
+        else item.get("video_duration")
+    )
+    content_ratio = item.get(
+        "content_50_plus_ratio",
+        content_normalization.get("older_ratio"),
+    )
+    content_tgi = item.get(
+        "content_50_plus_tgi",
+        content_normalization.get("older_tgi"),
+    )
+    account_ratio = item.get(
+        "account_50_plus_ratio",
+        account_normalization.get("older_ratio"),
+    )
+    account_tgi = item.get(
+        "account_50_plus_tgi",
+        account_normalization.get("older_tgi"),
+    )
+    temporal_evidence = item.get("temporal_evidence")
     mapping = {
         "candidate_id": candidate_id,
         "title": _clean_text(item.get("title"), max_length=512),
@@ -159,6 +209,10 @@ def _normalize_candidate_update(item: dict[str, Any]) -> dict[str, Any]:
         "author_name": _clean_text(item.get("author_name"), max_length=256),
         "author_sec_uid": _clean_text(item.get("author_sec_uid"), max_length=256),
         "tags_json": item.get("tags") if "tags" in item else None,
+        "publish_at": (
+            publish_at.replace(tzinfo=None) if publish_at is not None else None
+        ),
+        "duration_seconds": duration,
         "play_count": _nonnegative_int(item.get("play_count")),
         "like_count": _nonnegative_int(item.get("like_count")),
         "comment_count": _nonnegative_int(item.get("comment_count")),
@@ -173,6 +227,19 @@ def _normalize_candidate_update(item: dict[str, Any]) -> dict[str, Any]:
         "age_normalization_json": (
             _json(age_normalization) if age_normalization is not None else None
         ),
+        "content_50_plus_ratio": _optional_ratio_decimal(content_ratio),
+        "content_50_plus_tgi": _optional_decimal(content_tgi, 4),
+        "account_50_plus_ratio": _optional_ratio_decimal(account_ratio),
+        "account_50_plus_tgi": _optional_decimal(account_tgi, 4),
+        "temporal_type": _clean_text(item.get("temporal_type"), max_length=24),
+        "temporal_status": _clean_text(item.get("temporal_status"), max_length=16),
+        "temporal_evidence_json": (
+            _json(temporal_evidence) if temporal_evidence is not None else None
+        ),
+        "reject_reason_code": _clean_text(
+            item.get("reject_reason_code"),
+            max_length=64,
+        ),
         "relevance_score": _optional_decimal(item.get("relevance_score"), 6),
         "elder_score": _optional_decimal(item.get("elder_score"), 6),
         "share_score": _optional_decimal(item.get("share_score"), 6),
@@ -195,8 +262,9 @@ def batch_update_video_discovery_candidates(
     Args:
         run_id: 发现运行 id。
         items: 候选数组。每项必须包含搜索工具返回的 candidate_id，并直接提供
-            decision_bucket；可更新标题、链接、作者、互动量、标签、双侧年龄证据、
-            画像标准化结果、R/E/S/V 和最终分池理由。
+            decision_bucket；可更新标题、链接、作者、发布时间、时长、互动量、标签、
+            双侧年龄证据、画像标准化结果、时间判断、R/E/S/V 和最终分池理由。
+            primary 会在数据库事务内强制校验本次运行的 P0 规则快照。
 
     Returns:
         JSON，包含 updated_count 和按 candidate_id 更新后的候选记录。

agents/find_agent/tools/batch_search_and_record.py

@@ -12,7 +12,10 @@ from agents.find_agent.support.batch_search_and_record import (
     _positive_page_limit,
 )
 from agents.find_agent.support.douyin_search import douyin_search
-from agents.find_agent.support.douyin_search_tikhub import douyin_search_tikhub
+from agents.find_agent.support.douyin_search_tikhub import (
+    DEFAULT_MIN_DURATION_SECONDS,
+    douyin_search_tikhub,
+)
 from supply_agent.tools import tool
 
 
@@ -29,7 +32,8 @@ async def batch_search_and_record(
         searches: 搜索任务数组，最多 6 个。每项必须包含 keyword、query_reason、
             source_type；可包含 provider（internal_keyword / tikhub）、max_pages
             （1~2）、source_value、parent_search_id、content_type、sort_type、
-            publish_time、filter_duration、min_duration_seconds。
+            publish_time、filter_duration、min_duration_seconds。所有搜索源实际执行的
+            min_duration_seconds 固定不低于 30。
     """
     run_text = str(run_id or "").strip()
     if not run_text:
@@ -108,13 +112,18 @@ async def batch_search_and_record(
                     search_id=provider_search_id,
                     backtrace=backtrace,
                     min_duration_seconds=int(
-                        raw_task.get("min_duration_seconds") or 0
+                        raw_task.get("min_duration_seconds")
+                        or DEFAULT_MIN_DURATION_SECONDS
                     ),
                 )
             else:
                 raw_result = await douyin_search(
                     **common,
                     cursor=str(cursor or "0"),
+                    min_duration_seconds=int(
+                        raw_task.get("min_duration_seconds")
+                        or DEFAULT_MIN_DURATION_SECONDS
+                    ),
                 )
 
             result = _load_result(raw_result)

agents/find_agent/tools/douyin_search.py

@@ -7,6 +7,7 @@ from typing import Optional
 import httpx as httpx
 
 from agents.find_agent.support.douyin_search import (
+    DEFAULT_MIN_DURATION_SECONDS,
     DOUYIN_ACCOUNT_ID,
     _build_search_results as _build_search_results,
     _douyin_search_raw,
@@ -31,6 +32,7 @@ async def douyin_search(
     publish_time: str = "不限",
     cursor: str = "0",
     account_id: str = DOUYIN_ACCOUNT_ID,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: Optional[float] = None,
 ) -> str:
     """搜索一页抖音视频，创建搜索记录和候选记录，返回基础信息及数据库 ID。"""
@@ -41,6 +43,7 @@ async def douyin_search(
         publish_time=publish_time,
         cursor=cursor,
         account_id=account_id,
+        min_duration_seconds=min_duration_seconds,
         timeout=timeout,
     )
     return await asyncio.to_thread(

agents/find_agent/tools/douyin_search_tikhub.py

@@ -7,6 +7,7 @@ import os as os
 import httpx as httpx
 
 from agents.find_agent.support.douyin_search_tikhub import (
+    DEFAULT_MIN_DURATION_SECONDS,
     _douyin_search_tikhub_raw,
     _ensure_env_loaded as _ensure_env_loaded,
     _wait_rate_limit as _wait_rate_limit,
@@ -31,7 +32,7 @@ async def douyin_search_tikhub(
     filter_duration: str = "不限",
     search_id: str = "",
     backtrace: str = "",
-    min_duration_seconds: int = 0,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: float | None = None,
 ) -> str:
     """使用 TikHub 搜索一页抖音视频，返回基础信息、数据库 ID 和分页状态。"""

agents/find_agent/tools/douyin_user_videos.py

@@ -6,6 +6,7 @@ import asyncio
 import httpx as httpx
 
 from agents.find_agent.support.douyin_user_videos import (
+    DEFAULT_MIN_DURATION_SECONDS,
     _douyin_user_videos_raw,
     _wait_rate_limit as _wait_rate_limit,
 )
@@ -23,7 +24,7 @@ async def douyin_user_videos(
     page_no: int = 1,
     sort_type: str = "最热",
     cursor: str = "",
-    min_duration_seconds: int = 0,
+    min_duration_seconds: int = DEFAULT_MIN_DURATION_SECONDS,
     timeout: float | None = None,
 ) -> str:
     """获取一页作者作品，创建搜索记录和候选记录并返回对应数据库 ID。"""

agents/find_agent/优化迭代方案-v3.md

@@ -0,0 +1,561 @@
+# find_agent 优化迭代方案
+
+> 文档版本：v3.0  
+> 文档状态：待评审  
+> 编制日期：2026-07-30  
+> 适用范围：`agents/find_agent` 视频搜索、补证、筛选、分池与结果输出  
+> 迭代目标：解决本轮寻找视频过程中暴露的时效、时长、内容理解、画像混用和准入规则不明确问题
+
+## 1. 背景
+
+当前 `find_agent` 已具备关键词搜索、详情获取、视频点赞用户画像、作者粉丝画像、年龄
+标准化和候选分池能力，但候选是否进入 `primary` 仍主要由 Prompt 驱动，程序没有对关键
+条件做强校验。
+
+本轮结果暴露出以下问题：
+
+1. 没有判断内容在当前日期和当前时段是否仍然有效。例如中午仍保留“早上好”内容，
+   与当前日期不匹配的节日内容也可能被搜索或推荐。
+2. 搜索召回和最终分池没有统一执行最短时长规则，小于 30 秒的视频仍可能保留。
+3. 当前明确禁用视频理解，只能依赖标题、描述、标签和互动数据，容易出现标题符合但
+   实际视频内容不符合需求的情况。
+4. 视频点赞用户画像和账号粉丝画像虽然分别获取，但最终仍被压缩为同一个老年倾向
+   判断，结果不够清晰。
+5. 年龄判断混入了 40～50 岁代理信号，没有把 `50岁及以上占比` 作为核心准入指标。
+6. 分享数、50+ 占比等关键内容指标只有软判断，没有程序级强制门槛。
+
+## 2. 迭代目标
+
+本次迭代将候选判断从“模型综合打分后自由分池”调整为“硬门槛过滤后再综合排序”。
+
+目标结果：
+
+- 与当前日期、节日和时段不一致的内容不能进入主推荐；
+- 时长小于 30 秒的视频不能进入主推荐；
+- 主推荐视频必须完成一次轻量视频内容核验；
+- 视频画像与账号画像独立展示、独立判断，不做静默平均；
+- 老年性判断以视频侧 `50岁及以上占比` 为核心；
+- 分享数和视频侧 50+ 占比由代码强校验；
+- 所有淘汰结果都有明确、可检索的原因码；
+- 阈值可以按配置调整，并能通过历史样本回测。
+
+本次不解决：
+
+- 将点赞用户画像等同于真实转发用户画像；
+- 预测视频未来一定会爆发；
+- 仅凭视频画面中的人物年龄推断观众年龄；
+- 用账号整体表现代替单条视频表现。
+
+### 2.1 当前实现差距
+
+| 目标能力 | 当前实现 | 主要差距 |
+|---|---|---|
+| 发布时间判断 | 搜索参数支持部分时间范围，详情返回字段未完整落候选表 | 无法在最终分池时稳定校验发布日期和语义有效期 |
+| 30 秒时长门槛 | TikHub 和作者作品工具支持最短时长参数，但默认值为 0 | 不是全搜索源默认规则，保存 `primary` 时也不复核 |
+| 视频理解 | 已有历史千问分析模块，但未注册给 Agent | 无正式调用链、结构化结果和持久化字段 |
+| 双侧年龄画像 | 能分别获取视频与账号画像，并生成双侧标准化结果 | 最终仍收敛为一个 `E` 分，没有独立准入结果 |
+| 50+ 老年判断 | 标准化逻辑能识别 50+，也保留 40+ 成熟代理信号 | 保存层不校验视频侧 50+ 占比 |
+| 分享与画像硬门槛 | Prompt 要求模型综合判断分享和年龄证据 | 候选更新工具不校验阈值、分数或 `primary` 证据完整性 |
+
+## 3. 核心设计原则
+
+### 3.1 先准入，后评分
+
+候选必须依次通过时效、时长、内容指标、视频理解和视频侧老年画像等门槛，才允许计算
+综合价值并参与 `primary` 排序。评分不能补偿硬门槛失败。
+
+### 3.2 内容证据与账号先验分离
+
+- 视频画像回答“这条视频实际吸引了谁”，属于内容侧直接证据；
+- 账号画像回答“这个账号通常触达谁”，属于账号侧先验；
+- 两侧分别输出比例、结论和置信度；
+- 不允许把两侧比例相加、平均后再判断；
+- 两侧冲突时优先视频画像，并保留 `portrait_conflict` 标记；
+- 只有账号画像偏老、视频画像缺失时，不允许进入主推荐。
+
+### 3.3 时间有效性不等于发布时间新
+
+时效判断同时包含：
+
+1. **发布新鲜度**：视频发布距当前时间有多久；
+2. **语义有效期**：视频中的日期、节日、问候时段、事件状态在当前是否成立。
+
+一条刚发布的“早上好”视频在中午仍可能失效；一条较早发布但没有时间依赖的常青内容
+仍可能有效。因此不能只用发布时间做单一判断。
+
+### 3.4 视频理解不承担年龄画像职责
+
+视频理解用于核验实际主题、关键信息、时间语义、需求相关性和分享动机，不用于推断
+观众年龄。老年受众结论必须来自年龄画像。
+
+## 4. 目标流程
+
+```text
+运行上下文
+  -> 注入当前时间、日期、时区和阈值版本
+  -> 按需求生成搜索词与时间类型
+  -> 搜索召回
+  -> 基础字段完整性检查
+  -> 时长硬过滤（>= 30 秒）
+  -> 分享数硬过滤
+  -> 发布时间与时间语义初筛
+  -> 获取详情和可播放地址
+  -> 轻量视频理解
+  -> 视频点赞用户画像判断
+  -> 账号粉丝画像独立判断
+  -> 统一质量门禁
+  -> 对通过候选计算 R / S / V 并排序
+  -> 保存分池、原因码和证据快照
+  -> 完成守卫校验
+  -> 输出结果
+```
+
+模型仍负责搜索策略、语义理解和结果解释；程序负责不可绕过的门槛、字段校验和完成
+状态校验。
+
+## 5. 规则设计
+
+### 5.1 时效性规则
+
+#### 5.1.1 运行上下文
+
+每次运行必须向 Agent 和规则引擎提供：
+
+- `current_datetime`：运行开始时间；
+- `current_date`：当前业务日期；
+- `timezone`：默认 `Asia/Shanghai`；
+- `biz_dt`：调度业务日期；
+- `rule_version`：本次使用的规则配置版本。
+
+禁止只给 `biz_dt` 而不给当前具体时间。时段内容判断必须使用 `current_datetime`。
+
+#### 5.1.2 时间类型
+
+需求和候选分别标记以下时间类型：
+
+| 类型 | 典型内容 | 默认有效规则 |
+|---|---|---|
+| `daypart` | 早上好、午安、晚安 | 当前时间必须处于对应时段 |
+| `festival` | 春节、端午、中秋、国庆 | 当前日期必须处于该节日前置窗口或节日期间 |
+| `event` | 新闻、比赛、政策、突发事件 | 事件仍在发生，或内容仍有明确回顾价值 |
+| `seasonal` | 入伏、开学、换季 | 当前日期必须处于对应季节窗口 |
+| `evergreen` | 健康常识、家庭技巧、怀旧内容 | 不设统一绝对过期日，但仍检查信息是否失效 |
+
+建议首发时段配置：
+
+| 内容 | 有效时段 |
+|---|---|
+| 早上好/晨间问候 | 05:00～10:30 |
+| 午安/午间问候 | 11:00～14:00 |
+| 晚上好/晚安 | 18:00～次日 01:00 |
+
+建议首发节日配置：
+
+- 普通节日内容：节日前 7 天至节日结束；
+- 强当天语义内容，如“今天是中秋”：仅节日当天有效；
+- 节后复盘、祝福回顾类内容：仅在视频理解明确识别为回顾时放行；
+- 不在有效窗口内的节日词，不作为搜索扩展词；偶然召回后直接淘汰。
+
+#### 5.1.3 发布时间策略
+
+搜索时根据时间类型主动使用发布时间过滤：
+
+- `daypart`、突发 `event`：优先“一天内”；
+- 普通事件和强时效资讯：优先“一周内”；
+- 季节性和阶段性内容：优先“半年内”，再做语义有效性判断；
+- `evergreen`：可不限发布时间，但详情阶段必须取得发布时间并检查失效表述。
+
+候选必须保存真实 `publish_at`，并计算 `content_age_hours`、`content_age_days`。若上游搜索
+没有发布时间，候选只能处于待补证状态；详情仍无法获取时不得进入 `primary`。
+
+#### 5.1.4 时间判断结果
+
+每条候选输出：
+
+- `temporal_type`；
+- `publish_at`；
+- `semantic_time_refs`：识别到的日期、节日、时段或事件；
+- `temporal_status`：`pass / fail / unknown`；
+- `temporal_reason`；
+- `valid_from / valid_to`，能够确定时填写。
+
+`fail` 或 `unknown` 均不能进入主推荐。`unknown` 表示证据不足，不应描述为内容错误。
+
+### 5.2 时长规则
+
+- 主推荐视频必须满足 `duration_seconds >= 30`；
+- `29.999` 秒按不通过处理，`30.000` 秒按通过处理；
+- 搜索接口支持客户端过滤时，统一传 `min_duration_seconds=30`；
+- 作者作品扩展同样执行 30 秒过滤；
+- 搜索结果没有时长时不得直接保留为正式候选，必须通过详情补齐；
+- 详情仍无法获得时长时，标记 `DURATION_UNKNOWN`，不得进入 `primary`；
+- 搜索侧过滤只用于节省成本，最终保存时必须再次由代码校验，防止其他搜索源绕过。
+
+### 5.3 轻量视频理解
+
+#### 5.3.1 使用范围
+
+只对通过时长、分享数和初步时效检查的高潜候选执行，默认每个需求最多分析 8 条，降低
+模型与视频处理成本。
+
+现有 `qwen_video_analysis.py` 可作为实现基础，但需要从“历史未注册能力”改造成正式
+批量工具或确定性服务步骤，并把分析结果写入候选证据。
+
+#### 5.3.2 最小分析内容
+
+视频理解至少返回以下结构化字段：
+
+```json
+{
+  "content_summary": "视频实际讲了什么",
+  "main_topics": ["主题1", "主题2"],
+  "spoken_or_visual_time_refs": ["早上好", "2026年春节"],
+  "demand_match": "pass|fail|unknown",
+  "demand_match_reason": "实际内容与需求的关系",
+  "share_motives": ["实用提醒", "情感共鸣"],
+  "is_title_content_consistent": true,
+  "temporal_risk": "none|daypart|festival|event|seasonal|unknown",
+  "risk_flags": [],
+  "confidence": 0.0
+}
+```
+
+实现时可组合关键帧、OCR、ASR/字幕和多模态摘要。首版不要求深度剧情理解，只要求能
+识别：
+
+- 标题与实际内容是否一致；
+- 视频是否真正回答需求；
+- 是否出现“早上好”“今天”“某节日”等强时间表达；
+- 是否为纯广告、搬运拼接、无信息量画面；
+- 是否具备可解释的分享动机。
+
+#### 5.3.3 分池约束
+
+- `demand_match=fail`：淘汰；
+- `demand_match=unknown`：不得进入主推荐；
+- `temporal_risk` 命中后，交给时间规则结合当前时间复核；
+- 视频理解失败可重试一次；仍失败时标记 `VIDEO_ANALYSIS_UNAVAILABLE`，不得进入
+  `primary`；
+- 视频理解结果不得生成或修改 50+ 占比。
+
+### 5.4 视频画像规则
+
+视频画像使用“视频点赞用户年龄画像”，单独生成内容侧结论：
+
+- 核心指标：`content_50_plus_ratio`；
+- 辅助指标：`content_50_plus_tgi`；
+- `40～49` 或 `41～50` 只能记录为成熟人群辅助信息，不计入 50+ 准入占比；
+- 画像桶含义必须先标准化，接口中的 `50-` 按已确认的“50岁以上”解析；
+- 视频侧画像缺失时，标记 `CONTENT_PORTRAIT_MISSING`，不得由账号画像代替；
+- 视频侧 50+ 占比是主推荐的硬门槛。
+
+建议首发配置：
+
+```text
+min_content_50_plus_ratio = 0.20
+```
+
+即视频侧 50+ 占比至少为 20%。这是首轮试运行阈值，不代表永久业务常量；上线前应使用
+近期人工标注样本回测。
+
+### 5.5 账号画像规则
+
+账号画像使用“作者粉丝年龄画像”，单独生成账号侧结论：
+
+- 核心指标：`account_50_plus_ratio`；
+- 辅助指标：`account_50_plus_tgi`；
+- 输出 `account_portrait_status=pass / fail / missing`；
+- 首发参考线可同样配置为 `min_account_50_plus_ratio=0.20`；
+- 账号侧 `pass` 只增强置信度，不可使视频侧失败候选通过；
+- 账号侧 `fail` 不自动否决视频侧明确通过的单条内容，但必须标记画像冲突；
+- 账号侧缺失不等于年轻，只记录证据缺失。
+
+示例：
+
+| 视频侧 50+ | 账号侧 50+ | 结论 |
+|---:|---:|---|
+| 28% | 35% | 视频通过、账号通过、画像一致 |
+| 28% | 8% | 视频通过、账号不通过、画像冲突；仍可继续其他门禁 |
+| 缺失 | 55% | 视频证据不足，不得进入主推荐 |
+| 12% | 55% | 视频门槛失败，不得由账号画像补偿 |
+
+### 5.6 内容指标强制规则
+
+建议首发配置：
+
+```text
+min_duration_seconds = 30
+min_share_count = 1000
+min_content_50_plus_ratio = 0.20
+```
+
+主推荐必须同时满足：
+
+1. `share_count >= min_share_count`；
+2. `content_50_plus_ratio >= min_content_50_plus_ratio`；
+3. `duration_seconds >= min_duration_seconds`。
+
+分享数使用最新详情数据，不使用搜索页中的旧快照做最终判断。若详情没有分享数，标记
+`SHARE_COUNT_UNKNOWN`，不得进入主推荐。
+
+分享效率继续作为排序指标，不代替分享数门槛：
+
+- 有可靠播放数：`share_rate = share_count / play_count`；
+- 没有播放数但有点赞数：使用平滑后的 `share_count / like_count` 作为替代指标；
+- 分母过小的样本设置最小样本量保护；
+- 分享率再高，只要分享数未达到硬门槛，仍不能进入主推荐。
+
+`min_share_count=1000` 是用于阻止低传播规模内容进入主推荐的试运行值。应在上线后
+按需求类型、搜索来源和近 7～14 天样本分布校准，但任何配置都必须保留绝对分享数下限。
+
+### 5.7 统一质量门禁
+
+程序新增统一门禁函数，候选只有全部通过才可保存为 `primary`：
+
+| 门禁 | 通过条件 | 失败原因码 |
+|---|---|---|
+| 字段完整性 | ID、详情、发布时间等关键字段可用 | `REQUIRED_FIELD_MISSING` |
+| 时间有效性 | `temporal_status=pass` | `TEMPORAL_EXPIRED` / `TEMPORAL_UNKNOWN` |
+| 视频时长 | `duration_seconds >= 30` | `DURATION_TOO_SHORT` / `DURATION_UNKNOWN` |
+| 分享规模 | `share_count >= 1000` | `SHARE_COUNT_TOO_LOW` / `SHARE_COUNT_UNKNOWN` |
+| 视频理解 | 分析成功且需求匹配 | `CONTENT_MISMATCH` / `VIDEO_ANALYSIS_UNAVAILABLE` |
+| 视频侧画像 | 50+ 占比至少 20% | `CONTENT_50_PLUS_TOO_LOW` / `CONTENT_PORTRAIT_MISSING` |
+| 相关性 | 满足需求真实意图 | `LOW_RELEVANCE` |
+
+门禁阈值必须从版本化配置读取，表中的 30 秒、1000 次分享和 20% 为首发建议值。
+
+`batch_update_video_discovery_candidates` 保存 `primary` 时必须调用门禁函数。若模型请求
+将未通过候选设为 `primary`，程序应拒绝该条更新并返回具体失败字段，不应静默改写。
+
+## 6. 评分与排序调整
+
+硬门槛通过后再计算排序分。建议将原来的老年综合分拆开：
+
+- `R`：需求相关性；
+- `C50`：视频侧 50+ 占比归一化得分；
+- `S`：分享规模、分享效率和分享动机；
+- `Q`：视频理解置信度与内容质量；
+- `A50`：账号侧 50+ 先验，只作为小权重增强项。
+
+示例排序公式：
+
+```text
+V = R^0.35 × C50^0.30 × S^0.25 × Q^0.10
+V_final = V × account_adjustment
+account_adjustment ∈ [0.95, 1.05]
+```
+
+账号画像最多做小幅调整，不能改变任何硬门槛结果。首版也可以不启用
+`account_adjustment`，只展示冲突标记，待回测后再决定。
+
+## 7. 数据与接口改造
+
+### 7.1 候选表新增字段
+
+建议为 `video_discovery_candidate` 增加：
+
+| 字段 | 说明 |
+|---|---|
+| `publish_at` | 视频真实发布时间 |
+| `duration_seconds` | 视频时长，统一使用秒 |
+| `evaluated_at` | 本次候选评估时间 |
+| `content_age_hours_at_evaluation` | 评估时的视频年龄，避免被误解为实时值 |
+| `temporal_type` | 时间类型 |
+| `temporal_status` | `pass / fail / unknown` |
+| `temporal_evidence_json` | 时间表达、有效窗口和判断理由 |
+| `video_analysis_status` | 视频理解执行状态 |
+| `video_analysis_json` | 结构化视频理解结果 |
+| `video_analysis_version` | 模型、Prompt 和解析版本 |
+| `content_50_plus_ratio` | 视频侧 50+ 占比 |
+| `content_50_plus_tgi` | 视频侧 50+ TGI |
+| `content_portrait_status` | 视频侧画像独立结论 |
+| `account_50_plus_ratio` | 账号侧 50+ 占比 |
+| `account_50_plus_tgi` | 账号侧 50+ TGI |
+| `account_portrait_status` | 账号侧画像独立结论 |
+| `portrait_conflict` | 两侧是否冲突 |
+| `share_rate` | 分享效率 |
+| `gate_status` | `pass / fail / pending` |
+| `gate_results_json` | 每项门禁结果和使用阈值 |
+| `reject_reason_code` | 主要淘汰原因码 |
+| `rule_version` | 评估使用的规则版本 |
+
+可播放地址只需在分析阶段短期使用，避免长期保存可能失效或带鉴权参数的原始地址；持久化
+分析结果、输入视频 ID和分析版本即可。
+
+### 7.2 搜索与详情接口
+
+需要补齐：
+
+- TikHub 搜索标准化结果保留视频 `create_time`；
+- 所有召回工具统一支持并默认传入 `min_duration_seconds=30`；
+- 内部搜索没有时长或发布时间时，候选明确标记为待补证；
+- `douyin_detail` 明确保留视频真实发布时间，而不是只保存抓取系统的
+  `modify_timestamp/update_timestamp`；
+- 详情结果写回候选表，不再只返回给模型；
+- 详情写回和最终分池前都执行一次字段单位标准化。
+
+### 7.3 视频理解接口
+
+建议新增批量工具：
+
+```text
+batch_analyze_candidate_videos(run_id, candidate_ids, analysis_version)
+```
+
+工具内部负责：
+
+1. 按候选 ID 读取详情和临时播放地址；
+2. 校验时长、URL 和批量上限；
+3. 调用轻量视频理解；
+4. 校验结构化输出；
+5. 写回分析状态与结果；
+6. 返回每条候选的成功、失败和可重试状态。
+
+不建议让 Agent 自由拼接任意视频 URL 调用分析模型，以免分析对象与候选记录不一致。
+
+### 7.4 运行级配置
+
+新增版本化配置示例：
+
+```json
+{
+  "rule_version": "find-agent-gate-v1",
+  "timezone": "Asia/Shanghai",
+  "min_duration_seconds": 30,
+  "min_share_count": 1000,
+  "min_content_50_plus_ratio": 0.2,
+  "min_account_50_plus_ratio": 0.2,
+  "max_video_analysis_candidates": 8,
+  "festival_lead_days": 7
+}
+```
+
+运行记录保存完整配置快照，防止配置变更后无法解释历史结果。
+
+## 8. Prompt 调整
+
+Prompt 保留搜索策略与语义判断，但删除或改写以下内容：
+
+- 删除“当前流程明确不使用视频理解”；
+- 增加当前日期、具体时间和时区的使用要求；
+- 明确 30 秒、分享数和视频侧 50+ 占比是程序硬门槛；
+- 明确视频理解是主推荐必需证据；
+- 将 `E` 拆成视频侧结论和账号侧结论；
+- 明确 40～49 岁不计入 50+ 准入占比；
+- 明确账号画像不能补偿视频画像缺失或失败；
+- 要求最终理由分别陈述视频侧画像、账号侧画像和冲突情况；
+- 要求引用时间有效性结论，不能只写“近期发布”。
+
+Prompt 中不再让模型自行决定阈值，也不再让模型手工计算是否通过门禁。工具返回
+`gate_results` 后，模型负责解释，不负责改写。
+
+## 9. 完成守卫
+
+当前“存在任意候选即成功”的判定过弱。本次同步调整为：
+
+运行标记 `finished` 前必须满足：
+
+- 所有需要最终判断的候选均不再是 `pending_evaluation`；
+- 每条 `primary` 均通过统一质量门禁；
+- 每条 `primary` 都有详情、时效判断、视频理解和视频侧画像证据；
+- `primary_count` 与数据库实际数量一致；
+- 没有主推荐也可以正常完成，但必须保存“无候选通过硬门槛”的停止原因；
+- 数据库写入或门禁执行失败时不得标记 `finished`。
+
+## 10. 实施拆分
+
+### P0：基础硬门槛
+
+- 补齐发布时间、时长和分享数持久化；
+- 所有搜索源统一 30 秒过滤；
+- 增加时间上下文和基础时效规则；
+- 拆分视频画像与账号画像字段及结论；
+- 增加视频侧 50+ 占比、分享数硬门槛；
+- 在候选保存层增加统一门禁；
+- 增加原因码与规则配置快照；
+- 更新 Prompt 和完成守卫。
+
+交付标准：短于 30 秒、分享数不足、视频侧 50+ 不足、时间失效的候选无法被保存为
+`primary`。
+
+### P1：轻量视频理解
+
+- 将现有千问视频分析能力正式接入；
+- 定义结构化输出协议；
+- 增加批量分析工具、结果持久化和失败策略；
+- 将内容匹配和时间语义结果接入统一门禁；
+- 增加分析版本与成本监控。
+
+交付标准：所有 `primary` 均有成功的视频理解记录，标题与实际内容不一致的候选被淘汰。
+
+### P2：阈值校准与运营监控
+
+- 使用近期历史结果建立人工标注集；
+- 分析分享数、分享率、50+ 占比阈值对准确率和召回率的影响；
+- 按需求类型评估是否需要不同阈值；
+- 建立门禁淘汰分布、画像缺失率、分析失败率和主推荐人工通过率看板；
+- 根据回测结果发布 `rule_version=v2`，不直接覆盖历史配置。
+
+## 11. 验收用例
+
+| 场景 | 输入 | 期望结果 |
+|---|---|---|
+| 时段失效 | 当前 12:00，视频内容为“早上好” | `TEMPORAL_EXPIRED`，不得主推荐 |
+| 节日失效 | 当前不在中秋窗口，视频为当年中秋祝福 | `TEMPORAL_EXPIRED` |
+| 常青内容 | 较早发布但内容无过期信息 | 可继续后续门禁 |
+| 发布时间缺失 | 搜索、详情均无真实发布时间 | `TEMPORAL_UNKNOWN` |
+| 时长边界失败 | 29.999 秒 | `DURATION_TOO_SHORT` |
+| 时长边界通过 | 30.000 秒 | 通过时长门禁 |
+| 时长缺失 | 详情仍无法取得时长 | `DURATION_UNKNOWN` |
+| 分享数不足 | 分享 999，其他指标均通过 | `SHARE_COUNT_TOO_LOW` |
+| 分享数边界 | 分享 1000 | 通过分享数门禁 |
+| 视频侧占比不足 | 视频侧 50+ 为 19.9% | `CONTENT_50_PLUS_TOO_LOW` |
+| 视频侧占比边界 | 视频侧 50+ 为 20% | 通过视频画像门禁 |
+| 仅账号偏老 | 视频画像缺失，账号侧 50+ 为 55% | `CONTENT_PORTRAIT_MISSING` |
+| 双侧冲突 | 视频侧 28%，账号侧 8% | 视频通过、账号失败、记录冲突，不做平均 |
+| 题文不符 | 标题符合需求，视频理解显示实际为无关广告 | `CONTENT_MISMATCH` |
+| 分析失败 | 视频理解重试后仍失败 | `VIDEO_ANALYSIS_UNAVAILABLE` |
+| 模型绕过 | 模型请求把门禁失败候选保存为 `primary` | 保存层拒绝并返回失败项 |
+
+## 12. 监控指标
+
+上线后至少监控：
+
+- 搜索召回量；
+- 30 秒时长过滤率；
+- 时间失效率及其类型分布；
+- 分享数门槛淘汰率；
+- 视频侧画像缺失率；
+- 账号侧画像缺失率；
+- 视频侧 50+ 门槛淘汰率；
+- 双侧画像冲突率；
+- 视频理解成功率、平均耗时和单次成本；
+- 标题与内容不一致率；
+- 各原因码数量；
+- 每个需求最终主推荐数；
+- 人工审核通过率；
+- 因硬门槛造成的误杀率和漏放率。
+
+## 13. 风险与处理
+
+| 风险 | 影响 | 处理方式 |
+|---|---|---|
+| 分享数阈值过高 | 新视频和小众优质内容召回下降 | 用标注集回测，阈值配置化 |
+| 50+ 画像覆盖率低 | 主推荐数量下降 | 提升画像接口覆盖；不允许账号画像冒充内容画像 |
+| 视频理解成本和耗时增加 | 单需求运行时间变长 | 只分析基础门槛后的前 8 条，批量并发并缓存 |
+| 视频播放地址失效 | 分析失败 | 详情阶段即时获取，不长期依赖原始地址 |
+| 节日历法复杂 | 时间误判 | 使用版本化节日日历，保留人工可配置窗口 |
+| 模型输出结构不稳定 | 门禁无法判断 | JSON Schema 校验，失败一次修复重试 |
+| 过度依赖固定阈值 | 不同需求类型表现不一致 | 保留绝对底线，分需求类型做二阶段校准 |
+
+## 14. 上线判定
+
+满足以下条件后才进入全量运行：
+
+1. P0、P1 验收用例全部通过；
+2. 历史样本回放中不存在硬门槛绕过；
+3. 随机抽检能够看到视频画像和账号画像的独立结论；
+4. “早上好”和非当前节日等时间失效样本被稳定识别；
+5. 人工审核确认主推荐准确率较现状提升；
+6. 视频理解失败不会导致运行被误标成功；
+7. 规则配置、分析版本和淘汰原因可完整追溯。

alembic/versions/20260731_08_add_find_agent_p0_gates.py

@@ -0,0 +1,165 @@
+"""add find_agent P0 gate fields
+
+Revision ID: 20260731_08
+Revises: 20260730_07
+Create Date: 2026-07-31
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260731_08"
+down_revision: str | None = "20260730_07"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "video_discovery_run",
+        sa.Column(
+            "rule_version",
+            sa.String(length=64),
+            nullable=True,
+            comment="本次候选门禁规则版本",
+        ),
+    )
+    op.add_column(
+        "video_discovery_run",
+        sa.Column(
+            "rule_config_json",
+            sa.Text(),
+            nullable=True,
+            comment="运行时规则、当前时间和阈值快照 JSON",
+        ),
+    )
+
+    columns = (
+        sa.Column(
+            "publish_at",
+            sa.DateTime(),
+            nullable=True,
+            comment="视频真实发布时间，按运行时区解释",
+        ),
+        sa.Column(
+            "duration_seconds",
+            sa.Numeric(precision=10, scale=3),
+            nullable=True,
+            comment="视频时长（秒）",
+        ),
+        sa.Column(
+            "content_50_plus_ratio",
+            sa.Numeric(precision=8, scale=6),
+            nullable=True,
+            comment="视频点赞用户中 50+ 占比",
+        ),
+        sa.Column(
+            "content_50_plus_tgi",
+            sa.Numeric(precision=10, scale=4),
+            nullable=True,
+            comment="视频点赞用户 50+ TGI",
+        ),
+        sa.Column(
+            "content_portrait_status",
+            sa.String(length=16),
+            nullable=True,
+            comment="视频画像结论 pass / fail / missing",
+        ),
+        sa.Column(
+            "account_50_plus_ratio",
+            sa.Numeric(precision=8, scale=6),
+            nullable=True,
+            comment="账号粉丝中 50+ 占比",
+        ),
+        sa.Column(
+            "account_50_plus_tgi",
+            sa.Numeric(precision=10, scale=4),
+            nullable=True,
+            comment="账号粉丝 50+ TGI",
+        ),
+        sa.Column(
+            "account_portrait_status",
+            sa.String(length=16),
+            nullable=True,
+            comment="账号画像结论 pass / fail / missing",
+        ),
+        sa.Column(
+            "portrait_conflict",
+            sa.Integer(),
+            nullable=False,
+            server_default="0",
+            comment="视频与账号画像结论是否冲突",
+        ),
+        sa.Column(
+            "temporal_type",
+            sa.String(length=24),
+            nullable=True,
+            comment="时间内容类型",
+        ),
+        sa.Column(
+            "temporal_status",
+            sa.String(length=16),
+            nullable=True,
+            comment="时间有效性 pass / fail / unknown",
+        ),
+        sa.Column(
+            "temporal_evidence_json",
+            sa.Text(),
+            nullable=True,
+            comment="时间语义、有效窗口和判断理由 JSON",
+        ),
+        sa.Column(
+            "gate_status",
+            sa.String(length=16),
+            nullable=True,
+            comment="硬门槛 pass / fail / pending",
+        ),
+        sa.Column(
+            "gate_results_json",
+            sa.Text(),
+            nullable=True,
+            comment="逐项硬门槛结果 JSON",
+        ),
+        sa.Column(
+            "reject_reason_code",
+            sa.String(length=64),
+            nullable=True,
+            comment="主要淘汰原因码",
+        ),
+        sa.Column(
+            "rule_version",
+            sa.String(length=64),
+            nullable=True,
+            comment="候选评估使用的规则版本",
+        ),
+    )
+    for column in columns:
+        op.add_column("video_discovery_candidate", column)
+
+
+def downgrade() -> None:
+    candidate_columns = (
+        "rule_version",
+        "reject_reason_code",
+        "gate_results_json",
+        "gate_status",
+        "temporal_evidence_json",
+        "temporal_status",
+        "temporal_type",
+        "portrait_conflict",
+        "account_portrait_status",
+        "account_50_plus_tgi",
+        "account_50_plus_ratio",
+        "content_portrait_status",
+        "content_50_plus_tgi",
+        "content_50_plus_ratio",
+        "duration_seconds",
+        "publish_at",
+    )
+    for column in candidate_columns:
+        op.drop_column("video_discovery_candidate", column)
+    op.drop_column("video_discovery_run", "rule_config_json")
+    op.drop_column("video_discovery_run", "rule_version")

api/services/video_discovery_records.py

@@ -54,6 +54,8 @@ def _serialize_run(row: VideoDiscoveryRun, counts: dict[str, int]) -> dict[str,
         "rejected_count": counts.get("rejected", 0),
         "pending_count": counts.get("pending_evaluation", 0),
         "stop_reason": row.stop_reason,
+        "rule_version": row.rule_version,
+        "rule_config": _json_value(row.rule_config_json),
         "create_time": _timestamp(row.create_time),
         "update_time": _timestamp(row.update_time),
     }
@@ -67,6 +69,8 @@ def _serialize_search_candidate(row: VideoDiscoveryCandidate) -> dict[str, Any]:
         "content_link": row.content_link,
         "author_name": row.author_name,
         "decision_bucket": row.decision_bucket,
+        "publish_at": _timestamp(row.publish_at),
+        "duration_seconds": _number(row.duration_seconds),
         "play_count": int(row.play_count) if row.play_count is not None else None,
         "like_count": int(row.like_count) if row.like_count is not None else None,
         "comment_count": (
@@ -80,6 +84,14 @@ def _serialize_search_candidate(row: VideoDiscoveryCandidate) -> dict[str, Any]:
         "elder_score": _number(row.elder_score),
         "share_score": _number(row.share_score),
         "value_score": _number(row.value_score),
+        "content_50_plus_ratio": _number(row.content_50_plus_ratio),
+        "content_portrait_status": row.content_portrait_status,
+        "account_50_plus_ratio": _number(row.account_50_plus_ratio),
+        "account_portrait_status": row.account_portrait_status,
+        "portrait_conflict": bool(row.portrait_conflict),
+        "temporal_status": row.temporal_status,
+        "gate_status": row.gate_status,
+        "reject_reason_code": row.reject_reason_code,
     }
 
 
@@ -131,6 +143,8 @@ def _serialize_candidate(row: VideoDiscoveryCandidate) -> dict[str, Any]:
         "source_keywords": _json_value(row.source_keywords_json),
         "source_search_ids": _json_value(row.source_search_ids_json),
         "tags": _json_value(row.tags_json),
+        "publish_at": _timestamp(row.publish_at),
+        "duration_seconds": _number(row.duration_seconds),
         "play_count": int(row.play_count) if row.play_count is not None else None,
         "like_count": int(row.like_count) if row.like_count is not None else None,
         "comment_count": (
@@ -143,6 +157,20 @@ def _serialize_candidate(row: VideoDiscoveryCandidate) -> dict[str, Any]:
         "content_age_evidence": _json_value(row.content_age_evidence_json),
         "account_age_evidence": _json_value(row.account_age_evidence_json),
         "age_normalization": _json_value(row.age_normalization_json),
+        "content_50_plus_ratio": _number(row.content_50_plus_ratio),
+        "content_50_plus_tgi": _number(row.content_50_plus_tgi),
+        "content_portrait_status": row.content_portrait_status,
+        "account_50_plus_ratio": _number(row.account_50_plus_ratio),
+        "account_50_plus_tgi": _number(row.account_50_plus_tgi),
+        "account_portrait_status": row.account_portrait_status,
+        "portrait_conflict": bool(row.portrait_conflict),
+        "temporal_type": row.temporal_type,
+        "temporal_status": row.temporal_status,
+        "temporal_evidence": _json_value(row.temporal_evidence_json),
+        "gate_status": row.gate_status,
+        "gate_results": _json_value(row.gate_results_json),
+        "reject_reason_code": row.reject_reason_code,
+        "rule_version": row.rule_version,
         "relevance_score": _number(row.relevance_score),
         "elder_score": _number(row.elder_score),
         "share_score": _number(row.share_score),
@@ -322,6 +350,8 @@ def list_video_discovery_searches(
                         VideoDiscoveryCandidate.content_link,
                         VideoDiscoveryCandidate.author_name,
                         VideoDiscoveryCandidate.decision_bucket,
+                        VideoDiscoveryCandidate.publish_at,
+                        VideoDiscoveryCandidate.duration_seconds,
                         VideoDiscoveryCandidate.play_count,
                         VideoDiscoveryCandidate.like_count,
                         VideoDiscoveryCandidate.comment_count,
@@ -331,6 +361,14 @@ def list_video_discovery_searches(
                         VideoDiscoveryCandidate.elder_score,
                         VideoDiscoveryCandidate.share_score,
                         VideoDiscoveryCandidate.value_score,
+                        VideoDiscoveryCandidate.content_50_plus_ratio,
+                        VideoDiscoveryCandidate.content_portrait_status,
+                        VideoDiscoveryCandidate.account_50_plus_ratio,
+                        VideoDiscoveryCandidate.account_portrait_status,
+                        VideoDiscoveryCandidate.portrait_conflict,
+                        VideoDiscoveryCandidate.temporal_status,
+                        VideoDiscoveryCandidate.gate_status,
+                        VideoDiscoveryCandidate.reject_reason_code,
                     )
                 )
                 .where(VideoDiscoveryCandidate.search_id.in_(candidates_by_search))

sql/video_discovery_tables.sql

@@ -19,6 +19,8 @@ CREATE TABLE IF NOT EXISTS `video_discovery_run` (
   `search_count` INT NOT NULL DEFAULT 0 COMMENT '已保存搜索页数',
   `primary_count` INT NOT NULL DEFAULT 0 COMMENT '主推荐数',
   `stop_reason` TEXT NULL COMMENT '停止搜索的证据或失败原因',
+  `rule_version` VARCHAR(64) NULL COMMENT '本次候选门禁规则版本',
+  `rule_config_json` TEXT NULL COMMENT '运行时规则、当前时间和阈值快照 JSON',
   `create_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
   `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
     ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
@@ -80,6 +82,8 @@ CREATE TABLE IF NOT EXISTS `video_discovery_candidate` (
   `source_keywords_json` TEXT NULL COMMENT '命中过该视频的搜索词 JSON',
   `source_search_ids_json` TEXT NULL COMMENT '来源搜索轨迹 id JSON',
   `tags_json` TEXT NULL COMMENT '视频标签/话题 JSON',
+  `publish_at` DATETIME NULL COMMENT '视频真实发布时间，按运行时区解释',
+  `duration_seconds` DECIMAL(10,3) NULL COMMENT '视频时长（秒）',
   `play_count` BIGINT NULL COMMENT '播放数快照',
   `like_count` BIGINT NULL COMMENT '点赞数快照',
   `comment_count` BIGINT NULL COMMENT '评论数快照',
@@ -88,6 +92,20 @@ CREATE TABLE IF NOT EXISTS `video_discovery_candidate` (
   `content_age_evidence_json` TEXT NULL COMMENT '视频点赞用户年龄证据 JSON',
   `account_age_evidence_json` TEXT NULL COMMENT '作者粉丝年龄证据 JSON',
   `age_normalization_json` TEXT NULL COMMENT '双侧年龄画像标准化结果 JSON',
+  `content_50_plus_ratio` DECIMAL(8,6) NULL COMMENT '视频点赞用户中 50+ 占比',
+  `content_50_plus_tgi` DECIMAL(10,4) NULL COMMENT '视频点赞用户 50+ TGI',
+  `content_portrait_status` VARCHAR(16) NULL COMMENT '视频画像结论 pass / fail / missing',
+  `account_50_plus_ratio` DECIMAL(8,6) NULL COMMENT '账号粉丝中 50+ 占比',
+  `account_50_plus_tgi` DECIMAL(10,4) NULL COMMENT '账号粉丝 50+ TGI',
+  `account_portrait_status` VARCHAR(16) NULL COMMENT '账号画像结论 pass / fail / missing',
+  `portrait_conflict` TINYINT NOT NULL DEFAULT 0 COMMENT '视频与账号画像结论是否冲突',
+  `temporal_type` VARCHAR(24) NULL COMMENT '时间内容类型',
+  `temporal_status` VARCHAR(16) NULL COMMENT '时间有效性 pass / fail / unknown',
+  `temporal_evidence_json` TEXT NULL COMMENT '时间语义、有效窗口和判断理由 JSON',
+  `gate_status` VARCHAR(16) NULL COMMENT '硬门槛 pass / fail / pending',
+  `gate_results_json` TEXT NULL COMMENT '逐项硬门槛结果 JSON',
+  `reject_reason_code` VARCHAR(64) NULL COMMENT '主要淘汰原因码',
+  `rule_version` VARCHAR(64) 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',

supply_infra/config.py

@@ -109,6 +109,52 @@ class InfraSettings(BaseSettings):
         alias="SCHEDULER_CRON_MINUTE",
     )
 
+    # find_agent quality gates
+    find_agent_rule_version: str = Field(
+        default="find-agent-gate-v1",
+        alias="FIND_AGENT_RULE_VERSION",
+    )
+    find_agent_min_duration_seconds: int = Field(
+        default=30,
+        ge=30,
+        alias="FIND_AGENT_MIN_DURATION_SECONDS",
+    )
+    find_agent_min_share_count: int = Field(
+        default=1000,
+        ge=0,
+        alias="FIND_AGENT_MIN_SHARE_COUNT",
+    )
+    find_agent_min_content_50_plus_ratio: float = Field(
+        default=0.20,
+        ge=0,
+        le=1,
+        alias="FIND_AGENT_MIN_CONTENT_50_PLUS_RATIO",
+    )
+    find_agent_min_account_50_plus_ratio: float = Field(
+        default=0.20,
+        ge=0,
+        le=1,
+        alias="FIND_AGENT_MIN_ACCOUNT_50_PLUS_RATIO",
+    )
+    find_agent_festival_lead_days: int = Field(
+        default=7,
+        ge=0,
+        le=30,
+        alias="FIND_AGENT_FESTIVAL_LEAD_DAYS",
+    )
+    find_agent_event_max_age_days: int = Field(
+        default=7,
+        ge=1,
+        le=180,
+        alias="FIND_AGENT_EVENT_MAX_AGE_DAYS",
+    )
+    find_agent_seasonal_max_age_days: int = Field(
+        default=180,
+        ge=1,
+        le=366,
+        alias="FIND_AGENT_SEASONAL_MAX_AGE_DAYS",
+    )
+
     # Pipeline process/control plane
     process_role: str = Field(default="api", alias="PROCESS_ROLE")
     database_auto_create: bool = Field(default=False, alias="DATABASE_AUTO_CREATE")

supply_infra/db/models/video_discovery.py

@@ -71,6 +71,12 @@ class VideoDiscoveryRun(Base):
     stop_reason: Mapped[str | None] = mapped_column(
         Text, nullable=True, comment="停止搜索的证据或失败原因"
     )
+    rule_version: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, comment="本次候选门禁规则版本"
+    )
+    rule_config_json: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="运行时规则、当前时间和阈值快照 JSON"
+    )
     create_time: Mapped[datetime] = mapped_column(
         nullable=False, server_default=func.now(), comment="创建时间"
     )
@@ -216,6 +222,12 @@ class VideoDiscoveryCandidate(Base):
     tags_json: Mapped[str | None] = mapped_column(
         Text, nullable=True, comment="视频标签/话题 JSON"
     )
+    publish_at: Mapped[datetime | None] = mapped_column(
+        nullable=True, comment="视频真实发布时间，按运行时区解释"
+    )
+    duration_seconds: Mapped[Decimal | None] = mapped_column(
+        Numeric(10, 3), nullable=True, comment="视频时长（秒）"
+    )
     play_count: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
     like_count: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
     comment_count: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
@@ -230,6 +242,48 @@ class VideoDiscoveryCandidate(Base):
     age_normalization_json: Mapped[str | None] = mapped_column(
         Text, nullable=True, comment="双侧年龄画像标准化结果 JSON"
     )
+    content_50_plus_ratio: Mapped[Decimal | None] = mapped_column(
+        Numeric(8, 6), nullable=True, comment="视频点赞用户中 50+ 占比"
+    )
+    content_50_plus_tgi: Mapped[Decimal | None] = mapped_column(
+        Numeric(10, 4), nullable=True, comment="视频点赞用户 50+ TGI"
+    )
+    content_portrait_status: Mapped[str | None] = mapped_column(
+        String(16), nullable=True, comment="视频画像结论 pass / fail / missing"
+    )
+    account_50_plus_ratio: Mapped[Decimal | None] = mapped_column(
+        Numeric(8, 6), nullable=True, comment="账号粉丝中 50+ 占比"
+    )
+    account_50_plus_tgi: Mapped[Decimal | None] = mapped_column(
+        Numeric(10, 4), nullable=True, comment="账号粉丝 50+ TGI"
+    )
+    account_portrait_status: Mapped[str | None] = mapped_column(
+        String(16), nullable=True, comment="账号画像结论 pass / fail / missing"
+    )
+    portrait_conflict: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="视频与账号画像结论是否冲突"
+    )
+    temporal_type: Mapped[str | None] = mapped_column(
+        String(24), nullable=True, comment="daypart / festival / event / seasonal / evergreen"
+    )
+    temporal_status: Mapped[str | None] = mapped_column(
+        String(16), nullable=True, comment="时间有效性 pass / fail / unknown"
+    )
+    temporal_evidence_json: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="时间语义、有效窗口和判断理由 JSON"
+    )
+    gate_status: Mapped[str | None] = mapped_column(
+        String(16), nullable=True, comment="硬门槛 pass / fail / pending"
+    )
+    gate_results_json: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="逐项硬门槛结果 JSON"
+    )
+    reject_reason_code: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, comment="主要淘汰原因码"
+    )
+    rule_version: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, comment="候选评估使用的规则版本"
+    )
     relevance_score: Mapped[Decimal | None] = mapped_column(
         Numeric(8, 6), nullable=True, comment="R，范围 0~1"
     )

supply_infra/db/repositories/video_discovery_repo.py

@@ -6,6 +6,10 @@ from typing import Any
 from sqlalchemy import func, select
 from sqlalchemy.exc import OperationalError, ProgrammingError
 
+from supply_infra.video_discovery_gates import (
+    evaluate_candidate_gate,
+    load_rule_snapshot,
+)
 from supply_infra.db.models.video_discovery import (
     VideoDiscoveryCandidate,
     VideoDiscoveryRun,
@@ -177,6 +181,11 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         if len(candidate_ids) != len(set(candidate_ids)):
             raise ValueError("candidate_id 不能重复")
 
+        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.id.in_(candidate_ids),
@@ -190,6 +199,7 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
                 f"candidate_id 不存在或不属于 run_id={run_id}: {missing}"
             )
 
+        gate_failures: list[str] = []
         for row in rows:
             candidate_id = int(row["candidate_id"])
             entity = existing[candidate_id]
@@ -208,6 +218,56 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
                 elif hasattr(entity, key):
                     setattr(entity, key, value)
 
+            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"]:
+                    codes = ", ".join(gate["failed_reason_codes"])
+                    gate_failures.append(f"candidate_id={candidate_id}: {codes}")
+                else:
+                    entity.reject_reason_code = None
+            else:
+                failed_codes = gate["failed_reason_codes"]
+                if failed_codes:
+                    entity.reject_reason_code = str(failed_codes[0])
+                elif not entity.reject_reason_code:
+                    entity.reject_reason_code = "REJECTED_BY_AGENT"
+
+        if gate_failures:
+            raise ValueError(
+                "primary 候选未通过 P0 硬门槛: " + "; ".join(gate_failures)
+            )
         self.session.flush()
         return [existing[candidate_id] for candidate_id in candidate_ids]
 

supply_infra/services/video_discovery_service.py

@@ -48,6 +48,8 @@ def _serialize_run(run: Any) -> dict[str, Any]:
         "search_count": int(run.search_count or 0),
         "primary_count": int(run.primary_count or 0),
         "stop_reason": run.stop_reason,
+        "rule_version": run.rule_version,
+        "rule_config": _load_json(run.rule_config_json, {}),
     }
 
 
@@ -92,6 +94,10 @@ def _serialize_candidate(item: Any) -> dict[str, Any]:
         "source_keywords": _load_json(item.source_keywords_json, []),
         "source_search_ids": _load_json(item.source_search_ids_json, []),
         "tags": _load_json(item.tags_json, []),
+        "publish_at": item.publish_at.isoformat() if item.publish_at else None,
+        "duration_seconds": (
+            float(item.duration_seconds) if item.duration_seconds is not None else None
+        ),
         "play_count": item.play_count,
         "like_count": item.like_count,
         "comment_count": item.comment_count,
@@ -107,6 +113,36 @@ def _serialize_candidate(item: Any) -> dict[str, Any]:
         "content_age_evidence": _load_json(item.content_age_evidence_json, {}),
         "account_age_evidence": _load_json(item.account_age_evidence_json, {}),
         "age_normalization": _load_json(item.age_normalization_json, {}),
+        "content_50_plus_ratio": (
+            float(item.content_50_plus_ratio)
+            if item.content_50_plus_ratio is not None
+            else None
+        ),
+        "content_50_plus_tgi": (
+            float(item.content_50_plus_tgi)
+            if item.content_50_plus_tgi is not None
+            else None
+        ),
+        "content_portrait_status": item.content_portrait_status,
+        "account_50_plus_ratio": (
+            float(item.account_50_plus_ratio)
+            if item.account_50_plus_ratio is not None
+            else None
+        ),
+        "account_50_plus_tgi": (
+            float(item.account_50_plus_tgi)
+            if item.account_50_plus_tgi is not None
+            else None
+        ),
+        "account_portrait_status": item.account_portrait_status,
+        "portrait_conflict": bool(item.portrait_conflict),
+        "temporal_type": item.temporal_type,
+        "temporal_status": item.temporal_status,
+        "temporal_evidence": _load_json(item.temporal_evidence_json, {}),
+        "gate_status": item.gate_status,
+        "gate_results": _load_json(item.gate_results_json, {}),
+        "reject_reason_code": item.reject_reason_code,
+        "rule_version": item.rule_version,
         "decision_reason": item.decision_reason,
     }
 

supply_infra/video_discovery_gates.py

@@ -0,0 +1,519 @@
+"""find_agent P0 候选硬门槛与基础时效规则。"""
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import asdict, dataclass
+from datetime import date, datetime, time, timedelta
+from decimal import Decimal
+from typing import Any, Mapping
+from zoneinfo import ZoneInfo
+
+from supply_infra.config import get_infra_settings
+
+
+@dataclass(frozen=True)
+class FindAgentGateRules:
+    rule_version: str
+    timezone: str
+    min_duration_seconds: int
+    min_share_count: int
+    min_content_50_plus_ratio: float
+    min_account_50_plus_ratio: float
+    festival_lead_days: int
+    event_max_age_days: int
+    seasonal_max_age_days: int
+
+
+_DAYPART_RULES: tuple[tuple[str, tuple[str, ...], time, time], ...] = (
+    ("morning", ("早上好", "早安", "晨安", "早晨好"), time(5, 0), time(10, 30)),
+    ("noon", ("午安", "中午好", "午间好"), time(11, 0), time(14, 0)),
+    ("evening", ("晚上好", "晚安", "夜安"), time(18, 0), time(1, 0)),
+)
+
+_FIXED_FESTIVALS: dict[str, tuple[int, int]] = {
+    "元旦": (1, 1),
+    "情人节": (2, 14),
+    "妇女节": (3, 8),
+    "劳动节": (5, 1),
+    "儿童节": (6, 1),
+    "建党节": (7, 1),
+    "建军节": (8, 1),
+    "教师节": (9, 10),
+    "国庆": (10, 1),
+    "国庆节": (10, 1),
+    "圣诞": (12, 25),
+    "圣诞节": (12, 25),
+}
+
+# 业务运行期内使用的农历节日阳历日期。超出覆盖年份时返回 unknown，不猜测日期。
+_LUNAR_FESTIVALS: dict[int, dict[str, tuple[int, int]]] = {
+    2025: {
+        "春节": (1, 29),
+        "元宵": (2, 12),
+        "元宵节": (2, 12),
+        "端午": (5, 31),
+        "端午节": (5, 31),
+        "七夕": (8, 29),
+        "中秋": (10, 6),
+        "中秋节": (10, 6),
+        "重阳": (10, 29),
+        "重阳节": (10, 29),
+    },
+    2026: {
+        "春节": (2, 17),
+        "元宵": (3, 3),
+        "元宵节": (3, 3),
+        "端午": (6, 19),
+        "端午节": (6, 19),
+        "七夕": (8, 19),
+        "中秋": (9, 25),
+        "中秋节": (9, 25),
+        "重阳": (10, 18),
+        "重阳节": (10, 18),
+    },
+    2027: {
+        "春节": (2, 6),
+        "元宵": (2, 20),
+        "元宵节": (2, 20),
+        "端午": (6, 9),
+        "端午节": (6, 9),
+        "七夕": (8, 8),
+        "中秋": (9, 15),
+        "中秋节": (9, 15),
+        "重阳": (10, 8),
+        "重阳节": (10, 8),
+    },
+}
+
+_EVENT_MARKERS = (
+    "最新消息",
+    "突发",
+    "刚刚",
+    "今日新闻",
+    "赛事结果",
+    "比赛结果",
+)
+_RELATIVE_DATE_MARKERS = ("今天", "今日", "明天", "昨日", "昨天")
+
+
+def get_gate_rules() -> FindAgentGateRules:
+    settings = get_infra_settings()
+    return FindAgentGateRules(
+        rule_version=settings.find_agent_rule_version,
+        timezone=settings.scheduler_timezone,
+        min_duration_seconds=settings.find_agent_min_duration_seconds,
+        min_share_count=settings.find_agent_min_share_count,
+        min_content_50_plus_ratio=settings.find_agent_min_content_50_plus_ratio,
+        min_account_50_plus_ratio=settings.find_agent_min_account_50_plus_ratio,
+        festival_lead_days=settings.find_agent_festival_lead_days,
+        event_max_age_days=settings.find_agent_event_max_age_days,
+        seasonal_max_age_days=settings.find_agent_seasonal_max_age_days,
+    )
+
+
+def build_rule_snapshot(now: datetime | None = None) -> dict[str, Any]:
+    rules = get_gate_rules()
+    timezone = ZoneInfo(rules.timezone)
+    evaluated_at = now or datetime.now(timezone)
+    if evaluated_at.tzinfo is None:
+        evaluated_at = evaluated_at.replace(tzinfo=timezone)
+    else:
+        evaluated_at = evaluated_at.astimezone(timezone)
+    return {
+        **asdict(rules),
+        "current_datetime": evaluated_at.isoformat(timespec="seconds"),
+        "current_date": evaluated_at.strftime("%Y-%m-%d"),
+    }
+
+
+def load_rule_snapshot(value: Mapping[str, Any] | str | None) -> dict[str, Any]:
+    if isinstance(value, str):
+        try:
+            loaded = json.loads(value)
+        except (TypeError, ValueError):
+            loaded = {}
+    else:
+        loaded = dict(value or {})
+    fallback = build_rule_snapshot()
+    return {**fallback, **loaded}
+
+
+def parse_datetime_value(
+    value: Any,
+    *,
+    timezone_name: str = "Asia/Shanghai",
+) -> datetime | None:
+    if value in (None, "") or isinstance(value, bool):
+        return None
+    timezone = ZoneInfo(timezone_name)
+    if isinstance(value, datetime):
+        parsed = value
+    elif isinstance(value, date):
+        parsed = datetime.combine(value, time.min)
+    elif isinstance(value, (int, float, Decimal)):
+        timestamp = float(value)
+        if timestamp > 10_000_000_000:
+            timestamp /= 1000
+        try:
+            parsed = datetime.fromtimestamp(timestamp, timezone)
+        except (OSError, OverflowError, ValueError):
+            return None
+    else:
+        text = str(value).strip()
+        if not text:
+            return None
+        if re.fullmatch(r"\d{10,13}", text):
+            return parse_datetime_value(int(text), timezone_name=timezone_name)
+        normalized = text.replace("Z", "+00:00")
+        parsed = None
+        for candidate in (
+            normalized,
+            normalized.replace("/", "-"),
+        ):
+            try:
+                parsed = datetime.fromisoformat(candidate)
+                break
+            except ValueError:
+                continue
+        if parsed is None:
+            for pattern in ("%Y%m%d", "%Y-%m-%d", "%Y/%m/%d"):
+                try:
+                    parsed = datetime.strptime(text, pattern)
+                    break
+                except ValueError:
+                    continue
+        if parsed is None:
+            return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone)
+    return parsed.astimezone(timezone)
+
+
+def normalize_duration_seconds(value: Any, *, unit: str = "seconds") -> Decimal | None:
+    if value in (None, "") or isinstance(value, bool):
+        return None
+    try:
+        number = Decimal(str(value))
+    except (ArithmeticError, ValueError):
+        return None
+    if unit == "milliseconds":
+        number /= Decimal("1000")
+    if number < 0:
+        return None
+    return number.quantize(Decimal("0.001"))
+
+
+def _candidate_text(candidate: Mapping[str, Any]) -> str:
+    fragments = [str(candidate.get("title") or "")]
+    tags = candidate.get("tags")
+    if tags is None:
+        tags = candidate.get("tags_json")
+    if isinstance(tags, str):
+        try:
+            loaded = json.loads(tags)
+        except (TypeError, ValueError):
+            loaded = tags
+        tags = loaded
+    if isinstance(tags, list):
+        fragments.extend(str(item) for item in tags)
+    elif tags:
+        fragments.append(str(tags))
+    evidence = candidate.get("temporal_evidence")
+    if evidence is None:
+        evidence = candidate.get("temporal_evidence_json")
+    if evidence:
+        fragments.append(str(evidence))
+    return " ".join(fragments)
+
+
+def _time_in_window(current: time, start: time, end: time) -> bool:
+    if start <= end:
+        return start <= current <= end
+    return current >= start or current <= end
+
+
+def _festival_dates(year: int) -> dict[str, date]:
+    result = {
+        name: date(year, month, day)
+        for name, (month, day) in _FIXED_FESTIVALS.items()
+    }
+    for name, (month, day) in _LUNAR_FESTIVALS.get(year, {}).items():
+        result[name] = date(year, month, day)
+    return result
+
+
+def _nearest_festival_date(name: str, current: date) -> date | None:
+    candidates = [
+        festival_date
+        for year in (current.year - 1, current.year, current.year + 1)
+        if (festival_date := _festival_dates(year).get(name)) is not None
+    ]
+    if not candidates:
+        return None
+    return min(candidates, key=lambda value: abs((value - current).days))
+
+
+def evaluate_temporal_status(
+    candidate: Mapping[str, Any],
+    rule_snapshot: Mapping[str, Any] | str | None,
+) -> dict[str, Any]:
+    rules = load_rule_snapshot(rule_snapshot)
+    timezone_name = str(rules["timezone"])
+    timezone = ZoneInfo(timezone_name)
+    current = parse_datetime_value(
+        rules.get("current_datetime"),
+        timezone_name=timezone_name,
+    ) or datetime.now(timezone)
+    publish_at = parse_datetime_value(
+        candidate.get("publish_at"),
+        timezone_name=timezone_name,
+    )
+    text = _candidate_text(candidate)
+    inferred_type = str(candidate.get("temporal_type") or "").strip() or "evergreen"
+    evidence: dict[str, Any] = {
+        "evaluated_at": current.isoformat(timespec="seconds"),
+        "publish_at": publish_at.isoformat(timespec="seconds") if publish_at else None,
+        "matched_terms": [],
+    }
+
+    if publish_at is None:
+        return {
+            "temporal_type": inferred_type,
+            "status": "unknown",
+            "reason_code": "TEMPORAL_UNKNOWN",
+            "reason": "缺少真实发布时间，无法完成时效校验",
+            "evidence": evidence,
+        }
+
+    for daypart, terms, start, end in _DAYPART_RULES:
+        matched = [term for term in terms if term in text]
+        if not matched:
+            continue
+        inferred_type = "daypart"
+        evidence["matched_terms"].extend(matched)
+        evidence["valid_time"] = (
+            f"{start.strftime('%H:%M')}-{end.strftime('%H:%M')}"
+        )
+        if not _time_in_window(current.timetz().replace(tzinfo=None), start, end):
+            return {
+                "temporal_type": inferred_type,
+                "status": "fail",
+                "reason_code": "DAYPART_EXPIRED",
+                "reason": f"当前时间不在{daypart}内容有效时段",
+                "evidence": evidence,
+            }
+
+    matched_festivals: list[tuple[str, date]] = []
+    known_names = set(_FIXED_FESTIVALS)
+    for yearly in _LUNAR_FESTIVALS.values():
+        known_names.update(yearly)
+    for name in sorted(known_names, key=len, reverse=True):
+        if name not in text:
+            continue
+        festival_date = _nearest_festival_date(name, current.date())
+        if festival_date is None:
+            return {
+                "temporal_type": "festival",
+                "status": "unknown",
+                "reason_code": "TEMPORAL_UNKNOWN",
+                "reason": f"规则版本未覆盖当前相邻年份的{name}日期",
+                "evidence": {**evidence, "matched_terms": [name]},
+            }
+        matched_festivals.append((name, festival_date))
+        break
+    if matched_festivals:
+        inferred_type = "festival"
+        name, festival_date = matched_festivals[0]
+        lead_days = int(rules["festival_lead_days"])
+        valid_from = festival_date - timedelta(days=lead_days)
+        valid_to = festival_date + timedelta(days=1)
+        evidence.update(
+            {
+                "matched_terms": [name],
+                "valid_from": valid_from.isoformat(),
+                "valid_to": valid_to.isoformat(),
+            }
+        )
+        if not valid_from <= current.date() <= valid_to:
+            return {
+                "temporal_type": inferred_type,
+                "status": "fail",
+                "reason_code": "FESTIVAL_OUT_OF_WINDOW",
+                "reason": f"{name}内容不在当前有效窗口",
+                "evidence": evidence,
+            }
+
+    matched_relative = [term for term in _RELATIVE_DATE_MARKERS if term in text]
+    if matched_relative and publish_at.date() != current.date():
+        evidence["matched_terms"].extend(matched_relative)
+        return {
+            "temporal_type": "event",
+            "status": "fail",
+            "reason_code": "RELATIVE_DATE_EXPIRED",
+            "reason": "内容包含相对日期表述，但并非当天发布",
+            "evidence": evidence,
+        }
+
+    if any(marker in text for marker in _EVENT_MARKERS):
+        inferred_type = "event"
+    age_days = max(0.0, (current - publish_at).total_seconds() / 86400)
+    evidence["content_age_days"] = round(age_days, 3)
+    if inferred_type == "event" and age_days > int(rules["event_max_age_days"]):
+        return {
+            "temporal_type": inferred_type,
+            "status": "fail",
+            "reason_code": "EVENT_EXPIRED",
+            "reason": "事件型内容超过允许的新鲜度窗口",
+            "evidence": evidence,
+        }
+    if (
+        inferred_type == "seasonal"
+        and age_days > int(rules["seasonal_max_age_days"])
+    ):
+        return {
+            "temporal_type": inferred_type,
+            "status": "fail",
+            "reason_code": "SEASONAL_EXPIRED",
+            "reason": "季节型内容超过允许的新鲜度窗口",
+            "evidence": evidence,
+        }
+
+    explicit_status = str(candidate.get("temporal_status") or "").strip()
+    if explicit_status in {"fail", "unknown"}:
+        reason_code = (
+            "TEMPORAL_EXPIRED" if explicit_status == "fail" else "TEMPORAL_UNKNOWN"
+        )
+        return {
+            "temporal_type": inferred_type,
+            "status": explicit_status,
+            "reason_code": reason_code,
+            "reason": "候选补证结果未通过时间有效性判断",
+            "evidence": evidence,
+        }
+    return {
+        "temporal_type": inferred_type,
+        "status": "pass",
+        "reason_code": None,
+        "reason": "发布时间与基础时间语义有效",
+        "evidence": evidence,
+    }
+
+
+def _number(value: Any) -> float | None:
+    if value in (None, "") or isinstance(value, bool):
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def evaluate_candidate_gate(
+    candidate: Mapping[str, Any],
+    rule_snapshot: Mapping[str, Any] | str | None,
+) -> dict[str, Any]:
+    rules = load_rule_snapshot(rule_snapshot)
+    temporal = evaluate_temporal_status(candidate, rules)
+    duration = _number(candidate.get("duration_seconds"))
+    shares = _number(candidate.get("share_count"))
+    content_ratio = _number(candidate.get("content_50_plus_ratio"))
+    account_ratio = _number(candidate.get("account_50_plus_ratio"))
+
+    checks: list[dict[str, Any]] = [
+        {
+            "name": "temporal",
+            "status": temporal["status"],
+            "reason_code": temporal["reason_code"],
+            "actual": temporal["evidence"],
+        }
+    ]
+
+    def threshold_check(
+        name: str,
+        actual: float | None,
+        threshold: float,
+        *,
+        missing_code: str,
+        low_code: str,
+    ) -> None:
+        if actual is None:
+            status = "fail"
+            reason_code = missing_code
+        elif actual < threshold:
+            status = "fail"
+            reason_code = low_code
+        else:
+            status = "pass"
+            reason_code = None
+        checks.append(
+            {
+                "name": name,
+                "status": status,
+                "reason_code": reason_code,
+                "actual": actual,
+                "threshold": threshold,
+            }
+        )
+
+    threshold_check(
+        "duration_seconds",
+        duration,
+        float(rules["min_duration_seconds"]),
+        missing_code="DURATION_UNKNOWN",
+        low_code="DURATION_TOO_SHORT",
+    )
+    threshold_check(
+        "share_count",
+        shares,
+        float(rules["min_share_count"]),
+        missing_code="SHARE_COUNT_UNKNOWN",
+        low_code="SHARE_COUNT_TOO_LOW",
+    )
+    threshold_check(
+        "content_50_plus_ratio",
+        content_ratio,
+        float(rules["min_content_50_plus_ratio"]),
+        missing_code="CONTENT_PORTRAIT_MISSING",
+        low_code="CONTENT_50_PLUS_TOO_LOW",
+    )
+
+    failed_codes = [
+        str(check["reason_code"])
+        for check in checks
+        if check["status"] != "pass" and check.get("reason_code")
+    ]
+    content_status = (
+        "missing"
+        if content_ratio is None
+        else (
+            "pass"
+            if content_ratio >= float(rules["min_content_50_plus_ratio"])
+            else "fail"
+        )
+    )
+    account_status = (
+        "missing"
+        if account_ratio is None
+        else (
+            "pass"
+            if account_ratio >= float(rules["min_account_50_plus_ratio"])
+            else "fail"
+        )
+    )
+    portrait_conflict = (
+        content_status in {"pass", "fail"}
+        and account_status in {"pass", "fail"}
+        and content_status != account_status
+    )
+    return {
+        "rule_version": str(rules["rule_version"]),
+        "status": "pass" if not failed_codes else "fail",
+        "primary_eligible": not failed_codes,
+        "failed_reason_codes": failed_codes,
+        "checks": checks,
+        "temporal": temporal,
+        "content_portrait_status": content_status,
+        "account_portrait_status": account_status,
+        "portrait_conflict": portrait_conflict,
+    }

tests/supply_infra/scheduler/test_discover_videos_from_demands.py

@@ -4,6 +4,8 @@ import asyncio
 import importlib
 import json
 from contextlib import contextmanager
+from datetime import datetime
+from decimal import Decimal
 from unittest.mock import patch
 
 import pytest
@@ -21,6 +23,17 @@ from agents.find_agent.demand_run import (
 )
 from agents.find_agent.tools import video_discovery_store
 from agents.find_agent.tools.batch_search_and_record import batch_search_and_record
+from agents.find_agent.support.search_persistence import _candidate_from_search_result
+from agents.find_agent.support.douyin_search import (
+    DEFAULT_MIN_DURATION_SECONDS as INTERNAL_SEARCH_MIN_DURATION,
+)
+from agents.find_agent.support.douyin_search_tikhub import (
+    DEFAULT_MIN_DURATION_SECONDS as TIKHUB_SEARCH_MIN_DURATION,
+)
+from agents.find_agent.support.douyin_user_videos import (
+    DEFAULT_MIN_DURATION_SECONDS as AUTHOR_SEARCH_MIN_DURATION,
+)
+from supply_infra.video_discovery_gates import evaluate_candidate_gate
 from supply_infra.db.models.video_discovery import (
     VideoDiscoveryCandidate,
     VideoDiscoveryRun,
@@ -277,6 +290,153 @@ def test_find_agent_input_uses_reference_videos_without_seed_fields() -> None:
     assert '"title": "参考标题"' in user_input
     assert "create_video_discovery_run" not in user_input
     assert "relevant_points" not in user_input
+    assert "current_datetime：" in user_input
+    assert "timezone：Asia/Shanghai" in user_input
+    assert "quality_gate_rules：" in user_input
+
+
+_P0_RULES = {
+    "rule_version": "test-p0",
+    "timezone": "Asia/Shanghai",
+    "current_datetime": "2026-07-31T12:00:00+08:00",
+    "current_date": "2026-07-31",
+    "min_duration_seconds": 30,
+    "min_share_count": 1000,
+    "min_content_50_plus_ratio": 0.20,
+    "min_account_50_plus_ratio": 0.20,
+    "festival_lead_days": 7,
+    "event_max_age_days": 7,
+    "seasonal_max_age_days": 180,
+}
+
+
+def _p0_candidate(**overrides):
+    candidate = {
+        "title": "适合家庭分享的生活技巧",
+        "publish_at": "2026-07-31T09:00:00+08:00",
+        "duration_seconds": 30,
+        "share_count": 1000,
+        "content_50_plus_ratio": 0.20,
+        "account_50_plus_ratio": 0.30,
+    }
+    candidate.update(overrides)
+    return candidate
+
+
+def test_p0_gate_enforces_boundaries_and_time_context() -> None:
+    assert evaluate_candidate_gate(_p0_candidate(), _P0_RULES)["primary_eligible"]
+
+    short = evaluate_candidate_gate(
+        _p0_candidate(duration_seconds=Decimal("29.999")),
+        _P0_RULES,
+    )
+    assert "DURATION_TOO_SHORT" in short["failed_reason_codes"]
+
+    low_share = evaluate_candidate_gate(_p0_candidate(share_count=999), _P0_RULES)
+    assert "SHARE_COUNT_TOO_LOW" in low_share["failed_reason_codes"]
+
+    low_elder = evaluate_candidate_gate(
+        _p0_candidate(content_50_plus_ratio=0.199),
+        _P0_RULES,
+    )
+    assert "CONTENT_50_PLUS_TOO_LOW" in low_elder["failed_reason_codes"]
+
+    morning = evaluate_candidate_gate(
+        _p0_candidate(title="早上好，送给家人的祝福"),
+        _P0_RULES,
+    )
+    assert "DAYPART_EXPIRED" in morning["failed_reason_codes"]
+
+    festival = evaluate_candidate_gate(
+        _p0_candidate(title="春节祝福送给全家"),
+        _P0_RULES,
+    )
+    assert "FESTIVAL_OUT_OF_WINDOW" in festival["failed_reason_codes"]
+
+
+def test_p0_gate_keeps_content_and_account_portraits_separate() -> None:
+    conflict = evaluate_candidate_gate(
+        _p0_candidate(
+            content_50_plus_ratio=0.28,
+            account_50_plus_ratio=0.08,
+        ),
+        _P0_RULES,
+    )
+    assert conflict["content_portrait_status"] == "pass"
+    assert conflict["account_portrait_status"] == "fail"
+    assert conflict["portrait_conflict"] is True
+    assert conflict["primary_eligible"] is True
+
+    account_only = evaluate_candidate_gate(
+        _p0_candidate(
+            content_50_plus_ratio=None,
+            account_50_plus_ratio=0.55,
+        ),
+        _P0_RULES,
+    )
+    assert "CONTENT_PORTRAIT_MISSING" in account_only["failed_reason_codes"]
+
+
+def test_search_candidate_persists_publish_time_duration_and_shares() -> None:
+    candidate = _candidate_from_search_result(
+        {
+            "aweme_id": "video-p0",
+            "desc": "测试视频",
+            "duration_ms": 65000,
+            "publish_at": "2026-07-31T08:30:00+08:00",
+            "statistics": {"share_count": 45},
+        },
+        "测试关键词",
+    )
+    assert candidate is not None
+    assert candidate["duration_seconds"] == Decimal("65.000")
+    assert candidate["publish_at"] == datetime(2026, 7, 31, 8, 30)
+    assert candidate["share_count"] == 45
+
+
+def test_all_search_sources_default_to_thirty_seconds() -> None:
+    assert INTERNAL_SEARCH_MIN_DURATION == 30
+    assert TIKHUB_SEARCH_MIN_DURATION == 30
+    assert AUTHOR_SEARCH_MIN_DURATION == 30
+
+
+def test_repository_rejects_primary_that_fails_p0_gate() -> None:
+    factory = _expire_on_commit_session_factory()
+    with factory() as session:
+        session.add(
+            VideoDiscoveryRun(
+                id=1,
+                run_id="p0-gate-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="p0-gate-run",
+                aweme_id="video-p0",
+                title="适合家庭分享的生活技巧",
+                publish_at=datetime(2026, 7, 31, 9, 0),
+                duration_seconds=Decimal("30.000"),
+                share_count=999,
+                content_50_plus_ratio=Decimal("0.280000"),
+                account_50_plus_ratio=Decimal("0.080000"),
+                decision_bucket="pending_evaluation",
+            )
+        )
+        session.commit()
+
+    with factory() as session:
+        repo = VideoDiscoveryRepository(session)
+        with pytest.raises(ValueError, match="SHARE_COUNT_TOO_LOW"):
+            repo.update_candidates(
+                "p0-gate-run",
+                [{"candidate_id": 2, "decision_bucket": "primary"}],
+            )
 
 
 @pytest.mark.asyncio