refactor(subscribe): isolate queued search execution

app/application/subscription/execution.py

@@ -206,6 +206,17 @@ class SubscriptionSearchRepository(Protocol):
         """按订阅 ID 和各自到期时间建立或合并活动任务。"""
         ...
 
+    async def async_enqueue(
+        self,
+        *,
+        subscription_ids: tuple[int, ...],
+        source: str,
+        priority: int,
+        available_at_by_subscription: Optional[Mapping[int, str]] = None,
+    ) -> SearchEnqueueResult:
+        """在异步会话中建立或合并活动任务。"""
+        ...
+
     def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]:
         """按优先级和稳定游标认领下一条可执行任务。"""
         ...

app/chain/subscribe/contract.py

@@ -36,6 +36,7 @@ if TYPE_CHECKING:
         _SubscribeChain__async_apply_episodes_refresh: Callable[..., Awaitable[Any]]
         _SubscribeChain__async_notify_subscribe_create_failure: Callable[..., Awaitable[Any]]
         _SubscribeChain__async_post_subscribe_added: Callable[..., Awaitable[Any]]
+        _SubscribeChain__async_queue_new_subscription_search: Callable[..., Awaitable[Any]]
         _SubscribeChain__build_completion_notification: Callable[..., Any]
         _SubscribeChain__build_subscribe_notification: Callable[..., Any]
         _SubscribeChain__download_best_version_with_full_pack_first: Callable[..., Any]

app/chain/subscribe/notify.py

@@ -18,7 +18,6 @@ from app.domain.context import (
     MediaInfo,
 )
 from app.domain.meta.metabase import MetaBase
-from app.runtime.execution import run_in_threadpool
 from app.runtime.log import logger
 from app.schemas.common import JsonData
 from app.schemas.message import Message as _SchemaMessage
@@ -145,17 +144,16 @@ class SubscribeNotificationOwner(_SubscribeOwnerBase):
         )
         try:
             self._SubscribeChain__queue_new_subscription_search(subscribe_id)
-        except Exception as error:
-            logger.warning(
-                "订阅已保存，但自动搜索暂时没有安排成功，"
-                f"系统会在下一次检查时重试：{error}"
-            )
+        except Exception:
+            logger.warning("订阅已保存，但自动搜索暂时没有安排成功，系统会在下一次检查时重试")
+            logger.debug("安排订阅自动搜索失败", exc_info=True)
         try:
             report_delivered = _subscription_share_snapshot().report_added(
                 self._SubscribeChain__subscribe_report_payload(context)
             )
-        except Exception as error:
-            logger.warning(f"订阅新增统计上报失败，将由后台重试：{error}")
+        except Exception:
+            logger.warning("订阅新增统计暂时没有上报成功，系统会在后台重试")
+            logger.debug("订阅新增统计上报失败", exc_info=True)
             return False
         if not report_delivered:
             logger.warning("订阅新增统计上报未确认，将由后台重试")
@@ -186,21 +184,17 @@ class SubscribeNotificationOwner(_SubscribeOwnerBase):
             },
         )
         try:
-            await run_in_threadpool(
-                self._SubscribeChain__queue_new_subscription_search,
-                subscribe_id,
-            )
-        except Exception as error:
-            logger.warning(
-                "订阅已保存，但自动搜索暂时没有安排成功，"
-                f"系统会在下一次检查时重试：{error}"
-            )
+            await self._SubscribeChain__async_queue_new_subscription_search(subscribe_id)
+        except Exception:
+            logger.warning("订阅已保存，但自动搜索暂时没有安排成功，系统会在下一次检查时重试")
+            logger.debug("安排订阅自动搜索失败", exc_info=True)
         try:
             report_delivered = await _subscription_share_snapshot().async_report_added(
                 self._SubscribeChain__subscribe_report_payload(context)
             )
-        except Exception as error:
-            logger.warning(f"订阅新增统计上报失败，将由后台重试：{error}")
+        except Exception:
+            logger.warning("订阅新增统计暂时没有上报成功，系统会在后台重试")
+            logger.debug("订阅新增统计上报失败", exc_info=True)
             return False
         if not report_delivered:
             logger.warning("订阅新增统计上报未确认，将由后台重试")

app/chain/subscribe/search.py

@@ -19,7 +19,6 @@ from app.application.subscription.execution import (
     SearchTaskSnapshot,
     SubscriptionExecutionContext,
     SubscriptionSearchRepository,
-    handle_subscription_search_deferred,
     raise_subscription_site_budget_deferral,
     raise_subscription_site_budget_failures,
 )
@@ -28,20 +27,22 @@ from app.application.subscription.observability import (
     SearchTaskOutcome,
     batch_finished_count,
     batch_progress_text,
-    finish_returned_search_task,
     inline_search_result,
 )
 from app.application.subscription.query import SubscriptionQueryService
 from app.application.subscription.sitebudget import (
     SubscriptionSearchCancelled,
     SubscriptionSearchDeferred,
-    SubscriptionSiteBudget,
 )
 from app.chain.media import MediaChain
 from app.chain.search.facade import SearchChain
 from app.chain.subscribe.contract import _SubscribeOwnerBase
 from app.chain.subscribe.identity import subscribe_recognize_kwargs
 from app.chain.subscribe.metadata import apply_subscription_classification
+from app.chain.subscribe.searchtask import (
+    SubscriptionSearchTaskRunner,
+    retry_at_after,
+)
 from app.domain.context import (
     MediaInfo,
 )
@@ -54,8 +55,6 @@ from app.schemas.types import (
 )
 
 _NEW_SUBSCRIPTION_EDIT_SECONDS = 60
-_FOREGROUND_RETRY_SECONDS = 5
-_BACKGROUND_RETRY_SECONDS = 10
 
 
 def _ensure_execution_active(
@@ -70,22 +69,6 @@ def _ensure_execution_active(
         raise TimeoutError("这次搜索用时过长，已停止")
 
 
-def _update_search_task_phase(
-    queue: SubscriptionSearchRepository,
-    task_id: str,
-    lease_token: str,
-    phase: str,
-    current_site_id: Optional[int] = None,
-) -> None:
-    """以当前任务租约持久化业务阶段，过期执行者不得覆盖新状态。"""
-    queue.update_task_phase(
-        task_id=task_id,
-        lease_token=lease_token,
-        phase=phase,
-        current_site_id=current_site_id,
-    )
-
-
 def _search_source_and_priority(
     *,
     sid: Optional[int],
@@ -104,13 +87,6 @@ def _search_source_and_priority(
     return "new", 50
 
 
-def _retry_at_after(seconds: int) -> str:
-    """返回指定秒数后的 UTC 时间，供短暂等待任务重新入队。"""
-    return (datetime.now(timezone.utc) + timedelta(seconds=max(1, seconds))).isoformat(
-        timespec="seconds"
-    )
-
-
 def _search_task_available_at(
     source: str,
     subscription_ids: tuple[int, ...],
@@ -132,36 +108,6 @@ def _search_task_available_at(
     return available_at
 
 
-def _skip_search_task(
-    queue: SubscriptionSearchRepository,
-    task: SearchTaskSnapshot,
-    reason: str,
-) -> bool:
-    """以 skipped 终态收口未执行任务，并保留可见原因。"""
-    if task.lease_token is None:
-        return False
-    return queue.finish_task(
-        task_id=task.task_id,
-        lease_token=task.lease_token,
-        state="skipped",
-        error=reason,
-    )
-
-
-def _release_cancelled_or_stopped_search_task(
-    queue: SubscriptionSearchRepository,
-    task_id: str,
-    lease_token: str,
-    system_stopped: bool,
-) -> bool:
-    """用户取消落终态，系统停机仅释放任务以供重启恢复。"""
-    return queue.release_task(
-        task_id=task_id,
-        lease_token=lease_token,
-        cancelled=not system_stopped,
-    )
-
-
 class _SubscribeSearchQueueCoordinator(_SubscribeOwnerBase):
     """订阅主动搜索入口、队列提交与消费协调 owner。"""
 
@@ -528,212 +474,30 @@ class _SubscribeSearchQueueOwner(_SubscribeSearchQueueCoordinator):
         summary: SearchExecutionSummary,
     ) -> Optional[int]:
         """执行一条已认领任务，返回实际进入搜索处理的订阅 ID。"""
-        if task.lease_token is None:
-            logger.error("这次订阅搜索暂时无法开始，系统稍后会重新处理")
-            summary.record("failed", "missing_lease")
-            return None
-        task_id = str(task.task_id)
-        lease_token = task.lease_token
-        cancelled = partial(queue.is_cancel_requested, task_id)
         stop_state = getattr(self, "stop_state", runtime_stop_state)
-        if cancelled():
-            queue.release_task(task_id=task_id, lease_token=lease_token, cancelled=True)
-            summary.record("cancelled", "cancelled")
-            return None
-        subscribe = self.subscription_repository.get(task.subscription_id)
-        if subscribe is None:
-            queue.finish_task(
-                task_id=task_id,
-                lease_token=lease_token,
-                state="cancelled",
-                error="订阅已删除",
-            )
-            summary.record("cancelled", "missing_subscription")
-            return None
-        self._report_search_progress(progress_callback, subscribe, index, limit)
-        recent_retry_at = self._recent_subscription_retry_at(subscribe, task.source)
-        if recent_retry_at:
-            queue.defer_task(
-                task_id=task_id,
-                lease_token=lease_token,
-                available_at=recent_retry_at,
-                phase="scheduled",
-                message="订阅刚刚创建，保存好设置后会自动开始搜索",
-            )
-            summary.record("requeued", "recent_subscription")
-            return None
-        lease = self._subscription_execution_admission.try_acquire(
-            subscription_id=subscribe.id,
-            operation="search",
-            ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL,
+        runner = SubscriptionSearchTaskRunner(
+            queue=queue,
+            task=task,
+            owner=owner,
+            searchchain=searchchain,
+            index=index,
+            limit=limit,
+            progress_callback=progress_callback,
+            summary=summary,
+            subscription_repository=self.subscription_repository,
+            execution_admission=self._subscription_execution_admission,
+            execution_ttl=self._SUBSCRIPTION_EXECUTION_TTL,
+            recent_retry_at=self._recent_subscription_retry_at,
+            process_subscription=self._process_search_subscription,
+            reset_subscription=partial(
+                self._SubscribeChain__apply_subscribe_update,
+                update_data={"state": "R"},
+                scene="search_reset",
+            ),
+            report_progress=self._report_search_progress,
+            stop_state=stop_state,
         )
-        if lease is None:
-            if task.source in {"manual", "targeted", "new"}:
-                retry_seconds = (
-                    _FOREGROUND_RETRY_SECONDS
-                    if task.source in {"manual", "targeted"}
-                    else _BACKGROUND_RETRY_SECONDS
-                )
-                queue.defer_task(
-                    task_id=task_id,
-                    lease_token=lease_token,
-                    available_at=_retry_at_after(retry_seconds),
-                    phase="waiting_subscription",
-                    message="这个订阅正在处理，结束后会自动继续搜索",
-                )
-                summary.record("requeued", "admission_conflict")
-                logger.debug(f"订阅《{subscribe.name}》正在处理，搜索会在结束后自动继续")
-            else:
-                _skip_search_task(
-                    queue,
-                    task,
-                    "这个订阅正在处理，本次自动检查无需重复执行",
-                )
-                summary.record("skipped", "admission_conflict")
-            return None
-        phase_changed = partial(_update_search_task_phase, queue, task_id, lease_token)
-        execution_context = SubscriptionExecutionContext(
-            lease=lease,
-            admission=self._subscription_execution_admission,
-            task_id=task_id,
-            cancel_requested=lambda: cancelled() or stop_state.is_system_stopped,
-            phase_changed=phase_changed,
-        )
-        current = subscribe
-        try:
-            current = self.subscription_repository.get(task.subscription_id)
-            if current is None:
-                queue.finish_task(
-                    task_id=task_id,
-                    lease_token=lease_token,
-                    state="cancelled",
-                    error="订阅已删除",
-                )
-                summary.record("cancelled", "missing_subscription")
-                return None
-            if current.state == "S":
-                _skip_search_task(queue, task, "订阅已暂停，这次没有搜索")
-                summary.record("skipped", "paused")
-                return None
-            searchchain.configure_subscription_site_budget(
-                SubscriptionSiteBudget(
-                    repository=queue,
-                    owner=f"{owner}:{task_id}",
-                    cancelled=execution_context.should_stop,
-                    stop_state=stop_state,
-                    phase_changed=phase_changed,
-                    metrics=summary.site_metrics,
-                )
-            )
-            current = self._process_search_subscription(
-                current,
-                searchchain,
-                execution_context=execution_context,
-            )
-            system_stopped = stop_state.is_system_stopped
-            cancel_requested = False if system_stopped else cancelled()
-            subscription_id, outcome, reason = finish_returned_search_task(
-                queue=queue,
-                task_id=task_id,
-                lease_token=lease_token,
-                subscription_id=task.subscription_id,
-                execution_context=execution_context,
-                system_stopped=system_stopped,
-                cancel_requested=cancel_requested,
-            )
-            summary.record(outcome, reason)
-            return subscription_id
-        except SubscriptionSearchCancelled:
-            if execution_context.is_expired() and not execution_context.is_cancel_requested():
-                queue.finish_task(
-                    task_id=task_id,
-                    lease_token=lease_token,
-                    state="failed",
-                    error="这次搜索用时过长，已停止，可稍后重试",
-                )
-                summary.record("failed", "ttl_timeout")
-            else:
-                system_stopped = stop_state.is_system_stopped
-                _release_cancelled_or_stopped_search_task(
-                    queue,
-                    task_id,
-                    lease_token,
-                    system_stopped,
-                )
-                summary.record(
-                    "requeued" if system_stopped else "cancelled",
-                    "system_stop" if system_stopped else "cancelled",
-                )
-        except SubscriptionSearchDeferred as deferred:
-            handle_subscription_search_deferred(
-                queue, task_id, lease_token, deferred, summary.record
-            )
-        except Exception as err:
-            logger.error(f"订阅 {subscribe.name} 搜索失败：{str(err)}", exc_info=True)
-            queue.finish_task(
-                task_id=task_id,
-                lease_token=lease_token,
-                state="failed",
-                error=str(err),
-            )
-            summary.record("failed", "error")
-        finally:
-            self._cleanup_search_task(
-                queue=queue,
-                searchchain=searchchain,
-                subscribe=subscribe,
-                current=current,
-                lease=lease,
-                progress_callback=progress_callback,
-                index=index,
-                limit=limit,
-                summary=summary,
-            )
-        return None
-
-    def _cleanup_search_task(
-        self,
-        *,
-        queue: SubscriptionSearchRepository,
-        searchchain: SearchChain,
-        subscribe: SubscriptionSnapshot,
-        current: Optional[SubscriptionSnapshot],
-        lease: Any,
-        progress_callback: Optional[Callable[..., None]],
-        index: int,
-        limit: int,
-        summary: SearchExecutionSummary,
-    ) -> None:
-        """清理站点预算和订阅状态，并在所有异常路径释放 owner。"""
-        try:
-            searchchain.configure_subscription_site_budget(None)
-        except Exception as err:
-            logger.error(
-                f"订阅《{subscribe.name}》结束站点访问时遇到问题，"
-                f"系统稍后会继续处理：{str(err)}",
-                exc_info=True,
-            )
-        try:
-            if current and current.state == "N":
-                self._SubscribeChain__apply_subscribe_update(
-                    current,
-                    {"state": "R"},
-                    scene="search_reset",
-                )
-        except Exception as err:
-            logger.error(f"订阅《{subscribe.name}》搜索结束后未能恢复正常状态：{str(err)}", exc_info=True)
-        finally:
-            released = self._subscription_execution_admission.release(lease)
-            if not released:
-                summary.release_failures += 1
-                logger.error(f"订阅《{subscribe.name}》的搜索状态没有正常恢复，系统稍后会继续处理")
-            self._report_search_progress(
-                progress_callback,
-                subscribe,
-                index,
-                limit,
-                finished=True,
-            )
+        return runner.execute()
 
     def resume_search_queue(
         self,
@@ -837,7 +601,7 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
             return None
         retry_seconds = max(1, int(remaining) + 1)
         logger.debug(f"新订阅《{subscribe.name}》将在约 {retry_seconds} 秒后自动开始搜索")
-        return _retry_at_after(retry_seconds)
+        return retry_at_after(retry_seconds)
 
     def _SubscribeChain__queue_new_subscription_search(
         self,
@@ -865,6 +629,33 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner):
             logger.info(f"已安排新订阅《{subscribe.name}》自动搜索，保存好设置后会自动开始")
         return enqueued.active_batch_ids[0] if enqueued.active_batch_ids else None
 
+    async def _SubscribeChain__async_queue_new_subscription_search(
+        self,
+        subscribe_id: int,
+    ) -> Optional[str]:
+        """在异步订阅提交后安排自动搜索。"""
+        queue: Optional[SubscriptionSearchRepository] = getattr(
+            self, "subscription_search_repository", None
+        )
+        if queue is None:
+            return None
+        subscribe = await self.subscription_repository.async_get(subscribe_id)
+        if subscribe is None or subscribe.state != "N":
+            return None
+        available_at = self._recent_subscription_retry_at(subscribe, "new")
+        enqueued = await queue.async_enqueue(
+            subscription_ids=(subscribe_id,),
+            source="new",
+            priority=50,
+            available_at_by_subscription={
+                subscribe_id: available_at
+                or datetime.now(timezone.utc).isoformat(timespec="seconds")
+            },
+        )
+        if enqueued.created_count:
+            logger.info(f"已安排新订阅《{subscribe.name}》自动搜索，保存好设置后会自动开始")
+        return enqueued.active_batch_ids[0] if enqueued.active_batch_ids else None
+
     @staticmethod
     def _wait_before_scheduled_search(
         sid: Optional[int],

app/chain/subscribe/searchtask.py

@@ -0,0 +1,330 @@
+"""订阅搜索队列单任务执行。"""
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from functools import partial
+from typing import Callable, Optional
+
+from app.application.subscription.contract import (
+    SubscriptionRepository,
+    SubscriptionSnapshot,
+)
+from app.application.subscription.execution import (
+    SearchTaskSnapshot,
+    SubscriptionExecutionAdmission,
+    SubscriptionExecutionContext,
+    SubscriptionExecutionLease,
+    SubscriptionSearchRepository,
+    handle_subscription_search_deferred,
+)
+from app.application.subscription.observability import (
+    SearchExecutionSummary,
+    finish_returned_search_task,
+)
+from app.application.subscription.sitebudget import (
+    SubscriptionSearchCancelled,
+    SubscriptionSearchDeferred,
+    SubscriptionSiteBudget,
+)
+from app.chain.search.facade import SearchChain
+from app.runtime.log import logger
+from app.runtime.stop import StopState
+
+_FOREGROUND_RETRY_SECONDS = 5
+_BACKGROUND_RETRY_SECONDS = 10
+
+
+def retry_at_after(seconds: int) -> str:
+    """返回指定秒数后的 UTC 时间，供短暂等待任务重新入队。"""
+    return (datetime.now(timezone.utc) + timedelta(seconds=max(1, seconds))).isoformat(
+        timespec="seconds"
+    )
+
+
+def skip_search_task(
+    queue: SubscriptionSearchRepository,
+    task: SearchTaskSnapshot,
+    reason: str,
+) -> bool:
+    """以 skipped 终态收口未执行任务，并保留可见原因。"""
+    if task.lease_token is None:
+        return False
+    return queue.finish_task(
+        task_id=task.task_id,
+        lease_token=task.lease_token,
+        state="skipped",
+        error=reason,
+    )
+
+
+@dataclass(slots=True)
+class SubscriptionSearchTaskRunner:
+    """执行一条已认领的订阅搜索任务，并负责完整收尾。"""
+
+    queue: SubscriptionSearchRepository
+    task: SearchTaskSnapshot
+    owner: str
+    searchchain: SearchChain
+    index: int
+    limit: int
+    progress_callback: Optional[Callable[..., None]]
+    summary: SearchExecutionSummary
+    subscription_repository: SubscriptionRepository
+    execution_admission: SubscriptionExecutionAdmission
+    execution_ttl: int
+    recent_retry_at: Callable[[SubscriptionSnapshot, str], Optional[str]]
+    process_subscription: Callable[..., Optional[SubscriptionSnapshot]]
+    reset_subscription: Callable[[SubscriptionSnapshot], SubscriptionSnapshot]
+    report_progress: Callable[..., None]
+    stop_state: StopState
+
+    def execute(self) -> Optional[int]:
+        """执行当前任务，返回实际进入搜索处理的订阅 ID。"""
+        lease_token = self.task.lease_token
+        if lease_token is None:
+            logger.error("这次订阅搜索暂时无法开始，系统稍后会重新处理")
+            self.summary.record("failed", "missing_lease")
+            return None
+
+        task_id = str(self.task.task_id)
+        cancelled = partial(self.queue.is_cancel_requested, task_id)
+        if cancelled():
+            self.queue.release_task(task_id=task_id, lease_token=lease_token, cancelled=True)
+            self.summary.record("cancelled", "cancelled")
+            return None
+
+        subscribe = self.subscription_repository.get(self.task.subscription_id)
+        if subscribe is None:
+            self._finish_missing_subscription(task_id, lease_token)
+            return None
+
+        self.report_progress(
+            self.progress_callback,
+            subscribe,
+            self.index,
+            self.limit,
+        )
+        recent_retry_at = self.recent_retry_at(subscribe, self.task.source)
+        if recent_retry_at:
+            self.queue.defer_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                available_at=recent_retry_at,
+                phase="scheduled",
+                message="订阅刚刚创建，保存好设置后会自动开始搜索",
+            )
+            self.summary.record("requeued", "recent_subscription")
+            return None
+
+        execution_lease = self.execution_admission.try_acquire(
+            subscription_id=subscribe.id,
+            operation="search",
+            ttl_seconds=self.execution_ttl,
+        )
+        if execution_lease is None:
+            self._handle_active_subscription(subscribe, task_id, lease_token)
+            return None
+        return self._execute_owned_task(
+            subscribe=subscribe,
+            task_id=task_id,
+            lease_token=lease_token,
+            execution_lease=execution_lease,
+            cancelled=cancelled,
+        )
+
+    def _finish_missing_subscription(self, task_id: str, lease_token: str) -> None:
+        """把已经删除的订阅任务标记为取消。"""
+        self.queue.finish_task(
+            task_id=task_id,
+            lease_token=lease_token,
+            state="cancelled",
+            error="订阅已删除",
+        )
+        self.summary.record("cancelled", "missing_subscription")
+
+    def _handle_active_subscription(
+        self,
+        subscribe: SubscriptionSnapshot,
+        task_id: str,
+        lease_token: str,
+    ) -> None:
+        """根据搜索来源延后或跳过正在处理的订阅。"""
+        if self.task.source in {"manual", "targeted", "new"}:
+            retry_seconds = (
+                _FOREGROUND_RETRY_SECONDS
+                if self.task.source in {"manual", "targeted"}
+                else _BACKGROUND_RETRY_SECONDS
+            )
+            self.queue.defer_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                available_at=retry_at_after(retry_seconds),
+                phase="waiting_subscription",
+                message="这个订阅正在处理，结束后会自动继续搜索",
+            )
+            self.summary.record("requeued", "admission_conflict")
+            logger.debug(f"订阅《{subscribe.name}》正在处理，搜索会在结束后自动继续")
+            return
+        skip_search_task(
+            self.queue,
+            self.task,
+            "这个订阅正在处理，本次自动检查无需重复执行",
+        )
+        self.summary.record("skipped", "admission_conflict")
+
+    def _execute_owned_task(
+        self,
+        *,
+        subscribe: SubscriptionSnapshot,
+        task_id: str,
+        lease_token: str,
+        execution_lease: SubscriptionExecutionLease,
+        cancelled: Callable[[], bool],
+    ) -> Optional[int]:
+        """在持有订阅执行权时完成搜索、异常处理和资源释放。"""
+        phase_changed = partial(self._update_phase, task_id, lease_token)
+        execution_context = SubscriptionExecutionContext(
+            lease=execution_lease,
+            admission=self.execution_admission,
+            task_id=task_id,
+            cancel_requested=lambda: cancelled() or self.stop_state.is_system_stopped,
+            phase_changed=phase_changed,
+        )
+        current: Optional[SubscriptionSnapshot] = subscribe
+        try:
+            current = self.subscription_repository.get(self.task.subscription_id)
+            if current is None:
+                self._finish_missing_subscription(task_id, lease_token)
+                return None
+            if current.state == "S":
+                skip_search_task(self.queue, self.task, "订阅已暂停，这次没有搜索")
+                self.summary.record("skipped", "paused")
+                return None
+            self.searchchain.configure_subscription_site_budget(
+                SubscriptionSiteBudget(
+                    repository=self.queue,
+                    owner=f"{self.owner}:{task_id}",
+                    cancelled=execution_context.should_stop,
+                    stop_state=self.stop_state,
+                    phase_changed=phase_changed,
+                    metrics=self.summary.site_metrics,
+                )
+            )
+            current = self.process_subscription(
+                current,
+                self.searchchain,
+                execution_context=execution_context,
+            )
+            system_stopped = self.stop_state.is_system_stopped
+            cancel_requested = False if system_stopped else cancelled()
+            subscription_id, outcome, reason = finish_returned_search_task(
+                queue=self.queue,
+                task_id=task_id,
+                lease_token=lease_token,
+                subscription_id=self.task.subscription_id,
+                execution_context=execution_context,
+                system_stopped=system_stopped,
+                cancel_requested=cancel_requested,
+            )
+            self.summary.record(outcome, reason)
+            return subscription_id
+        except SubscriptionSearchCancelled:
+            self._handle_cancelled_search(task_id, lease_token, execution_context)
+        except SubscriptionSearchDeferred as deferred:
+            handle_subscription_search_deferred(
+                self.queue,
+                task_id,
+                lease_token,
+                deferred,
+                self.summary.record,
+            )
+        except Exception as err:
+            logger.error(f"订阅《{subscribe.name}》搜索失败：{str(err)}", exc_info=True)
+            self.queue.finish_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                state="failed",
+                error=str(err),
+            )
+            self.summary.record("failed", "error")
+        finally:
+            self._cleanup(subscribe, current, execution_lease)
+        return None
+
+    def _update_phase(
+        self,
+        task_id: str,
+        lease_token: str,
+        phase: str,
+        current_site_id: Optional[int] = None,
+    ) -> None:
+        """保存当前任务的用户可见阶段和正在访问的站点。"""
+        self.queue.update_task_phase(
+            task_id=task_id,
+            lease_token=lease_token,
+            phase=phase,
+            current_site_id=current_site_id,
+        )
+
+    def _handle_cancelled_search(
+        self,
+        task_id: str,
+        lease_token: str,
+        execution_context: SubscriptionExecutionContext,
+    ) -> None:
+        """区分搜索超时、用户停止和系统关闭。"""
+        if execution_context.is_expired() and not execution_context.is_cancel_requested():
+            self.queue.finish_task(
+                task_id=task_id,
+                lease_token=lease_token,
+                state="failed",
+                error="这次搜索用时过长，已停止，可稍后重试",
+            )
+            self.summary.record("failed", "ttl_timeout")
+            return
+        system_stopped = self.stop_state.is_system_stopped
+        self.queue.release_task(
+            task_id=task_id,
+            lease_token=lease_token,
+            cancelled=not system_stopped,
+        )
+        self.summary.record(
+            "requeued" if system_stopped else "cancelled",
+            "system_stop" if system_stopped else "cancelled",
+        )
+
+    def _cleanup(
+        self,
+        subscribe: SubscriptionSnapshot,
+        current: Optional[SubscriptionSnapshot],
+        execution_lease: SubscriptionExecutionLease,
+    ) -> None:
+        """清理站点访问、订阅状态和本轮执行权。"""
+        try:
+            self.searchchain.configure_subscription_site_budget(None)
+        except Exception as err:
+            logger.error(
+                f"订阅《{subscribe.name}》结束站点访问时遇到问题，"
+                f"系统稍后会继续处理：{str(err)}",
+                exc_info=True,
+            )
+        try:
+            if current and current.state == "N":
+                self.reset_subscription(current)
+        except Exception as err:
+            logger.error(
+                f"订阅《{subscribe.name}》搜索结束后未能恢复正常状态：{str(err)}",
+                exc_info=True,
+            )
+        finally:
+            released = self.execution_admission.release(execution_lease)
+            if not released:
+                self.summary.release_failures += 1
+                logger.error(f"订阅《{subscribe.name}》的搜索状态没有正常恢复，系统稍后会继续处理")
+            self.report_progress(
+                self.progress_callback,
+                subscribe,
+                self.index,
+                self.limit,
+                finished=True,
+            )

app/db/adapters/subscriptionsearch.py

@@ -1,9 +1,11 @@
 """订阅搜索持久队列的 SQLAlchemy 适配器。"""
 
 from collections.abc import Callable, Mapping
+from contextlib import AbstractAsyncContextManager
 from datetime import datetime, timedelta, timezone
 from typing import Optional, TypeVar
 
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Session
 
 from app.application.subscription.execution import (
@@ -17,7 +19,7 @@ from app.db.models.subscriptionsearch import (
     SubscriptionSearchTask,
 )
 from app.db.oper.subscriptionsearch import SubscriptionSearchOper
-from app.db.uow import SqlAlchemyUnitOfWork
+from app.db.uow import SqlAlchemyAsyncUnitOfWork, SqlAlchemyUnitOfWork
 
 T = TypeVar("T")
 _BUSY_SITE_RETRY_SECONDS = 10
@@ -68,12 +70,42 @@ def _task(record: SubscriptionSearchTask) -> SearchTaskSnapshot:
     )
 
 
-class TransactionalSubscriptionSearchRepository:
-    """使用短事务实现订阅搜索队列端口。"""
+def _enqueue_result(
+    repository: SubscriptionSearchOper,
+    *,
+    subscription_ids: tuple[int, ...],
+    source: str,
+    priority: int,
+    available_at_by_subscription: Optional[Mapping[int, str]],
+) -> SearchEnqueueResult:
+    """在当前事务中创建搜索批次并投影返回结果。"""
+    record, created, coalesced, active_batch_ids = repository.enqueue(
+        subscription_ids=subscription_ids,
+        source=source,
+        priority=priority,
+        available_at_by_subscription=available_at_by_subscription,
+    )
+    return SearchEnqueueResult(
+        batch=_batch(record),
+        created_count=created,
+        coalesced_count=coalesced,
+        active_batch_ids=active_batch_ids,
+    )
 
-    def __init__(self, session_factory: Callable[[], Session]) -> None:
-        """保存由组合根注入的同步 Session 工厂。"""
+
+class TransactionalSubscriptionSearchRepository:
+    """使用同步或异步短事务实现订阅搜索队列端口。"""
+
+    def __init__(
+        self,
+        session_factory: Callable[[], Session],
+        async_session_factory: Optional[
+            Callable[[], AbstractAsyncContextManager[AsyncSession]]
+        ] = None,
+    ) -> None:
+        """保存由组合根注入的同步和异步 Session 工厂。"""
         self._session_factory = session_factory
+        self._async_session_factory = async_session_factory
 
     def _read(self, operation: Callable[[SubscriptionSearchOper], T]) -> T:
         """在短 Session 中执行一次只读查询。"""
@@ -92,6 +124,22 @@ class TransactionalSubscriptionSearchRepository:
                 unit_of_work.rollback()
                 raise
 
+    async def _async_write(self, operation: Callable[[SubscriptionSearchOper], T]) -> T:
+        """在短异步事务中执行一次队列状态变更。"""
+        if self._async_session_factory is None:
+            raise RuntimeError("订阅搜索异步写入尚未配置")
+        async with self._async_session_factory() as session:
+            unit_of_work = SqlAlchemyAsyncUnitOfWork(session)
+            try:
+                result = await session.run_sync(
+                    lambda sync_session: operation(SubscriptionSearchOper(sync_session))
+                )
+                await unit_of_work.commit()
+                return result
+            except Exception:
+                await unit_of_work.rollback()
+                raise
+
     def enqueue(
         self,
         *,
@@ -101,22 +149,34 @@ class TransactionalSubscriptionSearchRepository:
         available_at_by_subscription: Optional[Mapping[int, str]] = None,
     ) -> SearchEnqueueResult:
         """创建批次并返回 single-flight 合并计数。"""
-        def operation(repository: SubscriptionSearchOper) -> SearchEnqueueResult:
-            """在同一事务内创建批次和任务。"""
-            record, created, coalesced, active_batch_ids = repository.enqueue(
+        return self._write(
+            lambda repository: _enqueue_result(
+                repository,
                 subscription_ids=subscription_ids,
                 source=source,
                 priority=priority,
                 available_at_by_subscription=available_at_by_subscription,
             )
-            return SearchEnqueueResult(
-                batch=_batch(record),
-                created_count=created,
-                coalesced_count=coalesced,
-                active_batch_ids=active_batch_ids,
-            )
+        )
 
-        return self._write(operation)
+    async def async_enqueue(
+        self,
+        *,
+        subscription_ids: tuple[int, ...],
+        source: str,
+        priority: int,
+        available_at_by_subscription: Optional[Mapping[int, str]] = None,
+    ) -> SearchEnqueueResult:
+        """在短异步事务中创建批次并返回合并计数。"""
+        return await self._async_write(
+            lambda repository: _enqueue_result(
+                repository,
+                subscription_ids=subscription_ids,
+                source=source,
+                priority=priority,
+                available_at_by_subscription=available_at_by_subscription,
+            )
+        )
 
     def claim_next(self, *, owner: str, lease_seconds: int = 900) -> Optional[SearchTaskSnapshot]:
         """认领下一任务并返回脱离 Session 的快照。"""

app/startup/composition/runtime.py

@@ -139,7 +139,10 @@ def compose_runtime_dependencies() -> RuntimeDependencies:
         transfer_execution=TransactionalTransferExecutionRepository(SessionFactory),
         message_helper=message_helper_factory(),
         message_queue=MessageQueueManager(auto_start=False),
-        subscription_search=TransactionalSubscriptionSearchRepository(SessionFactory),
+        subscription_search=TransactionalSubscriptionSearchRepository(
+            SessionFactory,
+            async_session_scope,
+        ),
     )
 
 

docs/architecture-overview.md

@@ -754,8 +754,8 @@ flowchart LR
 
 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 972 |
-| 内部导入边 | 8,208 |
+| Python 模块 | 973 |
+| 内部导入边 | 8,220 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 53（债务已清零，53 条精确 containment） |

docs/architecture/optimization-checklist.md

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 972 / 8,208 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 973 / 8,220 | `dependency-baseline.json` 当前快照 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8208,
-  "edge_sha256": "1a7eee327467c99451ce338d7d2670b74c2e71aeefcf36ac27b6668cb30a71aa",
+  "edge_count": 8220,
+  "edge_sha256": "4ef4fb9843d0e3a1d356f297ff2dafca35794b7811b7574d64bda71e114f2418",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -4779,7 +4779,6 @@
     "app.chain.subscribe.notify -> app.domain.meta",
     "app.chain.subscribe.notify -> app.domain.meta.metabase",
     "app.chain.subscribe.notify -> app.runtime",
-    "app.chain.subscribe.notify -> app.runtime.execution",
     "app.chain.subscribe.notify -> app.runtime.log",
     "app.chain.subscribe.notify -> app.schemas",
     "app.chain.subscribe.notify -> app.schemas.common",
@@ -4879,6 +4878,7 @@
     "app.chain.subscribe.search -> app.chain.subscribe.contract",
     "app.chain.subscribe.search -> app.chain.subscribe.identity",
     "app.chain.subscribe.search -> app.chain.subscribe.metadata",
+    "app.chain.subscribe.search -> app.chain.subscribe.searchtask",
     "app.chain.subscribe.search -> app.domain",
     "app.chain.subscribe.search -> app.domain.context",
     "app.chain.subscribe.search -> app.domain.meta",
@@ -4888,6 +4888,18 @@
     "app.chain.subscribe.search -> app.runtime.stop",
     "app.chain.subscribe.search -> app.schemas",
     "app.chain.subscribe.search -> app.schemas.types",
+    "app.chain.subscribe.searchtask -> app.application",
+    "app.chain.subscribe.searchtask -> app.application.subscription",
+    "app.chain.subscribe.searchtask -> app.application.subscription.contract",
+    "app.chain.subscribe.searchtask -> app.application.subscription.execution",
+    "app.chain.subscribe.searchtask -> app.application.subscription.observability",
+    "app.chain.subscribe.searchtask -> app.application.subscription.sitebudget",
+    "app.chain.subscribe.searchtask -> app.chain",
+    "app.chain.subscribe.searchtask -> app.chain.search",
+    "app.chain.subscribe.searchtask -> app.chain.search.facade",
+    "app.chain.subscribe.searchtask -> app.runtime",
+    "app.chain.subscribe.searchtask -> app.runtime.log",
+    "app.chain.subscribe.searchtask -> app.runtime.stop",
     "app.chain.system -> app.application",
     "app.chain.system -> app.application.configuration",
     "app.chain.system -> app.chain",
@@ -9301,7 +9313,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 972,
+  "module_count": 973,
   "modules": [
     "app",
     "app.adapters",
@@ -9712,6 +9724,7 @@
     "app.chain.subscribe.reconcile",
     "app.chain.subscribe.refresh",
     "app.chain.subscribe.search",
+    "app.chain.subscribe.searchtask",
     "app.chain.system",
     "app.chain.theaudiodb",
     "app.chain.tmdb",

tests/test_subscription_search_queue.py

@@ -1,8 +1,11 @@
 """订阅搜索持久队列、single-flight、租约和取消测试。"""
 
+from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
 
+import pytest
 from sqlalchemy import create_engine, select, update
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlalchemy.orm import Session, sessionmaker
 
 from app.db.adapters.subscriptionsearch import TransactionalSubscriptionSearchRepository
@@ -17,6 +20,43 @@ def _repository(tmp_path):
     return TransactionalSubscriptionSearchRepository(sessionmaker(bind=engine)), engine
 
 
+@pytest.mark.asyncio
+async def test_search_queue_async_enqueue_uses_async_session(tmp_path):
+    """异步新增订阅应通过 AsyncSession 入队，并可由同步消费者继续认领。"""
+    database_path = tmp_path / "async-search-queue.db"
+    async_engine = create_async_engine(f"sqlite+aiosqlite:///{database_path}")
+    async_factory = async_sessionmaker(bind=async_engine, expire_on_commit=False)
+
+    @asynccontextmanager
+    async def async_session_scope():
+        """为测试队列提供独立异步会话。"""
+        async with async_factory() as session:
+            yield session
+
+    async with async_engine.begin() as connection:
+        await connection.run_sync(Base.metadata.create_all)
+    sync_engine = create_engine(f"sqlite:///{database_path}")
+    repository = TransactionalSubscriptionSearchRepository(
+        sessionmaker(bind=sync_engine),
+        async_session_scope,
+    )
+
+    try:
+        enqueued = await repository.async_enqueue(
+            subscription_ids=(101,),
+            source="new",
+            priority=50,
+        )
+        claimed = repository.claim_next(owner="worker-async")
+
+        assert enqueued.created_count == 1
+        assert claimed is not None
+        assert claimed.subscription_id == 101
+    finally:
+        sync_engine.dispose()
+        await async_engine.dispose()
+
+
 def test_search_queue_coalesces_active_subscription_and_raises_priority(tmp_path):
     """重叠入口只保留一个活动任务，手工请求可提高优先级。"""
     repository, _engine = _repository(tmp_path)