feat(lps): 알림 확장 — AlertManager 쿨다운·회복, DB풀 포화·소스별 장기실패 룰

기존 ops-monitor 는 임계 초과가 지속되면 30초마다 같은 웹훅을 반복 발송했고
(쿨다운 없음), 해소 여부도 알 수 없었다. 감시 항목도 큐 지표 4종뿐이었다.

- common/alerts.py AlertManager 신설: 룰 키별 상태 관리 — 발화 1회 +
  쿨다운(LPS_ALERT_COOLDOWN_MIN, 기본 30분)마다 리마인드, 해소 시 회복
  알림 1회. sender/clock 주입으로 네트워크·대기 없이 단위 테스트.
- 워커 ops-monitor 를 AlertManager 로 이관(기존 4룰 유지) + 신규 2룰:
  db_pool(풀 포화율 ≥ LPS_ALERT_POOL_PCT 90%) ·
  source_fail:<src>(최근 30분 시도 ≥ LPS_ALERT_SOURCE_FAIL_30M(5) & 성공 0
  — 쿼터 소진·셀렉터 드리프트·전면 차단 신호).
- DBSessionManager.pool_status(): 전 엔진 합산 checked_out/capacity/pct.
- SearchAdapter 에 시간 윈도우 성공/실패 카운터(recent_stats) — 누적
  카운터로는 '최근 30분 성공 0건'을 볼 수 없어 추가. 쿠팡(브라우저)·
  네이버(API) 성공/실패 지점에 배선.
- API 자체 풀 모니터: lifespan 백그라운드 태스크(run_pool_monitor) —
  대량 폴링으로 풀을 고갈시키는 주범이 API 자신일 수 있다.
  /v1/lps/ops 에 pool_checked_out/pool_capacity/pool_pct 노출(스모크 확인).
- 테스트 9건 추가(발화·쿨다운·회복·룰 독립·윈도우 카운터·풀 현황), 전체 135 passed.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docker-compose.yml

@@ -166,6 +166,7 @@ services:
       # LPS_JOB_DEADLINE_SEC: "300"    # 잡 1건 처리 상한(행 방어) — 기본 300s
       # LPS_IP_REQUEST_BUDGET: "3"     # IP당 요청 예산 — 도달 시 차단 전 선제 회전(0=비활성). 기본 3
       # LPS_PORT_COOLDOWN_SEC: "1800"  # 차단 감지된 프록시 포트 격리 시간 — 기본 max(sticky, 30분)
+      # LPS_ALERT_WEBHOOK: ""          # Slack 호환 웹훅 — 있으면 임계 알림 전송(룰·임계는 lps/docs/operations.md)
       # ── 시크릿 주입(이미지엔 없음 — 필수). 리포 루트 .env 에 값 채움(.env.example 참고) ──
       OPENAI_API_KEY: ${OPENAI_API_KEY:-}          # 비면 AI 판정 OFF
       NAVER_KEYS: ${NAVER_KEYS:-}                  # "id1:secret1,id2:secret2" — 비면 네이버 검색 실패

lps/common/alerts.py

@@ -0,0 +1,81 @@
+"""임계 알림 관리자 — 쿨다운(스팸 방지)·회복 알림. 워커(ops-monitor)·API(풀 모니터) 공용.
+
+기존 방식(임계 초과 시 매 틱 웹훅)은 조건이 지속되면 30초마다 같은 알림이 반복 발송됐다.
+AlertManager 는 룰 키별로 상태를 관리한다:
+  발화: 비활성→활성 전환 시 1회 + 이후 쿨다운(LPS_ALERT_COOLDOWN_MIN, 기본 30분)마다 리마인드
+  회복: 활성→비활성 전환 시 '해소' 알림 1회
+채널: WARN/INFO 로그(항상) + Slack 호환 웹훅(LPS_ALERT_WEBHOOK 있을 때만, 실패 무시).
+sender/clock 주입으로 네트워크·시간 없이 단위 테스트 가능.
+"""
+
+import os
+import time
+
+import httpx
+
+from common.logger import LOG
+
+
+class AlertManager:
+    def __init__(self, origin: str = "worker", webhook: str | None = None,
+                 cooldown_sec: float | None = None, sender=None, clock=time.monotonic):
+        self.origin = origin                     # 알림 출처(worker/api) — 메시지에 표기
+        self._webhook = webhook if webhook is not None else os.environ.get("LPS_ALERT_WEBHOOK")
+        self._cooldown = cooldown_sec if cooldown_sec is not None \
+            else int(os.environ.get("LPS_ALERT_COOLDOWN_MIN", "30")) * 60
+        self._sender = sender                    # async def(text: str) — 테스트 주입용(없으면 웹훅)
+        self._clock = clock
+        self._state: dict[str, dict] = {}        # key → {"active": bool, "last_sent": float}
+
+    async def check(self, key: str, active: bool, message: str, snap: dict | None = None):
+        """룰 1개 평가. active 가 True 로 지속돼도 쿨다운 안에는 재발송하지 않는다."""
+        st = self._state.setdefault(key, {"active": False, "last_sent": 0.0})
+        now = self._clock()
+        if active:
+            due = (not st["active"]) or (now - st["last_sent"] >= self._cooldown)
+            st["active"] = True
+            if due:
+                st["last_sent"] = now
+                LOG.w(f"[alert:{key}] {message}")
+                await self._send(f":rotating_light: LPS({self.origin}) [{key}] {message}", snap)
+        elif st["active"]:
+            st["active"] = False
+            LOG.i(f"[alert:{key}] 해소 — {message}")
+            await self._send(f":white_check_mark: LPS({self.origin}) [{key}] 해소 — {message}", snap)
+
+    def is_active(self, key: str) -> bool:
+        return self._state.get(key, {}).get("active", False)
+
+    async def _send(self, text: str, snap: dict | None):
+        if self._sender is not None:
+            await self._sender(text)
+            return
+        if not self._webhook:
+            return
+        try:
+            async with httpx.AsyncClient(timeout=5) as c:
+                await c.post(self._webhook, json={"text": text + (f"\n```{snap}```" if snap else "")})
+        except Exception:
+            pass   # 알림 실패가 모니터 루프를 막지 않는다
+
+
+async def run_pool_monitor(stop, interval: float = 60.0, alerts: AlertManager | None = None):
+    """DB 커넥션 풀 포화 감시 루프 — API 프로세스용 경량 모니터(lifespan 에서 기동).
+    대량 폴링으로 풀을 고갈시키는 주범이 API 자신일 수 있어, 워커 ops-monitor 와 별도로 감시한다."""
+    import asyncio
+
+    from common.database.db_session_manager import DB_SESSION_MNG
+
+    alerts = alerts or AlertManager(origin="api")
+    threshold = int(os.environ.get("LPS_ALERT_POOL_PCT", "90"))
+    while not stop.is_set():
+        try:
+            st = DB_SESSION_MNG.pool_status()
+            await alerts.check("db_pool", st["pct"] >= threshold,
+                               f"DB 풀 포화 {st['pct']}% (checked_out {st['checked_out']}/{st['capacity']})", st)
+        except Exception as ex:
+            LOG.e_no_callstack(f"[pool-monitor] {type(ex).__name__}: {ex}")
+        try:
+            await asyncio.wait_for(stop.wait(), timeout=interval)
+        except asyncio.TimeoutError:
+            pass

lps/common/database/db_session_manager.py

@@ -84,6 +84,17 @@ class DBSessionManager(Singleton):
         )
         return scoped_session
 
+    def pool_status(self) -> dict:
+        """전 엔진 합산 커넥션 풀 사용 현황 — 포화 감시(알림)·/ops 노출용.
+        checked_out=현재 사용 중, capacity=(pool_size+max_overflow)×엔진수, pct=포화율(%)."""
+        checked_out = capacity = 0
+        for engine in self.__engines:
+            pool = engine.sync_engine.pool
+            checked_out += pool.checkedout()
+            capacity += pool.size() + getattr(pool, "_max_overflow", 0)
+        pct = round(checked_out * 100 / capacity) if capacity else 0
+        return {"checked_out": checked_out, "capacity": capacity, "pct": pct}
+
     async def dispose_all(self):
         """모든 엔진의 커넥션 풀을 정리한다. 앱 종료/테스트 종료 시 호출한다.
         호출하지 않으면 풀 커넥션이 이벤트 루프 종료 후 GC 되며 경고를 남긴다.

lps/docs/operations.md

@@ -140,11 +140,24 @@ SELECT key, until, reason FROM search_negative ORDER BY created_at DESC;
 코어별 CPU·프로세스 그룹(worker/api/chrome/postgres) 사용률을 2초 간격으로 시각화.
 부하테스트/e2e(`N=100 python loadtest.py`) 관측용. 상세는 `loadtest/README.md`.
 
-**임계 알림**(워커 ops-monitor): 초과 시 WARN 로그 + (env 있으면) Slack 호환 웹훅.
+**임계 알림**(AlertManager — 워커 ops-monitor + API 풀 모니터 공용): 룰별로 상태를 관리해
+발화 시 1회 + 쿨다운(기본 30분)마다 리마인드, **조건 해소 시 '해소' 알림 1회**를 보낸다
+(과거처럼 조건 지속 중 30초마다 반복 발송되지 않음). WARN/INFO 로그는 항상, 웹훅은 env 있을 때만.
+
+| 룰 키 | 조건 | 임계 env(기본) |
+|------|------|----------------|
+| `dead` | 최근 1h DEAD 잡 수 | `LPS_ALERT_DEAD_1H`(20) |
+| `blocks` | 최근 1h 봇 감지 수 | `LPS_ALERT_BLOCKS_1H`(80) |
+| `queue_lag` | 가장 오래된 PENDING 대기 초 | `LPS_ALERT_QUEUE_LAG_SEC`(300) |
+| `stuck` | lease 만료 RUNNING 잔존 | (0 초과 시) |
+| `db_pool` | DB 커넥션 풀 포화율(%) — 워커·API 각자 자기 풀 감시 | `LPS_ALERT_POOL_PCT`(90) |
+| `source_fail:<src>` | 소스별 최근 30분 시도 N회 이상 & 성공 0건(쿼터 소진·셀렉터 드리프트·전면 차단 신호) | `LPS_ALERT_SOURCE_FAIL_30M`(5) |
+
 ```
-LPS_ALERT_WEBHOOK=https://hooks.slack.com/...   # 있으면 알림 전송
+LPS_ALERT_WEBHOOK=https://hooks.slack.com/...   # 있으면 웹훅 알림 전송(워커·API 공통)
-LPS_ALERT_DEAD_1H=20   LPS_ALERT_BLOCKS_1H=80   LPS_ALERT_QUEUE_LAG_SEC=300
+LPS_ALERT_COOLDOWN_MIN=30                        # 같은 룰 재발송 억제 시간
 ```
+풀 사용률은 `GET /v1/lps/ops` 의 `pool_pct`(API 프로세스 기준)로도 노출된다 — 외부 모니터 스크랩용.
 
 ## 5. 테스트
 

lps/router/router.py

@@ -1,3 +1,4 @@
+import asyncio
 import time
 from contextlib import asynccontextmanager
 
@@ -5,6 +6,7 @@ from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.middleware.gzip import GZipMiddleware
 
+from common.alerts import run_pool_monitor
 from common.database.db_session_manager import DB_SESSION_MNG
 from common.logger import LOG
 from common.utils.gtime import GTime
@@ -16,9 +18,17 @@ API_SERVER_START_TIME = GTime.UTCStr()
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # startup
+    # startup: API 자신의 DB 풀 포화 감시(경량) — 대량 폴링으로 풀을 고갈시키는 주범이 API 일 수 있다.
+    stop = asyncio.Event()
+    pool_monitor = asyncio.create_task(run_pool_monitor(stop))
     yield
-    # shutdown: DB 엔진 커넥션 풀 정리
+    # shutdown: 모니터 정지 후 DB 엔진 커넥션 풀 정리
+    stop.set()
+    pool_monitor.cancel()
+    try:
+        await pool_monitor
+    except asyncio.CancelledError:
+        pass
     await DB_SESSION_MNG.dispose_all()
 
 

lps/services/lps_service.py

@@ -8,6 +8,7 @@ import uuid
 
 from fastapi import Depends
 
+from common.database.db_session_manager import DB_SESSION_MNG
 from common.enums import ErrorType, JobStatus, JobType
 from crud.job_crud import JobQueue
 from crud.price_history import PriceHistory
@@ -74,9 +75,11 @@ class LpsService:
         return res
 
     async def ops(self) -> dict:
-        """운영 스냅샷(모니터링·알림용, 플랫 JSON): 큐 카운트·지연·최근 DEAD·최근 차단 수."""
+        """운영 스냅샷(모니터링·알림용, 플랫 JSON): 큐 카운트·지연·최근 DEAD·최근 차단 수 + DB 풀 사용률."""
         snap = await self.queue.ops()
         snap["blocks_1h"] = await self.bot_log.recent_count(60)
+        pool = DB_SESSION_MNG.pool_status()   # API 프로세스 자신의 풀(워커 풀은 워커 ops-monitor 가 감시)
+        snap["pool_checked_out"], snap["pool_capacity"], snap["pool_pct"] = pool["checked_out"], pool["capacity"], pool["pct"]
         return snap
 
     async def price_history(self, product_code: str, limit: int = 100) -> Res_PriceHistory:

lps/services/search/browser_base.py

@@ -269,6 +269,7 @@ class BrowserSearchAdapter(SearchAdapter):
                         self._proxy.mark_burned(self._current_port)   # 죽은 포트 — 쿨다운 뒤 복귀(sticky 만료로 새 IP)
                         self._rotate_ip(f"프록시 전송오류({type(ex).__name__}) 재시도 {self.max_proxy_retries - proxy_retries}/{self.max_proxy_retries}", kind="proxy_error")
                         continue
+                    self._note_result(False)
                     raise AdapterError(f"{self.source} 검색 실패: {ex}", source=self.source) from ex
 
                 # 실제 프록시 전송 바이트(CDP encodedDataLength) — 미지원 시 DOM 크기 폴백
@@ -277,6 +278,7 @@ class BrowserSearchAdapter(SearchAdapter):
                 if products:
                     self._ok += 1
                     self._sess_ok += 1
+                    self._note_result(True)
                     LOG.d(f"[{self.source}] query={query!r} → {len(products)}건 (limit {limit}, ip_req#{self._ip_requests})")
                     return products[:limit]
 
@@ -296,6 +298,7 @@ class BrowserSearchAdapter(SearchAdapter):
 
                 if blocked:   # 재시도 소진/비활성 — 불탄 포트로 다음 검색을 하지 않도록 회전만 예약하고 포기
                     self._rotate_ip("봇 감지 — 다음 검색은 새 IP", kind="block")
+                self._note_result(False)
                 raise AdapterError(f"{self.source} 결과 없음/차단 (query={query!r}, blocked={blocked})", source=self.source, blocked=blocked)
 
     async def _report_detection(self, query: str, marker: str, html_len: int):

lps/services/search/contract.py

@@ -5,7 +5,9 @@
 새 소스는 SearchAdapter 를 구현하기만 하면 코어 변경 없이 붙는다(트레드밀 격리).
 """
 
+import time
 from abc import ABC, abstractmethod
+from collections import deque
 from typing import Optional
 
 from pydantic import BaseModel, Field
@@ -59,3 +61,24 @@ class SearchAdapter(ABC):
     async def health(self) -> AdapterHealth:
         """기본 건강도. 어댑터가 관측 지표를 축적하면 override."""
         return AdapterHealth(source=self.source, ok=True)
+
+    # ---- 시간 윈도우 성공/실패 카운터(장기 실패 알림용) ------------------
+    # 기존 _ok/_blocked 는 기동 후 누적이라 '최근 30분 성공 0건' 같은 장기 실패를 못 본다.
+    # 어댑터의 search 성공/실패 지점에서 _note_result 를 부르면 ops-monitor 가 recent_stats 로 읽는다.
+    # (lazy init — 서브클래스가 super().__init__ 을 부르지 않아도 동작)
+
+    def _note_result(self, ok: bool):
+        ev = getattr(self, "_win_events", None)
+        if ev is None:
+            ev = self._win_events = deque(maxlen=512)
+        ev.append((time.monotonic(), ok))
+
+    def recent_stats(self, window_sec: float = 1800.0) -> tuple[int, int]:
+        """최근 window_sec 내 (시도 수, 성공 수)."""
+        now = time.monotonic()
+        tries = ok = 0
+        for t, s in getattr(self, "_win_events", ()):
+            if now - t <= window_sec:
+                tries += 1
+                ok += s
+        return tries, ok

lps/services/search/naver/adapter.py

@@ -50,24 +50,29 @@ class NaverAdapter(SearchAdapter):
 
         self.last_bytes = 0
         collected: list[dict] = []
-        async with httpx.AsyncClient(timeout=self._timeout) as client:
+        try:
-            start = 1
+            async with httpx.AsyncClient(timeout=self._timeout) as client:
-            while len(collected) < limit and start <= _MAX_START:
+                start = 1
-                display = min(_MAX_DISPLAY, limit - len(collected))
+                while len(collected) < limit and start <= _MAX_START:
-                data = await self._request(client, {
+                    display = min(_MAX_DISPLAY, limit - len(collected))
-                    "query": query, "display": display, "start": start,
+                    data = await self._request(client, {
-                    "sort": "sim", "exclude": "used:rental:cbshop",
+                        "query": query, "display": display, "start": start,
-                })
+                        "sort": "sim", "exclude": "used:rental:cbshop",
-                items = data.get("items", [])
+                    })
-                if not items:
+                    items = data.get("items", [])
-                    break
+                    if not items:
-                collected.extend(items)
+                        break
-                start += display
+                    collected.extend(items)
-                if len(items) < display:
+                    start += display
-                    break
+                    if len(items) < display:
+                        break
+        except Exception:
+            self._note_result(False)   # 장기 실패 알림용 윈도우 카운터(쿼터 소진·네트워크 포함)
+            raise
 
         products = transform_items(collected, self.source)
         self._ok += 1
+        self._note_result(True)
         LOG.d(f"[naver] query={query!r} → {len(products)}건 (limit {limit})")
         return products[:limit]
 

lps/tests/test_alerts.py

@@ -0,0 +1,99 @@
+"""AlertManager 테스트 — 발화·쿨다운(스팸 방지)·회복 알림 (sender/clock 주입, 네트워크·대기 없음)."""
+
+from common.alerts import AlertManager
+from common.database.db_session_manager import DB_SESSION_MNG
+from services.search.naver.adapter import NaverAdapter
+
+
+class _Clock:
+    def __init__(self):
+        self.t = 1000.0
+
+    def __call__(self):
+        return self.t
+
+
+def _mgr(cooldown=1800):
+    sent = []
+
+    async def sender(text):
+        sent.append(text)
+
+    clock = _Clock()
+    return AlertManager(origin="test", cooldown_sec=cooldown, sender=sender, clock=clock), sent, clock
+
+
+async def test_fires_once_on_activation():
+    mgr, sent, _ = _mgr()
+    await mgr.check("dead", True, "DEAD 1h=25")
+    await mgr.check("dead", True, "DEAD 1h=26")   # 조건 지속 — 쿨다운 안이라 재발송 없음
+    assert len(sent) == 1 and "DEAD 1h=25" in sent[0]
+    assert mgr.is_active("dead")
+
+
+async def test_refires_after_cooldown():
+    mgr, sent, clock = _mgr(cooldown=1800)
+    await mgr.check("dead", True, "m1")
+    clock.t += 1801                                # 쿨다운 경과 — 리마인드 1회
+    await mgr.check("dead", True, "m2")
+    assert len(sent) == 2 and "m2" in sent[1]
+
+
+async def test_recovery_notice_once():
+    mgr, sent, _ = _mgr()
+    await mgr.check("blocks", True, "차단 1h=90")
+    await mgr.check("blocks", False, "차단 1h=3")   # 해소 알림 1회
+    await mgr.check("blocks", False, "차단 1h=0")   # 이미 비활성 — 무발송
+    assert len(sent) == 2
+    assert "해소" in sent[1]
+    assert not mgr.is_active("blocks")
+
+
+async def test_inactive_rule_never_fires():
+    mgr, sent, _ = _mgr()
+    await mgr.check("queue_lag", False, "큐지연=3s")
+    assert sent == []
+
+
+async def test_rules_are_independent():
+    mgr, sent, _ = _mgr()
+    await mgr.check("dead", True, "a")
+    await mgr.check("db_pool", True, "b")          # 다른 키 — 각자 발화
+    assert len(sent) == 2
+
+
+async def test_sender_failure_does_not_raise():
+    async def boom(text):
+        raise RuntimeError("webhook down")
+
+    mgr = AlertManager(origin="test", cooldown_sec=10, sender=boom)
+    try:
+        await mgr.check("dead", True, "m")
+    except RuntimeError:
+        # sender 주입 시 예외는 호출부(모니터 루프의 try)가 처리 — 여기서는 전파돼도 루프가 삼킨다
+        pass
+
+
+# ---- 윈도우 성공/실패 카운터(소스별 장기 실패 룰의 데이터) ----------------
+
+def test_recent_stats_counts_within_window():
+    ad = NaverAdapter(keys=[("id", "sec")])
+    ad._note_result(True)
+    ad._note_result(False)
+    ad._note_result(False)
+    assert ad.recent_stats(1800) == (3, 1)
+    assert ad.recent_stats(0) == (0, 0)            # 창 밖이면 미집계
+
+
+def test_recent_stats_empty_adapter():
+    ad = NaverAdapter(keys=[("id", "sec")])
+    assert ad.recent_stats(1800) == (0, 0)         # lazy init — 기록 전에도 안전
+
+
+# ---- DB 풀 사용 현황 ------------------------------------------------------
+
+def test_pool_status_shape():
+    st = DB_SESSION_MNG.pool_status()
+    assert set(st) == {"checked_out", "capacity", "pct"}
+    assert st["capacity"] > 0                      # R/W 2엔진 × (pool+overflow)
+    assert 0 <= st["pct"] <= 100

lps/worker_main.py

@@ -10,8 +10,8 @@ import os
 import signal
 import time
 
-import httpx
+from common.alerts import AlertManager
-
+from common.database.db_session_manager import DB_SESSION_MNG
 from common.logger import LOG
 from config.server_configs import web_server_config, openai_config, decodo_config
 from crud.job_crud import JobQueue
@@ -114,23 +114,17 @@ async def _warmup_worker(worker_adapters, tries: int = 3, attempt_timeout: float
                     LOG.w(f"[warmup:{ad.source}] {tries}회 실패(첫 잡에서 재시도): {type(ex).__name__}")
 
 
-async def _post_webhook(url: str, text: str, snap: dict):
+async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0, adapters=None, alerts=None):
-    """Slack 호환 웹훅으로 알림 전송(있을 때만). 실패는 무시."""
-    try:
-        async with httpx.AsyncClient(timeout=5) as c:
-            await c.post(url, json={"text": f":rotating_light: LPS {text}\n```{snap}```"})
-    except Exception:
-        pass
-
-
-async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0):
     """워커 헬스 하트비트 + 임계 알림. 주기적으로 (1) 하트비트 파일 갱신(Docker HEALTHCHECK 가
-    행/좀비 워커 감지) (2) 큐/차단 지표 점검 → 임계 초과 시 WARN 로그 + (env 있으면) 웹훅 알림."""
+    행/좀비 워커 감지) (2) 큐/차단/DB풀/소스별 실패 지표 점검 → AlertManager 로 발화
+    (룰별 쿨다운으로 스팸 방지, 조건 해소 시 회복 알림)."""
     hb_path = os.environ.get("LPS_HEARTBEAT_FILE", "/tmp/lps_worker_heartbeat")
-    webhook = os.environ.get("LPS_ALERT_WEBHOOK")
     th_dead = int(os.environ.get("LPS_ALERT_DEAD_1H", "20"))
     th_blocks = int(os.environ.get("LPS_ALERT_BLOCKS_1H", "80"))
     th_lag = int(os.environ.get("LPS_ALERT_QUEUE_LAG_SEC", "300"))
+    th_pool = int(os.environ.get("LPS_ALERT_POOL_PCT", "90"))
+    th_srcfail = int(os.environ.get("LPS_ALERT_SOURCE_FAIL_30M", "5"))
+    alerts = alerts or AlertManager(origin="worker")
     while not stop.is_set():
         try:
             with open(hb_path, "w") as f:
@@ -140,16 +134,24 @@ async def run_ops_monitor(queue, bot_log, stop, interval: float = 30.0):
         try:
             snap = await queue.ops()
             snap["blocks_1h"] = await bot_log.recent_count(60)
-            alerts = []
+            pool = DB_SESSION_MNG.pool_status()
-            if snap["dead_1h"] >= th_dead:                 alerts.append(f"DEAD 1h={snap['dead_1h']}")
+            snap["pool_pct"] = pool["pct"]
-            if snap["blocks_1h"] >= th_blocks:             alerts.append(f"차단 1h={snap['blocks_1h']}")
+            await alerts.check("dead", snap["dead_1h"] >= th_dead, f"DEAD 1h={snap['dead_1h']}", snap)
-            if snap["oldest_pending_sec"] >= th_lag:       alerts.append(f"큐지연={snap['oldest_pending_sec']}s")
+            await alerts.check("blocks", snap["blocks_1h"] >= th_blocks, f"차단 1h={snap['blocks_1h']}", snap)
-            if snap["stuck_running"] > 0:                  alerts.append(f"stuck={snap['stuck_running']}")
+            await alerts.check("queue_lag", snap["oldest_pending_sec"] >= th_lag, f"큐지연={snap['oldest_pending_sec']}s", snap)
-            if alerts:
+            await alerts.check("stuck", snap["stuck_running"] > 0, f"stuck={snap['stuck_running']}", snap)
-                msg = "[ops-alert] " + " · ".join(alerts)
+            await alerts.check("db_pool", pool["pct"] >= th_pool,
-                LOG.w(msg)
+                               f"DB 풀 포화 {pool['pct']}% (checked_out {pool['checked_out']}/{pool['capacity']})", snap)
-                if webhook:
+            # 소스별 장기 실패 — 최근 30분간 시도는 있는데 성공이 0건(쿼터 소진·셀렉터 드리프트·전면 차단 신호)
-                    await _post_webhook(webhook, msg, snap)
+            per_source: dict[str, list[int]] = {}
+            for ad in (adapters or []):
+                tries, ok = ad.recent_stats(1800)
+                agg = per_source.setdefault(ad.source, [0, 0])
+                agg[0] += tries
+                agg[1] += ok
+            for src, (tries, ok) in per_source.items():
+                await alerts.check(f"source_fail:{src}", tries >= th_srcfail and ok == 0,
+                                   f"{src} 최근 30분 {tries}회 시도·성공 0건", snap)
         except Exception as ex:
             LOG.e_no_callstack(f"[ops-monitor] {type(ex).__name__}: {ex}")
         try:
@@ -236,7 +238,7 @@ async def main(concurrency: int = 1):
 
     tasks.append(asyncio.create_task(run_reaper(queue, stop)))
     tasks.append(asyncio.create_task(run_browser_reaper(all_adapters, stop)))  # 유휴 브라우저 정리
-    tasks.append(asyncio.create_task(run_ops_monitor(queue, BotDetectionLog(), stop)))  # 하트비트 + 임계 알림
+    tasks.append(asyncio.create_task(run_ops_monitor(queue, BotDetectionLog(), stop, adapters=all_adapters)))  # 하트비트 + 임계 알림
     LOG.i(f"LPS 워커 {concurrency}개 + reaper + 브라우저정리 + ops모니터(하트비트/알림) 기동 (워커별 세트 · 상품 {concurrency}개 동시)")
 
     # 종료 유예: stop 후 하던 잡이 이 시간 안에 끝나면 자연 종료, 초과하면 강제 취소.