diff --git a/app/api/controllers/risk.py b/app/api/controllers/risk.py
index 18aec6a..025c97f 100644
--- a/app/api/controllers/risk.py
+++ b/app/api/controllers/risk.py
@@ -23,13 +23,14 @@ from app.api.schemas.risk import (
     RiskNotificationPageQuery,
 )
 from app.core.contracts import RequestContext
+from app.infrastructure.db import mysql_scan_lock
 from app.service.risk_action_service import RiskActionService
 from app.service.risk_daily_report_mail_service import RiskDailyReportMailService
 from app.service.risk_daily_report_service import RiskDailyReportService
 from app.service.risk_evidence_archive_service import RiskEvidenceArchiveService
 from app.service.risk_notification_service import RiskNotificationService
 from app.service.risk_query_service import RiskQueryService
-from app.service.risk_scan_service import RiskScanService
+from app.service.risk_scan_service import RiskScanBusyError, RiskScanService
 
 router = APIRouter(
     prefix="/api/v1/risk",
@@ -62,7 +63,14 @@ async def scan_risk_alerts(
     context: RequestContext = Depends(build_request_context),  # noqa: B008
     session: AsyncSession = Depends(get_session),  # noqa: B008
 ) -> dict[str, object]:
-    data = await RiskScanService(session).scan(context)
+    # 手工触发的扫描必须与定时扫描互斥，否则两条路径会同时查不到重复、同时插入。
+    # 锁加在**入口层**而不是 `RiskScanService.scan()` 内部：`GET_LOCK` 是连接级的，
+    # 而调度器已在它自己的 session 上持锁 —— 被两个入口共用的服务方法若再取同一把锁，
+    # 取锁的连接不是持锁的那一个、必然失败，会**把定时扫描自己挡死**。
+    async with mysql_scan_lock() as acquired:
+        if not acquired:
+            raise RiskScanBusyError("规则扫描正在执行，请稍后重试")
+        data = await RiskScanService(session).scan(context)
     return _envelope(data, context)
 
 
diff --git a/app/infrastructure/db.py b/app/infrastructure/db.py
index bc60deb..4201a31 100644
--- a/app/infrastructure/db.py
+++ b/app/infrastructure/db.py
@@ -1,3 +1,7 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from sqlalchemy import text
 from sqlalchemy.engine import make_url
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
@@ -29,3 +33,36 @@ if "init_command" not in _url.query:
 
 engine = create_async_engine(_url, pool_pre_ping=True)
 SessionFactory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+# 规则扫描的**跨进程**咨询锁名。用 MySQL 连接级 `GET_LOCK` 而不是进程内 `asyncio.Lock`：
+# 后者只在单个进程内互斥，多 Web worker、或 Worker 与 API 同时运行时形同虚设。
+SCAN_LOCK_NAME = "jr_risk_scan_schedule"
+
+
+@asynccontextmanager
+async def mysql_scan_lock() -> AsyncIterator[bool]:
+    """约束规则扫描的跨进程并发；yield 是否取得锁。
+
+    **为什么两个入口都必须用它**：定时扫描（`worker/risk_scan_scheduler.py`）与手工触发的
+    HTTP 端点（`POST /api/v1/risk/alerts/scan`）最终都调 `RiskScanService.scan()`，
+    而扫描的幂等只有应用层的 `_exists` 查重 —— 两条路径并发时会同时查不到、同时插入，
+    产生重复预警。锁要**共用同一个名字**才能互相排斥。
+
+    **为什么不放在 `scan()` 内部**：`GET_LOCK` 是**连接级**的。调度器已在自己的 session 上
+    持锁，再新建 session 去调 `scan()`；若 `scan()` 又去取同一把锁，取锁的连接并不是持锁的
+    那一个，`GET_LOCK` 返回 0 —— **调度器会把自己挡死**。所以锁加在**入口层**
+    （调度器 / 端点各取一次），而不是被两个入口共用的服务方法里。
+
+    `GET_LOCK(name, 0)` 的 0 表示不等待、立即返回：取不到就跳过本轮，而不是排队。
+    """
+    async with SessionFactory() as session:
+        acquired = bool(
+            await session.scalar(text("SELECT GET_LOCK(:name, 0)"), {"name": SCAN_LOCK_NAME})
+        )
+        try:
+            yield acquired
+        finally:
+            if acquired:
+                await session.scalar(
+                    text("SELECT RELEASE_LOCK(:name)"), {"name": SCAN_LOCK_NAME}
+                )
diff --git a/app/worker/risk_scan_scheduler.py b/app/worker/risk_scan_scheduler.py
index d21c6dd..9c5a4f1 100644
--- a/app/worker/risk_scan_scheduler.py
+++ b/app/worker/risk_scan_scheduler.py
@@ -9,17 +9,15 @@ from __future__ import annotations
 import argparse
 import asyncio
 import logging
-from collections.abc import AsyncIterator, Awaitable, Callable
-from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from collections.abc import Awaitable, Callable
+from contextlib import AbstractAsyncContextManager
 from datetime import UTC, datetime, timedelta
 from typing import Any
 from uuid import uuid4
 
-from sqlalchemy import text
-
 from app.core.config import get_settings
 from app.core.contracts import RequestContext
-from app.infrastructure.db import SessionFactory, engine
+from app.infrastructure.db import SessionFactory, engine, mysql_scan_lock
 from app.model.audit import InteractionAudit
 from app.service.risk_scan_schedule_config import (
     RiskScanScheduleConfig,
@@ -27,7 +25,6 @@ from app.service.risk_scan_schedule_config import (
 )
 from app.service.risk_scan_service import RiskScanService
 
-SCAN_LOCK_NAME = "jr_risk_scan_schedule"
 logger = logging.getLogger(__name__)
 
 ConfigLoader = Callable[[], Awaitable[RiskScanScheduleConfig]]
@@ -35,25 +32,8 @@ ScanExecutor = Callable[[], Awaitable[dict[str, int | str]]]
 AuditWriter = Callable[[str, dict[str, Any]], Awaitable[None]]
 LockFactory = Callable[[], AbstractAsyncContextManager[bool]]
 
-
-@asynccontextmanager
-async def mysql_scan_lock() -> AsyncIterator[bool]:
-    """使用 MySQL 连接级咨询锁约束跨进程并发。"""
-    async with SessionFactory() as session:
-        acquired = bool(
-            await session.scalar(
-                text("SELECT GET_LOCK(:name, 0)"),
-                {"name": SCAN_LOCK_NAME},
-            )
-        )
-        try:
-            yield acquired
-        finally:
-            if acquired:
-                await session.scalar(
-                    text("SELECT RELEASE_LOCK(:name)"),
-                    {"name": SCAN_LOCK_NAME},
-                )
+# 跨进程扫描锁已移到 `app/infrastructure/db.py`：端点与调度器**必须共用同一把锁**，
+# 放在基础设施层两个入口才都能引用（service 不该反向依赖 worker）。
 
 
 async def default_scan_executor() -> dict[str, int | str]: