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

agents/find_agent/PRD.md

@@ -142,6 +142,10 @@ Agent 对每个候选独立判断：
 最终报告只读取数据库最终状态，不通过文字反向修改候选分池。报告中的主推荐和淘汰
 候选必须与数据库状态一致。
 
+当 Agent 结合任务上下文、替代方案和工具反馈判断任务已经无法继续时，应停止无效重试，
+输出以 `任务未完成（工具故障）` 开头的失败摘要。完成守卫只识别该声明，不解析工具
+返回结构或错误文案，也不替 Agent 判断错误是否可恢复。
+
 ## 5. 当前判断规则
 
 ### 5.1 需求相关性是准入条件
@@ -233,6 +237,7 @@ Agent 对每个候选独立判断：
 | 评分与分池 | 部分可用 | 规则完整，但主要依赖模型遵守长提示词 |
 | 状态保存 | 可用 | 可保存运行、搜索页和候选状态 |
 | 完成审计 | 已接入 | 可校验完成顺序、审计新鲜度和报告分池 |
+| 故障终止 | 已接入 | Agent 明确声明工具故障时允许失败结束 |
 | 最终一致性 | 部分可用 | 报告只读并接受校验，尚无单一事务性最终化入口 |
 | 运行恢复 | 部分可用 | 能查询旧状态，但缺少执行代次和自动恢复机制 |
 | 可观测性 | 不完整 | 有日志和基础计数，缺少 Agent 级质量与成本指标 |
@@ -322,3 +327,11 @@ Agent 对每个候选独立判断：
 10. 最终报告与数据库 `primary / rejected` 完全一致；
 11. 不使用未注册的视频理解能力或编造缺失证据；
 12. 推荐不足 5 条时不降低质量标准。
+
+若工具故障导致任务无法继续，则不适用上述成功条件，但必须满足：
+
+1. Agent 已根据上下文判断继续调用无法产生有效进展；
+2. Agent 已停止重复调用失败工具；
+3. 最终摘要明确标记 `任务未完成（工具故障）`；
+4. 摘要包含失败工具、原始错误、已完成内容和未完成内容；
+5. 不输出看似有效的推荐结果。

agents/find_agent/README.md

@@ -78,6 +78,10 @@ TikHub 翻页必须同时沿用 `next_cursor、search_id、backtrace`。
 审计后如果又发生搜索、取证或评估，必须重新审计并重新查询状态。最终报告只读取数据库
 中的 `primary / rejected`，报告之后不再反向修改分池。
 
+当 Agent 根据任务上下文和已尝试方案判断工具故障已导致任务无法继续时，可以停止重复
+调用，输出以 `任务未完成（工具故障）` 开头的失败摘要。完成守卫只识别该失败声明，
+不解析工具返回结构、错误字段或错误文案，也不替 Agent 判断错误是否可恢复。
+
 ## 建议补充的外部数据工具
 
 以下工具依赖抖音爬虫或热点宝增加接口，当前仓库无法自行补出真实数据。建议按优先级

agents/find_agent/VALIDATION.md

@@ -18,11 +18,11 @@
 
 ## 本轮回归结果
 
-- 2026-07-28 当前契约定向测试：`tests/test_find_agent.py` 及完成控制、模型列、超时
-  回收测试为 `19 passed`；
-- 调度测试为 `3 passed, 1 failed`；剩余失败是既有测试夹具使用普通 `object()`
-  代替 `FindDemandContext`，日志读取 `demand_grade_id / demand_name` 时失败，与本次
-  “不使用视频理解、仅 primary/rejected”修改无关，本次未越界修复；
+- 2026-07-28 移除工具错误分类后，find_agent、AgentLoop、工具框架和失败声明
+  定向测试合计 `39 passed`；Ruff 与 `git diff --check` 通过；
+- 调度测试文件全量为 `4 passed, 3 failed`；失败来自既有测试环境问题：两个 SQLite
+  用例未为 MySQL `BIGINT` 主键提供自增兼容，一个测试夹具使用普通 `object()` 代替
+  `FindDemandContext`。三项均与本次故障终止修改无关；
 - 2026-07-23 历史 find_agent 与 LLM 重试定向测试：`19 passed`；
 - Ruff 与 `git diff --check`：通过；
 - 三张 MySQL 表结构、列、唯一键和索引：真实检查通过；
@@ -104,8 +104,9 @@
 3. 持久化候选需要保存 `detail_verified / content_portrait_attempted /
    account_portrait_attempted / age_portraits_normalized` 等审计状态；视频理解字段不
    属于当前审计契约。
-4. find_agent 完成守卫已启用；`can_finish=false`、审计后状态变更或审计前状态查询
-   都会拒绝最终文本。
+4. find_agent 完成守卫已启用；正常结果仍要求完整成功顺序。Agent 明确输出
+   `任务未完成（工具故障）` 时允许失败结束；程序不解析工具返回，也不判断错误是否
+   可恢复。
 
 ## 下一步修复
 

agents/find_agent/agent.py

@@ -30,6 +30,8 @@ _SEARCH_TOOLS = {
     "douyin_search_tikhub",
     "douyin_user_videos",
 }
+_FAILURE_REPORT_PREFIX = "任务未完成（工具故障）"
+_MIN_PRIMARY_TARGET = 5
 _VIDEO_ID_PATTERN = re.compile(r"(?<!\d)\d{15,22}(?!\d)")
 
 
@@ -109,6 +111,91 @@ def _validate_report_buckets(
     return None
 
 
+def _primary_count_from_state(state: dict[str, object]) -> int:
+    run = state.get("run")
+    if isinstance(run, dict):
+        return int(run.get("primary_count") or 0)
+    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]]]:
@@ -126,12 +213,38 @@ def _successful_tool_events(
     return events
 
 
+def _last_final_assistant_message(messages: list[Message]) -> Message | None:
+    return next(
+        (
+            message
+            for message in reversed(messages)
+            if message.role == Role.ASSISTANT and not message.tool_calls
+        ),
+        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):
+        return None
+
     events = _successful_tool_events(messages)
     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
@@ -196,14 +309,7 @@ def find_agent_completion_guard(messages: list[Message]) -> str | None:
     if state_index < audit_index:
         return "最终状态查询必须在审计通过之后执行"
 
-    last_assistant = next(
-        (
-            message
-            for message in reversed(messages)
-            if message.role == Role.ASSISTANT and not message.tool_calls
-        ),
-        None,
-    )
+    last_assistant = _last_final_assistant_message(messages)
     if last_assistant is None or not (last_assistant.content or "").strip():
         return "最终回答为空"
     report_error = _validate_report_buckets(last_assistant.content, state)
@@ -226,31 +332,6 @@ def create_find_agent(
         max_iterations=60,
         temperature=0.2,
         completion_guard=find_agent_completion_guard,
-        tool_call_budgets={
-            "search": (
-                {
-                    "douyin_search",
-                    "douyin_search_tikhub",
-                    "douyin_user_videos",
-                },
-                10,
-            ),
-            "detail": ({"douyin_detail"}, 2),
-            "portrait": (
-                {
-                    "get_content_fans_portrait",
-                    "get_account_fans_portrait",
-                    "batch_fetch_portraits",
-                },
-                3,
-            ),
-            "evaluation_save": (
-                {"batch_save_video_candidate_evaluations"},
-                6,
-            ),
-            "audit": ({"audit_video_discovery_run"}, 4),
-            "state_query": ({"query_video_discovery_state"}, 8),
-        },
         tool_repeat_requires_change={
             "query_video_discovery_state": {
                 "record_video_search_page",

agents/find_agent/demand_run.py

@@ -19,15 +19,12 @@ from supply_infra.db.repositories.demand_video_expansion_repo import (
 from supply_infra.db.repositories.multi_demand_video_detail_repo import (
     MultiDemandVideoDetailRepository,
 )
-from supply_infra.db.repositories.video_discovery_repo import VideoDiscoveryRepository
 from supply_infra.db.session import get_session
+from supply_infra.services.video_discovery_service import get_video_discovery_service
 
 logger = logging.getLogger(__name__)
 
 _POINT_TYPES = {"inspiration", "purpose", "key"}
-_SKIP_STATUSES = frozenset({"running", "finished"})
-
-
 @dataclass
 class FindDemandPoint:
     point: str
@@ -275,30 +272,25 @@ def prepare_video_discovery_run(
     """执行 Agent 前预创建 video_discovery_run，返回 (run_id, skip_reason)。"""
     payload = build_run_input_payload(ctx)
     primary = ctx.primary_video
-
-    with get_session() as session:
-        repo = VideoDiscoveryRepository(session)
-        existing = repo.get_by_biz_dt_and_grade(ctx.biz_dt, ctx.demand_grade_id)
-        if existing is not None and not force and existing.status in _SKIP_STATUSES:
-            return None, (
-                f"biz_dt={ctx.biz_dt} demand_grade_id={ctx.demand_grade_id} "
-                f"已执行过 run_id={existing.run_id} status={existing.status}"
-            )
-
-        run_id = existing.run_id if existing is not None else uuid.uuid4().hex
-        values = {
-            "run_id": run_id,
-            "biz_dt": ctx.biz_dt,
-            "demand_grade_id": ctx.demand_grade_id,
-            "demand_word": ctx.demand_name,
-            "seed_video_id": primary.video_id if primary else None,
-            "seed_video_title": primary.title if primary else None,
-            "relevant_points_json": json.dumps(payload, ensure_ascii=False),
-            "intent_summary": None,
-            "status": "running",
-            "stop_reason": None,
-        }
-        repo.upsert_scheduled_run(values)
+    values = {
+        "run_id": uuid.uuid4().hex,
+        "biz_dt": ctx.biz_dt,
+        "demand_grade_id": ctx.demand_grade_id,
+        "demand_word": ctx.demand_name,
+        "seed_video_id": primary.video_id if primary else None,
+        "seed_video_title": primary.title if primary else None,
+        "relevant_points_json": json.dumps(payload, ensure_ascii=False),
+        "intent_summary": None,
+        "status": "running",
+        "stop_reason": None,
+    }
+    run_id, skip_reason = get_video_discovery_service().prepare_scheduled_run(
+        biz_dt=ctx.biz_dt,
+        demand_grade_id=ctx.demand_grade_id,
+        values=values,
+        force=force,
+    )
+    if skip_reason is None and run_id:
         logger.info(
             "prepared video_discovery_run: biz_dt=%s demand_grade_id=%s run_id=%s demand=%s",
             ctx.biz_dt,
@@ -306,7 +298,7 @@ def prepare_video_discovery_run(
             run_id,
             ctx.demand_name,
         )
-        return run_id, None
+    return run_id, skip_reason
 
 
 def filter_pending_contexts(
@@ -320,8 +312,7 @@ def filter_pending_contexts(
     if not skip_finished or not contexts:
         return contexts, stats
 
-    with get_session() as session:
-        skip_ids = VideoDiscoveryRepository(session).list_skip_grade_ids(biz_dt)
+    skip_ids = get_video_discovery_service().list_skip_grade_ids(biz_dt)
 
     pending = [ctx for ctx in contexts if ctx.demand_grade_id not in skip_ids]
     stats["skipped_already_done"] = len(contexts) - len(pending)
@@ -393,11 +384,10 @@ def discover_videos_for_demand(
         agent_result = run_find_agent(user_input)
         return FindDemandExecutionResult(run_id=run_id, agent_result=agent_result)
     except Exception as exc:
-        with get_session() as session:
-            VideoDiscoveryRepository(session).mark_run_failed(
-                run_id,
-                stop_reason=str(exc),
-            )
+        get_video_discovery_service().mark_run_failed(
+            run_id,
+            stop_reason=str(exc),
+        )
         raise
 
 

agents/find_agent/prompt/system_prompt.md

@@ -179,8 +179,24 @@
 所有工具失败都保留原始错误语义，不得编造缺失字段。
 
 若 `create_video_discovery_run` 明确返回数据库表未初始化或数据库不可用，只尝试一次：
-保留原始错误且不得反复调用或假装完成。当前确定性完成控制要求持久化、数据库审计和
-最终状态查询全部成功；数据库不可用时不能输出已完成报告。
+保留原始错误且不得反复调用或假装完成。数据库不可用时不能输出已完成报告，应立即按
+下方“工具故障终止规则”结束任务。
+
+# 工具故障终止规则
+
+由你结合任务上下文、已尝试的替代方案和工具反馈判断任务是否已经无法继续。程序不解析
+工具错误字段，也不根据错误文案替你判断错误是否可恢复。参数可以修正或仍有替代工具时
+应继续；继续调用已经确认无效的工具不会产生新信息时，应停止重试。
+
+确认无法继续后：
+
+1. 立即停止调用失败工具，不再为了满足正常成功守卫重复调用；
+2. 直接输出失败摘要，第一行必须是 `任务未完成（工具故障）`；
+3. 说明失败工具、原始错误、已完成内容和未完成内容；
+4. 明确写出“未产出有效推荐”，不得输出看似正常的主推荐或淘汰候选报告。
+
+完成守卫只识别上述失败摘要前缀并允许任务结束，不读取或分类工具返回。该出口表示任务
+失败结束，不是成功完成。
 
 # 成本与迭代预算
 
@@ -193,6 +209,7 @@
 - 执行新搜索、翻页、详情或画像后，应重新保存受影响候选；
 - 不得用“接下来我会继续”作为最终回答。运行时完成守卫会拒绝顺序不完整、审计未
   通过或使用旧状态生成的最终报告。
+- 你已按“工具故障终止规则”判断无法继续时，不再尝试正常完成条件，直接输出失败摘要。
 - 数据库保留数量达到 5 条后，若没有明显更高价值的搜索前沿，优先结束任务。
 - 保留数量不足 5 条时，优先继续有效的搜索、翻页或扩标签；合理前沿已经耗尽，或剩余
   候选明显不值得保留时，可以少于 5 条结束。

agents/find_agent/tools/decision_support.py

@@ -314,6 +314,8 @@ def audit_video_discovery_process(
         "rejected": 0,
         "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(
@@ -357,15 +359,24 @@ def audit_video_discovery_process(
             and share >= 0.50
         )
         if bucket == "rejected" and meets_primary:
-            critical.append(f"{aweme_id} 达到 primary 线却被错误淘汰")
+            rejected_misclass_messages.append(
+                f"{aweme_id} 达到 primary 线却被错误淘汰"
+            )
         if bucket == "pending_evaluation":
             message = f"{aweme_id} 等待 Agent 补证和评估"
             if intended_status == "finished":
-                critical.append(message)
+                pending_evaluation_messages.append(message)
             else:
                 warnings.append(message)
 
     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(
             f"当前保留 {retained_count} 条，低于优先目标 5 条；"

agents/find_agent/tools/video_discovery_store.py

@@ -8,16 +8,15 @@ import uuid
 from decimal import Decimal
 from typing import Any
 
-from sqlalchemy.exc import OperationalError, ProgrammingError
-
 from agents.find_agent.tools.decision_support import (
     audit_video_discovery_process,
 )
 from supply_agent.tools import tool
-from supply_infra.db.repositories.video_discovery_repo import (
-    VideoDiscoveryRepository,
+from supply_infra.services.video_discovery_service import (
+    RunNotFoundError,
+    format_db_error,
+    get_video_discovery_service,
 )
-from supply_infra.db.session import get_session
 
 logger = logging.getLogger(__name__)
 
@@ -52,15 +51,6 @@ def _load_json(value: str | None, default: Any) -> Any:
         return default
 
 
-def _db_error_message(error: Exception) -> str:
-    if isinstance(error, (OperationalError, ProgrammingError)):
-        return (
-            f"数据库表尚未初始化或不可用: {error}。"
-            "请先运行 `.venv/bin/python -m supply_infra.db`。"
-        )
-    return str(error)
-
-
 def _clean_text(value: Any, *, max_length: int | None = None) -> str | None:
     if value is None:
         return None
@@ -146,71 +136,6 @@ def _candidate_from_search_result(item: dict[str, Any], keyword: str) -> dict[st
     }
 
 
-def _serialize_candidate(item: Any) -> dict[str, Any]:
-    decision_bucket = (
-        "pending_evaluation"
-        if item.decision_bucket == "unreviewed"
-        else item.decision_bucket
-    )
-    return {
-        "aweme_id": item.aweme_id,
-        "title": item.title,
-        "content_link": item.content_link,
-        "author_name": item.author_name,
-        "author_sec_uid": item.author_sec_uid,
-        "source_keywords": _load_json(item.source_keywords_json, []),
-        "source_search_ids": _load_json(item.source_search_ids_json, []),
-        "tags": _load_json(item.tags_json, []),
-        "hit_points": _load_json(item.hit_points_json, []),
-        "play_count": item.play_count,
-        "like_count": item.like_count,
-        "comment_count": item.comment_count,
-        "collect_count": item.collect_count,
-        "share_count": item.share_count,
-        "relevance_score": (
-            float(item.relevance_score) if item.relevance_score is not None else None
-        ),
-        "elder_score": float(item.elder_score) if item.elder_score is not None else None,
-        "share_score": float(item.share_score) if item.share_score is not None else None,
-        "value_score": float(item.value_score) if item.value_score is not None else None,
-        "confidence": item.confidence,
-        "decision_bucket": decision_bucket,
-        "content_age_evidence": _load_json(item.content_age_evidence_json, {}),
-        "account_age_evidence": _load_json(item.account_age_evidence_json, {}),
-        "age_normalization": _load_json(item.age_normalization_json, {}),
-        "detail_verified": bool(item.detail_verified),
-        "content_portrait_attempted": bool(item.content_portrait_attempted),
-        "account_portrait_attempted": bool(item.account_portrait_attempted),
-        "age_portraits_normalized": bool(item.age_portraits_normalized),
-        "expansion_worthy_tags": _load_json(
-            item.expansion_worthy_tags_json, []
-        ),
-        "relevance_reason": item.relevance_reason,
-        "elder_reason": item.elder_reason,
-        "share_reason": item.share_reason,
-        "decision_reason": item.decision_reason,
-    }
-
-
-def _serialize_search(item: Any) -> dict[str, Any]:
-    return {
-        "search_id": int(item.id),
-        "keyword": item.keyword,
-        "query_reason": item.query_reason,
-        "source_type": item.source_type,
-        "source_value": item.source_value,
-        "parent_search_id": item.parent_search_id,
-        "provider": item.provider,
-        "provider_state": _load_json(item.provider_state_json, {}),
-        "cursor": item.cursor,
-        "page_no": item.page_no,
-        "results_count": item.results_count,
-        "new_candidate_count": item.new_candidate_count,
-        "has_more": bool(item.has_more),
-        "next_cursor": item.next_cursor,
-    }
-
-
 @tool
 def create_video_discovery_run(
     demand_word: str,
@@ -239,24 +164,24 @@ def create_video_discovery_run(
     Returns:
         JSON，包含后续存储工具必须使用的 run_id。
     """
+    service = get_video_discovery_service()
     cleaned_run_id = _clean_text(run_id, max_length=64)
     if cleaned_run_id:
         try:
-            with get_session() as session:
-                existing = VideoDiscoveryRepository(session).get_run(cleaned_run_id)
+            existing = service.lookup_run(cleaned_run_id)
             if existing is not None:
                 return _json(
                     {
                         "title": "视频发现运行已存在",
                         "run_id": cleaned_run_id,
-                        "status": existing.status,
+                        "status": existing["status"],
                         "pre_created": True,
                         "output": f"run_id={cleaned_run_id}",
                     }
                 )
         except Exception as exc:
             logger.error("create_video_discovery_run lookup failed: %s", exc, exc_info=True)
-            return _json({"error": _db_error_message(exc), "title": "查询视频发现运行失败"})
+            return _json({"error": format_db_error(exc), "title": "查询视频发现运行失败"})
 
     demand = _clean_text(demand_word, max_length=256)
     if not demand:
@@ -274,19 +199,18 @@ def create_video_discovery_run(
         "status": "running",
     }
     try:
-        with get_session() as session:
-            VideoDiscoveryRepository(session).create_run(values)
+        created = service.create_run(values)
         return _json(
             {
                 "title": "视频发现运行已创建",
-                "run_id": new_run_id,
-                "status": "running",
-                "output": f"run_id={new_run_id}",
+                "run_id": created["run_id"],
+                "status": created["status"],
+                "output": f"run_id={created['run_id']}",
             }
         )
     except Exception as exc:
         logger.error("create_video_discovery_run failed: %s", exc, exc_info=True)
-        return _json({"error": _db_error_message(exc), "title": "创建视频发现运行失败"})
+        return _json({"error": format_db_error(exc), "title": "创建视频发现运行失败"})
 
 
 @tool
@@ -378,32 +302,25 @@ def record_video_search_page(
     search_values["search_key"] = _search_key(search_values)
 
     try:
-        with get_session() as session:
-            repo = VideoDiscoveryRepository(session)
-            if repo.get_run(run_text) is None:
-                return _input_error(f"run_id 不存在: {run_text}")
-            search, new_count = repo.save_search_page(search_values, candidate_rows)
-            payload = {
-                "title": "搜索页已保存",
-                "search_id": int(search.id),
-                "run_id": run_text,
-                "keyword": keyword_text,
-                "source_type": search.source_type,
-                "parent_search_id": search.parent_search_id,
-                "page_no": search.page_no,
-                "results_count": search.results_count,
-                "new_candidate_count": new_count,
-                "has_more": bool(search.has_more),
-                "next_cursor": search.next_cursor,
-                "output": (
-                    f"search_id={search.id}，本页 {search.results_count} 条，"
-                    f"新增候选 {new_count} 条"
-                ),
-            }
+        saved = get_video_discovery_service().save_search_page(
+            run_text,
+            search_values,
+            candidate_rows,
+        )
+        payload = {
+            "title": "搜索页已保存",
+            **saved,
+            "output": (
+                f"search_id={saved['search_id']}，本页 {saved['results_count']} 条，"
+                f"新增候选 {saved['new_candidate_count']} 条"
+            ),
+        }
         return _json(payload)
+    except RunNotFoundError as exc:
+        return _input_error(str(exc))
     except Exception as exc:
         logger.error("record_video_search_page failed: %s", exc, exc_info=True)
-        return _json({"error": _db_error_message(exc), "title": "保存搜索页失败"})
+        return _json({"error": format_db_error(exc), "title": "保存搜索页失败"})
 
 
 def _normalize_evaluation(item: dict[str, Any]) -> dict[str, Any]:
@@ -513,44 +430,37 @@ def batch_save_video_candidate_evaluations(
             errors.append(f"[{index}] {exc}")
 
     try:
-        with get_session() as session:
-            repo = VideoDiscoveryRepository(session)
-            if repo.get_run(run_text) is None:
-                return _input_error(f"run_id 不存在: {run_text}")
-            if rows:
-                saved, audit_relevant_changed = repo.save_candidate_evaluations(
-                    run_text, rows
-                )
-            else:
-                saved, audit_relevant_changed = 0, False
-            run = repo.finish_run(
-                run_text,
-                status=run_status,
-                intent_summary=_clean_text(intent_summary),
-                stop_reason=_clean_text(stop_reason),
-            )
-            payload = {
-                "title": "候选评估已保存",
-                "run_id": run_text,
-                "saved_count": saved,
-                "error_count": len(errors),
-                "errors": errors,
-                "input_error": bool(errors and not rows),
-                "audit_relevant_changed": audit_relevant_changed,
-                "status": run.status,
-                "search_count": run.search_count,
-                "primary_count": run.primary_count,
-                "output": (
-                    f"保存 {saved} 条；主推荐 {run.primary_count} 条；"
-                    f"状态 {run.status}"
-                ),
-            }
+        saved = get_video_discovery_service().save_evaluations_and_finish(
+            run_text,
+            rows,
+            status=run_status,
+            intent_summary=_clean_text(intent_summary),
+            stop_reason=_clean_text(stop_reason),
+        )
+        payload = {
+            "title": "候选评估已保存",
+            "run_id": run_text,
+            "saved_count": saved["saved_count"],
+            "error_count": len(errors),
+            "errors": errors,
+            "input_error": bool(errors and not rows),
+            "audit_relevant_changed": saved["audit_relevant_changed"],
+            "status": saved["status"],
+            "search_count": saved["search_count"],
+            "primary_count": saved["primary_count"],
+            "output": (
+                f"保存 {saved['saved_count']} 条；主推荐 {saved['primary_count']} 条；"
+                f"状态 {saved['status']}"
+            ),
+        }
         return _json(payload)
+    except RunNotFoundError as exc:
+        return _input_error(str(exc))
     except Exception as exc:
         logger.error(
             "batch_save_video_candidate_evaluations failed: %s", exc, exc_info=True
         )
-        return _json({"error": _db_error_message(exc), "title": "保存候选评估失败"})
+        return _json({"error": format_db_error(exc), "title": "保存候选评估失败"})
 
 
 @tool
@@ -568,41 +478,27 @@ def query_video_discovery_state(
     if not run_text:
         return _json({"error": "run_id 不能为空"})
     try:
-        with get_session() as session:
-            repo = VideoDiscoveryRepository(session)
-            run = repo.get_run(run_text)
-            if run is None:
-                return _json({"error": f"run_id 不存在: {run_text}"})
-            searches = repo.list_searches(run_text)
-            buckets = (
-                ("primary", "rejected", "pending_evaluation", "unreviewed")
-                if include_rejected
-                else ("primary", "pending_evaluation", "unreviewed")
-            )
-            candidates = repo.list_candidates(
-                run_text, buckets=buckets, limit=max(1, min(int(limit), 500))
-            )
-            payload = {
-                "title": f"视频发现状态: {run_text}",
-                "run": {
-                    "run_id": run.run_id,
-                    "demand_word": run.demand_word,
-                    "intent_summary": run.intent_summary,
-                    "status": run.status,
-                    "search_count": run.search_count,
-                    "primary_count": run.primary_count,
-                    "stop_reason": run.stop_reason,
-                },
-                "searches": [_serialize_search(item) for item in searches],
-                "candidates": [_serialize_candidate(item) for item in candidates],
-                "output": (
-                    f"搜索页 {run.search_count}；主推荐 {run.primary_count}"
-                ),
-            }
+        state = get_video_discovery_service().get_full_state(
+            run_text,
+            include_rejected=include_rejected,
+            limit=limit,
+        )
+        run = state["run"]
+        payload = {
+            "title": f"视频发现状态: {run_text}",
+            "run": run,
+            "searches": state["searches"],
+            "candidates": state["candidates"],
+            "output": (
+                f"搜索页 {run['search_count']}；主推荐 {run['primary_count']}"
+            ),
+        }
         return _json(payload)
+    except RunNotFoundError:
+        return _json({"error": f"run_id 不存在: {run_text}"})
     except Exception as exc:
         logger.error("query_video_discovery_state failed: %s", exc, exc_info=True)
-        return _json({"error": _db_error_message(exc), "title": "查询视频发现状态失败"})
+        return _json({"error": format_db_error(exc), "title": "查询视频发现状态失败"})
 
 
 @tool
@@ -633,29 +529,12 @@ def audit_video_discovery_run(
         )
 
     try:
-        with get_session() as session:
-            repo = VideoDiscoveryRepository(session)
-            run = repo.get_run(run_text)
-            if run is None:
-                return _input_error(f"run_id 不存在: {run_text}")
-            searches = [
-                _serialize_search(item)
-                for item in repo.list_searches(run_text)
-            ]
-            candidates = [
-                _serialize_candidate(item)
-                for item in repo.list_candidates(run_text, limit=500)
-            ]
-            persisted_run = {
-                "status": run.status,
-                "search_count": run.search_count,
-                "primary_count": run.primary_count,
-            }
-
+        snapshot = get_video_discovery_service().get_audit_snapshot(run_text)
+        persisted_run = snapshot["persisted_run"]
         result = _load_json(
             audit_video_discovery_process(
-                searches=searches,
-                candidates=candidates,
+                searches=snapshot["searches"],
+                candidates=snapshot["candidates"],
                 intended_status=intended_status,
             ),
             {},
@@ -679,6 +558,8 @@ def audit_video_discovery_run(
             result["critical_violations"] = list(dict.fromkeys(violations))
             result["can_finish"] = False
         return _json(result)
+    except RunNotFoundError as exc:
+        return _input_error(str(exc))
     except Exception as exc:
         logger.error(
             "audit_video_discovery_run failed: %s",
@@ -686,5 +567,5 @@ def audit_video_discovery_run(
             exc_info=True,
         )
         return _json(
-            {"error": _db_error_message(exc), "title": "数据库运行审计失败"}
+            {"error": format_db_error(exc), "title": "数据库运行审计失败"}
         )

supply_infra/scheduler/jobs/discover_videos_from_demands.py

@@ -22,8 +22,8 @@ from agents.find_agent.demand_run import (
 )
 from supply_infra.config import get_infra_settings
 from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
-from supply_infra.db.repositories.video_discovery_repo import VideoDiscoveryRepository
 from supply_infra.db.session import ensure_mysql_pool_capacity, get_session
+from supply_infra.services.video_discovery_service import get_video_discovery_service
 
 logger = logging.getLogger(__name__)
 
@@ -107,8 +107,7 @@ def process_single_discover(
 
 
 def _count_passed_videos(biz_dt: str) -> int:
-    with get_session() as session:
-        return VideoDiscoveryRepository(session).count_passed_videos(biz_dt)
+    return get_video_discovery_service().count_passed_videos(biz_dt)
 
 
 def discover_videos_from_demands(

supply_infra/scheduler/jobs/publish_videos_from_discovery.py

@@ -16,8 +16,11 @@ from supply_infra.aigc.plan_map import (
 )
 from supply_infra.config import get_infra_settings
 from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
-from supply_infra.db.repositories.video_discovery_repo import VideoDiscoveryRepository
 from supply_infra.db.session import get_session
+from supply_infra.services.video_discovery_service import (
+    PublishableCandidate,
+    get_video_discovery_service,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -25,14 +28,6 @@ logger = logging.getLogger(__name__)
 _PUBLISHABLE_BUCKETS = ("primary",)
 
 
-@dataclass(frozen=True)
-class PublishCandidate:
-    """脱离 ORM Session 后仍可安全使用的发布候选快照。"""
-
-    id: int
-    aweme_id: str
-
-
 @dataclass
 class PublishBatchResult:
     plan_label: str
@@ -75,11 +70,11 @@ def _resolve_biz_dt(biz_dt: str | None) -> str | None:
 
 
 def _group_candidates_by_plan(
-    candidates: list[PublishCandidate],
+    candidates: list[PublishableCandidate],
     plan_pairs: list[AigcPlanPair],
-) -> list[tuple[AigcPlanPair, list[PublishCandidate]]]:
+) -> list[tuple[AigcPlanPair, list[PublishableCandidate]]]:
     buckets = distribute_evenly(candidates, len(plan_pairs))
-    grouped: list[tuple[AigcPlanPair, list[PublishCandidate]]] = []
+    grouped: list[tuple[AigcPlanPair, list[PublishableCandidate]]] = []
     for plan_pair, bucket in zip(plan_pairs, buckets, strict=False):
         if bucket:
             grouped.append((plan_pair, bucket))
@@ -104,18 +99,12 @@ def publish_videos_from_discovery(
     if not plan_pairs:
         raise RuntimeError("AIGC 计划映射为空，无法分发")
 
-    with get_session() as session:
-        repo = VideoDiscoveryRepository(session)
-        orm_candidates = repo.list_publishable_candidates(
-            run_id=run_id,
-            biz_dt=resolved_biz_dt,
-            skip_published=skip_published,
-            limit=limit,
-        )
-        candidates = [
-            PublishCandidate(id=int(item.id), aweme_id=str(item.aweme_id))
-            for item in orm_candidates
-        ]
+    candidates = get_video_discovery_service().list_publishable_candidates(
+        run_id=run_id,
+        biz_dt=resolved_biz_dt,
+        skip_published=skip_published,
+        limit=limit,
+    )
 
     if not candidates:
         return {
@@ -181,14 +170,13 @@ def publish_videos_from_discovery(
                 batch.bind_error = str(bind_result.get("error") or "绑定生成计划失败")
 
             if crawler_plan_id and batch.bind_success:
-                with get_session() as session:
-                    updated = VideoDiscoveryRepository(session).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,
-                    )
+                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,
+                )
                 logger.info(
                     "published batch: plan=%s crawler=%s videos=%d db_updated=%d",
                     plan_pair.label,

tests/supply_infra/scheduler/test_discover_videos_from_demands.py

@@ -1,13 +1,25 @@
 from __future__ import annotations
 
 import asyncio
+import json
+from contextlib import contextmanager
 from unittest.mock import patch
 
 import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
 
 from agents.find_agent import create_find_agent
 from agents.find_agent.agent import find_agent_completion_guard
 from agents.find_agent.async_runner import arun_find_agent
+from agents.find_agent.demand_run import (
+    FindDemandContext,
+    FindDemandPoint,
+    FindDemandVideo,
+    prepare_video_discovery_run,
+)
+from agents.find_agent.tools import video_discovery_store
+from supply_agent.types import Message, Role
 from supply_infra.db.models.video_discovery import (
     VideoDiscoveryCandidate,
     VideoDiscoveryRun,
@@ -17,6 +29,145 @@ from supply_infra.scheduler.jobs.discover_videos_from_demands import (
 )
 
 
+def _expire_on_commit_session_factory() -> sessionmaker[Session]:
+    """Match production get_session: commit expires ORM instances."""
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    # SQLite 对 BigInteger PK 不会自增，测试里显式写入 id。
+    VideoDiscoveryRun.__table__.create(engine)
+    return sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+
+def _patch_service_session(
+    monkeypatch: pytest.MonkeyPatch,
+    factory: sessionmaker[Session],
+) -> None:
+    @contextmanager
+    def get_test_session():
+        session = factory()
+        try:
+            yield session
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+    monkeypatch.setattr(
+        "supply_infra.services.video_discovery_service.get_session",
+        get_test_session,
+    )
+    import supply_infra.services.video_discovery_service as service_module
+
+    service_module._default_service = None
+
+
+def _seed_run(
+    factory: sessionmaker[Session],
+    *,
+    run_id: str,
+    demand_grade_id: int,
+    status: str = "running",
+    row_id: int = 1,
+) -> None:
+    with factory() as session:
+        session.add(
+            VideoDiscoveryRun(
+                id=row_id,
+                run_id=run_id,
+                biz_dt="20260728",
+                demand_grade_id=demand_grade_id,
+                demand_word="广场舞",
+                seed_video_id="vid-1",
+                seed_video_title="参考标题",
+                relevant_points_json="[]",
+                status=status,
+            )
+        )
+        session.commit()
+
+
+def test_create_video_discovery_run_reuses_precreated_run(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """定时任务预创建 run 后，Agent 复用 run_id 不得触发 DetachedInstanceError。"""
+    factory = _expire_on_commit_session_factory()
+    _patch_service_session(monkeypatch, factory)
+    _seed_run(factory, run_id="precreated-run", demand_grade_id=101)
+
+    payload = json.loads(
+        video_discovery_store.create_video_discovery_run(
+            demand_word="广场舞",
+            seed_video_title="参考标题",
+            relevant_points=[],
+            run_id="precreated-run",
+            demand_grade_id=101,
+        )
+    )
+
+    assert "error" not in payload
+    assert payload["run_id"] == "precreated-run"
+    assert payload["status"] == "running"
+    assert payload["pre_created"] is True
+
+
+def test_prepare_then_reuse_run_id_scheduled_flow(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """完整调度衔接：prepare 得到 run_id → create 复用，全程无 DetachedInstanceError。"""
+    factory = _expire_on_commit_session_factory()
+    _patch_service_session(monkeypatch, factory)
+    # 先写入一条 failed 记录，prepare 会原地重置为 running 并复用 run_id。
+    _seed_run(
+        factory,
+        run_id="scheduled-run",
+        demand_grade_id=202,
+        status="failed",
+    )
+    ctx = FindDemandContext(
+        biz_dt="20260728",
+        demand_grade_id=202,
+        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 skip_reason is None
+    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
+
+    payload = json.loads(
+        video_discovery_store.create_video_discovery_run(
+            demand_word=ctx.demand_name,
+            seed_video_title="参考标题",
+            relevant_points=[{"point": "动作简单", "point_type": "key"}],
+            run_id=run_id,
+            demand_grade_id=ctx.demand_grade_id,
+            seed_video_id="vid-1",
+        )
+    )
+    assert "error" not in payload
+    assert payload["pre_created"] is True
+    assert payload["run_id"] == run_id
+
+
 def test_video_discovery_models_exclude_unused_columns() -> None:
     run_columns = set(VideoDiscoveryRun.__table__.columns.keys())
     candidate_columns = set(VideoDiscoveryCandidate.__table__.columns.keys())
@@ -40,6 +191,95 @@ def test_find_agent_enables_deterministic_completion_control() -> None:
     )
 
 
+def test_find_agent_guard_allows_agent_declared_tool_failure() -> None:
+    messages = [
+        Message(
+            role=Role.ASSISTANT,
+            content=(
+                "任务未完成（工具故障）\n"
+                "失败工具：douyin_search\n"
+                "未产出有效推荐。"
+            ),
+        )
+    ]
+
+    assert find_agent_completion_guard(messages) is None
+
+
+def _tool_message(name: str, payload: dict[str, object]) -> Message:
+    return Message(
+        role=Role.TOOL,
+        name=name,
+        content=json.dumps(payload, ensure_ascii=False),
+    )
+
+
+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"}
+            for aweme_id in primary_ids
+        ],
+    }
+
+
+def test_find_agent_guard_allows_relaxed_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}"
+        for index, aweme_id in enumerate(primary_ids)
+    )
+    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=f"需求理解：广场舞。\n\n主推荐\n{primary_lines}",
+        ),
+    ]
+
+    assert find_agent_completion_guard(messages) is None
+
+
+def test_find_agent_guard_relaxed_path_skips_audit_and_search_order() -> None:
+    primary_ids = [f"7123456789012345{i}" for i in range(5)]
+    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(
+            "audit_video_discovery_run",
+            {
+                "can_finish": False,
+                "critical_violations": ["独立根搜索词少于2个"],
+            },
+        ),
+        _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 primary_ids)
+            ),
+        ),
+    ]
+
+    assert find_agent_completion_guard(messages) is None
+
+
 @patch(
     "supply_infra.scheduler.jobs.discover_videos_from_demands.process_single_discover"
 )
@@ -56,7 +296,20 @@ def test_stops_discovery_after_200_passed_videos(
     mock_count_passed,
     mock_process,
 ) -> None:
-    contexts = [object(), object()]
+    contexts = [
+        FindDemandContext(
+            biz_dt="20260727",
+            demand_grade_id=1,
+            demand_name="需求A",
+            grade="S",
+        ),
+        FindDemandContext(
+            biz_dt="20260727",
+            demand_grade_id=2,
+            demand_name="需求B",
+            grade="A",
+        ),
+    ]
     mock_list_contexts.return_value = ("20260727", contexts)
     mock_filter_contexts.return_value = (
         contexts,