寻找Agent 增加并发

agents/find_agent/LOCAL_TEST.md

@@ -99,9 +99,10 @@ python -m agents.find_agent.run --demand-id 12345 --force
 
 | 项 | 本地 `--demand-id` | 定时任务 `discover_videos_from_demands` |
 |---|---|---|
-| 选任务方式 | 指定单个需求 id | 按业务日批量拉 S/A |
+| 选任务方式 | 指定单个需求 id | 按业务日拉齐 S/A，先入待执行队列 |
 | 等级限制 | 不限制等级 | 默认仅 S/A |
-| 并发 | 单条串行 | 可多 worker |
+| 并发 | 单条串行 | 默认 5 worker 循环取任务；超时退出后自动补齐 |
+| 落库时机 | 开始执行时创建 `video_discovery_run` | 同左（入队时不预建记录） |
 | 用途 | 联调 / 复现 / 单条验收 | 日常生产调度 |
 
 ## 代码内调用

agents/find_agent/support/douyin_detail.py

@@ -14,6 +14,7 @@ from typing import Any, Optional
 
 import httpx
 
+from agents.find_agent.support.rate_limit import make_async_interval_limiter
 from supply_infra.video_discovery_gates import parse_datetime_value
 from supply_infra.services.video_discovery_service import (
     RunNotFoundError,
@@ -24,8 +25,7 @@ from supply_infra.services.video_discovery_service import (
 logger = logging.getLogger(__name__)
 
 _MIN_REQUEST_INTERVAL_SECONDS = 10.1
-_rate_limit_lock = asyncio.Lock()
-_last_request_monotonic: float = 0.0
+_wait_rate_limit = make_async_interval_limiter(_MIN_REQUEST_INTERVAL_SECONDS)
 
 DOUYIN_DETAIL_API = "http://8.217.190.241:8888/crawler/dou_yin/detail"
 DEFAULT_TIMEOUT = 60.0
@@ -229,16 +229,6 @@ def _error_result(
     )
 
 
-async def _wait_rate_limit() -> None:
-    global _last_request_monotonic
-    async with _rate_limit_lock:
-        now_mono = time.monotonic()
-        wait_seconds = _MIN_REQUEST_INTERVAL_SECONDS - (now_mono - _last_request_monotonic)
-        if wait_seconds > 0:
-            await asyncio.sleep(wait_seconds)
-        _last_request_monotonic = time.monotonic()
-
-
 async def _fetch_one_detail(
     client: httpx.AsyncClient,
     content_id: str,

agents/find_agent/support/douyin_search.py

@@ -13,13 +13,13 @@ from typing import Any, Optional
 
 import httpx
 
+from agents.find_agent.support.rate_limit import make_async_interval_limiter
 from agents.find_agent.support.search_persistence import persist_search_payload
 
 logger = logging.getLogger(__name__)
 
 _MIN_REQUEST_INTERVAL_SECONDS = 10.1
-_rate_limit_lock = asyncio.Lock()
-_last_request_monotonic: float = 0.0
+_wait_rate_limit = make_async_interval_limiter(_MIN_REQUEST_INTERVAL_SECONDS)
 
 # API 基础配置
 DOUYIN_SEARCH_API = "http://crawapi.piaoquantv.com/crawler/dou_yin/keyword"
@@ -191,13 +191,7 @@ async def _douyin_search_raw(
     request_timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
 
     try:
-        global _last_request_monotonic
-        async with _rate_limit_lock:
-            now_mono = time.monotonic()
-            wait_seconds = _MIN_REQUEST_INTERVAL_SECONDS - (now_mono - _last_request_monotonic)
-            if wait_seconds > 0:
-                await asyncio.sleep(wait_seconds)
-            _last_request_monotonic = time.monotonic()
+        await _wait_rate_limit()
 
         payload = {
             "keyword": keyword,

agents/find_agent/support/douyin_search_tikhub.py

@@ -11,6 +11,7 @@ from typing import Any
 import httpx
 from dotenv import load_dotenv
 
+from agents.find_agent.support.rate_limit import make_async_interval_limiter
 from agents.find_agent.support.search_persistence import persist_search_payload
 from supply_agent.paths import find_project_root
 
@@ -22,8 +23,7 @@ DOUYIN_SEARCH_TIKHUB_API = (
 DEFAULT_TIMEOUT = 60.0
 DEFAULT_MIN_DURATION_SECONDS = 30
 _MIN_REQUEST_INTERVAL_SECONDS = 1.0
-_rate_limit_lock = asyncio.Lock()
-_last_request_monotonic = 0.0
+_wait_rate_limit = make_async_interval_limiter(_MIN_REQUEST_INTERVAL_SECONDS)
 _env_loaded = False
 
 _CONTENT_TYPE_MAP = {
@@ -224,15 +224,6 @@ def _error_result(error: str, *, raw_response: Any = None) -> str:
     return json.dumps(payload, ensure_ascii=False)
 
 
-async def _wait_rate_limit() -> None:
-    global _last_request_monotonic
-    async with _rate_limit_lock:
-        elapsed = time.monotonic() - _last_request_monotonic
-        if elapsed < _MIN_REQUEST_INTERVAL_SECONDS:
-            await asyncio.sleep(_MIN_REQUEST_INTERVAL_SECONDS - elapsed)
-        _last_request_monotonic = time.monotonic()
-
-
 async def _douyin_search_tikhub_raw(
     keyword: str,
     content_type: str = "视频",

agents/find_agent/support/douyin_user_videos.py

@@ -9,6 +9,7 @@ from typing import Any
 
 import httpx
 
+from agents.find_agent.support.rate_limit import make_async_interval_limiter
 from agents.find_agent.support.search_persistence import persist_search_payload
 
 logger = logging.getLogger(__name__)
@@ -17,8 +18,7 @@ DOUYIN_USER_VIDEOS_API = "http://crawapi.piaoquantv.com/crawler/dou_yin/blogger"
 DEFAULT_TIMEOUT = 60.0
 DEFAULT_MIN_DURATION_SECONDS = 30
 _MIN_REQUEST_INTERVAL_SECONDS = 10.1
-_rate_limit_lock = asyncio.Lock()
-_last_request_monotonic = 0.0
+_wait_rate_limit = make_async_interval_limiter(_MIN_REQUEST_INTERVAL_SECONDS)
 _SORT_TYPES = {"最新", "最热"}
 
 
@@ -135,15 +135,6 @@ def _error_result(error: str, *, raw_response: Any = None) -> str:
     return json.dumps(payload, ensure_ascii=False)
 
 
-async def _wait_rate_limit() -> None:
-    global _last_request_monotonic
-    async with _rate_limit_lock:
-        elapsed = time.monotonic() - _last_request_monotonic
-        if elapsed < _MIN_REQUEST_INTERVAL_SECONDS:
-            await asyncio.sleep(_MIN_REQUEST_INTERVAL_SECONDS - elapsed)
-        _last_request_monotonic = time.monotonic()
-
-
 async def _douyin_user_videos_raw(
     account_id: str,
     sort_type: str = "最热",

agents/find_agent/support/rate_limit.py

@@ -0,0 +1,30 @@
+"""跨线程安全的异步请求间隔限速。
+
+find_agent 以多线程跑时，每个 worker 自建事件循环；不能用 asyncio.Lock 做进程级限速。
+"""
+from __future__ import annotations
+
+import asyncio
+import threading
+import time
+from collections.abc import Awaitable, Callable
+
+
+def make_async_interval_limiter(
+    min_interval_seconds: float,
+) -> Callable[[], Awaitable[None]]:
+    """返回 awaitable 限速函数：保证两次调用间隔不少于 min_interval_seconds。"""
+    lock = threading.Lock()
+    state = {"next_allowed": 0.0}
+
+    async def _wait() -> None:
+        with lock:
+            now = time.monotonic()
+            wait_seconds = state["next_allowed"] - now
+            if wait_seconds < 0:
+                wait_seconds = 0.0
+            state["next_allowed"] = now + wait_seconds + float(min_interval_seconds)
+        if wait_seconds > 0:
+            await asyncio.sleep(wait_seconds)
+
+    return _wait

supply_infra/scheduler/constants.py

@@ -7,5 +7,5 @@ SUPPLY_PIPELINE_JOB_NAME = "供给数据流水线"
 AIGC_PUBLISH_JOB_ID = "publish_videos_from_discovery"
 AIGC_PUBLISH_JOB_NAME = "AIGC候选分发"
 
-# find_agent：当日全部 S/A 需求（有拓展点位），单线程串行找视频
-PIPELINE_FIND_AGENT_WORKERS = 1
+# find_agent：当日全部 S/A 需求（有拓展点位），任务队列 + 固定 worker 并发找视频
+PIPELINE_FIND_AGENT_WORKERS = 5

supply_infra/scheduler/jobs/discover_videos_from_demands.py

@@ -2,12 +2,25 @@
 从全部 S/A 级需求及其拓展点位触发 find_agent 视频发现；有效视频满 300 提前结束。
 
 任务层负责查库与组装上下文；Agent 负责搜索、画像与分池落库。
+
+调度模型：
+1. 当天全部待执行需求先入内存任务队列（不预建 DB 记录）；
+2. 维护最多 N 个 worker 线程循环取任务；
+3. 执行前检查当日通过视频数，达标后停止领取新任务（允许略超）；
+4. 单任务超时（默认 600s）后该 worker 退出，监督线程自动补齐到 N。
 """
 from __future__ import annotations
 
 import logging
+import queue
+import threading
 import uuid
-from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
+from concurrent.futures import (
+    FIRST_COMPLETED,
+    ThreadPoolExecutor,
+    TimeoutError as FuturesTimeoutError,
+    wait,
+)
 from datetime import datetime
 from time import monotonic
 from typing import Any
@@ -20,6 +33,7 @@ from agents.find_agent.demand_run import (
     list_find_demand_contexts,
     serialize_find_demand_context,
 )
+from agents.find_agent.runtime import find_agent_timeout_seconds
 from supply_infra.config import get_infra_settings
 from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
 from supply_infra.db.session import ensure_mysql_pool_capacity, get_session
@@ -27,8 +41,10 @@ from supply_infra.services.video_discovery_service import get_video_discovery_se
 
 logger = logging.getLogger(__name__)
 
-_DEFAULT_WORKERS = 1
+_DEFAULT_WORKERS = 5
 _DAILY_PASSED_VIDEO_LIMIT = 300
+_SUPERVISOR_POLL_SECONDS = 0.5
+_QUEUE_GET_TIMEOUT_SECONDS = 1.0
 
 
 def _resolve_biz_dt(biz_dt: str | None) -> str:
@@ -52,7 +68,10 @@ def process_single_discover(
     *,
     force: bool = False,
 ) -> dict[str, Any]:
-    """并发 worker：对单条需求记录执行 find_agent。"""
+    """并发 worker：对单条需求记录执行 find_agent。
+
+    video_discovery_run 记录在 discover_videos_for_demand 开始执行时创建。
+    """
     summary = serialize_find_demand_context(ctx)
     logger.info(
         "discover videos started: grade_id=%s demand=%s videos=%d points=%d",
@@ -127,6 +146,327 @@ def _count_passed_videos(biz_dt: str) -> int:
     return get_video_discovery_service().count_passed_videos(biz_dt)
 
 
+def _apply_item_result(result: dict[str, Any], item_result: dict[str, Any]) -> None:
+    result["processed"] += 1
+    if item_result.get("skipped"):
+        result["skipped"] += 1
+        return
+    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
+        return
+
+    result["failed"] += 1
+    result["errors"].append(
+        {
+            "demand_grade_id": item_result.get("demand_grade_id"),
+            "demand_name": item_result.get("demand_name"),
+            "video_count": item_result.get("video_count"),
+            "run_id": item_result.get("run_id"),
+            "error": item_result.get("error"),
+        }
+    )
+
+
+def _run_task_with_timeout(
+    ctx: FindDemandContext,
+    *,
+    force: bool,
+    timeout_seconds: float,
+) -> dict[str, Any]:
+    """在独立线程中执行单任务，超时后向上抛出 FuturesTimeoutError。
+
+    超时后不阻塞等待底层线程（shutdown wait=False），以便 worker 能退出并由监督方补齐。
+    """
+    executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="find-task")
+    try:
+        future = executor.submit(process_single_discover, ctx, force=force)
+        return future.result(timeout=timeout_seconds)
+    finally:
+        executor.shutdown(wait=False, cancel_futures=True)
+
+
+def _drain_task_queue(task_queue: queue.Queue[FindDemandContext]) -> int:
+    discarded = 0
+    while True:
+        try:
+            task_queue.get_nowait()
+        except queue.Empty:
+            break
+        discarded += 1
+        task_queue.task_done()
+    return discarded
+
+
+def _run_worker_pool(
+    *,
+    contexts: list[FindDemandContext],
+    biz_dt: str,
+    worker_count: int,
+    force: bool,
+    result: dict[str, Any],
+    task_timeout_seconds: float,
+) -> None:
+    """任务队列 + 固定规模 worker 池；超时退出后自动补齐线程。"""
+    task_queue: queue.Queue[FindDemandContext] = queue.Queue()
+    for ctx in contexts:
+        task_queue.put(ctx)
+
+    stop_event = threading.Event()
+    stats_lock = threading.Lock()
+    worker_seq = 0
+    started_at = monotonic()
+
+    def worker_loop(worker_id: int) -> str:
+        """循环取任务；单任务超时则退出，由监督方补线程。"""
+        logger.info(
+            "discover worker started: worker_id=%s biz_dt=%s timeout_seconds=%.0f",
+            worker_id,
+            biz_dt,
+            task_timeout_seconds,
+        )
+        exit_reason = "queue_empty"
+        while not stop_event.is_set():
+            try:
+                ctx = task_queue.get(timeout=_QUEUE_GET_TIMEOUT_SECONDS)
+            except queue.Empty:
+                if stop_event.is_set() or task_queue.empty():
+                    exit_reason = "queue_empty"
+                    break
+                continue
+
+            try:
+                if stop_event.is_set():
+                    exit_reason = "stopped"
+                    break
+
+                try:
+                    passed_videos = _count_passed_videos(biz_dt)
+                except Exception:
+                    logger.exception(
+                        "discover worker count_passed_videos failed: worker_id=%s",
+                        worker_id,
+                    )
+                    with stats_lock:
+                        result["failed"] += 1
+                        result["processed"] += 1
+                        result["errors"].append(
+                            {
+                                "demand_grade_id": ctx.demand_grade_id,
+                                "demand_name": ctx.demand_name,
+                                "error": "count_passed_videos_failed",
+                            }
+                        )
+                    continue
+
+                with stats_lock:
+                    result["passed_videos"] = passed_videos
+
+                if passed_videos >= _DAILY_PASSED_VIDEO_LIMIT:
+                    with stats_lock:
+                        result["stopped_by_passed_video_limit"] = True
+                    stop_event.set()
+                    exit_reason = "passed_video_limit"
+                    logger.info(
+                        "discover worker stop by daily limit before execute: "
+                        "worker_id=%s grade_id=%s demand=%s passed_videos=%s",
+                        worker_id,
+                        ctx.demand_grade_id,
+                        ctx.demand_name,
+                        passed_videos,
+                    )
+                    break
+
+                logger.info(
+                    "discover worker pickup: worker_id=%s grade_id=%s demand=%s "
+                    "queue_remaining~=%s passed_videos=%s",
+                    worker_id,
+                    ctx.demand_grade_id,
+                    ctx.demand_name,
+                    task_queue.qsize(),
+                    passed_videos,
+                )
+
+                try:
+                    item_result = _run_task_with_timeout(
+                        ctx,
+                        force=force,
+                        timeout_seconds=task_timeout_seconds,
+                    )
+                except FuturesTimeoutError:
+                    error_text = (
+                        f"find_agent worker timed out after {task_timeout_seconds:.0f}s"
+                    )
+                    logger.error(
+                        "discover worker task timeout, worker will exit: "
+                        "worker_id=%s grade_id=%s demand=%s timeout_seconds=%.0f",
+                        worker_id,
+                        ctx.demand_grade_id,
+                        ctx.demand_name,
+                        task_timeout_seconds,
+                    )
+                    with stats_lock:
+                        result["failed"] += 1
+                        result["processed"] += 1
+                        result["errors"].append(
+                            {
+                                "demand_grade_id": ctx.demand_grade_id,
+                                "demand_name": ctx.demand_name,
+                                "video_count": ctx.video_count,
+                                "error": error_text,
+                            }
+                        )
+                    exit_reason = "task_timeout"
+                    break
+                except Exception as exc:
+                    logger.exception(
+                        "discover worker unexpected error: worker_id=%s grade_id=%s demand=%s",
+                        worker_id,
+                        ctx.demand_grade_id,
+                        ctx.demand_name,
+                    )
+                    with stats_lock:
+                        result["failed"] += 1
+                        result["processed"] += 1
+                        result["errors"].append(
+                            {
+                                "demand_grade_id": ctx.demand_grade_id,
+                                "demand_name": ctx.demand_name,
+                                "error": str(exc),
+                            }
+                        )
+                    continue
+
+                with stats_lock:
+                    _apply_item_result(result, item_result)
+                    logger.info(
+                        "discover worker item finished: worker_id=%s grade_id=%s demand=%s "
+                        "success=%s skipped=%s total_processed=%d",
+                        worker_id,
+                        ctx.demand_grade_id,
+                        ctx.demand_name,
+                        item_result.get("success"),
+                        item_result.get("skipped"),
+                        result["processed"],
+                    )
+
+                try:
+                    passed_videos = _count_passed_videos(biz_dt)
+                except Exception:
+                    logger.exception(
+                        "discover worker post-count failed: worker_id=%s",
+                        worker_id,
+                    )
+                    continue
+
+                with stats_lock:
+                    result["passed_videos"] = passed_videos
+                    if passed_videos >= _DAILY_PASSED_VIDEO_LIMIT:
+                        result["stopped_by_passed_video_limit"] = True
+                        stop_event.set()
+                        exit_reason = "passed_video_limit"
+                        break
+            finally:
+                task_queue.task_done()
+
+        logger.info(
+            "discover worker exited: worker_id=%s reason=%s elapsed_seconds=%.1f",
+            worker_id,
+            exit_reason,
+            monotonic() - started_at,
+        )
+        return exit_reason
+
+    with ThreadPoolExecutor(
+        max_workers=worker_count,
+        thread_name_prefix="find-agent",
+    ) as executor:
+        futures: set[Any] = set()
+
+        def _spawn_worker() -> None:
+            nonlocal worker_seq
+            worker_seq += 1
+            futures.add(executor.submit(worker_loop, worker_seq))
+
+        for _ in range(worker_count):
+            _spawn_worker()
+
+        while futures:
+            done, futures = wait(
+                futures,
+                timeout=_SUPERVISOR_POLL_SECONDS,
+                return_when=FIRST_COMPLETED,
+            )
+            if not done:
+                alive = len(futures)
+                logger.info(
+                    "discover workers heartbeat: biz_dt=%s alive=%d/%d "
+                    "processed=%d queue_remaining~=%s passed_videos=%s "
+                    "elapsed_seconds=%d",
+                    biz_dt,
+                    alive,
+                    worker_count,
+                    result["processed"],
+                    task_queue.qsize(),
+                    result.get("passed_videos"),
+                    int(monotonic() - started_at),
+                )
+                # 超时/异常退出后补齐到目标线程数
+                if (
+                    not stop_event.is_set()
+                    and not task_queue.empty()
+                    and alive < worker_count
+                ):
+                    for _ in range(worker_count - alive):
+                        if task_queue.empty():
+                            break
+                        logger.info(
+                            "discover worker replenish: biz_dt=%s alive=%d target=%d",
+                            biz_dt,
+                            alive,
+                            worker_count,
+                        )
+                        _spawn_worker()
+                        alive += 1
+                continue
+
+            for future in done:
+                try:
+                    exit_reason = future.result()
+                except Exception:
+                    logger.exception(
+                        "discover worker future crashed: biz_dt=%s",
+                        biz_dt,
+                    )
+                    exit_reason = "crashed"
+
+                if (
+                    not stop_event.is_set()
+                    and not task_queue.empty()
+                    and len(futures) < worker_count
+                ):
+                    logger.info(
+                        "discover worker replenish after exit: biz_dt=%s "
+                        "exit_reason=%s alive=%d target=%d",
+                        biz_dt,
+                        exit_reason,
+                        len(futures),
+                        worker_count,
+                    )
+                    _spawn_worker()
+
+        if stop_event.is_set():
+            discarded = _drain_task_queue(task_queue)
+            if discarded:
+                logger.info(
+                    "discover discarded pending tasks after stop: biz_dt=%s discarded=%d",
+                    biz_dt,
+                    discarded,
+                )
+
+
 def discover_videos_from_demands(
     biz_dt: str | None = None,
     *,
@@ -138,16 +478,16 @@ def discover_videos_from_demands(
     force: bool = False,
 ) -> dict[str, Any]:
     """
-    对指定业务日全部 S/A 级需求（有拓展点位）逐条调用 find_agent。
+    对指定业务日全部 S/A 级需求（有拓展点位）以任务队列方式并发调用 find_agent。
 
-    每条记录对应一个 demand_grade + 其下全部视频与全部拓展点位。
-    执行前会预写 video_discovery_run，并按 biz_dt + demand_grade_id 跳过已执行记录。
-    默认处理全部 S/A（S 优先于 A、再按 score 排序）；仅 CLI --top-limit 可人为截断。
-    当日 primary 去重视频达到上限时提前结束，否则跑完待处理队列。
-    默认单线程串行执行 find_agent（每条 S/A 需求依次处理）。
+    - 先把当天待执行需求全部放入内存队列（此时不写 video_discovery_run）
+    - 固定 worker 线程循环取任务；单任务开始执行时再创建 run 记录
+    - 执行前检查当日 primary 通过数，达标后不再领取新任务（允许略超）
+    - 单任务默认 600s 超时后该 worker 退出，监督方自动补齐线程
     """
     started_at = datetime.now()
     batch_run_id = uuid.uuid4().hex
+    task_timeout_seconds = find_agent_timeout_seconds()
 
     try:
         resolved_biz_dt, contexts = list_find_demand_contexts(
@@ -180,7 +520,9 @@ def discover_videos_from_demands(
     passed_videos = _count_passed_videos(resolved_biz_dt)
 
     logger.info(
-        "discover_videos_from_demands start: biz_dt=%s batch_run_id=%s workers=%s top_limit=%s offset=%s pending=%s skipped=%s passed_videos=%s passed_video_limit=%s",
+        "discover_videos_from_demands start: biz_dt=%s batch_run_id=%s workers=%s "
+        "top_limit=%s offset=%s pending=%s skipped=%s passed_videos=%s "
+        "passed_video_limit=%s task_timeout_seconds=%.0f",
         resolved_biz_dt,
         batch_run_id,
         workers,
@@ -190,6 +532,7 @@ def discover_videos_from_demands(
         preload_stats.get("skipped_already_done", 0),
         passed_videos,
         _DAILY_PASSED_VIDEO_LIMIT,
+        task_timeout_seconds,
     )
 
     result: dict[str, Any] = {
@@ -202,6 +545,7 @@ def discover_videos_from_demands(
         **preload_stats,
         "total_records": len(contexts),
         "workers": 0,
+        "task_timeout_seconds": task_timeout_seconds,
         "processed": 0,
         "succeeded": 0,
         "goal_met": 0,
@@ -227,131 +571,22 @@ def discover_videos_from_demands(
     ensure_mysql_pool_capacity(worker_count)
     result["workers"] = worker_count
 
-    with ThreadPoolExecutor(max_workers=worker_count) as executor:
-        next_context = 0
-        while next_context < len(contexts):
-            remaining_slots = _DAILY_PASSED_VIDEO_LIMIT - passed_videos
-            if remaining_slots <= 0:
-                result["stopped_by_passed_video_limit"] = True
-                break
-
-            batch_size = min(
-                worker_count,
-                remaining_slots,
-                len(contexts) - next_context,
-            )
-            batch = contexts[next_context : next_context + batch_size]
-            next_context += batch_size
-            logger.info(
-                "discover videos batch start: biz_dt=%s batch_size=%d "
-                "completed_batches=%d/%d demands=%s",
-                resolved_biz_dt,
-                len(batch),
-                next_context // batch_size,
-                (len(contexts) + batch_size - 1) // batch_size,
-                [
-                    f"{ctx.demand_grade_id}:{ctx.demand_name}"
-                    for ctx in batch
-                ],
-            )
-            future_to_ctx = {
-                executor.submit(process_single_discover, ctx, force=force): ctx
-                for ctx in batch
-            }
-            pending = set(future_to_ctx.keys())
-            batch_started = monotonic()
-            while pending:
-                completed, pending = wait(
-                    pending,
-                    timeout=60,
-                    return_when=FIRST_COMPLETED,
-                )
-                if not completed:
-                    pending_demands = [
-                        f"{future_to_ctx[future].demand_grade_id}:"
-                        f"{future_to_ctx[future].demand_name}"
-                        for future in pending
-                    ]
-                    logger.info(
-                        "discover videos still running: biz_dt=%s "
-                        "batch_pending=%d processed=%d elapsed_seconds=%d "
-                        "pending_demands=%s",
-                        resolved_biz_dt,
-                        len(pending),
-                        result["processed"],
-                        int(monotonic() - batch_started),
-                        pending_demands,
-                    )
-                    continue
-
-                for future in completed:
-                    ctx = future_to_ctx[future]
-                    try:
-                        item_result = future.result()
-                    except Exception as exc:
-                        logger.exception(
-                            "discover videos worker 出现未捕获错误: biz_dt=%s "
-                            "grade_id=%s demand=%s",
-                            resolved_biz_dt,
-                            ctx.demand_grade_id,
-                            ctx.demand_name,
-                        )
-                        result["failed"] += 1
-                        result["processed"] += 1
-                        result["errors"].append(
-                            {
-                                "demand_grade_id": ctx.demand_grade_id,
-                                "demand_name": ctx.demand_name,
-                                "error": str(exc),
-                            }
-                        )
-                        logger.info(
-                            "discover videos batch item failed: grade_id=%s "
-                            "demand=%s batch_remaining=%d total_processed=%d",
-                            ctx.demand_grade_id,
-                            ctx.demand_name,
-                            len(pending),
-                            result["processed"],
-                        )
-                        continue
-
-                    result["processed"] += 1
-                    logger.info(
-                        "discover videos batch item finished: grade_id=%s demand=%s "
-                        "success=%s skipped=%s batch_remaining=%d total_processed=%d",
-                        ctx.demand_grade_id,
-                        ctx.demand_name,
-                        item_result.get("success"),
-                        item_result.get("skipped"),
-                        len(pending),
-                        result["processed"],
-                    )
-                    if item_result.get("skipped"):
-                        result["skipped"] += 1
-                        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
-                    result["errors"].append(
-                        {
-                            "demand_grade_id": item_result.get("demand_grade_id"),
-                            "demand_name": item_result.get("demand_name"),
-                            "video_count": item_result.get("video_count"),
-                            "run_id": item_result.get("run_id"),
-                            "error": item_result.get("error"),
-                        }
-                    )
+    _run_worker_pool(
+        contexts=contexts,
+        biz_dt=resolved_biz_dt,
+        worker_count=worker_count,
+        force=force,
+        result=result,
+        task_timeout_seconds=task_timeout_seconds,
+    )
 
-            passed_videos = _count_passed_videos(resolved_biz_dt)
-            result["passed_videos"] = passed_videos
-            if passed_videos >= _DAILY_PASSED_VIDEO_LIMIT:
-                result["stopped_by_passed_video_limit"] = True
-                break
+    # 最终再读一次，避免结果里的 passed_videos 过旧
+    try:
+        result["passed_videos"] = _count_passed_videos(resolved_biz_dt)
+        if result["passed_videos"] >= _DAILY_PASSED_VIDEO_LIMIT:
+            result["stopped_by_passed_video_limit"] = True
+    except Exception:
+        logger.exception("discover_videos_from_demands final count_passed_videos failed")
 
     finished_at = datetime.now()
     result["finished_at"] = finished_at.isoformat()

tests/supply_agent/test_find_agent_rate_limit.py

@@ -0,0 +1,43 @@
+"""跨线程异步限速单元测试。"""
+from __future__ import annotations
+
+import asyncio
+import threading
+import time
+
+from agents.find_agent.support.rate_limit import make_async_interval_limiter
+
+
+def test_async_interval_limiter_is_safe_across_threads() -> None:
+    wait = make_async_interval_limiter(0.05)
+    hits: list[float] = []
+    lock = threading.Lock()
+    errors: list[BaseException] = []
+
+    def worker() -> None:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            for _ in range(2):
+                loop.run_until_complete(wait())
+                with lock:
+                    hits.append(time.monotonic())
+        except BaseException as exc:  # noqa: BLE001 - collect for assertion
+            with lock:
+                errors.append(exc)
+        finally:
+            loop.close()
+            asyncio.set_event_loop(None)
+
+    threads = [threading.Thread(target=worker) for _ in range(3)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=5)
+        assert not thread.is_alive()
+
+    assert not errors
+    assert len(hits) == 6
+    ordered = sorted(hits)
+    for prev, curr in zip(ordered, ordered[1:]):
+        assert curr - prev >= 0.04

tests/supply_infra/scheduler/test_discover_videos_from_demands.py

@@ -1460,8 +1460,15 @@ def test_stops_discovery_after_300_passed_videos(
         contexts,
         {"total_loaded": 2, "skipped_already_done": 0},
     )
-    mock_count_passed.side_effect = [299, 300]
-    mock_process.return_value = {"success": True, "skipped": False}
+    # 初始检查 / 执行前检查 / 执行后检查 / 收尾检查
+    mock_count_passed.side_effect = [299, 299, 300, 300]
+    mock_process.return_value = {
+        "success": True,
+        "skipped": False,
+        "demand_grade_id": 1,
+        "demand_name": "需求A",
+        "business_outcome": "goal_met",
+    }
 
     result = discover_videos_from_demands("20260727", workers=1)
 
@@ -1471,6 +1478,121 @@ def test_stops_discovery_after_300_passed_videos(
     assert result["stopped_by_passed_video_limit"] is True
 
 
+@patch(
+    "supply_infra.scheduler.jobs.discover_videos_from_demands.process_single_discover"
+)
+@patch("supply_infra.scheduler.jobs.discover_videos_from_demands._count_passed_videos")
+@patch(
+    "supply_infra.scheduler.jobs.discover_videos_from_demands.filter_pending_contexts"
+)
+@patch(
+    "supply_infra.scheduler.jobs.discover_videos_from_demands.list_find_demand_contexts"
+)
+def test_queue_workers_process_all_pending_tasks(
+    mock_list_contexts,
+    mock_filter_contexts,
+    mock_count_passed,
+    mock_process,
+) -> None:
+    contexts = [
+        FindDemandContext(
+            biz_dt="20260727",
+            demand_grade_id=i,
+            demand_name=f"需求{i}",
+            grade="S",
+        )
+        for i in range(1, 8)
+    ]
+    mock_list_contexts.return_value = ("20260727", contexts)
+    mock_filter_contexts.return_value = (
+        contexts,
+        {"total_loaded": 7, "skipped_already_done": 0},
+    )
+    mock_count_passed.return_value = 0
+
+    def _fake_process(ctx, *, force=False):
+        return {
+            "success": True,
+            "skipped": False,
+            "demand_grade_id": ctx.demand_grade_id,
+            "demand_name": ctx.demand_name,
+            "business_outcome": "partial",
+        }
+
+    mock_process.side_effect = _fake_process
+
+    result = discover_videos_from_demands("20260727", workers=5)
+
+    assert mock_process.call_count == 7
+    assert result["processed"] == 7
+    assert result["succeeded"] == 7
+    assert result["partial"] == 7
+    assert result["workers"] == 5
+    assert result["stopped_by_passed_video_limit"] is False
+
+
+@patch(
+    "supply_infra.scheduler.jobs.discover_videos_from_demands._run_task_with_timeout"
+)
+@patch("supply_infra.scheduler.jobs.discover_videos_from_demands._count_passed_videos")
+@patch(
+    "supply_infra.scheduler.jobs.discover_videos_from_demands.filter_pending_contexts"
+)
+@patch(
+    "supply_infra.scheduler.jobs.discover_videos_from_demands.list_find_demand_contexts"
+)
+def test_worker_timeout_marks_failed_and_continues_queue(
+    mock_list_contexts,
+    mock_filter_contexts,
+    mock_count_passed,
+    mock_run_timeout,
+) -> None:
+    from concurrent.futures import TimeoutError as FuturesTimeoutError
+
+    contexts = [
+        FindDemandContext(
+            biz_dt="20260727",
+            demand_grade_id=1,
+            demand_name="超时任务",
+            grade="S",
+        ),
+        FindDemandContext(
+            biz_dt="20260727",
+            demand_grade_id=2,
+            demand_name="后续任务",
+            grade="A",
+        ),
+    ]
+    mock_list_contexts.return_value = ("20260727", contexts)
+    mock_filter_contexts.return_value = (
+        contexts,
+        {"total_loaded": 2, "skipped_already_done": 0},
+    )
+    mock_count_passed.return_value = 0
+
+    def _side_effect(ctx, *, force=False, timeout_seconds=600.0):
+        if ctx.demand_grade_id == 1:
+            raise FuturesTimeoutError()
+        return {
+            "success": True,
+            "skipped": False,
+            "demand_grade_id": ctx.demand_grade_id,
+            "demand_name": ctx.demand_name,
+            "business_outcome": "goal_met",
+        }
+
+    mock_run_timeout.side_effect = _side_effect
+
+    result = discover_videos_from_demands("20260727", workers=1)
+
+    assert mock_run_timeout.call_count == 2
+    assert result["processed"] == 2
+    assert result["failed"] == 1
+    assert result["succeeded"] == 1
+    assert result["goal_met"] == 1
+    assert any("timed out" in str(err.get("error", "")) for err in result["errors"])
+
+
 class _AsyncClient:
     def __init__(self) -> None:
         self.closed = False