修复数据库写入和结束守卫

agents/find_agent/agent.py

@@ -18,20 +18,7 @@ from agents.find_agent.tools import register_all_tools
 _PROMPT_PATH = Path(__file__).parent / "prompt" / "system_prompt.md"
 FIND_AGENT_SYSTEM_PROMPT = _PROMPT_PATH.read_text(encoding="utf-8")
 
-_EVIDENCE_TOOLS = {
-    "douyin_detail",
-    "get_content_fans_portrait",
-    "get_account_fans_portrait",
-    "batch_fetch_portraits",
-    "normalize_age_portraits",
-}
-_SEARCH_TOOLS = {
-    "douyin_search",
-    "douyin_search_tikhub",
-    "douyin_user_videos",
-}
 _FAILURE_REPORT_PREFIX = "任务未完成（工具故障）"
-_MIN_PRIMARY_TARGET = 5
 _VIDEO_ID_PATTERN = re.compile(r"(?<!\d)\d{15,22}(?!\d)")
 
 
@@ -52,10 +39,11 @@ def _report_section(
     return content[start:end]
 
 
-def _validate_report_buckets(
+def _validate_report_primary_section(
     content: str,
     state: dict[str, object],
 ) -> str | None:
+    """Only validate aweme_ids listed in the primary report section."""
     candidates = state.get("candidates")
     if not isinstance(candidates, list):
         return "最终状态缺少 candidates，无法校验报告分池"
@@ -64,50 +52,30 @@ def _validate_report_buckets(
         for item in candidates
         if isinstance(item, dict) and item.get("aweme_id")
     }
-    run = state.get("run")
-    run_state = run if isinstance(run, dict) else {}
 
     primary_section = _report_section(
         content,
         "主推荐",
-        ("淘汰候选",),
-    )
-    rejected_section = _report_section(
-        content,
-        "淘汰候选",
-        ("搜索树", "缺失数据", "总结"),
+        ("淘汰候选", "搜索树", "缺失数据", "总结"),
     )
-    if primary_section is None or rejected_section is None:
-        return "最终报告必须分别包含“主推荐”和“淘汰候选”段"
+    if primary_section is None:
+        return "最终报告必须包含“主推荐”段"
 
     primary_ids = set(_VIDEO_ID_PATTERN.findall(primary_section))
-    rejected_ids = set(_VIDEO_ID_PATTERN.findall(rejected_section))
+    if not primary_ids:
+        return "主推荐段必须输出至少一个 aweme_id"
+
     wrong_primary = sorted(
         video_id
         for video_id in primary_ids
         if bucket_by_id.get(video_id) != "primary"
     )
-    wrong_rejected = sorted(
-        video_id
-        for video_id in rejected_ids
-        if bucket_by_id.get(video_id) != "rejected"
-    )
     if wrong_primary:
         return (
             "主推荐段包含非 primary 候选: "
             + ", ".join(wrong_primary[:5])
             + "。必须按数据库 decision_bucket 输出"
         )
-    if wrong_rejected:
-        return (
-            "淘汰候选段包含非 rejected 候选: "
-            + ", ".join(wrong_rejected[:5])
-            + "。必须按数据库 decision_bucket 输出"
-        )
-
-    primary_count = int(run_state.get("primary_count") or 0)
-    if primary_count > 0 and not primary_ids:
-        return "数据库存在 primary 候选，但主推荐段没有输出 aweme_id"
     return None
 
 
@@ -118,84 +86,6 @@ def _primary_count_from_state(state: dict[str, object]) -> int:
     return 0
 
 
-def _validate_report_primary_section(
-    content: str,
-    state: dict[str, object],
-) -> str | None:
-    candidates = state.get("candidates")
-    if not isinstance(candidates, list):
-        return "最终状态缺少 candidates，无法校验报告分池"
-    bucket_by_id = {
-        str(item.get("aweme_id")): str(item.get("decision_bucket"))
-        for item in candidates
-        if isinstance(item, dict) and item.get("aweme_id")
-    }
-    run = state.get("run")
-    run_state = run if isinstance(run, dict) else {}
-
-    primary_section = _report_section(
-        content,
-        "主推荐",
-        ("淘汰候选", "搜索树", "缺失数据", "总结"),
-    )
-    if primary_section is None:
-        return "最终报告必须包含“主推荐”段"
-
-    primary_ids = set(_VIDEO_ID_PATTERN.findall(primary_section))
-    wrong_primary = sorted(
-        video_id
-        for video_id in primary_ids
-        if bucket_by_id.get(video_id) != "primary"
-    )
-    if wrong_primary:
-        return (
-            "主推荐段包含非 primary 候选: "
-            + ", ".join(wrong_primary[:5])
-            + "。必须按数据库 decision_bucket 输出"
-        )
-
-    primary_count = int(run_state.get("primary_count") or 0)
-    if primary_count > 0 and not primary_ids:
-        return "数据库存在 primary 候选，但主推荐段没有输出 aweme_id"
-    return None
-
-
-def _guard_relaxed_primary_target(
-    events: list[tuple[int, str, dict[str, object]]],
-    messages: list[Message],
-) -> str | None:
-    """Allow finish when enough primary candidates are persisted."""
-    evaluation_events = [
-        (index, payload)
-        for index, name, payload in events
-        if name == "batch_save_video_candidate_evaluations"
-    ]
-    if not evaluation_events:
-        return "尚未保存候选评估"
-    _, evaluation = evaluation_events[-1]
-    if evaluation.get("status") != "finished":
-        return "最后一次候选保存尚未把运行状态设置为 finished"
-
-    state_events = [
-        (index, payload)
-        for index, name, payload in events
-        if name == "query_video_discovery_state"
-    ]
-    if not state_events:
-        return "尚未查询最终数据库状态"
-    _, state = state_events[-1]
-    if _primary_count_from_state(state) < _MIN_PRIMARY_TARGET:
-        return f"数据库 primary 数量不足 {_MIN_PRIMARY_TARGET} 条"
-
-    last_assistant = _last_final_assistant_message(messages)
-    if last_assistant is None or not (last_assistant.content or "").strip():
-        return "最终回答为空"
-    report_error = _validate_report_primary_section(last_assistant.content, state)
-    if report_error:
-        return report_error
-    return None
-
-
 def _successful_tool_events(
     messages: list[Message],
 ) -> list[tuple[int, str, dict[str, object]]]:
@@ -225,7 +115,7 @@ def _last_final_assistant_message(messages: list[Message]) -> Message | None:
 
 
 def find_agent_completion_guard(messages: list[Message]) -> str | None:
-    """Enforce the final search → evidence → evaluation → audit → state → report order."""
+    """返回完成提示；默认仅提示、不阻断结束。"""
     last_assistant = _last_final_assistant_message(messages)
     content = (last_assistant.content or "").strip() if last_assistant else ""
     if content.startswith(_FAILURE_REPORT_PREFIX):
@@ -235,39 +125,6 @@ def find_agent_completion_guard(messages: list[Message]) -> str | None:
     if not any(name == "create_video_discovery_run" for _, name, _ in events):
         return "尚未成功创建视频发现运行"
 
-    state_events = [
-        (index, payload)
-        for index, name, payload in events
-        if name == "query_video_discovery_state"
-    ]
-    if state_events:
-        _, latest_state = state_events[-1]
-        if _primary_count_from_state(latest_state) >= _MIN_PRIMARY_TARGET:
-            return _guard_relaxed_primary_target(events, messages)
-
-    search_indexes = [
-        index
-        for index, name, _ in events
-        if name == "record_video_search_page"
-    ]
-    if not search_indexes:
-        return "尚未持久化任何搜索页"
-    last_search_index = max(search_indexes)
-    raw_search_indexes = [
-        index for index, name, _ in events if name in _SEARCH_TOOLS
-    ]
-    if raw_search_indexes and last_search_index < max(raw_search_indexes):
-        return "最后一次搜索结果尚未通过 record_video_search_page 持久化"
-
-    evidence_indexes = [
-        index for index, name, _ in events if name in _EVIDENCE_TOOLS
-    ]
-    if not evidence_indexes:
-        return "尚未获取并整理候选证据"
-    last_evidence_index = max(evidence_indexes)
-    if last_evidence_index < last_search_index:
-        return "最后一次搜索后尚未重新获取并整理候选证据"
-
     evaluation_events = [
         (index, payload)
         for index, name, payload in events
@@ -276,28 +133,9 @@ def find_agent_completion_guard(messages: list[Message]) -> str | None:
     if not evaluation_events:
         return "尚未保存候选评估"
     evaluation_index, evaluation = evaluation_events[-1]
-    if evaluation_index < last_evidence_index:
-        return "最后一次证据获取发生在候选评估之后，证据尚未重新评估并保存"
     if evaluation.get("status") != "finished":
         return "最后一次候选保存尚未把运行状态设置为 finished"
 
-    audit_events = [
-        (index, payload)
-        for index, name, payload in events
-        if name == "audit_video_discovery_run"
-    ]
-    if not audit_events:
-        return "尚未调用 audit_video_discovery_run 执行数据库完成审计"
-    audit_index, audit = audit_events[-1]
-    if audit_index < evaluation_index:
-        return "候选评估晚于最后一次审计，请重新调用 audit_video_discovery_run"
-    if audit.get("can_finish") is not True:
-        violations = audit.get("critical_violations")
-        if isinstance(violations, list) and violations:
-            summary = "；".join(str(item) for item in violations[:5])
-            return f"最后一次审计未通过：{summary}"
-        return "最后一次审计未通过"
-
     state_events = [
         (index, payload)
         for index, name, payload in events
@@ -306,15 +144,19 @@ def find_agent_completion_guard(messages: list[Message]) -> str | None:
     if not state_events:
         return "尚未查询最终数据库状态"
     state_index, state = state_events[-1]
-    if state_index < audit_index:
-        return "最终状态查询必须在审计通过之后执行"
+    if state_index < evaluation_index:
+        return "最终状态查询必须在候选保存为 finished 之后执行"
 
-    last_assistant = _last_final_assistant_message(messages)
-    if last_assistant is None or not (last_assistant.content or "").strip():
+    run = state.get("run")
+    run_state = run if isinstance(run, dict) else {}
+    if run_state.get("status") != "finished":
+        return "运行状态尚未持久化为 finished"
+
+    if last_assistant is None or not content:
         return "最终回答为空"
-    report_error = _validate_report_buckets(last_assistant.content, state)
-    if report_error:
-        return report_error
+
+    if _primary_count_from_state(state) > 0:
+        return _validate_report_primary_section(content, state)
     return None
 
 

agents/find_agent/demand_run.py

@@ -70,6 +70,8 @@ class FindDemandExecutionResult:
     skipped: bool = False
     skip_reason: str | None = None
     agent_result: AgentResult | None = None
+    succeeded: bool = False
+    failure_reason: str | None = None
 
 
 def _resolve_biz_dt(biz_dt: str | None) -> str:
@@ -381,8 +383,25 @@ def discover_videos_for_demand(
 
     user_input = build_find_agent_user_input(ctx, run_id)
     try:
+        from agents.find_agent.run_outcome import evaluate_find_agent_run
+
         agent_result = run_find_agent(user_input)
-        return FindDemandExecutionResult(run_id=run_id, agent_result=agent_result)
+        outcome = evaluate_find_agent_run(run_id, agent_result)
+        if not outcome.succeeded:
+            stop_reason = outcome.failure_reason or "incomplete"
+            content_preview = (agent_result.content or "").strip()[:500]
+            if content_preview:
+                stop_reason = f"{stop_reason}: {content_preview}"
+            get_video_discovery_service().mark_run_failed(
+                run_id,
+                stop_reason=stop_reason,
+            )
+        return FindDemandExecutionResult(
+            run_id=run_id,
+            agent_result=agent_result,
+            succeeded=outcome.succeeded,
+            failure_reason=outcome.failure_reason,
+        )
     except Exception as exc:
         get_video_discovery_service().mark_run_failed(
             run_id,

agents/find_agent/output_sync.py

@@ -1,52 +0,0 @@
-"""Parse recommendation buckets from the model's final response."""
-from __future__ import annotations
-
-import re
-
-_VIDEO_ID_PATTERN = re.compile(r"(?<!\d)\d{15,22}(?!\d)")
-_PRIMARY_LABELS = ("主推荐", "正式推荐")
-_REJECTED_LABELS = ("淘汰候选", "淘汰")
-_END_LABELS = (
-    "搜索轨迹",
-    "搜索树",
-    "缺失数据",
-    "局限",
-    "总结",
-)
-
-
-def _normalized_heading(line: str) -> str:
-    text = line.strip()
-    text = re.sub(r"^#{1,6}\s*", "", text)
-    text = re.sub(r"^\d+\s*[.、]\s*", "", text)
-    return text.replace("*", "").strip()
-
-
-def extract_model_recommendation_ids(
-    content: str,
-) -> tuple[list[str], list[str]]:
-    """Extract video ids from the model's primary and rejected sections."""
-    primary: list[str] = []
-    rejected: list[str] = []
-    section: str | None = None
-
-    for line in (content or "").splitlines():
-        heading = _normalized_heading(line)
-        if any(heading.startswith(label) for label in _PRIMARY_LABELS):
-            section = "primary"
-        elif any(heading.startswith(label) for label in _REJECTED_LABELS):
-            section = "rejected"
-        elif any(heading.startswith(label) for label in _END_LABELS):
-            section = None
-
-        if section is None:
-            continue
-        target = primary if section == "primary" else rejected
-        for aweme_id in _VIDEO_ID_PATTERN.findall(line):
-            if aweme_id not in target:
-                target.append(aweme_id)
-
-    primary_set = set(primary)
-    return primary, [
-        aweme_id for aweme_id in rejected if aweme_id not in primary_set
-    ]

agents/find_agent/prompt/system_prompt.md

@@ -113,11 +113,12 @@
 
 `V(v) = 100 × R(v)^0.40 × E(v)^0.35 × S(v)^0.25`
 
-这意味着任一维度接近零，整体价值都会被明显压低。评分用于保持排序一致，不得制造虚假精确性；最终报告应展示整数分、原始数据和理由。
+这意味着任一维度接近零，整体价值都会被明显压低。评分用于保持排序一致，不得制造虚假精确性；**数据库保存与最终报告中的 `R/E/S/V` 均使用 `0~1` 小数，原样写入，不做百分制换算**。
 
-`R/E/S/V` 是帮助你保持判断一致的参考量，不是程序校验线。候选最终进入
-`primary / rejected` 完全由你根据全部证据判断。保存工具不会重算分数、
-设置画像上限或改写你的 `decision_bucket`。
+`R/E/S/V` 是帮助你保持判断一致的参考量，**不是程序校验线**。候选最终进入
+`primary / rejected` 完全由你根据全部证据判断。`batch_save_video_candidate_evaluations`
+会 **原样保存** 你给出的 `0~1` 分数，不会重算、换算或改写 `decision_bucket`。
+`audit_video_discovery_run` **不校验分数**，只审计证据完备性、分池合法性和搜索覆盖。
 
 优先目标是保留至少 5 条质量可靠的 `primary`，可以超过 5 条。5 条是搜索和筛选的
 优先目标，不是硬性准入线：
@@ -165,12 +166,13 @@
   翻页来源、作者来源、供应方分页状态和本页结果；TikHub 搜索及作者作品页也必须保存，
   任何搜索页都不能只存在于上下文中。新召回候选初始状态是
   `pending_evaluation`，表示等待 Agent 补证和评分，不能直接输出。
-- `batch_save_video_candidate_evaluations`：原样保存你给出的详情、证据、评分和
-  `decision_bucket`。工具只接受 `primary / rejected`，不会重算分数或替你改池。
+- `batch_save_video_candidate_evaluations`：原样保存你给出的详情、证据、**0~1 的 R/E/S/V 评分**
+  和 `decision_bucket`。工具只接受 `primary / rejected`，不会重算分数或替你改池。
   每个候选至少传入 `aweme_id` 和你决定的 `decision_bucket`；其他证据、理由和分数
   尽量完整传入。
 - `audit_video_discovery_run`：候选评估完成后按 `run_id` 从数据库读取完整搜索和候选
-  状态，执行确定性完成审计。只有返回 `can_finish=true` 才能进入最终状态查询。
+  状态，执行确定性完成审计（证据、分池、搜索覆盖；**不校验 R/E/S/V 分数**）。只有返回
+  `can_finish=true` 才能进入最终状态查询。
 - `query_video_discovery_state`：恢复长搜索的已探索关键词、翻页状态、主推荐和淘汰
   候选，也用于查看已经保存的模型决定。结束前的最后一次查询必须发生在审计通过后，
   最终报告只能依据这次查询结果生成。
@@ -253,7 +255,7 @@
 - 视频点赞年龄画像证据；
 - 作者粉丝年龄画像证据；
 - 老年人可能愿意分享的内容动机；
-- `R / E / S / V` 整数分、置信度（高/中/低）；
+- `R / E / S / V` 的 `0~1` 分值、置信度（高/中/低）；
 - 一句包含正证与主要限制的推荐理由。
 
 输出顺序：

agents/find_agent/run_outcome.py

@@ -0,0 +1,49 @@
+"""判定 find_agent 调度执行是否真正成功完成。"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from supply_agent.types import AgentResult
+from supply_infra.services.video_discovery_service import get_video_discovery_service
+
+TOOL_FAILURE_PREFIX = "任务未完成（工具故障）"
+MAX_ITERATIONS_PREFIX = "Max iterations reached"
+
+
+@dataclass(frozen=True)
+class FindAgentRunOutcome:
+    succeeded: bool
+    failure_reason: str | None = None
+
+
+def evaluate_find_agent_run(run_id: str, agent_result: AgentResult) -> FindAgentRunOutcome:
+    """根据 Agent 输出与 DB run 状态判断是否算成功完成。"""
+    content = (agent_result.content or "").strip()
+
+    if content.startswith(TOOL_FAILURE_PREFIX):
+        return FindAgentRunOutcome(
+            succeeded=False,
+            failure_reason="agent_declared_tool_failure",
+        )
+
+    if content.startswith(MAX_ITERATIONS_PREFIX):
+        return FindAgentRunOutcome(
+            succeeded=False,
+            failure_reason="max_iterations_reached",
+        )
+
+    run = get_video_discovery_service().lookup_run(run_id)
+    if run is None:
+        return FindAgentRunOutcome(
+            succeeded=False,
+            failure_reason="run_not_found",
+        )
+
+    status = str(run.get("status") or "")
+    if status != "finished":
+        return FindAgentRunOutcome(
+            succeeded=False,
+            failure_reason=f"run_status_{status or 'unknown'}",
+        )
+
+    return FindAgentRunOutcome(succeeded=True)

agents/find_agent/tools/decision_support.py

@@ -222,15 +222,6 @@ def normalize_age_portraits(
     return json.dumps(result, ensure_ascii=False)
 
 
-def _float_score(candidate: dict[str, Any], key: str) -> float | None:
-    value = candidate.get(key)
-    try:
-        number = float(value)
-    except (TypeError, ValueError):
-        return None
-    return number if 0 <= number <= 1 else None
-
-
 @tool
 def audit_video_discovery_process(
     searches: list[dict[str, Any]],
@@ -240,6 +231,8 @@ def audit_video_discovery_process(
     """
     在结束找片前审计搜索树、翻页、标签扩展、证据完备性和最终分流。
 
+    不校验 R/E/S/V 分数，也不根据分数质疑 decision_bucket；分数由 Agent 原样保存。
+
     Args:
         searches: 已执行搜索页。建议包含 search_id、keyword、source_type、
             parent_search_id、cursor、page_no、has_more、new_candidate_count。
@@ -315,7 +308,6 @@ def audit_video_discovery_process(
         "pending_evaluation": 0,
     }
     pending_evaluation_messages: list[str] = []
-    rejected_misclass_messages: list[str] = []
     for candidate in valid_candidates:
         aweme_id = str(candidate.get("aweme_id") or "unknown")
         bucket = str(
@@ -327,9 +319,6 @@ def audit_video_discovery_process(
             critical.append(f"{aweme_id} 使用了不支持的分池: {bucket}")
             continue
         bucket_counts[bucket] += 1
-        relevance = _float_score(candidate, "relevance_score")
-        elder = _float_score(candidate, "elder_score")
-        share = _float_score(candidate, "share_score")
 
         if bucket == "primary":
             if not candidate.get("detail_verified"):
@@ -341,27 +330,6 @@ def audit_video_discovery_process(
             if not candidate.get("age_portraits_normalized"):
                 critical.append(f"{aweme_id} 未标准化年龄画像")
 
-        if bucket == "primary" and not (
-            relevance is not None
-            and relevance >= 0.55
-            and elder is not None
-            and elder >= 0.55
-            and share is not None
-            and share >= 0.50
-        ):
-            critical.append(f"{aweme_id} 不满足 primary 阈值")
-        meets_primary = (
-            relevance is not None
-            and relevance >= 0.55
-            and elder is not None
-            and elder >= 0.55
-            and share is not None
-            and share >= 0.50
-        )
-        if bucket == "rejected" and meets_primary:
-            rejected_misclass_messages.append(
-                f"{aweme_id} 达到 primary 线却被错误淘汰"
-            )
         if bucket == "pending_evaluation":
             message = f"{aweme_id} 等待 Agent 补证和评估"
             if intended_status == "finished":
@@ -372,10 +340,8 @@ def audit_video_discovery_process(
     retained_count = bucket_counts.get("primary", 0)
     if retained_count >= 5 and intended_status == "finished":
         warnings.extend(pending_evaluation_messages)
-        warnings.extend(rejected_misclass_messages)
     else:
         critical.extend(pending_evaluation_messages)
-        critical.extend(rejected_misclass_messages)
 
     if retained_count < 5:
         warnings.append(

agents/find_agent/tools/video_discovery_store.py

@@ -400,7 +400,8 @@ def batch_save_video_candidate_evaluations(
     原样保存 Agent 给出的候选详情、证据、评分和分池。
 
     本工具不重算 R/E/S/V，不执行画像证据上限，也不根据阈值修改
-    decision_bucket。最终分池只接受 primary / rejected。
+    decision_bucket。分数按 Agent 提供的原始值写入（`0~1` 小数）。
+    最终分池只接受 primary / rejected。
 
     Args:
         run_id: 发现运行 id。

supply_agent/agent/core.py

@@ -61,6 +61,7 @@ class Agent:
         reasoning_effort: str | None = None,
         logger: AgentLogger | None = None,
         completion_guard: Callable[[list[Message]], str | None] | None = None,
+        completion_guard_blocks: bool = False,
         tool_call_budgets: dict[str, tuple[set[str], int]] | None = None,
         tool_repeat_requires_change: dict[str, set[str]] | None = None,
     ) -> None:
@@ -81,6 +82,7 @@ class Agent:
         self.max_iterations = max_iterations or self.settings.agent_max_iterations
         self.temperature = temperature
         self.completion_guard = completion_guard
+        self.completion_guard_blocks = completion_guard_blocks
         self.tool_call_budgets = tool_call_budgets or {}
         self.tool_repeat_requires_change = tool_repeat_requires_change or {}
 
@@ -140,6 +142,7 @@ class Agent:
             logger=self.logger,
             active_skills=self._active_skills,
             completion_guard=self.completion_guard,
+            completion_guard_blocks=self.completion_guard_blocks,
             tool_call_budgets=self.tool_call_budgets,
             tool_repeat_requires_change=self.tool_repeat_requires_change,
         )

supply_agent/agent/loop.py

@@ -40,6 +40,7 @@ class AgentLoop:
         active_skills: list[str] | None = None,
         system_message_builder: Callable[[], Message] | None = None,
         completion_guard: Callable[[list[Message]], str | None] | None = None,
+        completion_guard_blocks: bool = False,
         tool_call_budgets: dict[str, tuple[set[str], int]] | None = None,
         tool_repeat_requires_change: dict[str, set[str]] | None = None,
     ) -> None:
@@ -53,6 +54,7 @@ class AgentLoop:
         self.logger = logger
         self.active_skills = active_skills or []
         self.completion_guard = completion_guard
+        self.completion_guard_blocks = completion_guard_blocks
         self.tool_call_budgets = tool_call_budgets or {}
         self.tool_repeat_requires_change = tool_repeat_requires_change or {}
         self._tool_budget_counts = {
@@ -81,6 +83,15 @@ class AgentLoop:
             return None
         return self.completion_guard(self.messages)
 
+    def _should_block_completion(self, reason: str | None) -> bool:
+        return bool(reason and self.completion_guard_blocks)
+
+    def _completion_response_content(self, content: str) -> str:
+        reason = self._completion_block_reason()
+        if reason and not self.completion_guard_blocks:
+            return f"{content}\n\n---\n完成提示（未阻断结束）：{reason}"
+        return content
+
     def _append_completion_feedback(self, reason: str) -> None:
         self.messages.append(
             Message(
@@ -214,6 +225,30 @@ class AgentLoop:
             content = f"Max iterations reached before completion: {reason}"
         return self._build_result(content, iterations)
 
+    def _iterations_exhausted_result(self, iterations: int) -> AgentResult:
+        if self.completion_guard is not None and self.completion_guard_blocks:
+            return self._max_iterations_result(iterations)
+        last_assistant = next(
+            (
+                message
+                for message in reversed(self.messages)
+                if message.role == Role.ASSISTANT and not message.tool_calls
+            ),
+            None,
+        )
+        content = (
+            (last_assistant.content or "").strip()
+            if last_assistant and (last_assistant.content or "").strip()
+            else "Max iterations reached"
+        )
+        return self._build_result(
+            self._completion_response_content(content),
+            iterations,
+        )
+
+    def _uses_blocking_completion_guard(self) -> bool:
+        return self.completion_guard is not None and self.completion_guard_blocks
+
     def run(self) -> AgentResult:
         iterations = 0
         while iterations < self.max_iterations:
@@ -232,10 +267,13 @@ class AgentLoop:
 
             if not response.tool_calls:
                 reason = self._completion_block_reason()
-                if reason:
+                if self._should_block_completion(reason):
                     self._append_completion_feedback(reason)
                     continue
-                return self._build_result(response.content or "", iterations)
+                return self._build_result(
+                    self._completion_response_content(response.content or ""),
+                    iterations,
+                )
 
             for tc in response.tool_calls:
                 self.tool_calls_made += 1
@@ -276,7 +314,7 @@ class AgentLoop:
                     )
                 )
 
-        if self.completion_guard is not None:
+        if self._uses_blocking_completion_guard():
             return self._max_iterations_result(iterations)
 
         self.messages.append(
@@ -311,10 +349,13 @@ class AgentLoop:
 
             if not response.tool_calls:
                 reason = self._completion_block_reason()
-                if reason:
+                if self._should_block_completion(reason):
                     self._append_completion_feedback(reason)
                     continue
-                return self._build_result(response.content or "", iterations)
+                return self._build_result(
+                    self._completion_response_content(response.content or ""),
+                    iterations,
+                )
 
             for tc in response.tool_calls:
                 self.tool_calls_made += 1
@@ -357,7 +398,7 @@ class AgentLoop:
                     )
                 )
 
-        if self.completion_guard is not None:
+        if self._uses_blocking_completion_guard():
             return self._max_iterations_result(iterations)
 
         self.messages.append(
@@ -397,16 +438,19 @@ class AgentLoop:
 
             if not response.tool_calls:
                 reason = self._completion_block_reason()
-                if reason:
+                if self._should_block_completion(reason):
                     self._append_completion_feedback(reason)
                     continue
+                final_content = self._completion_response_content(
+                    response.content or ""
+                )
                 yield AgentEvent(
                     type=AgentEventType.MESSAGE,
-                    data={"content": response.content or ""},
+                    data={"content": final_content},
                 )
                 yield AgentEvent(
                     type=AgentEventType.DONE,
-                    data=self._build_result(response.content or "", iterations).model_dump(),
+                    data=self._build_result(final_content, iterations).model_dump(),
                 )
                 return
 
@@ -455,7 +499,7 @@ class AgentLoop:
 
         yield AgentEvent(
             type=AgentEventType.DONE,
-            data=self._max_iterations_result(iterations).model_dump(),
+            data=self._iterations_exhausted_result(iterations).model_dump(),
         )
 
     async def astream(self) -> AsyncIterator[AgentEvent]:
@@ -481,16 +525,19 @@ class AgentLoop:
 
             if not response.tool_calls:
                 reason = self._completion_block_reason()
-                if reason:
+                if self._should_block_completion(reason):
                     self._append_completion_feedback(reason)
                     continue
+                final_content = self._completion_response_content(
+                    response.content or ""
+                )
                 yield AgentEvent(
                     type=AgentEventType.MESSAGE,
-                    data={"content": response.content or ""},
+                    data={"content": final_content},
                 )
                 yield AgentEvent(
                     type=AgentEventType.DONE,
-                    data=self._build_result(response.content or "", iterations).model_dump(),
+                    data=self._build_result(final_content, iterations).model_dump(),
                 )
                 return
 
@@ -541,7 +588,7 @@ class AgentLoop:
 
         yield AgentEvent(
             type=AgentEventType.DONE,
-            data=self._max_iterations_result(iterations).model_dump(),
+            data=self._iterations_exhausted_result(iterations).model_dump(),
         )
 
     def _build_result(self, content: str, iterations: int) -> AgentResult:

supply_infra/db/repositories/video_discovery_repo.py

@@ -72,11 +72,11 @@ class VideoDiscoveryRepository(BaseRepository[VideoDiscoveryRun]):
         return self.session.scalar(stmt)
 
     def list_skip_grade_ids(self, biz_dt: str) -> set[int]:
-        """返回指定业务日已有 running/finished 记录的 demand_grade.id。"""
+        """返回指定业务日已成功完成（finished）的 demand_grade.id。"""
         stmt = select(VideoDiscoveryRun.demand_grade_id).where(
             VideoDiscoveryRun.biz_dt == biz_dt,
             VideoDiscoveryRun.demand_grade_id.is_not(None),
-            VideoDiscoveryRun.status.in_(("running", "finished")),
+            VideoDiscoveryRun.status == "finished",
         )
         try:
             return {

supply_infra/scheduler/jobs/discover_videos_from_demands.py

@@ -71,6 +71,20 @@ def process_single_discover(
                 "skip_reason": execution.skip_reason,
             }
 
+        if not execution.succeeded:
+            return {
+                "success": False,
+                "skipped": False,
+                "run_id": execution.run_id,
+                **summary,
+                "error": execution.failure_reason or "find_agent_incomplete",
+                "iterations": (
+                    execution.agent_result.iterations
+                    if execution.agent_result is not None
+                    else None
+                ),
+            }
+
         agent_result = execution.agent_result
         assert agent_result is not None
         logger.info(

supply_infra/scheduler/jobs/publish_videos_from_discovery.py

@@ -169,20 +169,24 @@ def publish_videos_from_discovery(
             if not batch.bind_success:
                 batch.bind_error = str(bind_result.get("error") or "绑定生成计划失败")
 
-            if crawler_plan_id and batch.bind_success:
+            if crawler_plan_id:
+                plan_label = plan_pair.label
+                if not batch.bind_success:
+                    plan_label = f"BIND_FAILED:{batch.bind_error}"
                 updated = get_video_discovery_service().mark_candidates_aigc_plans(
                     id_batch,
                     crawler_plan_id=crawler_plan_id,
                     produce_plan_id=plan_pair.produce_plan_id,
                     publish_plan_id=plan_pair.publish_plan_id,
-                    plan_label=plan_pair.label,
+                    plan_label=plan_label,
                 )
                 logger.info(
-                    "published batch: plan=%s crawler=%s videos=%d db_updated=%d",
+                    "published batch: plan=%s crawler=%s videos=%d db_updated=%d bind_success=%s",
                     plan_pair.label,
                     crawler_plan_id,
                     len(aweme_batch),
                     updated,
+                    batch.bind_success,
                 )
 
             batch_results.append(batch)

supply_infra/services/video_discovery_service.py

@@ -10,7 +10,7 @@ from sqlalchemy.exc import OperationalError, ProgrammingError
 from supply_infra.db.repositories.video_discovery_repo import VideoDiscoveryRepository
 from supply_infra.db.session import get_session
 
-_SKIP_STATUSES = frozenset({"running", "finished"})
+_SKIP_STATUSES = frozenset({"finished"})
 
 
 class RunNotFoundError(LookupError):

tests/supply_infra/scheduler/test_discover_videos_from_demands.py

@@ -148,10 +148,8 @@ 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 is None
-    assert reuse_skip is not None
-    assert "scheduled-run" in reuse_skip
-    assert "running" in reuse_skip
+    assert reuse_id == "scheduled-run"
+    assert reuse_skip is None
 
     payload = json.loads(
         video_discovery_store.create_video_discovery_run(
@@ -168,6 +166,37 @@ def test_prepare_then_reuse_run_id_scheduled_flow(
     assert payload["run_id"] == run_id
 
 
+def test_prepare_skips_when_run_already_finished(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    factory = _expire_on_commit_session_factory()
+    _patch_service_session(monkeypatch, factory)
+    _seed_run(
+        factory,
+        run_id="finished-run",
+        demand_grade_id=303,
+        status="finished",
+    )
+    ctx = FindDemandContext(
+        biz_dt="20260728",
+        demand_grade_id=303,
+        demand_name="广场舞",
+        grade="S",
+        videos=[
+            FindDemandVideo(
+                video_id="vid-1",
+                title="参考标题",
+                points=[FindDemandPoint(point="动作简单", point_type="key")],
+            )
+        ],
+    )
+
+    run_id, skip_reason = prepare_video_discovery_run(ctx)
+    assert run_id is None
+    assert skip_reason is not None
+    assert "finished" in skip_reason
+
+
 def test_video_discovery_models_exclude_unused_columns() -> None:
     run_columns = set(VideoDiscoveryRun.__table__.columns.keys())
     candidate_columns = set(VideoDiscoveryCandidate.__table__.columns.keys())
@@ -185,6 +214,7 @@ def test_find_agent_enables_deterministic_completion_control() -> None:
 
     assert "audit_video_discovery_run" in agent.tools.list_tools()
     assert agent.completion_guard is find_agent_completion_guard
+    assert agent.completion_guard_blocks is False
     assert (
         "audit_video_discovery_run"
         in agent.tool_repeat_requires_change["query_video_discovery_state"]
@@ -218,16 +248,23 @@ def _primary_state_payload(primary_ids: list[str]) -> dict[str, object]:
     return {
         "run": {"primary_count": len(primary_ids), "status": "finished"},
         "candidates": [
-            {"aweme_id": aweme_id, "decision_bucket": "primary"}
+            {
+                "aweme_id": aweme_id,
+                "decision_bucket": "primary",
+                "detail_verified": True,
+                "content_portrait_attempted": True,
+                "account_portrait_attempted": True,
+                "age_portraits_normalized": True,
+            }
             for aweme_id in primary_ids
         ],
     }
 
 
-def test_find_agent_guard_allows_relaxed_finish_with_five_primary() -> None:
+def test_find_agent_guard_allows_finish_with_five_primary() -> None:
     primary_ids = [f"7123456789012345{i}" for i in range(5)]
     primary_lines = "\n".join(
-        f"{index + 1}. 标题{i} aweme_id={aweme_id}"
+        f"{index + 1}. 标题{index} aweme_id={aweme_id}"
         for index, aweme_id in enumerate(primary_ids)
     )
     messages = [
@@ -280,6 +317,134 @@ def test_find_agent_guard_relaxed_path_skips_audit_and_search_order() -> None:
     assert find_agent_completion_guard(messages) is None
 
 
+def test_find_agent_guard_allows_partial_primary_in_report() -> None:
+    primary_ids = [f"7123456789012345{i}" for i in range(5)]
+    reported_ids = primary_ids[:2]
+    messages = [
+        _tool_message("create_video_discovery_run", {"run_id": "run-1"}),
+        _tool_message(
+            "batch_save_video_candidate_evaluations",
+            {"status": "finished", "primary_count": 5},
+        ),
+        _tool_message(
+            "query_video_discovery_state",
+            _primary_state_payload(primary_ids),
+        ),
+        Message(
+            role=Role.ASSISTANT,
+            content="主推荐\n"
+            + "\n".join(f"- aweme_id={aweme_id}" for aweme_id in reported_ids),
+        ),
+    ]
+
+    assert find_agent_completion_guard(messages) is None
+
+
+def test_find_agent_guard_allows_zero_primary_finish() -> None:
+    messages = [
+        _tool_message("create_video_discovery_run", {"run_id": "run-1"}),
+        _tool_message(
+            "batch_save_video_candidate_evaluations",
+            {"status": "finished", "primary_count": 0},
+        ),
+        _tool_message(
+            "query_video_discovery_state",
+            {"run": {"primary_count": 0, "status": "finished"}, "candidates": []},
+        ),
+        Message(
+            role=Role.ASSISTANT,
+            content="未找到满足条件的候选，本轮结束。",
+        ),
+    ]
+
+    assert find_agent_completion_guard(messages) is None
+
+
+def test_find_agent_completion_guard_reports_stale_state_hint() -> None:
+    primary_ids = [f"7123456789012345{i}" for i in range(5)]
+    messages = [
+        _tool_message("create_video_discovery_run", {"run_id": "run-1"}),
+        _tool_message(
+            "query_video_discovery_state",
+            _primary_state_payload(primary_ids),
+        ),
+        _tool_message(
+            "batch_save_video_candidate_evaluations",
+            {"status": "finished", "primary_count": 5},
+        ),
+        Message(
+            role=Role.ASSISTANT,
+            content="主推荐\n" + "\n".join(f"- aweme_id={pid}" for pid in primary_ids),
+        ),
+    ]
+
+    reason = find_agent_completion_guard(messages)
+    assert reason is not None
+    assert "finished 之后" in reason
+
+
+def test_list_skip_grade_ids_only_finished(monkeypatch: pytest.MonkeyPatch) -> None:
+    factory = _expire_on_commit_session_factory()
+    _patch_service_session(monkeypatch, factory)
+    _seed_run(factory, run_id="running-run", demand_grade_id=301, status="running")
+    _seed_run(
+        factory,
+        run_id="finished-run",
+        demand_grade_id=302,
+        status="finished",
+        row_id=2,
+    )
+
+    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}
+
+
+def test_evaluate_find_agent_run_marks_max_iterations_as_failed() -> None:
+    from agents.find_agent.run_outcome import evaluate_find_agent_run
+    from supply_agent.types import AgentResult
+
+    outcome = evaluate_find_agent_run(
+        "run-1",
+        AgentResult(
+            content="Max iterations reached before completion: 尚未保存候选评估",
+            messages=[],
+            iterations=60,
+            tool_calls_made=0,
+        ),
+    )
+    assert outcome.succeeded is False
+    assert outcome.failure_reason == "max_iterations_reached"
+
+
+def test_evaluate_find_agent_run_requires_finished_status(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from agents.find_agent.run_outcome import evaluate_find_agent_run
+    from supply_agent.types import AgentResult
+
+    monkeypatch.setattr(
+        "agents.find_agent.run_outcome.get_video_discovery_service",
+        lambda: type(
+            "Svc",
+            (),
+            {
+                "lookup_run": staticmethod(
+                    lambda _run_id: {"status": "running", "primary_count": 5}
+                )
+            },
+        )(),
+    )
+
+    outcome = evaluate_find_agent_run(
+        "run-1",
+        AgentResult(content="主推荐\n完成", messages=[], iterations=10, tool_calls_made=0),
+    )
+    assert outcome.succeeded is False
+    assert outcome.failure_reason == "run_status_running"
+
+
 @patch(
     "supply_infra.scheduler.jobs.discover_videos_from_demands.process_single_discover"
 )

tests/supply_infra/scheduler/test_publish_videos_from_discovery.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from supply_infra.scheduler.jobs.publish_videos_from_discovery import (
+    publish_videos_from_discovery,
+)
+from supply_infra.services.video_discovery_service import PublishableCandidate
+
+
+@patch("supply_infra.scheduler.jobs.publish_videos_from_discovery.AigcClient")
+@patch(
+    "supply_infra.scheduler.jobs.publish_videos_from_discovery.get_video_discovery_service"
+)
+@patch(
+    "supply_infra.scheduler.jobs.publish_videos_from_discovery.list_unique_plan_pairs"
+)
+def test_publish_marks_candidates_when_bind_fails(
+    mock_plan_pairs,
+    mock_get_service,
+    mock_client_cls,
+) -> None:
+    mock_plan_pairs.return_value = [
+        type(
+            "Plan",
+            (),
+            {
+                "label": "plan-a",
+                "produce_plan_id": "produce-1",
+                "publish_plan_id": "publish-1",
+            },
+        )()
+    ]
+    mock_get_service.return_value = MagicMock(
+        list_publishable_candidates=MagicMock(
+            return_value=[PublishableCandidate(id=1, aweme_id="71234567890123450")]
+        ),
+        mark_candidates_aigc_plans=MagicMock(return_value=1),
+    )
+    mock_client_cls.return_value = MagicMock(
+        create_video_crawler_plan=MagicMock(
+            return_value={
+                "success": True,
+                "crawler_plan_id": "crawler-1",
+                "crawler_plan_name": "test-plan",
+            }
+        ),
+        bind_crawler_to_produce_plan=MagicMock(
+            return_value={"success": False, "error": "bind failed"}
+        ),
+    )
+
+    result = publish_videos_from_discovery(biz_dt="20260728")
+
+    assert result["failed_batch_count"] == 1
+    mock_get_service.return_value.mark_candidates_aigc_plans.assert_called_once()
+    call_kwargs = (
+        mock_get_service.return_value.mark_candidates_aigc_plans.call_args.kwargs
+    )
+    assert call_kwargs["crawler_plan_id"] == "crawler-1"
+    assert call_kwargs["plan_label"].startswith("BIND_FAILED:")