修改寻找agent

agents/find_agent/PRD.md

@@ -1,7 +1,7 @@
 # find_agent 当前执行逻辑 PRD
 
-> 文档版本：v2.0  
-> 代码基线：2026-07-29  
+> 文档版本：v2.1
+> 代码基线：2026-08-03
 > 文档性质：As-Is 现状说明
 
 ## 1. 文档范围
@@ -68,12 +68,11 @@ Agent 初始化时注册一个内置 `load_skill` 工具和 12 个 `find_agent`
 1. 新任务生成随机 `run_id`，初始状态为 `running`；
 2. 已有记录且未启用 `force` 时：
    - 状态为 `finished`，跳过；
-   - 或该运行下已经存在任意候选记录，跳过；
+   - 或 `attempt_count >= 1`，跳过；一次技术失败也不会自动重试；
 3. 已有记录但不满足跳过条件时，复用原 `run_id`，并将运行状态重置为 `running`；
-4. 启用 `force` 时复用已有 `run_id` 并重置运行输入，不新建第二条同需求运行。
+4. 启用 `force` 时可人工重跑，复用已有 `run_id` 并重置运行输入，不新建第二条同需求运行。
 
-复用或强制运行不会清空原 `run_id` 下的搜索页和候选，也没有“执行代次”字段。新一次
-模型执行会继续读写同一组持久化记录。
+复用或强制运行不会清空原 `run_id` 下的搜索页和候选；`attempt_count` 记录执行次数。
 
 ### 3.3 模型用户消息
 
@@ -225,6 +224,8 @@ TikHub 翻页需要原样复用上一页的 `next_cursor`、`search_id`、`backt
 
 `batch_fetch_portraits` 单次最多处理 8 个候选，按输入顺序逐条请求内容画像；设置
 `fetch_account_portrait=true` 时同时请求作者画像。每条结果自动生成标准化年龄结果。
+任一上游画像请求可能单独失败；保存时未获取侧保持 `missing`，只要视频侧或作者侧任一侧
+达到对应 50+ 门槛，画像门禁即通过。
 
 单条画像工具只返回原始画像，需要额外调用 `normalize_age_portraits`。
 
@@ -302,8 +303,9 @@ Prompt 定义联合价值关系：
 - 最终理由是否齐全。
 
 候选更新必须使用搜索工具返回的 `candidate_id`。不存在或不属于当前 `run_id` 时整批
-失败，不允许补建候选，也不允许用 `aweme_id` 更新同视频的其他搜索记录。候选更新工具
-只修改 `video_discovery_candidate`；运行状态由独立工具修改。
+失败，不允许补建候选，也不允许用 `aweme_id` 更新同视频的其他搜索记录。请求为
+`primary` 但未通过 P0 的候选会逐条自动保存为 `rejected`，同批其他合格候选不会回滚。
+候选更新工具只修改 `video_discovery_candidate`；运行状态由独立工具修改。
 
 ## 9. 结束与输出流程
 
@@ -312,9 +314,9 @@ Prompt 规定的正常结束流程为：
 `停止搜索 → 获取必要证据 → 按 candidate_id 更新已作出的候选判断
 → 将运行设为 finished → 输出`
 
-`query_video_discovery_state` 保留为按需恢复和查看已持久化状态的工具，不是正常结束的
-强制步骤。AgentLoop 没有完成守卫：模型任何一轮只要返回不含工具调用的普通消息，就会
-立即结束。
+`update_video_discovery_run_status(status="finished")` 会把剩余过程候选归档为 rejected，
+按 `aweme_id` 去重统计 `gate_status=pass` 的 primary，并计算：`goal_met`（至少 5 条）、
+`partial`（1～4 条）或 `no_match`（0 条）。完成守卫要求 finished 运行已经生成该业务结果。
 
 ## 10. 结束、超时与失败状态
 
@@ -352,27 +354,17 @@ Agent 运行或结果判定期间出现未处理异常时，调度包装层把
 工具返回的结构化错误不是未处理异常，不会自动将运行标记为 `failed`。是否停止、切换
 工具或输出“任务未完成（工具故障）”由模型判断。
 
-### 11.4 当前成功判定
+### 11.4 当前完成与业务结果判定
 
-模型循环返回后，调度侧的当前成功判定只检查：
+技术完成要求 `status=finished` 且已生成业务结果。业务结果独立计算：
 
-`该 run_id 下是否存在至少一条候选记录`
+- `goal_met`：有效且去重的 primary 至少 5 条；
+- `partial`：有效且去重的 primary 为 1～4 条；
+- `no_match`：没有有效 primary；
+- `failed`：超时、异常、未正确结束或模型显式标记技术失败。
 
-只要存在任意候选，即判定本次执行成功。该候选可以是：
-
-- `pending_evaluation`；
-- `primary`；
-- `rejected`；
-- 仅由搜索页自动创建、尚未补证的候选。
-
-成功判定不检查：
-
-- 运行状态是否为 `finished`；
-- 是否存在 `primary`；
-- 最终文本是否符合输出契约。
-
-如果一条候选都不存在，调度侧将运行标记为 `failed`，失败原因为 `no_candidates`。模型
-最终文本只作为失败原因预览附加保存，不参与成功判定。
+`partial` 和 `no_match` 都是正常完成，不触发自动重试。最终文本只用于展示，必须以状态
+工具返回的 `outcome_status` 和 `valid_primary_count` 为准。
 
 ## 11. 当前约束归属
 
@@ -393,7 +385,8 @@ Agent 运行或结果判定期间出现未处理异常时，调度包装层把
 | `primary` 具备足够详情和双画像证据 | 是 | 否 |
 | 运行最大 60 次模型迭代 | 否 | 是 |
 | 单次运行总超时 | 否 | 是 |
-| 调度成功必须存在候选 | 否 | 是 |
+| 技术完成与业务达标分开 | 否 | 是 |
+| 自动执行最多一次，人工 force 可重跑 | 否 | 是 |
 
 ## 12. 当前执行伪代码
 
@@ -404,7 +397,7 @@ load contexts
 
 for each context:
   run_id = prepare_or_reuse_run(status="running")
-  if finished or has_any_candidate and not force:
+  if (finished or attempt_count >= 1) and not force:
       skip
 
   user_input = build_agent_input(context, run_id)
@@ -424,9 +417,12 @@ for each context:
       if 60 iterations exhausted:
           agent_result = LLM("provide best answer", tools=None)
 
-      success = database.has_any_candidate(run_id)
+      run = database.lookup_run(run_id)
+      success = run.status == "finished" and run.outcome_status in {
+          "goal_met", "partial", "no_match"
+      }
       if not success:
-          mark_run_failed("no_candidates")
+          mark_run_failed("run_not_finished")
   except timeout or exception:
       mark_run_failed(error)
       raise

agents/find_agent/completion_guard.py

@@ -12,7 +12,7 @@ from supply_infra.services.video_discovery_service import (
 if TYPE_CHECKING:
     from supply_agent.agent.core import Agent
 
-_TERMINAL_STATUSES = {"finished", "failed"}
+_BUSINESS_OUTCOMES = {"goal_met", "partial", "no_match"}
 
 
 def create_find_completion_guard(run_id: str) -> CompletionGuard:
@@ -34,8 +34,17 @@ def create_find_completion_guard(run_id: str) -> CompletionGuard:
             )
 
         status = str(run.get("status") or "").strip()
-        if status in _TERMINAL_STATUSES:
+        if status == "failed":
             return None
+        if status == "finished":
+            outcome = str(run.get("outcome_status") or "").strip()
+            if outcome in _BUSINESS_OUTCOMES:
+                return None
+            return (
+                "run.status 已是 finished，但缺少程序计算的业务结果。"
+                "请再次调用 update_video_discovery_run_status，"
+                "并以返回的 outcome_status 与 valid_primary_count 为准。"
+            )
 
         return (
             f"当前 run.status 仍为 {status or 'unknown'}。"

agents/find_agent/demand_run.py

@@ -72,6 +72,9 @@ class FindDemandExecutionResult:
     skip_reason: str | None = None
     agent_result: AgentResult | None = None
     succeeded: bool = False
+    business_outcome: str | None = None
+    goal_met: bool = False
+    valid_primary_count: int = 0
     failure_reason: str | None = None
 
 
@@ -286,6 +289,7 @@ def prepare_video_discovery_run(
         "relevant_points_json": json.dumps(payload, ensure_ascii=False),
         "intent_summary": None,
         "status": "running",
+        "attempt_count": 1,
         "stop_reason": None,
         "rule_version": rule_snapshot["rule_version"],
         "rule_config_json": json.dumps(rule_snapshot, ensure_ascii=False),
@@ -412,6 +416,9 @@ def discover_videos_for_demand(
             run_id=run_id,
             agent_result=agent_result,
             succeeded=outcome.succeeded,
+            business_outcome=outcome.business_outcome,
+            goal_met=outcome.goal_met,
+            valid_primary_count=outcome.valid_primary_count,
             failure_reason=outcome.failure_reason,
         )
     except Exception as exc:

agents/find_agent/prompt/system_prompt.md

@@ -63,10 +63,10 @@
 
 ## P0 与缺失数据的判断原则
 
-P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消息中的 `quality_gate_rules`
-快照（`min_duration_seconds`、`min_share_count`、`min_content_50_plus_ratio`、时效规则等），
-再结合本次需求、候选整体质量和可观测证据做综合判断。经验判断不能脱离给定条件，
-也不能把给定条件当成无需理解的死板公式。
+P0 门槛是 `primary` 的程序准入底线。判断时**必须严格参考**用户消息中的
+`quality_gate_rules` 快照（`min_duration_seconds`、`min_share_count`、
+`min_content_50_plus_ratio`、时效规则等）。语义评分和经验判断用于排序与解释，不能覆盖
+已经明确不达标的数值。
 
 - 缺失不等于不合格。接口失败、字段未返回、画像拉取失败都应视为“未知”，降低置信度，
   但不能把未知直接写成“不适合”或“证据不足必须拒绝”。
@@ -77,8 +77,8 @@ P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消
 - 明确不达标的数值不能靠“综合表现好”绕过；补偿只适用于未知，不适用于已知失败。
 - 低相关、画像明显偏年轻、时效明显失效的候选，即使其他指标亮眼，也不应进入 `primary`。
 
-保存层对部分缺失字段留有补偿空间。下列是**参考条件**，达到时可支持补偿通过，未完全
-达到但综合证据仍很强时，也可结合经验审慎判断，并在 `decision_reason` 中说明：
+保存层只对缺失字段留有确定性的补偿空间。下列条件只适用于字段为未知；字段已经返回且
+明确低于门槛时不得补偿，并应保存为 `rejected`：
 
 1. **双侧画像均缺失**：参考任一侧 `50+` 达标的间接证据，或 `R/E/S` 综合优秀；
 2. **仅一侧画像缺失**：以另一侧是否达标为准；达标即可，不必强求双侧齐全；
@@ -196,17 +196,16 @@ P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消
 `decision_reason` 中写清缺失了什么、靠什么间接证据支撑结论，并参照上文 P0 缺失数据
 判断原则与补偿参考条件。
 
-## 9. 五条主推荐目标与相对择优
+## 9. 五条主推荐目标与正常部分完成
 
 每次任务的目标是形成 **5 条** `decision_bucket=primary` 的推荐视频。
 
-1. **优先**保留完全满足 `R/E/S` 且通过 P0（或经补偿参考条件支持）的候选；
-2. 经合理搜索、翻页、补证后仍不足 5 条时，从全部已评估候选中按综合效果**相对择优**，
-   补足至 5 条（候选总数不足 5 条时保留全部相对较优者）；
-3. 相对择优排序依据：`V` 降序为主，其次看 `R`、分享规模/效率、老年倾向证据完整度；
-4. 择优补足时仍须满足基本相关性，可结合 P0 缺失数据补偿参考条件；**不得**为凑满 5 条
-   纳入低相关、画像明显偏年轻或时效明显失效的候选；
-5. 相对择优入选的候选须在 `decision_reason` 中说明“未完全达标但相对较优”及主要依据。
+1. **只**保留满足 `R/E/S` 且通过 P0（或符合确定性缺失字段补偿条件）的候选；
+2. 经合理搜索、翻页、补证后仍不足 5 条时，以实际合格数量结束，不得把已知不达标候选
+   补入 `primary`；
+3. 合格候选内部按 `V` 降序为主，其次看 `R`、分享规模/效率和老年倾向证据完整度；
+4. 有 1～4 条合格候选是 `partial`，0 条是 `no_match`，两者均为正常完成；
+5. **不得**为凑满 5 条纳入低相关、画像明显偏年轻、分享数已知不足或时效明显失效的候选。
 
 ## 10. 多样性边际定理
 
@@ -216,7 +215,7 @@ P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消
 
 只有当一次额外搜索、详情或画像有可能改变准入、排序或置信度时，它才有价值。
 在 `primary` 不足 5 条且仍有合理搜索前沿时，应继续搜索、翻页或补证；当前沿耗尽或
-剩余候选明显无法改善 Top 5 时，按「五条主推荐目标与相对择优」从已评估候选中补足后结束。
+剩余候选明显无法改善 Top 5 时，按「五条主推荐目标与正常部分完成」保留实际合格数量并结束。
 
 # 工具的证据含义
 
@@ -244,7 +243,9 @@ P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消
   `publish_at / duration_seconds / share_count` 一并写回。
 - `batch_fetch_portraits`：用于批量取得视频点赞画像；对正式候选应设置
   `fetch_account_portrait=true`，同时取得作者粉丝画像。批量结果会自动附带
-  `age_normalization`，无需为同一候选再单独调用标准化工具。
+  `age_normalization`，无需为同一候选再单独调用标准化工具。上游可能只返回视频侧或只返回
+  作者侧画像；必须保留已成功的一侧，另一侧记为 `missing`，不得因为未同时获得双侧画像而
+  拒绝候选。任一已获取侧达到对应 50+ 门槛即可通过画像门禁。
 - `get_content_fans_portrait` / `get_account_fans_portrait`：用于补充或复核单条画像。
 - `normalize_age_portraits`：把视频与作者画像中的 `50- / 50+ / 50岁以上 / 41-50`
   等年龄桶统一为直接老年比例、TGI、成熟代理比例和证据强度；取得画像后必须调用，
@@ -253,9 +254,11 @@ P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消
   `elder_score` 不得超过该值。
 - `batch_update_video_discovery_candidates`：严格按搜索结果返回的 `candidate_id`
   更新详情、证据、**0~1 的 R/E/S/V 评分**和 `decision_bucket`。更新正式候选时必须
-  提供最新的 `publish_at / duration_seconds / share_count`、双侧画像及
-  `age_normalization`。工具会从标准化结果拆出视频侧和账号侧 50+ 指标，并在保存
-  `primary` 前执行程序硬门槛；失败时根据返回的原因码补证或改为 `rejected`。工具不会
+  提供最新的 `publish_at / duration_seconds / share_count`、已经获取到的画像及
+  `age_normalization`。不要求双侧同时存在；工具会从标准化结果拆出视频侧和账号侧 50+
+  指标，并在保存
+  `primary` 前执行程序硬门槛；未通过的 `primary` 会自动保存为 `rejected` 并返回原因码，
+  同一批中其他合格候选仍会成功保存。工具不会
   新增候选、修改搜索记录或修改运行状态。`decision_reason` 必须分别覆盖相关性、视频
   侧与作者侧 50+（说明采纳哪一侧或双侧均达标）、分享价值、时间有效性和主要限制。
   同一 `aweme_id` 对应多个 `candidate_id` 时必须分别判断和更新。
@@ -294,16 +297,17 @@ P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消
   预筛，然后进入 `douyin_detail` 和双画像阶段；
 - 内容相关性与分享动机主要依据标题、描述、`topic_list`、话题标签和详情字段判断
 - 执行新搜索、翻页、详情或画像后，应重新保存受影响候选；
-- `primary` 不足 5 条时，优先继续有效搜索；前沿耗尽后按 `V` 与证据完整度从已评估候选
-  中相对择优补足至 5 条（不足 5 条候选时保留全部较优者）。
+- `primary` 不足 5 条时优先继续仍有信息价值的搜索；完成 2～3 个语义根搜索、每个根搜索
+  至多两页，或连续两个搜索前沿没有新增高潜候选时，应结束本轮，不得为了凑数无限搜索。
 
 # 结束流程
 
-结果要求：形成 **5 条** `decision_bucket=primary` 的推荐视频。
+业务目标：形成 **5 条去重且通过 P0** 的 `decision_bucket=primary` 推荐视频。5 条是
+`goal_met` 的判定标准，不是继续执行的无限循环条件。
 
 1. 优先保留完全达标或通过 P0 补偿参考条件支持的候选；
-2. 合理搜索后仍不足 5 条时，从已评估候选中按综合效果相对择优补足至 5 条；
-3. 候选池本身不足 5 条时，保留全部相对较优者即可，不得纳入明显不合格内容。
+2. 合理搜索后仍不足 5 条时，只保留实际通过 P0 的候选，并以 `partial` 正常结束；
+3. 没有候选通过 P0 时以 `no_match` 正常结束；不得纳入明显不合格内容。
 
 正常结束时按以下流程执行：
 
@@ -311,8 +315,10 @@ P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消
 2. 对仍值得判断的候选获取并整理必要证据；
 3. 使用 `batch_update_video_discovery_candidates` 保存已经作出的候选判断；
 4. 使用 `update_video_discovery_run_status` 将运行更新为 `finished`，并记录意图摘要和
-   停止原因；
-5. 输出本次搜索与判断结果。
+   停止原因；程序会自动将未最终选择的过程候选归档为 `rejected`，并根据去重后的有效
+   primary 数计算 `goal_met / partial / no_match`；
+5. 最终输出必须严格使用状态工具返回的 `outcome_status` 和 `valid_primary_count`，不得按
+   模型记忆声称未落库的主推荐。
 
 因工具故障无法继续时，按照“工具故障终止规则”更新运行状态并输出失败摘要。
 
@@ -323,7 +329,7 @@ P0 门槛是底线，不是机械 checklist。判断时**必须参考**用户消
 
 语义相关性、分享动机和排序由 Agent 判断；时效、时长、分享数，以及视频侧或作者侧
 50+ 占比（任一侧达标即可）由程序按本次 `quality_gate_rules` 快照做准入校验。程序层对部分缺失字段留有补偿空间；
-不足 5 条时应相对择优补足，并参照 P0 缺失数据判断原则做审慎综合判断。
+不足 5 条时允许以 `partial` 或 `no_match` 正常结束，并参照 P0 缺失数据判断原则做审慎判断。
 
 禁止输出没有证据支撑的年龄结论，禁止把“内容讲老人”写成“观看者是老人”，禁止为凑满
 5 条而推荐低相关视频。

agents/find_agent/run_outcome.py

@@ -9,7 +9,12 @@ from supply_infra.services.video_discovery_service import get_video_discovery_se
 
 @dataclass(frozen=True)
 class FindAgentRunOutcome:
+    """一次运行的技术完成状态与独立业务结果。"""
+
     succeeded: bool
+    business_outcome: str
+    goal_met: bool
+    valid_primary_count: int
     failure_reason: str | None = None
 
 
@@ -17,14 +22,38 @@ def evaluate_find_agent_run(
     run_id: str,
     agent_result: AgentResult | None = None,
 ) -> FindAgentRunOutcome:
-    """有候选写入即成功：``video_discovery_candidate`` 按 run_id 存在记录。
+    """技术完成与业务达标分开判定；业务不足 5 条不会触发重跑。"""
+    del agent_result
+    run = get_video_discovery_service().lookup_run(run_id)
+    if run is None:
+        return FindAgentRunOutcome(
+            succeeded=False,
+            business_outcome="failed",
+            goal_met=False,
+            valid_primary_count=0,
+            failure_reason="run_not_found",
+        )
+
+    status = str(run.get("status") or "")
+    valid_primary_count = int(run.get("valid_primary_count") or 0)
+    business_outcome = str(run.get("outcome_status") or "")
+    if status == "finished" and business_outcome in {
+        "goal_met",
+        "partial",
+        "no_match",
+    }:
+        return FindAgentRunOutcome(
+            succeeded=True,
+            business_outcome=business_outcome,
+            goal_met=business_outcome == "goal_met",
+            valid_primary_count=valid_primary_count,
+        )
 
-    ``agent_result`` 保留给调用方兼容，不再参与成败判定。
-    """
-    del agent_result  # 成败只看库表，不看模型最终文案
-    if get_video_discovery_service().has_candidates(run_id):
-        return FindAgentRunOutcome(succeeded=True)
+    failure_reason = "run_failed" if status == "failed" else "run_not_finished"
     return FindAgentRunOutcome(
         succeeded=False,
-        failure_reason="no_candidates",
+        business_outcome="failed",
+        goal_met=False,
+        valid_primary_count=valid_primary_count,
+        failure_reason=failure_reason,
     )

agents/find_agent/support/video_discovery.py

@@ -78,6 +78,28 @@ def _age_side(
     return value if isinstance(value, dict) else {}
 
 
+def _unwrap_age_normalization(value: Any) -> Any:
+    """兼容模型把标准化结果包在额外 result 字段中的常见形态。"""
+    if not isinstance(value, dict):
+        return value
+    if isinstance(value.get("content"), dict) or isinstance(value.get("account"), dict):
+        return value
+    for key in ("age_normalization_result", "normalization_result", "result"):
+        nested = value.get(key)
+        if isinstance(nested, dict):
+            return nested
+    return value
+
+
+def _normalized_age_metric(side: dict[str, Any], field: str) -> Any:
+    """画像未获取时保持 None，避免把标准化占位值 0 当作真实低占比。"""
+    if not side:
+        return None
+    if side.get("has_age_portrait") is False or side.get("strength") == "missing":
+        return None
+    return side.get(field)
+
+
 def create_video_discovery_run(
     demand_word: str,
     relevant_points: list[dict[str, Any]],
@@ -139,6 +161,7 @@ def create_video_discovery_run(
         "relevant_points_json": _json(relevant_points or []),
         "intent_summary": _clean_text(intent_summary),
         "status": "running",
+        "attempt_count": 1,
         "rule_version": rule_snapshot["rule_version"],
         "rule_config_json": _json(rule_snapshot),
     }
@@ -167,7 +190,7 @@ 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")
+    age_normalization = _unwrap_age_normalization(item.get("age_normalization"))
     content_normalization = _age_side(age_normalization, "content")
     account_normalization = _age_side(age_normalization, "account")
     decision_bucket = (
@@ -185,22 +208,18 @@ def _normalize_candidate_update(item: dict[str, Any]) -> dict[str, Any]:
         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"),
-    )
+    content_ratio = item.get("content_50_plus_ratio")
+    if content_ratio is None:
+        content_ratio = _normalized_age_metric(content_normalization, "older_ratio")
+    content_tgi = item.get("content_50_plus_tgi")
+    if content_tgi is None:
+        content_tgi = _normalized_age_metric(content_normalization, "older_tgi")
+    account_ratio = item.get("account_50_plus_ratio")
+    if account_ratio is None:
+        account_ratio = _normalized_age_metric(account_normalization, "older_ratio")
+    account_tgi = item.get("account_50_plus_tgi")
+    if account_tgi is None:
+        account_tgi = _normalized_age_metric(account_normalization, "older_tgi")
     temporal_evidence = item.get("temporal_evidence")
     mapping = {
         "candidate_id": candidate_id,
@@ -299,12 +318,20 @@ def batch_update_video_discovery_candidates(
             run_text,
             rows,
         )
+        reclassified = list(updated.get("reclassified") or [])
         payload = {
             "title": "候选已更新",
             "run_id": run_text,
             "updated_count": updated["updated_count"],
+            "reclassified_count": int(
+                updated.get("reclassified_count", len(reclassified))
+            ),
+            "reclassified": reclassified,
             "candidates": updated["candidates"],
-            "output": f"更新 {updated['updated_count']} 条候选",
+            "output": (
+                f"更新 {updated['updated_count']} 条候选；"
+                f"P0 自动降级 {int(updated.get('reclassified_count', len(reclassified)))} 条"
+            ),
         }
         return _json(payload)
     except RunNotFoundError as exc:

alembic/versions/20260803_09_add_find_agent_outcomes.py

@@ -0,0 +1,79 @@
+"""add find_agent business outcomes and bounded attempts
+
+Revision ID: 20260803_09
+Revises: 20260731_08
+Create Date: 2026-08-03
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "20260803_09"
+down_revision: str | None = "20260731_08"
+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(
+            "valid_primary_count",
+            sa.Integer(),
+            nullable=False,
+            server_default="0",
+            comment="通过 P0 且按 aweme_id 去重后的主推荐数",
+        ),
+    )
+    op.add_column(
+        "video_discovery_run",
+        sa.Column(
+            "outcome_status",
+            sa.String(length=24),
+            nullable=True,
+            comment="goal_met / partial / no_match / failed",
+        ),
+    )
+    op.add_column(
+        "video_discovery_run",
+        sa.Column(
+            "attempt_count",
+            sa.Integer(),
+            nullable=False,
+            server_default="0",
+            comment="调度执行尝试次数",
+        ),
+    )
+    op.execute(
+        sa.text(
+            "UPDATE video_discovery_run "
+            "SET valid_primary_count = ("
+            "SELECT COUNT(DISTINCT c.aweme_id) "
+            "FROM video_discovery_candidate c "
+            "WHERE c.run_id = video_discovery_run.run_id "
+            "AND c.decision_bucket = 'primary' "
+            "AND c.gate_status = 'pass' "
+            "AND c.aweme_id IS NOT NULL AND TRIM(c.aweme_id) <> ''"
+            ")"
+        )
+    )
+    op.execute(
+        sa.text(
+            "UPDATE video_discovery_run SET outcome_status = CASE "
+            "WHEN status = 'failed' THEN 'failed' "
+            "WHEN status = 'finished' AND valid_primary_count >= 5 THEN 'goal_met' "
+            "WHEN status = 'finished' AND valid_primary_count > 0 THEN 'partial' "
+            "WHEN status = 'finished' THEN 'no_match' "
+            "ELSE NULL END, "
+            "attempt_count = 1"
+        )
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("video_discovery_run", "attempt_count")
+    op.drop_column("video_discovery_run", "outcome_status")
+    op.drop_column("video_discovery_run", "valid_primary_count")

api/services/video_discovery_records.py

@@ -51,6 +51,9 @@ def _serialize_run(row: VideoDiscoveryRun, counts: dict[str, int]) -> dict[str,
         "search_count": counts.get("search", int(row.search_count or 0)),
         "candidate_count": counts.get("candidate", 0),
         "primary_count": counts.get("primary", int(row.primary_count or 0)),
+        "valid_primary_count": int(row.valid_primary_count or 0),
+        "outcome_status": row.outcome_status,
+        "attempt_count": int(row.attempt_count or 0),
         "rejected_count": counts.get("rejected", 0),
         "pending_count": counts.get("pending_evaluation", 0),
         "stop_reason": row.stop_reason,

supply_infra/db/models/video_discovery.py

@@ -68,6 +68,23 @@ class VideoDiscoveryRun(Base):
     primary_count: Mapped[int] = mapped_column(
         Integer, nullable=False, default=0, comment="主推荐数"
     )
+    valid_primary_count: Mapped[int] = mapped_column(
+        Integer,
+        nullable=False,
+        default=0,
+        comment="通过 P0 且按 aweme_id 去重后的主推荐数",
+    )
+    outcome_status: Mapped[str | None] = mapped_column(
+        String(24),
+        nullable=True,
+        comment="goal_met / partial / no_match / failed",
+    )
+    attempt_count: Mapped[int] = mapped_column(
+        Integer,
+        nullable=False,
+        default=0,
+        comment="调度执行尝试次数",
+    )
     stop_reason: Mapped[str | None] = mapped_column(
         Text, nullable=True, comment="停止搜索的证据或失败原因"
     )

supply_infra/db/repositories/video_discovery_repo.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 import json
+from dataclasses import dataclass
 from typing import Any
 
-from sqlalchemy import func, select
+from sqlalchemy import func, select, update
 from sqlalchemy.exc import OperationalError, ProgrammingError
 
 from supply_infra.video_discovery_gates import (
@@ -42,6 +43,14 @@ def _merge_json_list(raw: str | None, values: list[Any]) -> str | None:
 
 
 _PUBLISHABLE_BUCKETS = ("primary",)
+_BUSINESS_GOAL_PRIMARY_COUNT = 5
+_PENDING_FINAL_REJECT_REASON = "NOT_SELECTED_AFTER_EVALUATION"
+
+
+@dataclass(frozen=True)
+class CandidateUpdateBatch:
+    candidates: list[VideoDiscoveryCandidate]
+    reclassified: list[dict[str, Any]]
 
 
 class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
@@ -63,18 +72,14 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         return self.session.scalar(stmt)
 
     def list_skip_grade_ids(self, biz_dt: str) -> set[int]:
-        """返回指定业务日已有候选落库的 demand_grade.id（与 run_outcome 成功口径一致）。"""
-        stmt = (
-            select(VideoDiscoveryRun.demand_grade_id)
-            .join(
-                VideoDiscoveryCandidate,
-                VideoDiscoveryCandidate.run_id == VideoDiscoveryRun.run_id,
-            )
-            .where(
-                VideoDiscoveryRun.biz_dt == biz_dt,
-                VideoDiscoveryRun.demand_grade_id.is_not(None),
-            )
-            .distinct()
+        """返回正常结束或已经耗尽重试次数的 demand_grade.id。"""
+        stmt = select(VideoDiscoveryRun.demand_grade_id).where(
+            VideoDiscoveryRun.biz_dt == biz_dt,
+            VideoDiscoveryRun.demand_grade_id.is_not(None),
+            (
+                (VideoDiscoveryRun.status == "finished")
+                | (VideoDiscoveryRun.attempt_count >= 1)
+            ),
         )
         try:
             return {
@@ -98,12 +103,17 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
             self.session.flush()
             return entity
 
+        next_attempt = int(existing.attempt_count or 0) + 1
         for key, value in values.items():
             if key in {"id", "create_time"}:
                 continue
+            if key == "attempt_count":
+                continue
             if hasattr(existing, key):
                 setattr(existing, key, value)
         existing.status = str(values.get("status") or "running")
+        existing.outcome_status = None
+        existing.attempt_count = next_attempt
         existing.stop_reason = values.get("stop_reason")
         self.session.flush()
         return existing
@@ -113,6 +123,7 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         if run is None:
             return
         run.status = "failed"
+        run.outcome_status = "failed"
         if stop_reason:
             run.stop_reason = stop_reason[:2000]
         self.session.flush()
@@ -175,8 +186,8 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         self,
         run_id: str,
         rows: list[dict[str, Any]],
-    ) -> list[VideoDiscoveryCandidate]:
-        """严格按 candidate_id 更新候选；不新增记录、不修改运行状态。"""
+    ) -> CandidateUpdateBatch:
+        """严格按 candidate_id 更新候选；P0 不合格 primary 自动降级为 rejected。"""
         candidate_ids = [int(row["candidate_id"]) for row in rows]
         if len(candidate_ids) != len(set(candidate_ids)):
             raise ValueError("candidate_id 不能重复")
@@ -199,7 +210,7 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
                 f"candidate_id 不存在或不属于 run_id={run_id}: {missing}"
             )
 
-        gate_failures: list[str] = []
+        reclassified: list[dict[str, Any]] = []
         for row in rows:
             candidate_id = int(row["candidate_id"])
             entity = existing[candidate_id]
@@ -253,8 +264,19 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
 
             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}")
+                    failed_codes = gate["failed_reason_codes"]
+                    entity.decision_bucket = "rejected"
+                    entity.reject_reason_code = str(
+                        failed_codes[0] if failed_codes else "P0_GATE_FAILED"
+                    )
+                    reclassified.append(
+                        {
+                            "candidate_id": candidate_id,
+                            "requested_bucket": "primary",
+                            "saved_bucket": "rejected",
+                            "failed_reason_codes": list(failed_codes),
+                        }
+                    )
                 else:
                     entity.reject_reason_code = None
             else:
@@ -264,12 +286,12 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
                 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]
+        self._refresh_run_counts(run_id)
+        return CandidateUpdateBatch(
+            candidates=[existing[candidate_id] for candidate_id in candidate_ids],
+            reclassified=reclassified,
+        )
 
     def finish_run(
         self,
@@ -283,14 +305,44 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         if run is None:
             raise ValueError(f"run_id 不存在: {run_id}")
         run.status = status
+        if status == "finished":
+            self._finalize_pending_candidates(run_id)
+        elif status == "failed":
+            run.outcome_status = "failed"
         if intent_summary is not None:
             run.intent_summary = intent_summary
         if stop_reason is not None:
             run.stop_reason = stop_reason
         self.session.flush()
         self._refresh_run_counts(run_id)
+        if status == "finished":
+            run.outcome_status = self._business_outcome(run.valid_primary_count)
         return run
 
+    @staticmethod
+    def _business_outcome(valid_primary_count: int) -> str:
+        if valid_primary_count >= _BUSINESS_GOAL_PRIMARY_COUNT:
+            return "goal_met"
+        if valid_primary_count > 0:
+            return "partial"
+        return "no_match"
+
+    def _finalize_pending_candidates(self, run_id: str) -> None:
+        self.session.execute(
+            update(VideoDiscoveryCandidate)
+            .where(
+                VideoDiscoveryCandidate.run_id == run_id,
+                VideoDiscoveryCandidate.decision_bucket.in_(
+                    ("pending_evaluation", "unreviewed")
+                ),
+            )
+            .values(
+                decision_bucket="rejected",
+                reject_reason_code=_PENDING_FINAL_REJECT_REASON,
+            )
+        )
+        self.session.flush()
+
     def list_searches(self, run_id: str) -> list[VideoDiscoverySearch]:
         stmt = (
             select(VideoDiscoverySearch)
@@ -342,6 +394,7 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
                 VideoDiscoveryRun.run_id == VideoDiscoveryCandidate.run_id,
             )
             .where(VideoDiscoveryCandidate.decision_bucket.in_(_PUBLISHABLE_BUCKETS))
+            .where(VideoDiscoveryCandidate.gate_status == "pass")
             .where(VideoDiscoveryCandidate.aweme_id.is_not(None))
             .where(func.trim(VideoDiscoveryCandidate.aweme_id) != "")
             .order_by(
@@ -370,6 +423,7 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
             )
             .where(VideoDiscoveryRun.biz_dt == biz_dt)
             .where(VideoDiscoveryCandidate.decision_bucket.in_(_PUBLISHABLE_BUCKETS))
+            .where(VideoDiscoveryCandidate.gate_status == "pass")
             .where(VideoDiscoveryCandidate.aweme_id.is_not(None))
             .where(func.trim(VideoDiscoveryCandidate.aweme_id) != "")
         )
@@ -419,4 +473,16 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         )
         counts = {str(bucket): int(count) for bucket, count in self.session.execute(bucket_stmt)}
         run.primary_count = counts.get("primary", 0)
+        valid_primary_stmt = select(
+            func.count(func.distinct(VideoDiscoveryCandidate.aweme_id))
+        ).where(
+            VideoDiscoveryCandidate.run_id == run_id,
+            VideoDiscoveryCandidate.decision_bucket == "primary",
+            VideoDiscoveryCandidate.gate_status == "pass",
+            VideoDiscoveryCandidate.aweme_id.is_not(None),
+            func.trim(VideoDiscoveryCandidate.aweme_id) != "",
+        )
+        run.valid_primary_count = int(
+            self.session.scalar(valid_primary_stmt) or 0
+        )
         self.session.flush()

supply_infra/scheduler/jobs/discover_videos_from_demands.py

@@ -101,6 +101,9 @@ def process_single_discover(
             "skipped": False,
             "run_id": execution.run_id,
             **summary,
+            "business_outcome": execution.business_outcome,
+            "goal_met": execution.goal_met,
+            "valid_primary_count": execution.valid_primary_count,
             "iterations": agent_result.iterations,
             "content_chars": len(agent_result.content or ""),
         }
@@ -201,6 +204,9 @@ def discover_videos_from_demands(
         "workers": 0,
         "processed": 0,
         "succeeded": 0,
+        "goal_met": 0,
+        "partial": 0,
+        "no_match": 0,
         "skipped": 0,
         "failed": 0,
         "errors": [],
@@ -325,6 +331,9 @@ def discover_videos_from_demands(
                         continue
                     if item_result.get("success"):
                         result["succeeded"] += 1
+                        business_outcome = item_result.get("business_outcome")
+                        if business_outcome in {"goal_met", "partial", "no_match"}:
+                            result[str(business_outcome)] += 1
                         continue
 
                     result["failed"] += 1

supply_infra/services/video_discovery_service.py

@@ -11,6 +11,7 @@ from supply_infra.db.repositories.video_discovery_repo import VideoDiscoveryRepo
 from supply_infra.db.session import get_session
 
 _SKIP_STATUSES = frozenset({"finished"})
+_MAX_SCHEDULED_ATTEMPTS = 1
 
 
 class RunNotFoundError(LookupError):
@@ -47,6 +48,9 @@ def _serialize_run(run: Any) -> dict[str, Any]:
         "status": run.status,
         "search_count": int(run.search_count or 0),
         "primary_count": int(run.primary_count or 0),
+        "valid_primary_count": int(run.valid_primary_count or 0),
+        "outcome_status": run.outcome_status,
+        "attempt_count": int(run.attempt_count or 0),
         "stop_reason": run.stop_reason,
         "rule_version": run.rule_version,
         "rule_config": _load_json(run.rule_config_json, {}),
@@ -212,11 +216,13 @@ class VideoDiscoveryService:
             repo = VideoDiscoveryRepository(session)
             if repo.get_run(run_id) is None:
                 raise RunNotFoundError(f"run_id 不存在: {run_id}")
-            candidates = repo.update_candidates(run_id, rows)
+            batch = repo.update_candidates(run_id, rows)
             return {
-                "updated_count": len(candidates),
+                "updated_count": len(batch.candidates),
+                "reclassified_count": len(batch.reclassified),
+                "reclassified": batch.reclassified,
                 "candidates": [
-                    _serialize_candidate(candidate) for candidate in candidates
+                    _serialize_candidate(candidate) for candidate in batch.candidates
                 ],
             }
 
@@ -282,12 +288,13 @@ class VideoDiscoveryService:
             if existing is not None and not force:
                 already_done = (
                     existing.status in _SKIP_STATUSES
-                    or repo.has_candidates(str(existing.run_id))
+                    or int(existing.attempt_count or 0) >= _MAX_SCHEDULED_ATTEMPTS
                 )
                 if already_done:
                     return None, (
                         f"biz_dt={biz_dt} demand_grade_id={demand_grade_id} "
-                        f"已执行过 run_id={existing.run_id} status={existing.status}"
+                        f"已执行过 run_id={existing.run_id} status={existing.status} "
+                        f"attempt_count={int(existing.attempt_count or 0)}"
                     )
             run_id = existing.run_id if existing is not None else str(values["run_id"])
             payload = {**values, "run_id": run_id}

supply_infra/video_discovery_gates.py

@@ -612,12 +612,7 @@ def evaluate_candidate_gate(
             portrait_reason_code = "CONTENT_PORTRAIT_MISSING"
     else:
         portrait_status = "fail"
-        if content_ratio is not None and account_ratio is not None:
-            portrait_reason_code = "PORTRAIT_50_PLUS_TOO_LOW"
-        elif content_ratio is None:
-            portrait_reason_code = "CONTENT_PORTRAIT_MISSING"
-        else:
-            portrait_reason_code = "ACCOUNT_50_PLUS_TOO_LOW"
+        portrait_reason_code = "PORTRAIT_50_PLUS_TOO_LOW"
 
     checks.append(
         {

tests/api/test_video_discovery_records.py

@@ -58,6 +58,9 @@ def _seed(factory: sessionmaker[Session]) -> None:
                     status="finished",
                     search_count=0,
                     primary_count=0,
+                    valid_primary_count=1,
+                    outcome_status="partial",
+                    attempt_count=1,
                     create_time=now,
                     update_time=now,
                 ),
@@ -149,6 +152,9 @@ def test_lists_runs_with_live_relation_counts(monkeypatch) -> None:
     assert run["search_count"] == 1
     assert run["candidate_count"] == 2
     assert run["primary_count"] == 1
+    assert run["valid_primary_count"] == 1
+    assert run["outcome_status"] == "partial"
+    assert run["attempt_count"] == 1
     assert run["rejected_count"] == 1
     assert run["relevant_points"] == [{"point": "大字模式"}]
 

tests/supply_agent/test_completion_guard.py

@@ -159,7 +159,10 @@ def test_find_agent_completion_guard_accepts_terminal_status(
     class _Service:
         @staticmethod
         def lookup_run(_run_id: str) -> dict[str, str]:
-            return {"status": status}
+            return {
+                "status": status,
+                "outcome_status": "partial" if status == "finished" else "failed",
+            }
 
     monkeypatch.setattr(
         "agents.find_agent.completion_guard.get_video_discovery_service",
@@ -197,6 +200,26 @@ def test_find_agent_completion_guard_rejects_running_status(
     assert "update_video_discovery_run_status" in feedback
 
 
+def test_find_agent_completion_guard_rejects_finished_without_business_outcome(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class _Service:
+        @staticmethod
+        def lookup_run(_run_id: str) -> dict[str, str | None]:
+            return {"status": "finished", "outcome_status": None}
+
+    monkeypatch.setattr(
+        "agents.find_agent.completion_guard.get_video_discovery_service",
+        lambda: _Service(),
+    )
+    guard = create_find_completion_guard("run-1")
+
+    feedback = guard(Message(role=Role.ASSISTANT, content="final"), ())
+
+    assert feedback is not None
+    assert "缺少程序计算的业务结果" in feedback
+
+
 def test_configure_find_agent_completion_guard_only_when_missing() -> None:
     agent = Agent()
 

tests/supply_infra/scheduler/test_discover_videos_from_demands.py

@@ -52,6 +52,7 @@ def _expire_on_commit_session_factory() -> sessionmaker[Session]:
     engine = create_engine("sqlite+pysqlite:///:memory:")
     # SQLite 对 BigInteger PK 不会自增，测试里显式写入 id。
     VideoDiscoveryRun.__table__.create(engine)
+    VideoDiscoverySearch.__table__.create(engine)
     VideoDiscoveryCandidate.__table__.create(engine)
     return sessionmaker(bind=engine, autoflush=False, autocommit=False)
 
@@ -167,8 +168,20 @@ def test_prepare_then_reuse_run_id_scheduled_flow(
     assert run_id == "scheduled-run"
 
     reuse_id, reuse_skip = prepare_video_discovery_run(ctx)
-    assert reuse_id == "scheduled-run"
-    assert reuse_skip is None
+    assert reuse_id is None
+    assert reuse_skip is not None
+    assert "attempt_count=1" in reuse_skip
+
+    forced_id, forced_skip = prepare_video_discovery_run(ctx, force=True)
+    assert forced_id == "scheduled-run"
+    assert forced_skip is None
+    from supply_infra.services.video_discovery_service import (
+        get_video_discovery_service,
+    )
+
+    forced_run = get_video_discovery_service().lookup_run("scheduled-run")
+    assert forced_run is not None
+    assert forced_run["attempt_count"] == 2
 
     payload = json.loads(
         video_discovery_store.create_video_discovery_run(
@@ -380,12 +393,23 @@ def test_p0_gate_keeps_content_and_account_portraits_separate() -> None:
     content_only = evaluate_candidate_gate(
         _p0_candidate(
             content_50_plus_ratio=0.28,
-            account_50_plus_ratio=0.08,
+            account_50_plus_ratio=None,
         ),
         _P0_RULES,
     )
     assert content_only["primary_eligible"] is True
 
+    one_side_low = evaluate_candidate_gate(
+        _p0_candidate(
+            content_50_plus_ratio=None,
+            account_50_plus_ratio=0.08,
+        ),
+        _P0_RULES,
+    )
+    assert one_side_low["primary_eligible"] is False
+    assert "PORTRAIT_50_PLUS_TOO_LOW" in one_side_low["failed_reason_codes"]
+    assert "CONTENT_PORTRAIT_MISSING" not in one_side_low["failed_reason_codes"]
+
     both_low = evaluate_candidate_gate(
         _p0_candidate(
             content_50_plus_ratio=0.10,
@@ -453,7 +477,7 @@ def test_all_search_sources_default_to_thirty_seconds() -> None:
     assert AUTHOR_SEARCH_MIN_DURATION == 30
 
 
-def test_repository_rejects_primary_that_fails_p0_gate() -> None:
+def test_repository_reclassifies_primary_that_fails_p0_gate() -> None:
     factory = _expire_on_commit_session_factory()
     with factory() as session:
         session.add(
@@ -485,11 +509,233 @@ def test_repository_rejects_primary_that_fails_p0_gate() -> None:
 
     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"}],
+        batch = repo.update_candidates(
+            "p0-gate-run",
+            [{"candidate_id": 2, "decision_bucket": "primary"}],
+        )
+        session.commit()
+
+        assert batch.candidates[0].decision_bucket == "rejected"
+        assert batch.candidates[0].reject_reason_code == "SHARE_COUNT_TOO_LOW"
+        assert batch.reclassified == [
+            {
+                "candidate_id": 2,
+                "requested_bucket": "primary",
+                "saved_bucket": "rejected",
+                "failed_reason_codes": ["SHARE_COUNT_TOO_LOW"],
+            }
+        ]
+
+
+def test_repository_keeps_valid_items_when_same_batch_contains_gate_failure() -> None:
+    factory = _expire_on_commit_session_factory()
+    with factory() as session:
+        session.add(
+            VideoDiscoveryRun(
+                id=1,
+                run_id="mixed-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_all(
+            [
+                VideoDiscoveryCandidate(
+                    id=2,
+                    run_id="mixed-gate-run",
+                    aweme_id="valid-video",
+                    decision_bucket="pending_evaluation",
+                ),
+                VideoDiscoveryCandidate(
+                    id=3,
+                    run_id="mixed-gate-run",
+                    aweme_id="low-share-video",
+                    decision_bucket="pending_evaluation",
+                ),
+            ]
+        )
+        session.commit()
+
+    base = {
+        "publish_at": datetime(2026, 7, 31, 9, 0),
+        "duration_seconds": Decimal("30.000"),
+        "content_50_plus_ratio": Decimal("0.280000"),
+        "account_50_plus_ratio": Decimal("0.080000"),
+        "decision_bucket": "primary",
+    }
+    with factory() as session:
+        batch = VideoDiscoveryRepository(session).update_candidates(
+            "mixed-gate-run",
+            [
+                {"candidate_id": 2, "share_count": 1000, **base},
+                {"candidate_id": 3, "share_count": 999, **base},
+            ],
+        )
+        session.commit()
+
+        assert [item.decision_bucket for item in batch.candidates] == [
+            "primary",
+            "rejected",
+        ]
+        assert [item["candidate_id"] for item in batch.reclassified] == [3]
+
+
+def test_finish_run_computes_partial_from_distinct_valid_primary_and_archives_pending() -> None:
+    factory = _expire_on_commit_session_factory()
+    with factory() as session:
+        session.add(
+            VideoDiscoveryRun(
+                id=1,
+                run_id="finish-partial",
+                demand_word="生活技巧",
+                relevant_points_json="[]",
+                status="running",
             )
+        )
+        session.add_all(
+            [
+                VideoDiscoveryCandidate(
+                    id=2,
+                    run_id="finish-partial",
+                    aweme_id="same-video",
+                    decision_bucket="primary",
+                    gate_status="pass",
+                ),
+                VideoDiscoveryCandidate(
+                    id=3,
+                    run_id="finish-partial",
+                    aweme_id="same-video",
+                    decision_bucket="primary",
+                    gate_status="pass",
+                ),
+                VideoDiscoveryCandidate(
+                    id=4,
+                    run_id="finish-partial",
+                    aweme_id="not-selected",
+                    decision_bucket="pending_evaluation",
+                ),
+            ]
+        )
+        session.commit()
+
+    with factory() as session:
+        repo = VideoDiscoveryRepository(session)
+        run = repo.finish_run("finish-partial", status="finished")
+        session.commit()
+
+        assert run.primary_count == 2
+        assert run.valid_primary_count == 1
+        assert run.outcome_status == "partial"
+        pending = session.scalar(
+            select(VideoDiscoveryCandidate).where(VideoDiscoveryCandidate.id == 4)
+        )
+        assert pending is not None
+        assert pending.decision_bucket == "rejected"
+        assert pending.reject_reason_code == "NOT_SELECTED_AFTER_EVALUATION"
+
+
+def test_new_run_lifecycle_accepts_one_portrait_side_and_finishes_partial(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """贯穿工具、Service、P0、结束归档和运行结果判定的完整新流程。"""
+    factory = _expire_on_commit_session_factory()
+    _patch_service_session(monkeypatch, factory)
+    with factory() as session:
+        session.add(
+            VideoDiscoveryRun(
+                id=1,
+                run_id="lifecycle-one-side",
+                demand_word="生活技巧",
+                relevant_points_json="[]",
+                status="running",
+                attempt_count=1,
+                rule_version="test-p0",
+                rule_config_json=json.dumps(_P0_RULES, ensure_ascii=False),
+            )
+        )
+        session.add_all(
+            [
+                VideoDiscoveryCandidate(
+                    id=2,
+                    run_id="lifecycle-one-side",
+                    aweme_id="account-portrait-only",
+                    title="适合家庭分享的生活技巧",
+                    decision_bucket="pending_evaluation",
+                ),
+                VideoDiscoveryCandidate(
+                    id=3,
+                    run_id="lifecycle-one-side",
+                    aweme_id="not-evaluated",
+                    decision_bucket="pending_evaluation",
+                ),
+            ]
+        )
+        session.commit()
+
+    updated = json.loads(
+        video_discovery_store.batch_update_video_discovery_candidates(
+            "lifecycle-one-side",
+            [
+                {
+                    "candidate_id": 2,
+                    "publish_at": "2026-07-31T09:00:00+08:00",
+                    "duration_seconds": 30,
+                    "share_count": 1000,
+                    "content_50_plus_ratio": None,
+                    "account_50_plus_ratio": None,
+                    "age_normalization": {
+                        "content": {
+                            "has_age_portrait": False,
+                            "older_ratio": 0,
+                            "strength": "missing",
+                        },
+                        "account": {
+                            "has_age_portrait": True,
+                            "older_ratio": 0.55,
+                            "strength": "strong",
+                        },
+                    },
+                    "decision_bucket": "primary",
+                }
+            ],
+        )
+    )
+    assert "error" not in updated
+    assert updated["reclassified_count"] == 0
+    assert updated["candidates"][0]["decision_bucket"] == "primary"
+    assert updated["candidates"][0]["gate_status"] == "pass"
+    assert updated["candidates"][0]["content_50_plus_ratio"] is None
+    assert updated["candidates"][0]["account_50_plus_ratio"] == 0.55
+
+    finished = json.loads(
+        video_discovery_store.update_video_discovery_run_status(
+            "lifecycle-one-side",
+            "finished",
+            stop_reason="合理搜索前沿已耗尽",
+        )
+    )
+    assert "error" not in finished
+    assert finished["run"]["valid_primary_count"] == 1
+    assert finished["run"]["outcome_status"] == "partial"
+
+    from agents.find_agent.run_outcome import evaluate_find_agent_run
+
+    outcome = evaluate_find_agent_run("lifecycle-one-side")
+    assert outcome.succeeded is True
+    assert outcome.business_outcome == "partial"
+    assert outcome.goal_met is False
+    assert outcome.valid_primary_count == 1
+
+    with factory() as session:
+        pending = session.scalar(
+            select(VideoDiscoveryCandidate).where(VideoDiscoveryCandidate.id == 3)
+        )
+        assert pending is not None
+        assert pending.decision_bucket == "rejected"
+        assert pending.reject_reason_code == "NOT_SELECTED_AFTER_EVALUATION"
 
 
 @pytest.mark.asyncio
@@ -646,6 +892,12 @@ def test_batch_update_candidates_uses_database_candidate_id(
                     "relevance_score": 0.8,
                     "elder_score": 0.7,
                     "share_score": 0.6,
+                    "age_normalization": {
+                        "age_normalization_result": {
+                            "content": {"older_ratio": 0.62},
+                            "account": {"older_ratio": 0.60},
+                        }
+                    },
                 }
             ],
         )
@@ -654,9 +906,69 @@ def test_batch_update_candidates_uses_database_candidate_id(
     assert result["updated_count"] == 1
     assert captured["run_id"] == "run-update"
     assert captured["rows"][0]["candidate_id"] == 901
+    assert captured["rows"][0]["content_50_plus_ratio"] == Decimal("0.620000")
+    assert captured["rows"][0]["account_50_plus_ratio"] == Decimal("0.600000")
     assert "aweme_id" not in captured["rows"][0]
 
 
+@pytest.mark.parametrize(
+    ("age_normalization", "expected_content", "expected_account"),
+    [
+        (
+            {
+                "content": {
+                    "has_age_portrait": False,
+                    "older_ratio": 0.0,
+                    "strength": "missing",
+                },
+                "account": {
+                    "has_age_portrait": True,
+                    "older_ratio": 0.55,
+                    "strength": "strong",
+                },
+            },
+            None,
+            Decimal("0.550000"),
+        ),
+        (
+            {
+                "content": {
+                    "has_age_portrait": True,
+                    "older_ratio": 0.42,
+                    "strength": "strong",
+                },
+                "account": {
+                    "has_age_portrait": False,
+                    "older_ratio": 0.0,
+                    "strength": "missing",
+                },
+            },
+            Decimal("0.420000"),
+            None,
+        ),
+    ],
+)
+def test_candidate_update_keeps_unavailable_portrait_side_missing(
+    age_normalization,
+    expected_content,
+    expected_account,
+) -> None:
+    from agents.find_agent.support.video_discovery import (
+        _normalize_candidate_update,
+    )
+
+    row = _normalize_candidate_update(
+        {
+            "candidate_id": 1,
+            "decision_bucket": "primary",
+            "age_normalization": age_normalization,
+        }
+    )
+
+    assert row["content_50_plus_ratio"] == expected_content
+    assert row["account_50_plus_ratio"] == expected_account
+
+
 def test_each_search_inserts_new_candidate_occurrences() -> None:
     engine = create_engine("sqlite+pysqlite:///:memory:")
     VideoDiscoveryRun.__table__.create(engine)
@@ -756,28 +1068,34 @@ def _seed_candidate(
         session.commit()
 
 
-def test_list_skip_grade_ids_when_candidates_exist(
+def test_list_skip_grade_ids_uses_finished_or_attempt_limit(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     factory = _expire_on_commit_session_factory()
     _patch_service_session(monkeypatch, factory)
-    _seed_run(factory, run_id="running-empty", demand_grade_id=301, status="running")
+    _seed_run(factory, run_id="finished", demand_grade_id=301, status="finished")
     _seed_run(
         factory,
-        run_id="running-with-candidates",
+        run_id="attempted",
         demand_grade_id=302,
         status="running",
         row_id=2,
     )
-    _seed_candidate(factory, run_id="running-with-candidates")
+    with factory() as session:
+        attempted = session.scalar(
+            select(VideoDiscoveryRun).where(VideoDiscoveryRun.run_id == "attempted")
+        )
+        assert attempted is not None
+        attempted.attempt_count = 1
+        session.commit()
 
     from supply_infra.services.video_discovery_service import get_video_discovery_service
 
     skip_ids = get_video_discovery_service().list_skip_grade_ids("20260728")
-    assert skip_ids == {302}
+    assert skip_ids == {301, 302}
 
 
-def test_evaluate_find_agent_run_succeeds_when_candidates_exist(
+def test_evaluate_find_agent_run_accepts_partial_as_completed(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     from agents.find_agent.run_outcome import evaluate_find_agent_run
@@ -788,7 +1106,15 @@ def test_evaluate_find_agent_run_succeeds_when_candidates_exist(
         lambda: type(
             "Svc",
             (),
-            {"has_candidates": staticmethod(lambda _run_id: True)},
+            {
+                "lookup_run": staticmethod(
+                    lambda _run_id: {
+                        "status": "finished",
+                        "outcome_status": "partial",
+                        "valid_primary_count": 3,
+                    }
+                )
+            },
         )(),
     )
 
@@ -797,10 +1123,13 @@ def test_evaluate_find_agent_run_succeeds_when_candidates_exist(
         AgentResult(content="任意文案", messages=[], iterations=3, tool_calls_made=0),
     )
     assert outcome.succeeded is True
+    assert outcome.business_outcome == "partial"
+    assert outcome.goal_met is False
+    assert outcome.valid_primary_count == 3
     assert outcome.failure_reason is None
 
 
-def test_evaluate_find_agent_run_fails_without_candidates(
+def test_evaluate_find_agent_run_fails_for_technical_failure(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     from agents.find_agent.run_outcome import evaluate_find_agent_run
@@ -811,7 +1140,15 @@ def test_evaluate_find_agent_run_fails_without_candidates(
         lambda: type(
             "Svc",
             (),
-            {"has_candidates": staticmethod(lambda _run_id: False)},
+            {
+                "lookup_run": staticmethod(
+                    lambda _run_id: {
+                        "status": "failed",
+                        "outcome_status": "failed",
+                        "valid_primary_count": 0,
+                    }
+                )
+            },
         )(),
     )
 
@@ -825,7 +1162,8 @@ def test_evaluate_find_agent_run_fails_without_candidates(
         ),
     )
     assert outcome.succeeded is False
-    assert outcome.failure_reason == "no_candidates"
+    assert outcome.business_outcome == "failed"
+    assert outcome.failure_reason == "run_failed"
 
 
 @patch(

web/src/types/videoDiscoveryRecords.ts

@@ -14,6 +14,9 @@ export interface VideoDiscoveryRunRecord {
   search_count: number
   candidate_count: number
   primary_count: number
+  valid_primary_count: number
+  outcome_status: 'goal_met' | 'partial' | 'no_match' | 'failed' | null
+  attempt_count: number
   rejected_count: number
   pending_count: number
   stop_reason: string | null

web/src/views/FindAgentRecordsView.vue

@@ -232,6 +232,16 @@ function statusLabel(value: string): string {
   return { running: '运行中', finished: '已完成', failed: '失败' }[value] || value
 }
 
+function outcomeLabel(value: VideoDiscoveryRunRecord['outcome_status']): string {
+  if (!value) return '待计算'
+  return {
+    goal_met: '完整达标',
+    partial: '部分完成',
+    no_match: '无有效结果',
+    failed: '技术失败',
+  }[value]
+}
+
 function bucketLabel(value: string): string {
   return {
     primary: '主推荐',
@@ -487,6 +497,8 @@ function openCandidate(candidate: VideoDiscoveryCandidateRecord) {
             <p v-else class="intent muted">Agent 尚未写入意图总结。</p>
             <div class="identity-row">
               <code>{{ selected.run_id }}</code>
+              <span>业务结果：{{ outcomeLabel(selected.outcome_status) }}</span>
+              <span>执行次数：{{ selected.attempt_count }}</span>
               <span v-if="selected.demand_grade_id">需求分级 #{{ selected.demand_grade_id }}</span>
               <span v-if="selected.seed_video_id">种子视频 {{ selected.seed_video_id }}</span>
             </div>
@@ -563,7 +575,7 @@ function openCandidate(candidate: VideoDiscoveryCandidateRecord) {
           <div class="metric-grid">
             <article><span>搜索页</span><strong>{{ selected.search_count }}</strong></article>
             <article><span>全部候选</span><strong>{{ selected.candidate_count }}</strong></article>
-            <article class="metric-primary"><span>主推荐</span><strong>{{ selected.primary_count }}</strong></article>
+            <article class="metric-primary"><span>有效主推荐 / 记录</span><strong>{{ selected.valid_primary_count }}<small> / {{ selected.primary_count }}</small></strong></article>
             <article class="metric-rejected"><span>淘汰 / 待评估</span><strong>{{ selected.rejected_count }}<small> / {{ selected.pending_count }}</small></strong></article>
           </div>
         </section>