修改可视化页面

api/routers/find_agent_v2.py

@@ -3,7 +3,10 @@ from __future__ import annotations
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, Request, status
 
 from api.schemas.find_agent_v2 import CreateFindAgentV2TestBody
-from api.services.find_agent_v2 import execute_test_run, get_run_detail, list_runs, prepare_test_run
+from api.services.find_agent_v2 import (
+    execute_test_run, get_run_candidates, get_run_detail, get_search_candidates,
+    list_runs, prepare_test_run,
+)
 
 router = APIRouter(prefix="/api/find-agent-v2", tags=["find-agent-v2"])
 
@@ -12,10 +15,12 @@ router = APIRouter(prefix="/api/find-agent-v2", tags=["find-agent-v2"])
 def runs(
     status_filter: str | None = Query(default=None, alias="status"),
     keyword: str | None = Query(default=None, max_length=256),
-    limit: int = Query(default=30, ge=1, le=100),
-    offset: int = Query(default=0, ge=0),
+    page: int = Query(default=1, ge=1),
+    page_size: int = Query(default=30, ge=1, le=100),
 ) -> dict:
-    return list_runs(status=status_filter, keyword=keyword, limit=limit, offset=offset)
+    return list_runs(
+        status=status_filter, keyword=keyword, page=page, page_size=page_size,
+    )
 
 
 @router.get("/runs/{run_id}")
@@ -26,6 +31,22 @@ def run_detail(run_id: str) -> dict:
     return result
 
 
+@router.get("/runs/{run_id}/candidates")
+def run_candidates(run_id: str) -> dict:
+    items = get_run_candidates(run_id)
+    if items is None:
+        raise HTTPException(status_code=404, detail="find_agent_v2 run not found")
+    return {"items": items, "count": len(items)}
+
+
+@router.get("/runs/{run_id}/searches/{search_id}/candidates")
+def search_candidates(run_id: str, search_id: int) -> dict:
+    items = get_search_candidates(run_id, search_id)
+    if items is None:
+        raise HTTPException(status_code=404, detail="find_agent_v2 search not found")
+    return {"items": items, "count": len(items)}
+
+
 @router.post("/tests", status_code=status.HTTP_202_ACCEPTED)
 def create_test(
     body: CreateFindAgentV2TestBody,

api/services/find_agent_v2.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import json
+from datetime import datetime, timedelta
 from typing import Any
+from zoneinfo import ZoneInfo
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import func, or_, select, update
 
 from find_agent_v2.demand_context import prepare_latest_v2_demand_run_by_name
 from find_agent_v2.models import (
@@ -16,10 +18,33 @@ from find_agent_v2.models import (
     FindAgentV2Search,
 )
 from find_agent_v2.runner import run_prepared_find_agent_v2
-from find_agent_v2.service import get_find_agent_v2_service
+from find_agent_v2.service import (
+    RUN_TIMEOUT_MINUTES,
+    RUN_TIMEOUT_REASON,
+    get_find_agent_v2_service,
+)
 from supply_infra.db.session import get_session
 
 
+def _expire_overdue_runs(session, *, now: datetime | None = None) -> int:
+    current = now or datetime.now(ZoneInfo("Asia/Shanghai")).replace(tzinfo=None)
+    cutoff = current - timedelta(minutes=RUN_TIMEOUT_MINUTES)
+    result = session.execute(
+        update(FindAgentV2Run)
+        .where(
+            FindAgentV2Run.status == "running",
+            FindAgentV2Run.create_time < cutoff,
+        )
+        .values(
+            status="failed",
+            outcome_status="failed",
+            stop_reason=RUN_TIMEOUT_REASON,
+            update_time=current,
+        )
+    )
+    return int(result.rowcount or 0)
+
+
 def _json(raw: str | None, default: Any) -> Any:
     try:
         return json.loads(raw) if raw else default
@@ -62,11 +87,69 @@ def _run(row: FindAgentV2Run) -> dict[str, Any]:
     }
 
 
+def _evidence_item(row: FindAgentV2Evidence) -> dict[str, Any]:
+    return {
+        "id": int(row.id), "candidate_id": row.candidate_id,
+        "evidence_type": row.evidence_type, "provider": row.provider,
+        "status": row.status, "raw": _json(row.raw_json, {}),
+        "normalized": _json(row.normalized_json, {}),
+        "error_message": row.error_message, "create_time": _time(row.create_time),
+    }
+
+
+def _candidate_item(
+    row: FindAgentV2Candidate, evidence: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    return {
+        "id": int(row.id), "first_search_id": row.first_search_id,
+        "aweme_id": row.aweme_id, "title": row.title, "content_link": row.content_link,
+        "author_name": row.author_name, "author_sec_uid": row.author_sec_uid,
+        "source_keywords": _json(row.source_keywords_json, []), "tags": _json(row.tags_json, []),
+        "publish_at": _time(row.publish_at), "duration_seconds": _number(row.duration_seconds),
+        "play_count": row.play_count, "like_count": row.like_count,
+        "comment_count": row.comment_count, "collect_count": row.collect_count,
+        "share_count": row.share_count, "detail": _json(row.detail_json, {}),
+        "portrait": _json(row.portrait_json, {}), "detail_status": row.detail_status,
+        "portrait_status": row.portrait_status,
+        "content_50_plus_ratio": _number(row.content_50_plus_ratio),
+        "account_50_plus_ratio": _number(row.account_50_plus_ratio),
+        "relevance_score": _number(row.relevance_score), "elder_score": _number(row.elder_score),
+        "share_score": _number(row.share_score), "value_score": _number(row.value_score),
+        "gate_status": row.gate_status, "gate_result": _json(row.gate_result_json, {}),
+        "decision_bucket": row.decision_bucket, "decision_reason": row.decision_reason,
+        "reject_reason_code": row.reject_reason_code, "evidence": evidence or [],
+        "create_time": _time(row.create_time), "update_time": _time(row.update_time),
+    }
+
+
+def _candidate_rows_with_evidence(session, query) -> list[dict[str, Any]]:
+    rows = list(session.scalars(query))
+    ids = [int(row.id) for row in rows]
+    evidence_rows = list(session.scalars(
+        select(FindAgentV2Evidence).where(FindAgentV2Evidence.candidate_id.in_(ids))
+        .order_by(FindAgentV2Evidence.id)
+    )) if ids else []
+    evidence_by_candidate: dict[int, list[dict[str, Any]]] = {}
+    for row in evidence_rows:
+        if row.candidate_id is not None:
+            evidence_by_candidate.setdefault(int(row.candidate_id), []).append(
+                _evidence_item(row)
+            )
+    return [
+        _candidate_item(row, evidence_by_candidate.get(int(row.id), []))
+        for row in rows
+    ]
+
+
 def list_runs(
     *, status: str | None = None, keyword: str | None = None,
-    limit: int = 30, offset: int = 0,
+    page: int = 1, page_size: int = 30,
 ) -> dict[str, Any]:
+    normalized_page = max(1, int(page))
+    normalized_page_size = min(100, max(1, int(page_size)))
+    offset = (normalized_page - 1) * normalized_page_size
     with get_session() as session:
+        _expire_overdue_runs(session)
         conditions = []
         if status:
             conditions.append(FindAgentV2Run.status == status)
@@ -84,13 +167,23 @@ def list_runs(
         rows = list(session.scalars(
             select(FindAgentV2Run).where(*conditions)
             .order_by(FindAgentV2Run.create_time.desc(), FindAgentV2Run.id.desc())
-            .limit(limit).offset(offset)
+            .limit(normalized_page_size).offset(offset)
         ))
-        return {"items": [_run(row) for row in rows], "total": total, "limit": limit, "offset": offset}
+        total_pages = max(1, (total + normalized_page_size - 1) // normalized_page_size)
+        return {
+            "items": [_run(row) for row in rows],
+            "total": total,
+            "page": normalized_page,
+            "page_size": normalized_page_size,
+            "total_pages": total_pages,
+            "has_previous": normalized_page > 1,
+            "has_next": normalized_page < total_pages,
+        }
 
 
 def get_run_detail(run_id: str) -> dict[str, Any] | None:
     with get_session() as session:
+        _expire_overdue_runs(session)
         run = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
         if run is None:
             return None
@@ -102,14 +195,9 @@ def get_run_detail(run_id: str) -> dict[str, Any] | None:
             select(FindAgentV2Search).where(FindAgentV2Search.run_id == run_id)
             .order_by(FindAgentV2Search.round_index, FindAgentV2Search.id)
         ))
-        candidates = list(session.scalars(
-            select(FindAgentV2Candidate).where(FindAgentV2Candidate.run_id == run_id)
-            .order_by(FindAgentV2Candidate.id)
-        ))
-        evidence = list(session.scalars(
-            select(FindAgentV2Evidence).where(FindAgentV2Evidence.run_id == run_id)
-            .order_by(FindAgentV2Evidence.id)
-        ))
+        candidate_summaries = list(session.execute(select(
+            FindAgentV2Candidate.aweme_id, FindAgentV2Candidate.decision_bucket,
+        ).where(FindAgentV2Candidate.run_id == run_id)))
         round_items = [{
             "id": int(row.id), "round_index": row.round_index, "phase": row.phase,
             "status": row.status, "plan": _json(row.plan_json, {}),
@@ -123,76 +211,73 @@ def get_run_detail(run_id: str) -> dict[str, Any] | None:
             "query_reason": row.query_reason, "source_type": row.source_type,
             "provider": row.provider, "cursor": row.cursor, "page_no": row.page_no,
             "has_more": bool(row.has_more), "next_cursor": row.next_cursor,
-            "provider_state": _json(row.provider_state_json, {}),
             "result_count": row.result_count, "status": row.status,
             "error_message": row.error_message,
-            "raw_response": _json(row.raw_response_json, {}),
             "create_time": _time(row.create_time),
         } for row in searches]
-        evidence_by_candidate: dict[int, list[dict[str, Any]]] = {}
-        evidence_items = []
-        for row in evidence:
-            item = {
-                "id": int(row.id), "candidate_id": row.candidate_id,
-                "evidence_type": row.evidence_type, "provider": row.provider,
-                "status": row.status, "raw": _json(row.raw_json, {}),
-                "normalized": _json(row.normalized_json, {}),
-                "error_message": row.error_message, "create_time": _time(row.create_time),
-            }
-            evidence_items.append(item)
-            if row.candidate_id is not None:
-                evidence_by_candidate.setdefault(int(row.candidate_id), []).append(item)
-        candidate_items = [{
-            "id": int(row.id), "first_search_id": row.first_search_id,
-            "aweme_id": row.aweme_id, "title": row.title, "content_link": row.content_link,
-            "author_name": row.author_name, "author_sec_uid": row.author_sec_uid,
-            "source_keywords": _json(row.source_keywords_json, []), "tags": _json(row.tags_json, []),
-            "publish_at": _time(row.publish_at), "duration_seconds": _number(row.duration_seconds),
-            "play_count": row.play_count, "like_count": row.like_count,
-            "comment_count": row.comment_count, "collect_count": row.collect_count,
-            "share_count": row.share_count, "detail": _json(row.detail_json, {}),
-            "portrait": _json(row.portrait_json, {}), "detail_status": row.detail_status,
-            "portrait_status": row.portrait_status,
-            "content_50_plus_ratio": _number(row.content_50_plus_ratio),
-            "account_50_plus_ratio": _number(row.account_50_plus_ratio),
-            "relevance_score": _number(row.relevance_score), "elder_score": _number(row.elder_score),
-            "share_score": _number(row.share_score), "value_score": _number(row.value_score),
-            "gate_status": row.gate_status, "gate_result": _json(row.gate_result_json, {}),
-            "decision_bucket": row.decision_bucket, "decision_reason": row.decision_reason,
-            "reject_reason_code": row.reject_reason_code,
-            "evidence": evidence_by_candidate.get(int(row.id), []),
-            "create_time": _time(row.create_time), "update_time": _time(row.update_time),
-        } for row in candidates]
-        candidates_by_aweme = {item["aweme_id"]: item for item in candidate_items}
+        bucket_by_aweme = {
+            str(aweme_id): str(bucket) for aweme_id, bucket in candidate_summaries
+        }
+        search_rows_by_id = {int(row.id): row for row in searches}
         for search in search_items:
-            raw_results = search["raw_response"].get("search_results", [])
-            result_ids = [
-                str(item.get("aweme_id") or "")
+            row = search_rows_by_id[search["id"]]
+            raw_results = _json(row.raw_response_json, {}).get("search_results", [])
+            result_ids = list(dict.fromkeys(
+                str(item.get("aweme_id"))
                 for item in raw_results
                 if isinstance(item, dict) and item.get("aweme_id")
-            ]
-            search["result_aweme_ids"] = result_ids
-            search["matched_candidates"] = [
-                candidates_by_aweme[aweme_id]
-                for aweme_id in result_ids
-                if aweme_id in candidates_by_aweme
-            ]
+            ))
+            buckets = [bucket_by_aweme.get(item, "pending_evaluation") for item in result_ids]
+            search["video_count"] = len(result_ids)
+            search["primary_count"] = sum(item == "primary" for item in buckets)
+            search["rejected_count"] = sum(item == "rejected" for item in buckets)
+            search["pending_count"] = (
+                search["video_count"] - search["primary_count"] - search["rejected_count"]
+            )
         timeline = []
         for row in round_items:
             timeline.append({"type": "round", "time": row["create_time"], "data": row})
         for row in search_items:
             timeline.append({"type": "search", "time": row["create_time"], "data": row})
-        for row in evidence_items:
-            timeline.append({"type": "evidence", "time": row["create_time"], "data": row})
         timeline.sort(key=lambda item: str(item.get("time") or ""))
         return {
             "run": {**_run(run), "input": _json(run.input_json, {}),
                     "rule_config": _json(run.rule_config_json, {})},
-            "rounds": round_items, "searches": search_items, "candidates": candidate_items,
-            "evidence": evidence_items, "timeline": timeline,
+            "rounds": round_items, "searches": search_items, "timeline": timeline,
         }
 
 
+def get_run_candidates(run_id: str) -> list[dict[str, Any]] | None:
+    with get_session() as session:
+        if session.scalar(select(FindAgentV2Run.id).where(FindAgentV2Run.run_id == run_id)) is None:
+            return None
+        return _candidate_rows_with_evidence(session, select(FindAgentV2Candidate).where(
+            FindAgentV2Candidate.run_id == run_id,
+        ).order_by(FindAgentV2Candidate.id))
+
+
+def get_search_candidates(run_id: str, search_id: int) -> list[dict[str, Any]] | None:
+    with get_session() as session:
+        search = session.scalar(select(FindAgentV2Search).where(
+            FindAgentV2Search.run_id == run_id, FindAgentV2Search.id == int(search_id),
+        ))
+        if search is None:
+            return None
+        raw_results = _json(search.raw_response_json, {}).get("search_results", [])
+        aweme_ids = list(dict.fromkeys(
+            str(item.get("aweme_id")) for item in raw_results
+            if isinstance(item, dict) and item.get("aweme_id")
+        ))
+        if not aweme_ids:
+            return []
+        items = _candidate_rows_with_evidence(session, select(FindAgentV2Candidate).where(
+            FindAgentV2Candidate.run_id == run_id,
+            FindAgentV2Candidate.aweme_id.in_(aweme_ids),
+        ))
+        by_aweme = {item["aweme_id"]: item for item in items}
+        return [by_aweme[item] for item in aweme_ids if item in by_aweme]
+
+
 def prepare_test_run(demand_word: str, current_user: dict[str, Any]) -> dict[str, Any]:
     prepared = prepare_latest_v2_demand_run_by_name(
         demand_word,

find_agent_v2/AGENT_FLOW.md

@@ -27,14 +27,11 @@ flowchart TD
         Q --> X["结束本轮"]
     end
 
-    X --> T{"valid primary ≥ 目标数?"}
-    T -- 是 --> F["finalize: goal_met"]
-    T -- 否 --> N{"仍有 pending?"}
+    X --> N{"仍有 pending?"}
     N -- 是 --> Z["finalize: failed"]
-    N -- 否 --> I{"本轮有新增候选且未达最大轮数?"}
+    N -- 否 --> I{"候选增量和评估通过率<br/>支持继续探索?"}
     I -- 是 --> C
-    I -- 否 --> O["finalize: partial 或 no_match"]
-    F --> RP["Report 只读生成报告"]
+    I -- 否 --> O["finalize: goal_met 或 no_match"]
     O --> RP
     Z --> END["结束，不执行 Report"]
     RP --> END
@@ -43,7 +40,6 @@ flowchart TD
 默认参数：
 
 - 最大业务轮数：`2`
-- 目标有效 Primary：`5`
 - 默认模型：`google/gemini-3-flash-preview`
 - 节点温度：`0.2`
 
@@ -201,10 +197,6 @@ portrait_status = pending
 
 | 原因码 | 含义 |
 |---|---|
-| `TEMPORAL_UNKNOWN` | 发布时间缺失且未满足补偿 |
-| `RELATIVE_DATE_EXPIRED` | 标题含相对日期，但内容已非当天 |
-| `EVENT_EXPIRED` | 事件型内容超出时效 |
-| `SEASONAL_EXPIRED` | 季节型内容超出时效 |
 | `DURATION_UNKNOWN` | 时长缺失且未满足补偿 |
 | `DURATION_TOO_SHORT` | 时长低于阈值 |
 | `SHARE_COUNT_UNKNOWN` | 分享量缺失且未满足补偿 |
@@ -218,24 +210,23 @@ portrait_status = pending
 
 ```mermaid
 flowchart TD
-    A["单轮结束"] --> B{"valid_primary_count ≥ target?"}
-    B -- 是 --> G["goal_met / finished"]
-    B -- 否 --> C{"pending_count > 0?"}
+    A["单轮结束"] --> C{"pending_count > 0?"}
     C -- 是 --> F["failed / failed"]
-    C -- 否 --> D{"candidate_count 有增长?"}
-    D -- 否 --> S["停止扩展"]
+    C -- 否 --> D{"候选增量、评估样本量和<br/>通过率支持继续探索?"}
+    D -- 否 --> S["停止探索"]
     D -- 是 --> E{"达到 max_rounds?"}
     E -- 否 --> NR["进入下一轮"]
     E -- 是 --> S
     S --> H{"有效 primary 数量"}
-    H -- 大于 0 --> P["partial / finished"]
+    H -- 大于 0 --> P["goal_met / finished"]
     H -- 等于 0 --> N["no_match / finished"]
 ```
 
 技术异常统一进入 `failed/failed`。非技术失败的终态会继续运行只读 Report 节点；Report 失败只
 记录到内存 `state.failures`，不会反向改变已完成的业务终态。
 
-`finalize()` 使用与外层 Agent 相同的 `target_primary_count` 判断 `goal_met`，默认值为 5。
+`finalize()` 根据是否存在有效 Primary 判断 `goal_met`；探索是否继续由候选增量、评估样本量和
+通过率共同决定。
 
 ## 7. 数据表和状态归属
 

find_agent_v2/agent.py

@@ -2,19 +2,76 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
+from dataclasses import dataclass
 
 from find_agent_v2.context import build_node_slots
 from find_agent_v2.graph import FindAgentRoundGraph, NodeRunner
 from find_agent_v2.observability import ObagentObserver
 from find_agent_v2.prompts import REPORT_PROMPT
 from find_agent_v2.runtime import FindAgentNodeHost, normalize_models
-from find_agent_v2.service import FindAgentV2Service, get_find_agent_v2_service
-from find_agent_v2.state import FindAgentResult, FindAgentState
+from find_agent_v2.service import (
+    RUN_TIMEOUT_REASON,
+    RUN_TIMEOUT_SECONDS,
+    FindAgentV2Service,
+    get_find_agent_v2_service,
+)
+from find_agent_v2.state import DiscoverySnapshot, FindAgentResult, FindAgentState
 from find_agent_v2.tools import REPORT_TOOLS
 from supply_agent.config import Settings
 
 
+@dataclass(frozen=True)
+class ExplorationDecision:
+    continue_exploring: bool
+    reason: str
+    new_candidate_count: int
+    new_evaluated_count: int
+    round_pass_rate: float
+    cumulative_pass_rate: float
+
+
+def decide_continued_exploration(
+    previous: DiscoverySnapshot, current: DiscoverySnapshot,
+) -> ExplorationDecision:
+    """Decide whether to continue from discovery volume and evaluation yield."""
+    new_candidates = max(0, current.candidate_count - previous.candidate_count)
+    previous_evaluated = previous.primary_count + previous.rejected_count
+    current_evaluated = current.primary_count + current.rejected_count
+    new_evaluated = max(0, current_evaluated - previous_evaluated)
+    new_passed = max(0, current.valid_primary_count - previous.valid_primary_count)
+    round_rate = new_passed / new_evaluated if new_evaluated else 0.0
+    cumulative_rate = (
+        current.valid_primary_count / current_evaluated if current_evaluated else 0.0
+    )
+
+    if new_candidates == 0:
+        keep_going = False
+        reason = "本轮没有新增候选，搜索前沿已无信息增益"
+    elif current_evaluated < 8:
+        keep_going = True
+        reason = "已评估样本不足 8 条，继续探索以形成可靠通过率"
+    elif new_evaluated and round_rate == 0:
+        keep_going = False
+        reason = "本轮有足够评估样本但通过率为 0，继续搜索的预期收益低"
+    elif new_candidates >= 2 and (round_rate >= 0.10 or cumulative_rate >= 0.10):
+        keep_going = True
+        reason = "本轮仍有候选增量且评估保持正向通过率，继续探索"
+    else:
+        keep_going = False
+        reason = "候选增量或评估通过率不足，停止探索"
+
+    return ExplorationDecision(
+        continue_exploring=keep_going,
+        reason=reason,
+        new_candidate_count=new_candidates,
+        new_evaluated_count=new_evaluated,
+        round_pass_rate=round(round_rate, 4),
+        cumulative_pass_rate=round(cumulative_rate, 4),
+    )
+
+
 class FindAgentV2:
     """Python outer loop + guarded Supervisor graph + node-local ReAct."""
 
@@ -26,9 +83,9 @@ class FindAgentV2:
         settings: Settings | None = None,
         models_by_role: dict[str, str] | None = None,
         max_rounds: int = 2,
-        target_primary_count: int = 5,
         max_actions_per_round: int = 16,
         max_search_actions_per_round: int = 3,
+        max_runtime_seconds: float = RUN_TIMEOUT_SECONDS,
         observer: ObagentObserver | None = None,
     ) -> None:
         self.service = service or get_find_agent_v2_service()
@@ -40,9 +97,9 @@ class FindAgentV2:
         )
         self.models_by_role = dict(models_by_role or {})
         self.max_rounds = max(1, int(max_rounds))
-        self.target_primary_count = max(1, int(target_primary_count))
         self.max_actions_per_round = max(4, int(max_actions_per_round))
         self.max_search_actions_per_round = max(1, int(max_search_actions_per_round))
+        self.max_runtime_seconds = max(0.01, float(max_runtime_seconds))
 
     async def arun(
         self, *, run_id: str, user_input: str, resume: bool = False,
@@ -57,7 +114,6 @@ class FindAgentV2:
             service=self.service, runner=self.node_runner, observer=self.observer,
             max_actions=self.max_actions_per_round,
             max_search_actions=self.max_search_actions_per_round,
-            target_primary_count=self.target_primary_count,
         )
         reset_usage = getattr(self.node_runner, "reset_usage", None)
         if callable(reset_usage):
@@ -72,9 +128,28 @@ class FindAgentV2:
             self.service.set_obagent_run_uid(
                 run_id, getattr(observation_run, "run_uid", None),
             )
-            result = await self._arun_inner(
-                state=state, graph=graph, start_round=int(run.get("current_round") or 0) + 1,
-            )
+            try:
+                result = await asyncio.wait_for(
+                    self._arun_inner(
+                        state=state, graph=graph,
+                        start_round=int(run.get("current_round") or 0) + 1,
+                    ),
+                    timeout=self.max_runtime_seconds,
+                )
+            except TimeoutError:
+                self.service.fail_run(run_id, RUN_TIMEOUT_REASON)
+                final_run = self.service.require_run(run_id)
+                result = FindAgentResult(
+                    run_id=run_id,
+                    status="failed",
+                    succeeded=False,
+                    business_outcome="failed",
+                    valid_primary_count=int(final_run.get("valid_primary_count") or 0),
+                    rounds=state.round_index,
+                    final_output=f"find_agent_v2 failed：{RUN_TIMEOUT_REASON}",
+                    node_runs=tuple(state.node_runs),
+                    stop_reason=RUN_TIMEOUT_REASON,
+                )
             usage = getattr(self.node_runner, "usage", None)
             if isinstance(usage, dict):
                 self.service.add_usage(run_id, usage)
@@ -95,18 +170,23 @@ class FindAgentV2:
                 self.service.begin_round(run_id, round_index, state.previous_snapshot)
                 await graph.invoke(state)
                 current = state.snapshot or self.service.snapshot(run_id)
-                if current.valid_primary_count >= self.target_primary_count:
-                    end_reason = f"已获得 {current.valid_primary_count} 条有效 primary"
-                    break
                 if current.pending_count:
                     failed = True
                     end_reason = f"第 {round_index} 轮结束仍有 {current.pending_count} 条待评估候选"
                     break
-                if current.candidate_count <= state.previous_snapshot.candidate_count:
-                    end_reason = "本轮未发现新增候选，搜索前沿已无信息增益"
+                exploration = decide_continued_exploration(
+                    state.previous_snapshot, current,
+                )
+                if not exploration.continue_exploring:
+                    end_reason = exploration.reason
                     break
                 if round_index == end_round - 1:
-                    end_reason = f"达到最大业务轮数 {self.max_rounds}"
+                    end_reason = (
+                        f"达到最大业务轮数 {self.max_rounds}；{exploration.reason}；"
+                        f"本轮新增={exploration.new_candidate_count}，"
+                        f"本轮通过率={exploration.round_pass_rate:.1%}，"
+                        f"累计通过率={exploration.cumulative_pass_rate:.1%}"
+                    )
         except Exception as exc:
             failed = True
             end_reason = f"{type(exc).__name__}: {exc}"
@@ -123,7 +203,6 @@ class FindAgentV2:
             run_id,
             failed=failed,
             reason=end_reason,
-            target_primary_count=self.target_primary_count,
         )
         final_output = (
             f"find_agent_v2 {final_run['outcome_status']}，"

find_agent_v2/gates.py

@@ -89,58 +89,30 @@ def _number(value: Any) -> float | None:
         return None
 
 
-def _strong(candidate: Mapping[str, Any]) -> bool:
-    scores = [_number(candidate.get(key)) for key in ("relevance_score", "elder_score", "share_score")]
-    value = _number(candidate.get("value_score"))
-    return value is not None and value >= 0.65 or all(score is not None and score >= floor for score, floor in zip(scores, (0.70, 0.70, 0.65), strict=True))
-
-
-def _temporal(candidate: Mapping[str, Any], rules: Mapping[str, Any]) -> dict[str, Any]:
-    timezone_name = str(rules["timezone"])
-    current = parse_datetime_value(rules.get("current_datetime"), timezone_name=timezone_name) or datetime.now(ZoneInfo(timezone_name))
-    published = parse_datetime_value(candidate.get("publish_at"), timezone_name=timezone_name)
-    title = str(candidate.get("title") or "")
-    temporal_type = str(candidate.get("temporal_type") or "evergreen")
-    if published is None:
-        compensated = _strong(candidate) and not any(word in title for word in ("今天", "今日", "明天", "刚刚", "突发"))
-        return {"status": "pass" if compensated else "unknown", "reason_code": None if compensated else "TEMPORAL_UNKNOWN", "temporal_type": temporal_type, "evidence": {"publish_at": None, "compensated": compensated}}
-    age_days = max(0.0, (current - published).total_seconds() / 86400)
-    reason_code = None
-    if any(word in title for word in ("今天", "今日", "明天", "昨日", "昨天")) and published.date() != current.date():
-        reason_code = "RELATIVE_DATE_EXPIRED"
-    elif temporal_type == "event" and age_days > int(rules["event_max_age_days"]):
-        reason_code = "EVENT_EXPIRED"
-    elif temporal_type == "seasonal" and age_days > int(rules["seasonal_max_age_days"]):
-        reason_code = "SEASONAL_EXPIRED"
-    return {"status": "fail" if reason_code else "pass", "reason_code": reason_code, "temporal_type": temporal_type, "evidence": {"publish_at": published.isoformat(timespec="seconds"), "content_age_days": round(age_days, 3)}}
-
-
 def evaluate_candidate_gate(candidate: Mapping[str, Any], rule_snapshot: Mapping[str, Any] | str | None) -> dict[str, Any]:
     rules = _rules(rule_snapshot)
-    temporal = _temporal(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", **temporal}]
+    checks: list[dict[str, Any]] = []
 
-    def threshold(name: str, actual: float | None, minimum: float, missing: str, low: str, compensate: bool = False) -> None:
+    def threshold(name: str, actual: float | None, minimum: float, missing: str, low: str) -> None:
         if actual is None:
-            status, reason = ("pass", None) if compensate else ("fail", missing)
+            status, reason = "fail", missing
         else:
             status, reason = ("pass", None) if actual >= minimum else ("fail", low)
-        checks.append({"name": name, "status": status, "reason_code": reason, "actual": actual, "threshold": minimum, "compensated": actual is None and compensate})
+        checks.append({"name": name, "status": status, "reason_code": reason, "actual": actual, "threshold": minimum, "compensated": False})
 
     min_duration = float(rules["min_duration_seconds"])
     min_shares = float(rules["min_share_count"])
-    threshold("duration_seconds", duration, min_duration, "DURATION_UNKNOWN", "DURATION_TOO_SHORT", compensate=_strong(candidate) or shares is not None and shares >= min_shares * 1.5)
-    likes, plays = _number(candidate.get("like_count")), _number(candidate.get("play_count"))
-    threshold("share_count", shares, min_shares, "SHARE_COUNT_UNKNOWN", "SHARE_COUNT_TOO_LOW", compensate=_strong(candidate) or bool(likes and likes >= 5000) or bool(plays and plays >= 50000))
+    threshold("duration_seconds", duration, min_duration, "DURATION_UNKNOWN", "DURATION_TOO_SHORT")
+    threshold("share_count", shares, min_shares, "SHARE_COUNT_UNKNOWN", "SHARE_COUNT_TOO_LOW")
 
     min_content = float(rules["min_content_50_plus_ratio"])
     min_account = float(rules["min_account_50_plus_ratio"])
     portrait_pass = bool(content_ratio is not None and content_ratio >= min_content or account_ratio is not None and account_ratio >= min_account)
-    portrait_compensated = content_ratio is None and account_ratio is None and _strong(candidate)
+    portrait_compensated = False
     checks.append({
         "name": "elder_portrait", "status": "pass" if portrait_pass or portrait_compensated else "fail",
         "reason_code": None if portrait_pass or portrait_compensated else "CONTENT_PORTRAIT_MISSING" if content_ratio is None and account_ratio is None else "PORTRAIT_50_PLUS_TOO_LOW",
@@ -154,7 +126,7 @@ def evaluate_candidate_gate(candidate: Mapping[str, Any], rule_snapshot: Mapping
     return {
         "rule_version": str(rules["rule_version"]), "status": "pass" if not failed else "fail",
         "primary_eligible": not failed, "failed_reason_codes": failed, "checks": checks,
-        "temporal": temporal, "content_portrait_status": content_status,
+        "content_portrait_status": content_status,
         "account_portrait_status": account_status,
         "portrait_conflict": content_status in {"pass", "fail"} and account_status in {"pass", "fail"} and content_status != account_status,
     }

find_agent_v2/graph.py

@@ -10,6 +10,7 @@ from typing import Any, Protocol
 from langgraph.graph import END, START, StateGraph
 
 from find_agent_v2.context import build_node_slots, render_node_context
+from find_agent_v2.gates import evaluate_candidate_gate
 from find_agent_v2.observability import InputSlot, NullObserver, graph_spec_for
 from find_agent_v2.prompts import EVALUATOR_PROMPT, EVIDENCE_PROMPT, SEARCH_PROMPT, SUPERVISOR_PROMPT
 from find_agent_v2.service import FindAgentV2Service
@@ -46,14 +47,12 @@ class FindAgentRoundGraph:
     def __init__(
         self, *, service: FindAgentV2Service, runner: NodeRunner, observer=None,
         max_actions: int = 16, max_search_actions: int = 3,
-        target_primary_count: int = 5,
     ) -> None:
         self.service = service
         self.runner = runner
         self.observer = observer or NullObserver()
         self.max_actions = max(4, int(max_actions))
         self.max_search_actions = max(1, int(max_search_actions))
-        self.target_primary_count = max(1, int(target_primary_count))
         self.app = self._build_graph()
         self.obagent_spec = graph_spec_for(self.app)
 
@@ -106,6 +105,24 @@ class FindAgentRoundGraph:
             plan=state.get("plan", ""),
         )
 
+    def _video_understanding_ids(
+        self, state: FindAgentGraphState, items: list[dict[str, Any]],
+    ) -> list[int]:
+        """Only candidates passing every deterministic hard gate may use video understanding."""
+        run = self._full_state(state).get("run") or {}
+        rules = run.get("rule_config") or {}
+        selected: list[int] = []
+        for item in items:
+            gate = evaluate_candidate_gate(item, rules)
+            checks = list(gate.get("checks") or [])
+            hard_gate_passed = gate.get("status") == "pass" and all(
+                check.get("status") == "pass" and not check.get("compensated")
+                for check in checks
+            )
+            if str(item.get("video_url") or "").strip() and hard_gate_passed:
+                selected.append(int(item["candidate_id"]))
+        return selected
+
     @staticmethod
     def _parse_supervisor(content: str) -> dict[str, Any]:
         text = content.strip()
@@ -141,10 +158,6 @@ class FindAgentRoundGraph:
         if scope not in {"detail", "portrait", "both"}:
             scope = "both"
 
-        snapshot = self.service.snapshot(state["run_id"])
-        if not pending and snapshot.valid_primary_count >= self.target_primary_count:
-            return "finish", "有效 primary 已达到目标，结束本轮", worker_count, scope
-
         if actions > self.max_actions:
             if not pending:
                 return "finish", "安全收敛动作已完成", worker_count, scope
@@ -290,11 +303,37 @@ class FindAgentRoundGraph:
         node_runs = list(state.get("node_runs", []))
         before = self.service.snapshot(state["run_id"])
         pending = self._full_state(state, pending_only=True)["candidates"]
-        shards = [pending[index:index + 8] for index in range(0, len(pending), 8)]
+        run = self._full_state(state).get("run") or {}
+        rules = run.get("rule_config") or {}
+        eligible: list[dict[str, Any]] = []
+        gate_failures: list[tuple[int, dict[str, Any]]] = []
+        for item in pending:
+            gate = evaluate_candidate_gate(item, rules)
+            if gate.get("status") == "pass":
+                eligible.append(item)
+            else:
+                gate_failures.append((int(item["candidate_id"]), gate))
+        self.service.reject_failed_gates(state["run_id"], gate_failures)
+        shards = [eligible[index:index + 8] for index in range(0, len(eligible), 8)]
         semaphore = asyncio.Semaphore(max(1, min(8, int(state.get("worker_count") or 4))))
 
         async def run_shard(index: int, items: list[dict[str, Any]]) -> NodeRun:
             candidate_ids = [int(item["candidate_id"]) for item in items]
+            video_understanding_ids = self._video_understanding_ids(state, items)
+            video_tools = (
+                bound_candidate_tools(
+                    (EVALUATION_TOOLS[0],),
+                    run_id=state["run_id"],
+                    candidate_ids=video_understanding_ids,
+                )
+                if video_understanding_ids else ()
+            )
+            evaluation_instruction = (
+                f"你只负责当前分片 candidate_ids={candidate_ids}。"
+                f"这些候选均已通过硬门禁；有播放地址、允许按需视频理解的 "
+                f"candidate_ids={video_understanding_ids}。"
+                "视频理解不是硬性要求，可根据已有证据决定是否调用；只允许对这个列表中的候选调用。"
+            )
             async with semaphore:
                 structured_runner = getattr(self.runner, "run_evaluation", None)
                 if callable(structured_runner):
@@ -302,12 +341,13 @@ class FindAgentRoundGraph:
                         round_index=state["round_index"],
                         system_prompt=EVALUATOR_PROMPT,
                         user_content=(
-                            f"你只负责当前分片 candidate_ids={candidate_ids}。"
-                            "必须为这些 candidate_id 各输出一次结构化评估。\n\n"
+                            evaluation_instruction
+                            + "必须为分片内每个 candidate_id 各输出一次结构化评估。\n\n"
                             + self._shard_context(state, items)
                         ),
                         slots=self._shard_slots(state, items),
                         branch_key=f"step-{state.get('supervisor_step', 0)}-shard-{index}",
+                        tools=video_tools,
                     )
                     normalized = normalize_evaluation_items(
                         proposed, allowed_candidates=items,
@@ -325,12 +365,17 @@ class FindAgentRoundGraph:
                     round_index=state["round_index"],
                     system_prompt=EVALUATOR_PROMPT,
                     user_content=(
-                        f"你只负责当前分片 candidate_ids={candidate_ids}。"
-                        "必须把这些候选全部分池，不得评估其他候选。\n\n"
+                        evaluation_instruction
+                        + "必须把这些候选全部分池，不得评估其他候选。\n\n"
                         + self._shard_context(state, items)
                     ),
-                    tools=bound_candidate_tools(
-                        EVALUATION_TOOLS, run_id=state["run_id"], candidate_ids=candidate_ids,
+                    tools=(
+                        *video_tools,
+                        *bound_candidate_tools(
+                            EVALUATION_TOOLS[1:],
+                            run_id=state["run_id"],
+                            candidate_ids=candidate_ids,
+                        ),
                     ),
                     max_iterations=12,
                     slots=self._shard_slots(state, items),

find_agent_v2/prompts.py

@@ -16,8 +16,8 @@ COMMON_RULES = """
 - S（share_score）：转发意愿与传播价值，范围 0~1。
 - V（value_score）：综合价值，范围 0~1。
 
-写入 primary 只是模型的申请，程序还会执行确定性门禁：时效、视频时长、分享量以及内容侧或
-账号侧的 50+ 画像。门禁失败时程序会强制写入 rejected，模型不得绕过。证据不足时应明确说明，
+写入 primary 只是模型的申请，程序还会执行确定性门禁：视频时长、分享量以及内容侧或账号侧的
+50+ 画像。门禁失败时程序会强制写入 rejected，模型不得绕过。证据不足时应明确说明，
 不得把点赞画像描述成转发画像。
 
 ## 数据和生命周期边界
@@ -40,6 +40,9 @@ SUPERVISOR_PROMPT = COMMON_RULES + """
 - `evaluator`：证据已处理后评分分池。
 - `finish`：没有待处理候选且继续搜索已无信息增益时结束本轮。
 
+是否继续探索应综合判断候选新增数量、已评估样本量、本轮通过率和累计通过率：样本仍少或搜索
+仍持续产生有效候选时继续；新增枯竭，或样本已充分但通过率持续为零或很低时结束。
+
 你还可以用 `worker_count` 建议 1~8 个并行 Worker。宿主会依据真实状态、工具权限和预算审核提议；
 非法跳转会被改写为安全动作。只输出 JSON，不要 Markdown：
 {"next_action":"search|evidence|evaluator|finish","reason":"...","worker_count":4,
@@ -73,11 +76,48 @@ EVALUATOR_PROMPT = COMMON_RULES + """
 
 # 当前模块：评估与分池
 
-宿主会给你一个互斥的 pending_evaluation 候选分片。你只负责根据输入中已经保存的证据，输出
-严格结构化的 R/E/S/V 和分池判断；数据库写入、分片校验与完成确认由宿主负责。
+## 你拥有的能力
+
+- 宿主会提供候选详情、互动数据、内容侧/账号侧年龄画像和用户需求上下文。
+- 对宿主列入 `允许按需视频理解的 candidate_ids` 的候选，你可以调用
+  `understand_candidate_video_30s_v2` 查看视频前 30 秒的画面、字幕、对白、主题与观点。
+- 调用时，prompt 应包含当前需求及需要核验的疑点，优先核验标题和元数据无法回答的问题；同一候选
+  最多调用一次。未列入允许列表的 candidate_id 不得尝试调用。
+
+## 判断顺序和要求
+
+1. **内容时间匹配**：以输入中规则快照的 `current_datetime` 和 `timezone` 作为当前日期与时间，
+   判断视频实际内容是否适合现在发布和推荐。只判断内容表达与当前日期/时段是否匹配，不以视频
+   发布时间新旧直接判断。重点检查：
+   - 季节、二十四节气、传统/法定节日、纪念日、特定月份、日期倒计时以及阶段性事件，均属于强日期
+     依赖内容，必须与当前日期核对，不能当作常青内容处理。例如内容处于立秋前后且当前也在立秋
+     前后，可以通过；内容围绕春节，但当前不在春节适用阶段，应 rejected。
+   - 一天中的时段：例如内容表达“早上好”但当前是中午，应 rejected；内容表达“中午好”且当前
+     是中午，可以通过。
+   - 没有日期、节气、节日、季节或早中晚限制的常青内容，不因时间原因淘汰。
+   候选出现强日期依赖线索时，必须得出明确的匹配结论。标题和详情不足以判断时，应调用视频理解
+   核验画面、字幕和对白；工具未调用、失败或调用后仍无法确认时，应 rejected。确认内容时间不匹配
+   时必须 rejected，并在理由中写明视频表达的日期/时段语义和当前日期/时间。
+2. **内容相关性 R**：判断视频实际主题、主要观点、使用场景和用户需求/参考点位是否直接匹配。
+   只有标题沾边、画面或对白实际跑题时应降低 R；直接解决需求核心问题时才可给高分。
+3. **老年适配 E**：以真实年龄画像为主要依据，视频内容只辅助判断表达是否易懂、场景是否贴近目标
+   人群、有无明显不适配信息。不得把视频人物年龄当作受众画像。
+4. **传播价值 S**：结合真实分享量、互动数据及内容中是否存在实用性、情绪共鸣、观点表达或社交
+   转发理由判断。不得用点赞量冒充分享量。
+5. **综合价值 V**：综合 R/E/S，不得因单项数据突出而掩盖内容跑题或适配问题。
+6. **最终申请**：只有内容确实匹配、时间适用且值得保留时才申请 primary；判断理由必须指出使用了哪些事实，
+   若使用视频理解，需明确写出它补充或推翻了什么判断。
+
+宿主只会把已经通过硬门禁的候选交给你。视频理解是增强证据而不是硬性准入要求；对于明确列入
+`允许按需视频理解的 candidate_ids` 的候选，可以根据已有证据决定是否调用
+`understand_candidate_video_30s_v2` 理解前 30 秒，
+对没有强日期依赖线索的内容，未调用或调用失败不能单独作为淘汰理由。若已发现强日期依赖线索，
+则必须确认其与当前日期匹配；无法确认时按关键时间证据不足 rejected。工具明确失败时保留失败事实，
+不要重复调用。
+数据库写入、分片校验与完成确认由宿主负责。
 每一项必须填写输入中的数据库 `candidate_id`（整数），不得拿 `aweme_id` 代替。分片内每个候选
 必须恰好输出一次并进入 primary 或 rejected，不得遗漏、重复或访问分片外候选。
-不得调用工具、搜索、补证或修改运行终态。
+除 `understand_candidate_video_30s_v2` 外不得调用其他工具、搜索、补证或修改运行终态。
 """
 
 REPORT_PROMPT = COMMON_RULES + """

find_agent_v2/providers.py

@@ -85,6 +85,16 @@ def _safe_int(value: Any, default: int = 0) -> int:
         return default
 
 
+def _optional_int(value: Any) -> int | None:
+    """Preserve a missing upstream metric instead of conflating it with a real zero."""
+    if value is None or isinstance(value, bool) or str(value).strip() == "":
+        return None
+    try:
+        return int(float(str(value).strip()))
+    except (TypeError, ValueError):
+        return None
+
+
 def _error(exc: Exception | str) -> dict[str, Any]:
     return {"error": str(exc), "search_results": [], "has_more": False}
 
@@ -130,11 +140,11 @@ def _normalize_search_item(item: dict[str, Any]) -> dict[str, Any] | None:
             "sec_uid": str(author.get("sec_uid") or ""),
         },
         "statistics": {
-            "digg_count": _safe_int(stats.get("digg_count")),
-            "comment_count": _safe_int(stats.get("comment_count")),
-            "share_count": _safe_int(stats.get("share_count")),
-            "collect_count": _safe_int(stats.get("collect_count")),
-            "play_count": _safe_int(stats.get("play_count")),
+            "digg_count": _optional_int(stats.get("digg_count")),
+            "comment_count": _optional_int(stats.get("comment_count")),
+            "share_count": _optional_int(stats.get("share_count")),
+            "collect_count": _optional_int(stats.get("collect_count")),
+            "play_count": _optional_int(stats.get("play_count")),
         },
         "duration_ms": _safe_int(item.get("duration_ms") or item.get("duration") or video.get("duration")),
         "publish_at": item.get("publish_at") or item.get("create_time") or item.get("create_timestamp") or item.get("publish_timestamp"),

find_agent_v2/qwen_video_understanding_30s.py

@@ -0,0 +1,300 @@
+"""Independent 30-second Qwen video-understanding tool for find_agent_v2."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import importlib
+import json
+import logging
+import os
+import re
+import signal
+import shutil
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+from typing import Any
+
+from dotenv import load_dotenv
+from openai import APIStatusError, APITimeoutError, OpenAI
+
+from find_agent_v2.service import get_find_agent_v2_service
+from supply_agent.paths import find_project_root
+from supply_agent.tools import tool
+from supply_infra.config import get_infra_settings
+from supply_infra.oss.client import OssClient
+
+logger = logging.getLogger(__name__)
+
+DASHSCOPE_BASE_URL = (
+    "https://llm-33b86fznnpci2exm.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
+)
+MODEL = "qwen3.7-plus"
+DEFAULT_PROMPT = (
+    "客观提取这段视频前30秒的画面、人物、动作、对白或字幕、主题、主要观点、适用场景，"
+    "以及出现的日期、节日、时段或事件状态。不要根据画面中人物年龄推断观看受众年龄。"
+)
+CLIP_SECONDS = 30.0
+MIN_USABLE_CLIP_SECONDS = 15.0
+REMOTE_CLIP_TIMEOUT_SECONDS = 60.0
+REMOTE_CLIP_ATTEMPTS = 2
+UPLOAD_TIMEOUT_SECONDS = 60.0
+MODEL_TIMEOUT_SECONDS = 300.0
+TOOL_TIMEOUT_SECONDS = 480.0
+PROBE_TIMEOUT_SECONDS = 15.0
+_DURATION_RE = re.compile(r"Duration:\s*(\d+):(\d+):(\d+(?:\.\d+)?)")
+
+
+class RemoteClipError(RuntimeError):
+    pass
+
+
+def _result(**payload: Any) -> str:
+    return json.dumps(payload, ensure_ascii=False)
+
+
+def _safe_unlink(path: Path) -> None:
+    try:
+        path.unlink(missing_ok=True)
+    except OSError as exc:
+        logger.warning("failed to delete find_agent_v2 temp video %s: %s", path, exc)
+
+
+def _ffmpeg_executable() -> str:
+    executable = shutil.which("ffmpeg")
+    if executable:
+        return executable
+    try:
+        return importlib.import_module("imageio_ffmpeg").get_ffmpeg_exe()
+    except ImportError:
+        raise ValueError(
+            "30 秒视频理解需要 ffmpeg；请安装系统 ffmpeg 或 imageio-ffmpeg"
+        ) from None
+
+
+def _probe_clip_duration(path: Path) -> float | None:
+    if not path.is_file() or path.stat().st_size <= 0:
+        return None
+    try:
+        completed = subprocess.run(
+            [_ffmpeg_executable(), "-hide_banner", "-i", str(path)],
+            capture_output=True,
+            text=True,
+            check=False,
+            timeout=PROBE_TIMEOUT_SECONDS,
+        )
+    except subprocess.TimeoutExpired:
+        return None
+    match = _DURATION_RE.search(completed.stderr or "")
+    if not match:
+        return None
+    hours, minutes, seconds = match.groups()
+    return int(hours) * 3600 + int(minutes) * 60 + float(seconds)
+
+
+def _clip_remote_once(video_url: str, clipped: Path) -> tuple[float | None, str]:
+    """Let ffmpeg read only the remote media needed for the first 30 seconds."""
+    command = [
+        _ffmpeg_executable(),
+        "-y",
+        "-user_agent",
+        "curl/8.6.0",
+        "-i",
+        video_url,
+        "-t",
+        str(CLIP_SECONDS),
+        "-c",
+        "copy",
+        "-movflags",
+        "+faststart",
+        str(clipped),
+    ]
+    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    timed_out = False
+    try:
+        stdout, stderr = process.communicate(timeout=REMOTE_CLIP_TIMEOUT_SECONDS)
+    except subprocess.TimeoutExpired:
+        timed_out = True
+        process.send_signal(signal.SIGINT)
+        try:
+            stdout, stderr = process.communicate(timeout=5)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            stdout, stderr = process.communicate()
+    duration = _probe_clip_duration(clipped)
+    detail = (stderr or stdout or "")[-500:]
+    status = "timeout" if timed_out else f"exit={process.returncode}"
+    return duration, f"{status}: {detail}"
+
+
+async def _clip_remote_with_one_retry(
+    video_url: str, temp_dir: Path,
+) -> tuple[Path, float, bool]:
+    """Prefer a complete clip; after both attempts accept the longest clip over 15 seconds."""
+    candidates: list[tuple[float, Path]] = []
+    errors: list[str] = []
+    for attempt in range(1, REMOTE_CLIP_ATTEMPTS + 1):
+        attempt_path = temp_dir / f"clipped_attempt_{attempt}.mp4"
+        _safe_unlink(attempt_path)
+        duration, detail = await asyncio.to_thread(
+            _clip_remote_once, video_url, attempt_path,
+        )
+        if duration is not None and duration >= CLIP_SECONDS:
+            return attempt_path, duration, True
+        if duration is not None and duration > MIN_USABLE_CLIP_SECONDS:
+            candidates.append((duration, attempt_path))
+        errors.append(f"attempt {attempt}/{REMOTE_CLIP_ATTEMPTS}: duration={duration}, {detail}")
+        logger.warning("find_agent_v2 remote clip %s", errors[-1])
+
+    if candidates:
+        duration, path = max(candidates, key=lambda item: item[0])
+        return path, duration, False
+    raise RemoteClipError(
+        "远程截取失败：两次尝试均未获得超过 15 秒的有效视频；" + " | ".join(errors)
+    )
+
+
+def _upload_to_oss(clipped: Path, video_url: str) -> str:
+    settings = get_infra_settings()
+    if not settings.aliyun_oss_access_key_id or not settings.aliyun_oss_access_key_secret:
+        raise ValueError("视频上传需要配置 OSS access key id 和 secret")
+    digest = hashlib.sha256(video_url.encode("utf-8")).hexdigest()[:16]
+    client = OssClient()
+    object_key = client.object_key("find_agent_v2", "video_clips", f"{digest}_30s.mp4")
+    return client.upload_file(clipped, object_key)
+
+
+def _qwen_client() -> OpenAI:
+    load_dotenv(find_project_root() / ".env")
+    api_key = os.getenv("DASHSCOPE_API_KEY")
+    if not api_key:
+        raise ValueError("未设置环境变量 DASHSCOPE_API_KEY")
+    return OpenAI(api_key=api_key, base_url=DASHSCOPE_BASE_URL)
+
+
+def _understand_with_qwen(oss_url: str, prompt: str) -> str:
+    completion = _qwen_client().chat.completions.create(
+        model=MODEL,
+        messages=[{
+            "role": "user",
+            "content": [
+                {"type": "video_url", "video_url": {"url": oss_url}, "fps": 2.0},
+                {"type": "text", "text": prompt},
+            ],
+        }],
+        timeout=MODEL_TIMEOUT_SECONDS,
+    )
+    return completion.choices[0].message.content or ""
+
+
+async def _prepare_oss_video(video_url: str) -> tuple[str, float, bool]:
+    temp_dir = Path(tempfile.mkdtemp(prefix="find_agent_v2_video_"))
+    try:
+        clipped, duration, complete = await _clip_remote_with_one_retry(video_url, temp_dir)
+        try:
+            oss_url = await asyncio.wait_for(
+                asyncio.to_thread(_upload_to_oss, clipped, video_url),
+                timeout=UPLOAD_TIMEOUT_SECONDS,
+            )
+        except TimeoutError:
+            raise TimeoutError("OSS 上传视频超时（60 秒）") from None
+        return oss_url, duration, complete
+    finally:
+        for path in temp_dir.iterdir():
+            if path.is_file():
+                _safe_unlink(path)
+        try:
+            temp_dir.rmdir()
+        except OSError:
+            logger.warning("failed to remove find_agent_v2 temp directory %s", temp_dir)
+
+
+@tool(
+    name="understand_candidate_video_30s_v2",
+    description=(
+        "理解当前评估分片中的候选视频。按 candidate_id 读取播放地址，由 ffmpeg 远程截取前 30 秒"
+        "（单次 60 秒超时，失败重试 1 次）；两次均不足 30 秒时只接受超过 15 秒的最长片段。上传 OSS"
+        "并清理全部本地文件后调用千问返回内容理解。"
+        "仅用于核验实际主题、需求相关性、表达与场景、传播理由及视频中的时间线索；不能用画面人物"
+        "年龄推断受众年龄。prompt 应写明当前需求和需要核验的疑点。"
+    ),
+)
+async def understand_candidate_video_30s_v2(
+    run_id: str,
+    candidate_id: int,
+    prompt: str = DEFAULT_PROMPT,
+) -> str:
+    """Return Qwen's understanding of the first 30 seconds of one candidate video."""
+    started = time.monotonic()
+    try:
+        candidate = get_find_agent_v2_service().candidate_input(run_id, candidate_id)
+        video_url = str(candidate.get("video_url") or "").strip()
+        if not video_url:
+            return _result(
+                error="候选详情中没有可用的视频播放地址",
+                error_code="video_url_missing",
+                candidate_id=candidate_id,
+                retryable=False,
+            )
+        if not prompt.strip():
+            return _result(
+                error="prompt 不能为空",
+                error_code="invalid_argument",
+                candidate_id=candidate_id,
+                retryable=False,
+            )
+        async with asyncio.timeout(TOOL_TIMEOUT_SECONDS):
+            oss_url, clip_duration, complete_clip = await _prepare_oss_video(video_url)
+            try:
+                content = await asyncio.wait_for(
+                    asyncio.to_thread(_understand_with_qwen, oss_url, prompt),
+                    timeout=MODEL_TIMEOUT_SECONDS,
+                )
+            except TimeoutError:
+                raise TimeoutError("千问视频理解请求超时") from None
+        return _result(
+            candidate_id=candidate_id,
+            original_video_url=video_url,
+            analysis_video_url=oss_url,
+            clip_seconds=round(clip_duration, 3),
+            complete_30s_clip=complete_clip,
+            model=MODEL,
+            content=content,
+            output=content,
+            duration_ms=int((time.monotonic() - started) * 1000),
+        )
+    except (RemoteClipError, TimeoutError) as exc:
+        return _result(
+            error=str(exc),
+            error_code=(
+                "video_understanding_timeout"
+                if isinstance(exc, TimeoutError)
+                else "remote_clip_failed"
+            ),
+            candidate_id=candidate_id,
+            retryable=False,
+        )
+    except APIStatusError as exc:
+        return _result(
+            error=str(exc),
+            error_code="qwen_api_error",
+            candidate_id=candidate_id,
+            retryable=exc.status_code >= 500,
+        )
+    except APITimeoutError as exc:
+        return _result(
+            error=f"千问视频理解请求超时: {exc}",
+            error_code="video_understanding_timeout",
+            candidate_id=candidate_id,
+            retryable=True,
+        )
+    except Exception as exc:
+        logger.exception("find_agent_v2 30s video understanding failed")
+        return _result(
+            error=str(exc),
+            error_code="video_understanding_failed",
+            candidate_id=candidate_id,
+            retryable=False,
+        )

find_agent_v2/runtime.py

@@ -291,12 +291,14 @@ class FindAgentNodeHost:
         user_content: str,
         slots: tuple[InputSlot, ...] = (),
         branch_key: str = "",
+        tools: Iterable[ToolFn] = (),
     ) -> tuple[NodeRun, list[dict[str, Any]]]:
         """Get one structured decision batch; the graph owns validation and persistence."""
         model_name = self.models_by_role.get("evaluator", self.default_model)
+        langchain_tools = [_as_langchain_tool(fn) for fn in tools]
         agent = create_agent(
             model=self._model("evaluator"),
-            tools=[],
+            tools=langchain_tools,
             system_prompt=system_prompt,
             middleware=self._middleware("evaluator"),
             response_format=EvaluationBatch,
@@ -307,7 +309,7 @@ class FindAgentNodeHost:
                 fallback=user_content,
                 system_prompt=system_prompt,
                 slots=slots,
-                tools=(),
+                tools=tuple(langchain_tools),
                 model=model_name,
             )
             result = await agent.ainvoke(
@@ -327,7 +329,11 @@ class FindAgentNodeHost:
                 "events": _events(messages),
                 "usage": usage,
                 "iterations": sum(isinstance(message, AIMessage) for message in messages),
-                "tool_calls_made": 0,
+                "tool_calls_made": sum(
+                    len(message.tool_calls)
+                    for message in messages
+                    if isinstance(message, AIMessage)
+                ),
                 "structured_items": items,
             }
             observation.record_react(output=process, ok=True)
@@ -338,7 +344,7 @@ class FindAgentNodeHost:
                 round_index=round_index,
                 content=str(last_ai.content if last_ai is not None else ""),
                 iterations=int(process["iterations"]),
-                tool_calls_made=0,
+                tool_calls_made=int(process["tool_calls_made"]),
             )
             return run, items
 

find_agent_v2/service.py

@@ -26,6 +26,11 @@ from find_agent_v2.gates import (
 )
 
 
+RUN_TIMEOUT_MINUTES = 60
+RUN_TIMEOUT_SECONDS = RUN_TIMEOUT_MINUTES * 60
+RUN_TIMEOUT_REASON = "运行时间超过 60 分钟，系统自动标记失败"
+
+
 class FindAgentV2RunNotFound(LookupError):
     pass
 
@@ -52,12 +57,25 @@ def _ratio(value: Any) -> Decimal | None:
     return number.quantize(Decimal("0.000001"))
 
 
+def _fails_search_share_gate(
+    provider: str, share_count: Any, min_share_count: int,
+) -> bool:
+    """Only TikHub has a verified search-time share metric contract."""
+    return (
+        provider == "tikhub"
+        and share_count is not None
+        and int(share_count) < int(min_share_count)
+    )
+
+
 def _candidate_dict(row: FindAgentV2Candidate) -> dict[str, Any]:
+    detail = _loads(row.detail_json, {})
     return {
         "candidate_id": int(row.id),
         "aweme_id": row.aweme_id,
         "title": row.title,
         "content_link": row.content_link,
+        "video_url": detail.get("video_url") if isinstance(detail, dict) else None,
         "author_name": row.author_name,
         "author_sec_uid": row.author_sec_uid,
         "source_keywords": _loads(row.source_keywords_json, []),
@@ -295,6 +313,8 @@ class FindAgentV2Service:
             session.add(search)
             session.flush()
             new_count = 0
+            share_gate_rejected_count = 0
+            min_share_count = int(_loads(run.rule_config_json, {}).get("min_share_count", 1000))
             for item in results:
                 aweme_id = str(item.get("aweme_id") or "").strip()
                 if not aweme_id:
@@ -326,15 +346,64 @@ class FindAgentV2Service:
                 duration_ms = item.get("duration_ms")
                 if duration_ms:
                     candidate.duration_seconds = Decimal(str(duration_ms)) / 1000
-                candidate.play_count = stats.get("play_count") or candidate.play_count
-                candidate.like_count = stats.get("digg_count") or stats.get("like_count") or candidate.like_count
-                candidate.comment_count = stats.get("comment_count") or candidate.comment_count
-                candidate.collect_count = stats.get("collect_count") or candidate.collect_count
-                candidate.share_count = stats.get("share_count") or candidate.share_count
+                metric_fields = {
+                    "play_count": stats.get("play_count"),
+                    "like_count": (
+                        stats.get("digg_count")
+                        if stats.get("digg_count") is not None
+                        else stats.get("like_count")
+                    ),
+                    "comment_count": stats.get("comment_count"),
+                    "collect_count": stats.get("collect_count"),
+                    "share_count": stats.get("share_count"),
+                }
+                for field, value in metric_fields.items():
+                    if value is not None:
+                        setattr(candidate, field, int(value))
+
+                search_share_count = stats.get("share_count")
+                if _fails_search_share_gate(provider, search_share_count, min_share_count):
+                    gate = {
+                        "stage": "search",
+                        "status": "fail",
+                        "primary_eligible": False,
+                        "failed_reason_codes": ["SHARE_COUNT_TOO_LOW"],
+                        "checks": [{
+                            "name": "share_count",
+                            "status": "fail",
+                            "reason_code": "SHARE_COUNT_TOO_LOW",
+                            "actual": int(search_share_count),
+                            "threshold": min_share_count,
+                            "compensated": False,
+                        }],
+                    }
+                    candidate.gate_status = "fail"
+                    candidate.gate_result_json = _json(gate)
+                    candidate.decision_bucket = "rejected"
+                    candidate.decision_reason = (
+                        f"搜索结果分享数 {int(search_share_count)} 低于门槛 {min_share_count}"
+                    )
+                    candidate.reject_reason_code = "SHARE_COUNT_TOO_LOW"
+                    share_gate_rejected_count += 1
+                elif (
+                    search_share_count is not None
+                    and candidate.reject_reason_code == "SHARE_COUNT_TOO_LOW"
+                ):
+                    # A later search response may contain a refreshed metric.
+                    candidate.gate_status = None
+                    candidate.gate_result_json = None
+                    candidate.decision_bucket = "pending_evaluation"
+                    candidate.decision_reason = None
+                    candidate.reject_reason_code = None
             run.search_count = int(session.scalar(select(func.count()).select_from(FindAgentV2Search).where(FindAgentV2Search.run_id == run_id)) or 0)
             session.flush()
             run.candidate_count = int(session.scalar(select(func.count()).select_from(FindAgentV2Candidate).where(FindAgentV2Candidate.run_id == run_id)) or 0)
-            return {"search_id": int(search.id), "new_candidate_count": new_count, "result_count": len(results)}
+            return {
+                "search_id": int(search.id),
+                "new_candidate_count": new_count,
+                "result_count": len(results),
+                "share_gate_rejected_count": share_gate_rejected_count,
+            }
 
     def candidate_inputs(self, run_id: str, candidate_ids: list[int]) -> list[dict[str, Any]]:
         with get_session() as session:
@@ -344,6 +413,13 @@ class FindAgentV2Service:
             )))
             return [_candidate_dict(row) for row in rows]
 
+    def candidate_input(self, run_id: str, candidate_id: int) -> dict[str, Any]:
+        """Return one candidate only when it belongs to the requested v2 run."""
+        items = self.candidate_inputs(run_id, [candidate_id])
+        if not items or int(items[0]["candidate_id"]) != int(candidate_id):
+            raise ValueError(f"candidate_id 不属于 run: {candidate_id}")
+        return items[0]
+
     def save_details(self, run_id: str, details: list[dict[str, Any]], errors: list[dict[str, Any]]) -> None:
         by_id = {str(item.get("content_id") or ""): item for item in details}
         error_by_id = {str(item.get("content_id") or ""): item for item in errors}
@@ -451,6 +527,31 @@ class FindAgentV2Service:
             run.valid_primary_count = len({row.aweme_id for row in primaries if row.gate_status == "pass"})
         return output
 
+    def reject_failed_gates(
+        self, run_id: str, failures: list[tuple[int, dict[str, Any]]],
+    ) -> list[int]:
+        """Reject hard-gate failures before any evaluator model invocation."""
+        rejected: list[int] = []
+        if not failures:
+            return rejected
+        by_id = {int(candidate_id): gate for candidate_id, gate in failures}
+        with get_session() as session:
+            rows = list(session.scalars(select(FindAgentV2Candidate).where(
+                FindAgentV2Candidate.run_id == run_id,
+                FindAgentV2Candidate.id.in_(list(by_id)),
+                FindAgentV2Candidate.decision_bucket == "pending_evaluation",
+            )))
+            for row in rows:
+                gate = by_id[int(row.id)]
+                failed = list(gate.get("failed_reason_codes") or [])
+                row.gate_status = "fail"
+                row.gate_result_json = _json(gate)
+                row.decision_bucket = "rejected"
+                row.decision_reason = "硬门禁未通过：" + ", ".join(failed)
+                row.reject_reason_code = failed[0] if failed else "HARD_GATE_FAILED"
+                rejected.append(int(row.id))
+        return rejected
+
     def recount_valid_primary(self, run_id: str) -> int:
         """Recompute the denormalized primary counter after parallel worker writes."""
         with get_session() as session:
@@ -519,24 +620,22 @@ class FindAgentV2Service:
         *,
         failed: bool = False,
         reason: str = "",
-        target_primary_count: int = 5,
     ) -> dict[str, Any]:
         snapshot = self.snapshot(run_id)
         if failed:
             outcome, status = "failed", "failed"
-        elif snapshot.valid_primary_count >= max(1, int(target_primary_count)):
-            outcome, status = "goal_met", "finished"
         elif snapshot.valid_primary_count > 0:
-            outcome, status = "partial", "finished"
+            outcome, status = "goal_met", "finished"
         else:
             outcome, status = "no_match", "finished"
         with get_session() as session:
             run = session.scalar(select(FindAgentV2Run).where(FindAgentV2Run.run_id == run_id))
             if run is None:
                 raise FindAgentV2RunNotFound(run_id)
-            run.status = status
-            run.outcome_status = outcome
-            run.stop_reason = reason[:2000] or None
+            if not (run.status == "failed" and run.outcome_status == "failed"):
+                run.status = status
+                run.outcome_status = outcome
+                run.stop_reason = reason[:2000] or None
         return self.require_run(run_id)
 
 

find_agent_v2/tools.py

@@ -8,6 +8,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field, model_validator
 
+from find_agent_v2.qwen_video_understanding_30s import understand_candidate_video_30s_v2
 from find_agent_v2.providers import (
     fetch_details,
     fetch_portraits,
@@ -137,6 +138,22 @@ def bound_candidate_tools(
 
             return bound_async
 
+        def make_video_understanding(
+            fn: ToolFn, fn_name: str, worker_run_id: str, worker_allowed: set[int],
+        ):
+            @tool(name=fn_name, description=getattr(fn, "_tool_description", ""))
+            async def bound_video_understanding(
+                run_id: str, candidate_id: int, prompt: str,
+            ) -> str:
+                if run_id != worker_run_id or int(candidate_id) not in worker_allowed:
+                    return json.dumps({
+                        "error": "candidate_id 或 run_id 超出当前评估 Worker 分片",
+                        "allowed_candidate_ids": sorted(worker_allowed),
+                    }, ensure_ascii=False)
+                return await fn(run_id=run_id, candidate_id=int(candidate_id), prompt=prompt)
+
+            return bound_video_understanding
+
         def make_sync(fn: ToolFn, fn_name: str, worker_run_id: str, worker_allowed: set[int]):
             @tool(name=fn_name, description=getattr(fn, "_tool_description", ""))
             def bound_sync(run_id: str, items: list[CandidateEvaluation]) -> str:
@@ -157,7 +174,9 @@ def bound_candidate_tools(
             return bound_sync
 
         wrapper: ToolFn
-        if name in {"fetch_candidate_details_v2", "fetch_candidate_portraits_v2"}:
+        if name == "understand_candidate_video_30s_v2":
+            wrapper = make_video_understanding(original, name, run_id, allowed)
+        elif name in {"fetch_candidate_details_v2", "fetch_candidate_portraits_v2"}:
             wrapper = make_async(original, name, run_id, allowed)
         elif name == "evaluate_candidates_v2":
             wrapper = make_sync(original, name, run_id, allowed)
@@ -292,6 +311,7 @@ EVIDENCE_TOOLS: tuple[ToolFn, ...] = (
     query_pending_candidates_v2,
 )
 EVALUATION_TOOLS: tuple[ToolFn, ...] = (
+    understand_candidate_video_30s_v2,
     evaluate_candidates_v2,
     query_pending_candidates_v2,
 )

prd/README.md

@@ -45,6 +45,7 @@ SupplyAgent 每天自动接收变化的需求信号、外部信号和后验反
 | [07-核心业务图.md](07-核心业务图.md) | 全局 Harness、需求图、闭环和定义变更图 |
 | [08-需求汇总反馈机制设计.md](08-需求汇总反馈机制设计.md) | 需求、视频和命中内容的人工反馈机制 |
 | [09-找视频记录展示设计.md](09-找视频记录展示设计.md) | Find Agent 运行、搜索与候选记录的审计展示 |
+| [10-Find-Agent-v2业务优化方案.md](10-Find-Agent-v2业务优化方案.md) | Find Agent v2 的需求理解、搜索召回、候选评价、结果集和反馈闭环优化 |
 
 ## 5. 业务成功标准
 

tests/supply_agent/test_find_agent_v2.py

@@ -10,6 +10,7 @@ from langchain_core.language_models.fake_chat_models import FakeMessagesListChat
 from langchain_core.messages import AIMessage
 
 from find_agent_v2.graph import FindAgentRoundGraph
+from find_agent_v2.agent import decide_continued_exploration
 from find_agent_v2.runtime import DelegateArgs, FindAgentNodeHost
 from find_agent_v2.demand_context import (
     V2DemandContext,
@@ -35,9 +36,10 @@ from find_agent_v2.observability import (
     OBAGENT_ROUND_ANCHOR,
     NullObserver,
 )
-from find_agent_v2.prompts import COMMON_RULES
-from find_agent_v2.providers import normalize_age_pair
+from find_agent_v2.prompts import COMMON_RULES, EVALUATOR_PROMPT
+from find_agent_v2.providers import _normalize_search_item, normalize_age_pair
 from find_agent_v2.state import DiscoverySnapshot, FindAgentState, NodeRun
+from find_agent_v2.service import _fails_search_share_gate
 from find_agent_v2.tools import (
     CandidateEvaluation,
     EVALUATION_TOOLS,
@@ -157,6 +159,31 @@ def test_v2_owns_age_portrait_normalization() -> None:
     assert normalized["consistency"] == "aligned"
 
 
+def test_search_metrics_preserve_missing_share_count_and_real_zero() -> None:
+    missing = _normalize_search_item({"aweme_id": "missing", "statistics": {}})
+    zero = _normalize_search_item({
+        "aweme_id": "zero", "statistics": {"share_count": 0},
+    })
+
+    assert missing is not None and missing["statistics"]["share_count"] is None
+    assert zero is not None and zero["statistics"]["share_count"] == 0
+
+
+@pytest.mark.parametrize(
+    "provider,share_count,expected",
+    [
+        ("tikhub", 999, True),
+        ("tikhub", 1000, False),
+        ("tikhub", None, False),
+        ("internal_keyword", 0, False),
+        ("internal_keyword", 999, False),
+        ("internal_keyword", None, False),
+    ],
+)
+def test_search_share_gate_only_applies_to_tikhub(provider, share_count, expected) -> None:
+    assert _fails_search_share_gate(provider, share_count, 1000) is expected
+
+
 def test_v2_owns_primary_candidate_gate() -> None:
     rules = build_rule_snapshot()
     candidate = {
@@ -193,6 +220,69 @@ def test_v2_gate_rejects_explicitly_low_evidence() -> None:
     )
 
 
+def test_v2_gate_ignores_publish_time_but_rejects_missing_elder_portrait() -> None:
+    rules = build_rule_snapshot()
+    candidate = {
+        "title": "高分但缺少硬性证据的视频",
+        "publish_at": None,
+        "duration_seconds": 60,
+        "share_count": 2000,
+        "content_50_plus_ratio": None,
+        "account_50_plus_ratio": None,
+        "relevance_score": 0.99,
+        "elder_score": 0.99,
+        "share_score": 0.99,
+        "value_score": 0.99,
+    }
+
+    result = evaluate_candidate_gate(candidate, rules)
+
+    assert result["status"] == "fail"
+    assert result["failed_reason_codes"] == ["CONTENT_PORTRAIT_MISSING"]
+    assert all(check["name"] != "temporal" for check in result["checks"])
+
+
+@pytest.mark.parametrize(
+    "content_ratio,account_ratio",
+    [(0.30, None), (None, 0.30), (0.05, 0.30), (0.30, 0.05)],
+)
+def test_v2_gate_accepts_either_elder_portrait_side(content_ratio, account_ratio) -> None:
+    rules = build_rule_snapshot()
+    result = evaluate_candidate_gate({
+        "title": "符合要求的视频",
+        "publish_at": rules["current_datetime"],
+        "duration_seconds": 60,
+        "share_count": 2000,
+        "content_50_plus_ratio": content_ratio,
+        "account_50_plus_ratio": account_ratio,
+    }, rules)
+
+    assert result["status"] == "pass"
+    assert result["primary_eligible"] is True
+
+
+def test_v2_gate_does_not_compensate_missing_duration_or_share_count() -> None:
+    rules = build_rule_snapshot()
+    result = evaluate_candidate_gate({
+        "title": "其他信号很强但硬指标缺失",
+        "publish_at": rules["current_datetime"],
+        "duration_seconds": None,
+        "share_count": None,
+        "content_50_plus_ratio": 0.30,
+        "like_count": 100000,
+        "play_count": 1000000,
+        "relevance_score": 0.99,
+        "elder_score": 0.99,
+        "share_score": 0.99,
+        "value_score": 0.99,
+    }, rules)
+
+    assert result["status"] == "fail"
+    assert {"DURATION_UNKNOWN", "SHARE_COUNT_UNKNOWN"} <= set(
+        result["failed_reason_codes"]
+    )
+
+
 def test_v2_context_deduplicates_expansion_points() -> None:
     rows = [
         SimpleNamespace(
@@ -313,6 +403,7 @@ def test_stage_tool_allowlists_are_physical_and_isolated() -> None:
         "query_pending_candidates_v2",
     }
     assert _names(EVALUATION_TOOLS) == {
+        "understand_candidate_video_30s_v2",
         "evaluate_candidates_v2",
         "query_pending_candidates_v2",
     }
@@ -329,6 +420,82 @@ def test_common_prompt_points_to_v2_tables_and_tools() -> None:
     assert "batch_update_video_discovery_candidates" not in COMMON_RULES
 
 
+def test_evaluator_prompt_defines_video_capability_and_judgment_boundaries() -> None:
+    assert "understand_candidate_video_30s_v2" in EVALUATOR_PROMPT
+    assert "同一候选" in EVALUATOR_PROMPT and "最多调用一次" in EVALUATOR_PROMPT
+    assert "已经通过硬门禁" in EVALUATOR_PROMPT
+    assert "不得把视频人物年龄当作受众画像" in EVALUATOR_PROMPT
+    assert "不得用点赞量冒充分享量" in EVALUATOR_PROMPT
+    assert "没有强日期依赖线索" in EVALUATOR_PROMPT
+    assert "无法确认时按关键时间证据不足 rejected" in EVALUATOR_PROMPT
+    assert "均属于强日期" in EVALUATOR_PROMPT
+    assert "不能当作常青内容处理" in EVALUATOR_PROMPT
+    assert "仍无法确认时，应 rejected" in EVALUATOR_PROMPT
+
+
+@pytest.mark.parametrize(
+    "previous,current,expected,reason",
+    [
+        (
+            DiscoverySnapshot("running", 1, 10, 0, 3, 3, 7),
+            DiscoverySnapshot("running", 2, 10, 0, 3, 3, 7),
+            False,
+            "没有新增候选",
+        ),
+        (
+            DiscoverySnapshot("running", 0, 0, 0, 0, 0, 0),
+            DiscoverySnapshot("running", 1, 4, 0, 1, 1, 3),
+            True,
+            "样本不足",
+        ),
+        (
+            DiscoverySnapshot("running", 1, 4, 0, 1, 1, 3),
+            DiscoverySnapshot("running", 2, 12, 0, 1, 1, 11),
+            False,
+            "通过率为 0",
+        ),
+        (
+            DiscoverySnapshot("running", 1, 8, 0, 1, 1, 7),
+            DiscoverySnapshot("running", 2, 12, 0, 2, 2, 10),
+            True,
+            "正向通过率",
+        ),
+    ],
+)
+def test_exploration_decision_uses_volume_and_pass_rate(
+    previous, current, expected, reason,
+) -> None:
+    decision = decide_continued_exploration(previous, current)
+    assert decision.continue_exploring is expected
+    assert reason in decision.reason
+
+
+def test_video_understanding_only_allows_candidates_passing_all_hard_gates(monkeypatch) -> None:
+    graph = FindAgentRoundGraph(service=_FakeService(pending_after_search=0), runner=object())
+    monkeypatch.setattr(graph, "_full_state", lambda _state: {
+        "run": {"rule_config": build_rule_snapshot()},
+    })
+    common = {
+        "video_url": "https://example.test/video.mp4",
+        "publish_at": build_rule_snapshot()["current_datetime"],
+        "content_50_plus_ratio": 0.30,
+        "duration_seconds": 60,
+        "share_count": 2000,
+    }
+    items = [
+        {"candidate_id": 1, **common},
+        {"candidate_id": 2, **common, "content_50_plus_ratio": 0.05},
+        {"candidate_id": 3, **common, "publish_at": None},
+        {"candidate_id": 4, **common, "video_url": None},
+        {"candidate_id": 5, **common, "duration_seconds": 10},
+        {"candidate_id": 6, **common, "share_count": 1},
+        {"candidate_id": 7, **common, "duration_seconds": None},
+        {"candidate_id": 8, **common, "share_count": None},
+    ]
+
+    assert graph._video_understanding_ids({"run_id": "run"}, items) == [1, 3]
+
+
 class _FakeService:
     def __init__(self, *, pending_after_search: int) -> None:
         self.pending_after_search = pending_after_search
@@ -347,6 +514,10 @@ class _FakeService:
                     "decision_bucket": "pending_evaluation",
                     "detail_status": evidence_status,
                     "portrait_status": evidence_status,
+                    "publish_at": build_rule_snapshot()["current_datetime"],
+                    "duration_seconds": 60,
+                    "share_count": 2000,
+                    "content_50_plus_ratio": 0.30,
                 }
                 for index in range(1, snapshot.pending_count + 1)
             ],
@@ -371,6 +542,9 @@ class _FakeService:
     def recount_valid_primary(self, _run_id: str) -> int:
         return 0
 
+    def reject_failed_gates(self, _run_id: str, failures) -> list[int]:
+        return [candidate_id for candidate_id, _gate in failures]
+
 
 class _FakeRunner:
     def __init__(self, service: _FakeService) -> None:
@@ -423,7 +597,9 @@ async def test_round_graph_supervisor_routes_with_guarded_allowlists() -> None:
     assert runner.calls[2][1] == set()
     assert runner.calls[3][1] <= _names(EVIDENCE_TOOLS)
     assert runner.calls[4][1] <= _names(EVIDENCE_TOOLS)
-    assert runner.calls[6][1] == _names(EVALUATION_TOOLS)
+    assert runner.calls[6][1] == {
+        "evaluate_candidates_v2", "query_pending_candidates_v2",
+    }
     assert result.phase == "done"
     assert result.snapshot is not None and result.snapshot.pending_count == 0
 
@@ -441,6 +617,46 @@ async def test_round_graph_skips_evidence_and_evaluation_without_candidates() ->
     ]
 
 
+@pytest.mark.asyncio
+async def test_evaluator_model_never_receives_hard_gate_failures() -> None:
+    service = _FakeService(pending_after_search=1)
+    service.stage = "evidenced"
+    rejected: list[int] = []
+
+    def state(_run_id: str, **_kwargs):
+        return {
+            "run": {"run_id": "r", "rule_config": build_rule_snapshot()},
+            "searches": [],
+            "candidates": [{
+                "candidate_id": 1,
+                "decision_bucket": "pending_evaluation",
+                "detail_status": "success",
+                "portrait_status": "success",
+                "publish_at": build_rule_snapshot()["current_datetime"],
+                "duration_seconds": 10,
+                "share_count": 2000,
+                "content_50_plus_ratio": 0.30,
+                "video_url": "https://example.test/video.mp4",
+            }],
+        }
+
+    service.get_full_state = state  # type: ignore[method-assign]
+    service.reject_failed_gates = (  # type: ignore[method-assign]
+        lambda _run_id, failures: rejected.extend(candidate_id for candidate_id, _ in failures)
+        or [candidate_id for candidate_id, _ in failures]
+    )
+    runner = _FakeRunner(service)
+    graph = FindAgentRoundGraph(service=service, runner=runner)
+
+    await graph._evaluator({
+        "run_id": "r", "round_index": 1, "worker_count": 1,
+        "action_count": 0, "node_runs": [], "user_input": "需求",
+    })
+
+    assert rejected == [1]
+    assert not any(name == "evaluator" for name, _tools in runner.calls)
+
+
 @pytest.mark.asyncio
 async def test_round_graph_reenters_evaluator_until_pending_queue_is_empty() -> None:
     service = _FakeService(pending_after_search=3)

tests/supply_agent/test_find_agent_v2_run_timeout.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Generator
+from contextlib import contextmanager
+from datetime import datetime, timedelta
+
+import pytest
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from api.services import find_agent_v2 as api_service
+from find_agent_v2.agent import FindAgentV2
+from find_agent_v2.models import FindAgentV2Run
+from find_agent_v2.observability import NullObserver
+from find_agent_v2.service import RUN_TIMEOUT_REASON
+
+
+def test_admin_list_marks_runs_older_than_60_minutes_failed(monkeypatch) -> None:
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    FindAgentV2Run.__table__.create(engine)
+    factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+    @contextmanager
+    def session_scope() -> Generator[Session, None, None]:
+        session = factory()
+        try:
+            yield session
+            session.commit()
+        finally:
+            session.close()
+
+    now = datetime(2026, 8, 14, 12, 0)
+    with factory.begin() as session:
+        session.add_all([
+            FindAgentV2Run(
+                id=1, run_id="overdue", demand_word="超时任务", input_json="{}",
+                rule_config_json="{}", status="running",
+                create_time=now - timedelta(minutes=61), update_time=now,
+            ),
+            FindAgentV2Run(
+                id=2, run_id="active", demand_word="运行任务", input_json="{}",
+                rule_config_json="{}", status="running",
+                create_time=now - timedelta(minutes=59), update_time=now,
+            ),
+        ])
+
+    expire = api_service._expire_overdue_runs
+    monkeypatch.setattr(api_service, "get_session", session_scope)
+    monkeypatch.setattr(
+        api_service, "_expire_overdue_runs", lambda session: expire(session, now=now),
+    )
+
+    listed = api_service.list_runs()
+    by_id = {item["run_id"]: item for item in listed["items"]}
+    assert by_id["overdue"]["status"] == "failed"
+    assert by_id["overdue"]["outcome_status"] == "failed"
+    assert by_id["overdue"]["stop_reason"] == RUN_TIMEOUT_REASON
+    assert by_id["active"]["status"] == "running"
+
+    second_page = api_service.list_runs(page=2, page_size=1)
+    assert second_page["page"] == 2
+    assert second_page["page_size"] == 1
+    assert second_page["total"] == 2
+    assert second_page["total_pages"] == 2
+    assert second_page["has_previous"] is True
+    assert second_page["has_next"] is False
+    assert len(second_page["items"]) == 1
+
+    with factory() as session:
+        stored = session.scalar(select(FindAgentV2Run).where(
+            FindAgentV2Run.run_id == "overdue",
+        ))
+        assert stored is not None and stored.status == "failed"
+
+
+@pytest.mark.asyncio
+async def test_agent_hard_timeout_marks_run_failed(monkeypatch) -> None:
+    class Service:
+        failed_reason = ""
+
+        def require_run(self, _run_id):
+            return {
+                "status": "failed" if self.failed_reason else "running",
+                "outcome_status": "failed" if self.failed_reason else None,
+                "demand_word": "测试", "current_round": 0, "valid_primary_count": 0,
+            }
+
+        def set_obagent_run_uid(self, *_args):
+            return None
+
+        def fail_run(self, _run_id, reason):
+            self.failed_reason = reason
+
+        def add_usage(self, *_args):
+            return None
+
+    class Runner:
+        usage = {}
+
+        def reset_usage(self):
+            return None
+
+    service = Service()
+    agent = FindAgentV2(
+        service=service, node_runner=Runner(), observer=NullObserver(),
+        max_runtime_seconds=0.01,
+    )
+
+    async def slow_run(**_kwargs):
+        await asyncio.sleep(0.1)
+
+    monkeypatch.setattr(agent, "_arun_inner", slow_run)
+    result = await agent.arun(run_id="timeout", user_input="task")
+
+    assert result.status == "failed"
+    assert result.stop_reason == RUN_TIMEOUT_REASON
+    assert service.failed_reason == RUN_TIMEOUT_REASON

tests/supply_agent/test_find_agent_v2_video_understanding.py

@@ -0,0 +1,116 @@
+"""Tests for the independent find_agent_v2 remote clipping pipeline."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from find_agent_v2 import qwen_video_understanding_30s as video_tool
+
+
+@pytest.mark.asyncio
+async def test_remote_clip_retries_once_and_prefers_complete_clip(monkeypatch, tmp_path) -> None:
+    attempts = 0
+
+    def clip(_url: str, destination: Path):
+        nonlocal attempts
+        attempts += 1
+        destination.write_bytes(b"clip")
+        return (20.0, "timeout") if attempts == 1 else (30.0, "exit=0")
+
+    monkeypatch.setattr(video_tool, "_clip_remote_once", clip)
+
+    path, duration, complete = await video_tool._clip_remote_with_one_retry("url", tmp_path)
+
+    assert attempts == 2
+    assert path.name == "clipped_attempt_2.mp4"
+    assert duration == 30.0
+    assert complete is True
+    assert video_tool.REMOTE_CLIP_TIMEOUT_SECONDS == 60.0
+
+
+@pytest.mark.asyncio
+async def test_remote_clip_accepts_longest_partial_only_after_retry(monkeypatch, tmp_path) -> None:
+    durations = iter((18.0, 16.0))
+
+    def clip(_url: str, destination: Path):
+        duration = next(durations)
+        destination.write_bytes(str(duration).encode())
+        return duration, "timeout"
+
+    monkeypatch.setattr(video_tool, "_clip_remote_once", clip)
+
+    path, duration, complete = await video_tool._clip_remote_with_one_retry("url", tmp_path)
+
+    assert path.name == "clipped_attempt_1.mp4"
+    assert duration == 18.0
+    assert complete is False
+
+
+@pytest.mark.asyncio
+async def test_remote_clip_rejects_exactly_15_seconds_or_less(monkeypatch, tmp_path) -> None:
+    durations = iter((15.0, 14.9))
+
+    def clip(_url: str, destination: Path):
+        destination.write_bytes(b"partial")
+        return next(durations), "timeout"
+
+    monkeypatch.setattr(video_tool, "_clip_remote_once", clip)
+
+    with pytest.raises(video_tool.RemoteClipError, match="超过 15 秒"):
+        await video_tool._clip_remote_with_one_retry("url", tmp_path)
+
+
+@pytest.mark.asyncio
+async def test_prepare_uploads_selected_clip_and_cleans_all_files(monkeypatch) -> None:
+    observed: dict[str, object] = {}
+
+    async def clip(_url: str, temp_dir: Path):
+        selected = temp_dir / "clipped_attempt_1.mp4"
+        unused = temp_dir / "clipped_attempt_2.mp4"
+        selected.write_bytes(b"selected")
+        unused.write_bytes(b"unused")
+        observed["temp_dir"] = temp_dir
+        observed["selected"] = selected
+        observed["unused"] = unused
+        return selected, 18.2, False
+
+    def upload(path: Path, _url: str) -> str:
+        observed["uploaded"] = path.read_bytes()
+        return "https://oss.example/clip.mp4"
+
+    monkeypatch.setattr(video_tool, "_clip_remote_with_one_retry", clip)
+    monkeypatch.setattr(video_tool, "_upload_to_oss", upload)
+
+    result = await video_tool._prepare_oss_video("https://example.test/video.mp4")
+
+    assert result == ("https://oss.example/clip.mp4", 18.2, False)
+    assert observed["uploaded"] == b"selected"
+    assert not Path(observed["selected"]).exists()
+    assert not Path(observed["unused"]).exists()
+    assert not Path(observed["temp_dir"]).exists()
+
+
+@pytest.mark.asyncio
+async def test_prepare_cleans_all_files_when_upload_fails(monkeypatch) -> None:
+    observed: dict[str, Path] = {}
+
+    async def clip(_url: str, temp_dir: Path):
+        selected = temp_dir / "clipped_attempt_1.mp4"
+        selected.write_bytes(b"selected")
+        observed["temp_dir"] = temp_dir
+        observed["selected"] = selected
+        return selected, 30.0, True
+
+    def fail_upload(_path: Path, _url: str) -> str:
+        raise RuntimeError("OSS unavailable")
+
+    monkeypatch.setattr(video_tool, "_clip_remote_with_one_retry", clip)
+    monkeypatch.setattr(video_tool, "_upload_to_oss", fail_upload)
+
+    with pytest.raises(RuntimeError, match="OSS unavailable"):
+        await video_tool._prepare_oss_video("https://example.test/video.mp4")
+
+    assert not observed["selected"].exists()
+    assert not observed["temp_dir"].exists()

web/src/api/findAgentV2.ts

@@ -1,4 +1,6 @@
-import type { FindAgentV2Detail, PagedFindAgentV2Runs } from '../types/findAgentV2'
+import type {
+  FindAgentV2Candidates, FindAgentV2Detail, PagedFindAgentV2Runs,
+} from '../types/findAgentV2'
 
 async function readJson<T>(response: Response): Promise<T> {
   const body = await response.json().catch(() => null) as { detail?: string } | null
@@ -7,13 +9,13 @@ async function readJson<T>(response: Response): Promise<T> {
 }
 
 export function fetchFindAgentV2Runs(params: {
-  status?: string; keyword?: string; limit?: number; offset?: number
+  status?: string; keyword?: string; page?: number; pageSize?: number
 } = {}): Promise<PagedFindAgentV2Runs> {
   const query = new URLSearchParams()
   if (params.status) query.set('status', params.status)
   if (params.keyword) query.set('keyword', params.keyword)
-  query.set('limit', String(params.limit || 30))
-  query.set('offset', String(params.offset || 0))
+  query.set('page', String(params.page || 1))
+  query.set('page_size', String(params.pageSize || 30))
   return fetch(`/api/find-agent-v2/runs?${query}`).then(readJson<PagedFindAgentV2Runs>)
 }
 
@@ -22,6 +24,18 @@ export function fetchFindAgentV2Detail(runId: string): Promise<FindAgentV2Detail
     .then(readJson<FindAgentV2Detail>)
 }
 
+export function fetchFindAgentV2Candidates(runId: string): Promise<FindAgentV2Candidates> {
+  return fetch(`/api/find-agent-v2/runs/${encodeURIComponent(runId)}/candidates`)
+    .then(readJson<FindAgentV2Candidates>)
+}
+
+export function fetchFindAgentV2SearchCandidates(
+  runId: string, searchId: number,
+): Promise<FindAgentV2Candidates> {
+  return fetch(`/api/find-agent-v2/runs/${encodeURIComponent(runId)}/searches/${searchId}/candidates`)
+    .then(readJson<FindAgentV2Candidates>)
+}
+
 export function createFindAgentV2Test(demandWord: string): Promise<{ accepted: boolean; run_id: string }> {
   return fetch('/api/find-agent-v2/tests', {
     method: 'POST',

web/src/types/findAgentV2.ts

@@ -24,14 +24,20 @@ export interface FindAgentV2Detail {
   run: FindAgentV2Run & { input: unknown; rule_config: unknown }
   rounds: Array<Record<string, any>>
   searches: Array<Record<string, any>>
-  candidates: Array<Record<string, any>>
-  evidence: Array<Record<string, any>>
   timeline: Array<{ type: string; time: string | null; data: Record<string, any> }>
 }
 
+export interface FindAgentV2Candidates {
+  items: Array<Record<string, any>>
+  count: number
+}
+
 export interface PagedFindAgentV2Runs {
   items: FindAgentV2Run[]
   total: number
-  limit: number
-  offset: number
+  page: number
+  page_size: number
+  total_pages: number
+  has_previous: boolean
+  has_next: boolean
 }