diff --git a/lps/crud/bot_detection.py b/lps/crud/bot_detection.py
index e1692f1..804ff62 100644
--- a/lps/crud/bot_detection.py
+++ b/lps/crud/bot_detection.py
@@ -27,6 +27,27 @@ class BotDetectionLog:
         finally:
             await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value)
 
+    async def admin_stats(self, hours: int = 168) -> dict:
+        """관리자 FE 용 차단 통계 — 시간대별 건수 + 최근 감지 목록."""
+        p = {"h": hours}
+        hourly = text("""
+            SELECT date_trunc('hour', created_at) AS bucket, count(*) AS n FROM bot_detection
+             WHERE created_at > now() - make_interval(hours => :h) GROUP BY 1 ORDER BY 1
+        """)
+        recent = text("""
+            SELECT source, query, ip_request_no, proxy_port, elapsed_sec, marker, html_len, created_at
+              FROM bot_detection WHERE created_at > now() - make_interval(hours => :h)
+             ORDER BY created_at DESC LIMIT 50
+        """)
+        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
+        try:
+            return {
+                "hourly": [{"bucket": r[0].isoformat(), "count": int(r[1])} for r in (await s.execute(hourly, p)).all()],
+                "items": [dict(r) for r in (await s.execute(recent, p)).mappings().all()],
+            }
+        finally:
+            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
+
     async def recent_count(self, minutes: int = 60) -> int:
         """최근 N분간 봇 감지(차단) 건수 — 차단율 급증 알림·모니터링용."""
         sql = text("SELECT count(*) FROM bot_detection WHERE created_at > now() - make_interval(mins => :m)")
diff --git a/lps/crud/ip_session.py b/lps/crud/ip_session.py
index ee5e709..af66c6c 100644
--- a/lps/crud/ip_session.py
+++ b/lps/crud/ip_session.py
@@ -27,6 +27,38 @@ class IpSessionLog:
         finally:
             await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value)
 
+    async def admin_stats(self, hours: int = 168) -> dict:
+        """관리자 FE 용 IP 세션 통계 — 종료 사유 분포·세션당 요청 수 히스토그램·
+        차단 세션 최소 요청 수(예산 튜닝 기준선)·최근 세션 목록."""
+        p = {"h": hours}
+        by_reason = text("""
+            SELECT end_reason, count(*) AS n FROM ip_session
+             WHERE created_at > now() - make_interval(hours => :h) GROUP BY end_reason
+        """)
+        histogram = text("""
+            SELECT requests, count(*) AS n FROM ip_session
+             WHERE created_at > now() - make_interval(hours => :h) GROUP BY requests ORDER BY requests
+        """)
+        block_min = text("""
+            SELECT min(requests) FROM ip_session
+             WHERE created_at > now() - make_interval(hours => :h) AND end_reason = 'block'
+        """)
+        recent = text("""
+            SELECT source, proxy_port, requests, ok_count, blocked_count, elapsed_sec, end_reason, created_at
+              FROM ip_session WHERE created_at > now() - make_interval(hours => :h)
+             ORDER BY created_at DESC LIMIT 50
+        """)
+        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
+        try:
+            return {
+                "by_reason": {r[0]: int(r[1]) for r in (await s.execute(by_reason, p)).all()},
+                "histogram": [{"requests": int(r[0]), "count": int(r[1])} for r in (await s.execute(histogram, p)).all()],
+                "block_min_requests": (lambda v: int(v) if v is not None else None)((await s.execute(block_min, p)).scalar()),
+                "sessions": [dict(r) for r in (await s.execute(recent, p)).mappings().all()],
+            }
+        finally:
+            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
+
     async def recent_stats(self, minutes: int = 60) -> dict:
         """최근 N분 세션 요약 — 종료 사유별 건수(모니터링·알림용). 예: {"budget": 12, "block": 1}"""
         sql = text("""
diff --git a/lps/crud/job_crud.py b/lps/crud/job_crud.py
index ee48f4d..db2b989 100644
--- a/lps/crud/job_crud.py
+++ b/lps/crud/job_crud.py
@@ -236,6 +236,82 @@ class JobQueue:
         finally:
             await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
 
+    # ---- 관리자(FE) 조회/액션 -------------------------------------------
+    async def list_jobs(self, status: int | None = None, q: str | None = None,
+                        limit: int = 50, offset: int = 0) -> tuple[list[dict], int]:
+        """잡 목록(최신순) + 전체 건수. status(코드)·q(product_code/상품명 부분일치) 필터."""
+        where = ["TRUE"]
+        params: dict = {"limit": limit, "offset": offset}
+        if status is not None:
+            where.append("status = :st")
+            params["st"] = status
+        if q:
+            where.append("(payload->>'product_code' ILIKE :q OR payload->>'product_name' ILIKE :q)")
+            params["q"] = f"%{q}%"
+        cond = " AND ".join(where)
+        sql = text(f"""
+            SELECT job_id, job_type, status, priority, attempts, max_attempts,
+                   payload->>'product_code' AS product_code, payload->>'product_name' AS product_name,
+                   result->>'outcome' AS outcome,
+                   (result#>>'{{lowest,price}}')::int AS final_lowest,
+                   (result#>>'{{metrics,cost,total_usd}}')::float AS cost_usd,
+                   last_error, created_at, run_started_at, updated_at
+            FROM job WHERE {cond}
+            ORDER BY created_at DESC LIMIT :limit OFFSET :offset
+        """)
+        cnt = text(f"SELECT count(*) FROM job WHERE {cond}")
+        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
+        try:
+            rows = [dict(r) for r in (await s.execute(sql, params)).mappings().all()]
+            total = int((await s.execute(cnt, params)).scalar() or 0)
+            for d in rows:
+                d["job_id"] = str(d["job_id"])
+            return rows, total
+        finally:
+            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
+
+    async def requeue(self, job_id: str) -> str | None:
+        """DEAD 잡 재큐(관리자 액션): attempts 리셋 + PENDING 전이 + 워커 깨움.
+        DEAD 가 아니거나 없으면 None. 같은 dedupe_key 의 활성 잡이 있으면 부분 유니크
+        위반(IntegrityError) — 호출부가 '활성 중복'으로 안내한다."""
+        sql = text("""
+            UPDATE job SET status = 1, attempts = 0, run_after = now(),
+                           lease_until = NULL, worker_id = NULL, run_started_at = NULL,
+                           last_error = NULL, updated_at = now()
+            WHERE job_id = CAST(:jid AS uuid) AND status = 4
+            RETURNING job_id
+        """)
+
+        async def run(s):
+            row = (await s.execute(sql, {"jid": job_id})).first()
+            if row:
+                await s.execute(text("SELECT pg_notify(:ch, '')"), {"ch": JOB_NOTIFY_CHANNEL})
+            return str(row[0]) if row else None
+
+        return await self._tx(run)
+
+    async def cost_buckets(self, hours: int = 48) -> list[dict]:
+        """시간별 검색원가 집계(완료 잡의 metrics 합산) — 비용 차트용."""
+        sql = text("""
+            SELECT date_trunc('hour', updated_at) AS bucket,
+                   count(*) AS jobs,
+                   COALESCE(sum((result#>>'{metrics,cost,ai_usd}')::float), 0)    AS ai_usd,
+                   COALESCE(sum((result#>>'{metrics,cost,proxy_usd}')::float), 0) AS proxy_usd,
+                   COALESCE(sum((result#>>'{metrics,cost,total_usd}')::float), 0) AS total_usd,
+                   COALESCE(avg((result#>>'{metrics,duration_ms}')::float), 0)    AS avg_ms
+            FROM job
+            WHERE status = 3 AND updated_at > now() - make_interval(hours => :h)
+            GROUP BY 1 ORDER BY 1
+        """)
+        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
+        try:
+            return [{"bucket": r["bucket"].isoformat(), "jobs": int(r["jobs"]),
+                     "ai_usd": round(float(r["ai_usd"]), 4), "proxy_usd": round(float(r["proxy_usd"]), 4),
+                     "total_usd": round(float(r["total_usd"]), 4), "avg_ms": int(r["avg_ms"])}
+                    for r in (await s.execute(sql, {"h": hours})).mappings().all()]
+        finally:
+            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
+
     async def ping(self) -> bool:
         """DB 도달성 확인(readiness). 실패 시 예외."""
         s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
diff --git a/lps/crud/price_history.py b/lps/crud/price_history.py
index 9a40011..d4c74c9 100644
--- a/lps/crud/price_history.py
+++ b/lps/crud/price_history.py
@@ -65,3 +65,27 @@ class PriceHistory:
             return [dict(r) for r in rows]
         finally:
             await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
+
+    async def list_products(self, q: str | None = None, limit: int = 50) -> list[dict]:
+        """이력이 있는 상품 목록(관리자 FE) — 상품별 최신 스냅샷 + 검색 횟수, 최근 검색순.
+        상품명은 price_history 에 없어 최신 스냅샷의 매칭 상품명(네이버 우선)으로 대신한다."""
+        where = "WHERE product_code ILIKE :q" if q else ""
+        sql = text(f"""
+            SELECT * FROM (
+                SELECT DISTINCT ON (product_code)
+                       product_code, triggered_at, outcome,
+                       naver_lowest, coupang_lowest, final_lowest, final_source,
+                       COALESCE(naver_name, coupang_name) AS display_name,
+                       count(*) OVER (PARTITION BY product_code) AS searches
+                FROM price_history {where}
+                ORDER BY product_code, triggered_at DESC
+            ) t ORDER BY triggered_at DESC LIMIT :lim
+        """)
+        params: dict = {"lim": limit}
+        if q:
+            params["q"] = f"%{q}%"
+        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
+        try:
+            return [dict(r) for r in (await s.execute(sql, params)).mappings().all()]
+        finally:
+            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)
diff --git a/lps/router/router.py b/lps/router/router.py
index 23e2f11..4a637c6 100644
--- a/lps/router/router.py
+++ b/lps/router/router.py
@@ -12,6 +12,7 @@ from common.logger import LOG
 from common.utils.gtime import GTime
 from config.server_configs import web_server_config
 from router.v1.validator.auth import configured_keys, require_api_key
+import router.v1.lps.admin
 import router.v1.lps.search
 
 API_SERVER_START_TIME = GTime.UTCStr()
@@ -88,6 +89,7 @@ async def readyz():
 # 각 도메인 라우터를 등록한다. 새 기능 추가 시 router.v1.<domain>.<file> 를 import 후 include.
 # guard: [WebServerConfig].api_keys 설정 시 /v1 전체에 X-API-Key 검증(개발은 빈값=개방 — auth.py 참고).
 app.include_router(router.v1.lps.search.router, dependencies=[Depends(require_api_key)])
+app.include_router(router.v1.lps.admin.router, dependencies=[Depends(require_api_key)])
 
 if configured_keys():
     LOG.i(f"API guard ON — X-API-Key 검증({len(configured_keys())}개 키)")
diff --git a/lps/router/v1/lps/admin.py b/lps/router/v1/lps/admin.py
new file mode 100644
index 0000000..6e00dd4
--- /dev/null
+++ b/lps/router/v1/lps/admin.py
@@ -0,0 +1,78 @@
+"""관리자 FE 라우터 — 잡 목록/재큐·상품 목록·IP세션/차단/비용 통계.
+
+검증→service→응답만(backend 컨벤션). guard(X-API-Key)는 router.py 의 include 에서 일괄 적용.
+"""
+
+from fastapi import APIRouter, Depends, Query
+
+from router.v1.validator.dependencies import RemoveNoneResponse
+from services.admin_service import AdminService
+from router.v1.lps.admin_protocol import (
+    Res_BotStats, Res_CostStats, Res_IpSessionStats, Res_JobList, Res_ProductList, Res_Requeue,
+)
+
+router = APIRouter(prefix="/v1/lps", tags=["LPS Admin"], responses={404: {"description": "Not found"}})
+
+
+@router.get(
+    path="/jobs",
+    response_model=Res_JobList,
+    summary="잡 목록(관리자)",
+    description="최신순 잡 목록 + 전체 건수. status(PENDING/RUNNING/DONE/DEAD)·q(상품코드/상품명 부분일치) 필터.",
+)
+async def list_jobs(status: str | None = Query(None), q: str | None = Query(None),
+                    limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0),
+                    service: AdminService = Depends()):
+    return RemoveNoneResponse(await service.list_jobs(status, q, limit, offset))
+
+
+@router.post(
+    path="/jobs/{job_id}/requeue",
+    response_model=Res_Requeue,
+    summary="DEAD 잡 재큐(관리자)",
+    description="재시도 소진으로 죽은 잡을 attempts 리셋 후 다시 대기열에 넣는다(워커 즉시 깨움). "
+                "같은 상품의 활성 잡이 있으면 DB_ALREADY_SAME_KEY.",
+)
+async def requeue_job(job_id: str, service: AdminService = Depends()):
+    return RemoveNoneResponse(await service.requeue(job_id))
+
+
+@router.get(
+    path="/products",
+    response_model=Res_ProductList,
+    summary="검색 이력 상품 목록(관리자)",
+    description="price_history 에 이력이 있는 상품별 최신 스냅샷 + 누적 검색 수. 최근 검색순.",
+)
+async def list_products(q: str | None = Query(None), limit: int = Query(50, ge=1, le=200),
+                        service: AdminService = Depends()):
+    return RemoveNoneResponse(await service.list_products(q, limit))
+
+
+@router.get(
+    path="/stats/ip-sessions",
+    response_model=Res_IpSessionStats,
+    summary="IP 세션 통계(관리자)",
+    description="종료 사유 분포·세션당 요청 수 히스토그램·차단 세션 최소 요청 수(예산 튜닝 기준선)·최근 세션.",
+)
+async def ip_session_stats(hours: int = Query(168, ge=1, le=720), service: AdminService = Depends()):
+    return RemoveNoneResponse(await service.ip_session_stats(hours))
+
+
+@router.get(
+    path="/stats/bot",
+    response_model=Res_BotStats,
+    summary="차단(봇 감지) 통계(관리자)",
+    description="시간대별 차단 건수 + 최근 감지 목록(마커·포트·IP 요청순번).",
+)
+async def bot_stats(hours: int = Query(168, ge=1, le=720), service: AdminService = Depends()):
+    return RemoveNoneResponse(await service.bot_stats(hours))
+
+
+@router.get(
+    path="/stats/cost",
+    response_model=Res_CostStats,
+    summary="검색원가 시계열(관리자)",
+    description="완료 잡의 metrics 를 시간별 합산 — AI vs 프록시 대역폭 비용 분해 + 평균 소요.",
+)
+async def cost_stats(hours: int = Query(48, ge=1, le=720), service: AdminService = Depends()):
+    return RemoveNoneResponse(await service.cost_stats(hours))
diff --git a/lps/router/v1/lps/admin_protocol.py b/lps/router/v1/lps/admin_protocol.py
new file mode 100644
index 0000000..c0033fc
--- /dev/null
+++ b/lps/router/v1/lps/admin_protocol.py
@@ -0,0 +1,108 @@
+"""관리자 FE 응답 프로토콜 — 잡 목록/재큐·상품 목록·IP세션/차단/비용 통계."""
+
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from common.models.gmodel import Res_WebPacketProtocol
+
+
+class JobListItem(BaseModel):
+    job_id: str
+    job_type: int
+    status: str = Field(description="JobStatus 이름(PENDING/RUNNING/DONE/DEAD)")
+    priority: int
+    attempts: int
+    max_attempts: int
+    product_code: Optional[str] = None
+    product_name: Optional[str] = None
+    outcome: Optional[str] = Field(None, description="found / not_found (완료 시)")
+    final_lowest: Optional[int] = None
+    cost_usd: Optional[float] = Field(None, description="검색 원가($, metrics 합)")
+    last_error: Optional[str] = None
+    created_at: str
+    run_started_at: Optional[str] = None
+    updated_at: str
+
+
+class Res_JobList(Res_WebPacketProtocol):
+    items: list[JobListItem] = Field(default_factory=list)
+    total: int = 0
+
+
+class Res_Requeue(Res_WebPacketProtocol):
+    job_id: Optional[str] = None
+    requeued: bool = False
+
+
+class ProductItem(BaseModel):
+    product_code: str
+    display_name: Optional[str] = Field(None, description="최신 매칭 상품명(네이버 우선) — 표시용 근사값")
+    triggered_at: str = Field(description="마지막 검색 시각")
+    outcome: str
+    naver_lowest: Optional[int] = None
+    coupang_lowest: Optional[int] = None
+    final_lowest: Optional[int] = None
+    final_source: Optional[str] = None
+    searches: int = Field(0, description="누적 검색(이력) 수")
+
+
+class Res_ProductList(Res_WebPacketProtocol):
+    items: list[ProductItem] = Field(default_factory=list)
+
+
+class HistogramBin(BaseModel):
+    requests: int
+    count: int
+
+
+class IpSessionRow(BaseModel):
+    source: str
+    proxy_port: Optional[int] = None
+    requests: int
+    ok_count: int
+    blocked_count: int
+    elapsed_sec: Optional[int] = None
+    end_reason: str
+    created_at: str
+
+
+class Res_IpSessionStats(Res_WebPacketProtocol):
+    by_reason: dict[str, int] = Field(default_factory=dict, description="종료 사유별 세션 수")
+    histogram: list[HistogramBin] = Field(default_factory=list, description="세션당 요청 수 분포")
+    block_min_requests: Optional[int] = Field(None, description="차단 세션의 최소 요청 수 — 예산은 이보다 낮게")
+    sessions: list[IpSessionRow] = Field(default_factory=list, description="최근 세션 50건")
+
+
+class HourlyCount(BaseModel):
+    bucket: str
+    count: int
+
+
+class BotRow(BaseModel):
+    source: str
+    query: Optional[str] = None
+    ip_request_no: Optional[int] = None
+    proxy_port: Optional[int] = None
+    elapsed_sec: Optional[int] = None
+    marker: Optional[str] = None
+    html_len: Optional[int] = None
+    created_at: str
+
+
+class Res_BotStats(Res_WebPacketProtocol):
+    hourly: list[HourlyCount] = Field(default_factory=list)
+    items: list[BotRow] = Field(default_factory=list, description="최근 감지 50건")
+
+
+class CostBucket(BaseModel):
+    bucket: str = Field(description="시간(hour truncate)")
+    jobs: int
+    ai_usd: float
+    proxy_usd: float
+    total_usd: float
+    avg_ms: int = Field(description="검색 1건 평균 소요(ms)")
+
+
+class Res_CostStats(Res_WebPacketProtocol):
+    buckets: list[CostBucket] = Field(default_factory=list)
diff --git a/lps/services/admin_service.py b/lps/services/admin_service.py
new file mode 100644
index 0000000..4f9e22b
--- /dev/null
+++ b/lps/services/admin_service.py
@@ -0,0 +1,116 @@
+"""관리자 FE 도메인 로직 — 조회 중심 + 필수 액션(DEAD 재큐)만.
+
+설정 변경(예산·임계 등)은 FE 에 두지 않는다 — 설정은 toml 단일 소스(재시작 반영) 원칙.
+"""
+
+import uuid
+
+from fastapi import Depends
+from sqlalchemy.exc import IntegrityError
+
+from common.enums import ErrorType, JobStatus
+from crud.bot_detection import BotDetectionLog
+from crud.ip_session import IpSessionLog
+from crud.job_crud import JobQueue
+from crud.price_history import PriceHistory
+from router.v1.lps.admin_protocol import (
+    BotRow,
+    CostBucket,
+    HistogramBin,
+    HourlyCount,
+    IpSessionRow,
+    JobListItem,
+    ProductItem,
+    Res_BotStats,
+    Res_CostStats,
+    Res_IpSessionStats,
+    Res_JobList,
+    Res_ProductList,
+    Res_Requeue,
+)
+
+_STATUS_BY_NAME = {js.name: js.value for js in JobStatus}
+
+
+class AdminService:
+    def __init__(self, queue: JobQueue = Depends(JobQueue), history: PriceHistory = Depends(PriceHistory),
+                 ip_log: IpSessionLog = Depends(IpSessionLog), bot_log: BotDetectionLog = Depends(BotDetectionLog)):
+        self.queue = queue
+        self.history = history
+        self.ip_log = ip_log
+        self.bot_log = bot_log
+
+    async def list_jobs(self, status: str | None, q: str | None, limit: int, offset: int) -> Res_JobList:
+        res = Res_JobList()
+        status_code = _STATUS_BY_NAME.get(status.upper()) if status else None
+        if status and status_code is None:
+            res.result.SetResult(ErrorType.LPS_JOB_NOT_FOUND)   # 알 수 없는 상태 이름
+            return res
+        rows, total = await self.queue.list_jobs(status_code, q, limit, offset)
+        res.total = total
+        res.items = [JobListItem(
+            job_id=r["job_id"], job_type=r["job_type"], status=JobStatus(r["status"]).name,
+            priority=r["priority"], attempts=r["attempts"], max_attempts=r["max_attempts"],
+            product_code=r.get("product_code"), product_name=r.get("product_name"),
+            outcome=r.get("outcome"), final_lowest=r.get("final_lowest"), cost_usd=r.get("cost_usd"),
+            last_error=r.get("last_error"),
+            created_at=r["created_at"].isoformat(timespec="seconds"),
+            run_started_at=r["run_started_at"].isoformat(timespec="seconds") if r.get("run_started_at") else None,
+            updated_at=r["updated_at"].isoformat(timespec="seconds"),
+        ) for r in rows]
+        return res
+
+    async def requeue(self, job_id: str) -> Res_Requeue:
+        res = Res_Requeue(job_id=job_id)
+        try:
+            uuid.UUID(job_id)
+        except (ValueError, TypeError):
+            res.result.SetResult(ErrorType.LPS_JOB_NOT_FOUND)
+            return res
+        try:
+            requeued = await self.queue.requeue(job_id)
+        except IntegrityError:
+            # 같은 상품의 활성 잡(PENDING/RUNNING)이 이미 있음 — 부분 유니크(dedupe) 위반
+            res.result.SetResult(ErrorType.DB_ALREADY_SAME_KEY)
+            return res
+        if requeued is None:   # 없거나 DEAD 가 아님
+            res.result.SetResult(ErrorType.LPS_JOB_NOT_FOUND)
+            return res
+        res.requeued = True
+        return res
+
+    async def list_products(self, q: str | None, limit: int) -> Res_ProductList:
+        res = Res_ProductList()
+        res.items = [ProductItem(
+            product_code=r["product_code"], display_name=r.get("display_name"),
+            triggered_at=r["triggered_at"].isoformat(timespec="seconds"), outcome=r["outcome"],
+            naver_lowest=r.get("naver_lowest"), coupang_lowest=r.get("coupang_lowest"),
+            final_lowest=r.get("final_lowest"), final_source=r.get("final_source"),
+            searches=int(r.get("searches") or 0),
+        ) for r in await self.history.list_products(q, limit)]
+        return res
+
+    async def ip_session_stats(self, hours: int) -> Res_IpSessionStats:
+        res = Res_IpSessionStats()
+        st = await self.ip_log.admin_stats(hours)
+        res.by_reason = st["by_reason"]
+        res.histogram = [HistogramBin(**b) for b in st["histogram"]]
+        res.block_min_requests = st["block_min_requests"]
+        res.sessions = [IpSessionRow(
+            **{**s, "created_at": s["created_at"].isoformat(timespec="seconds")}
+        ) for s in st["sessions"]]
+        return res
+
+    async def bot_stats(self, hours: int) -> Res_BotStats:
+        res = Res_BotStats()
+        st = await self.bot_log.admin_stats(hours)
+        res.hourly = [HourlyCount(bucket=h["bucket"], count=h["count"]) for h in st["hourly"]]
+        res.items = [BotRow(
+            **{**b, "created_at": b["created_at"].isoformat(timespec="seconds")}
+        ) for b in st["items"]]
+        return res
+
+    async def cost_stats(self, hours: int) -> Res_CostStats:
+        res = Res_CostStats()
+        res.buckets = [CostBucket(**b) for b in await self.queue.cost_buckets(hours)]
+        return res
diff --git a/lps/tests/test_admin_api.py b/lps/tests/test_admin_api.py
new file mode 100644
index 0000000..9205377
--- /dev/null
+++ b/lps/tests/test_admin_api.py
@@ -0,0 +1,139 @@
+"""관리자 FE API 테스트 — 잡 목록/재큐·상품 목록·IP세션/차단/비용 통계 (실 lps_db)."""
+
+import pytest_asyncio
+from sqlalchemy import text
+
+from common.enums import JobType
+from crud.job_crud import JobQueue
+
+
+@pytest_asyncio.fixture
+async def clean_all(db_engine):
+    async with db_engine.begin() as conn:
+        for t in ("job", "price_history", "ip_session", "bot_detection"):
+            await conn.execute(text(f"TRUNCATE {t}"))
+    return db_engine
+
+
+@pytest_asyncio.fixture
+async def q(clean_all):
+    return JobQueue()
+
+
+# ---- 잡 목록 --------------------------------------------------------------
+
+async def test_jobs_list_with_filters(client, q):
+    await q.enqueue(JobType.SEARCH.value, {"product_code": "A1", "product_name": "맥심 커피"})
+    await q.enqueue(JobType.SEARCH.value, {"product_code": "B2", "product_name": "생수"})
+    job = await q.claim("w1")
+    await q.complete(job["job_id"], "w1", {"outcome": "found", "lowest": {"price": 12000},
+                                           "metrics": {"cost": {"total_usd": 0.01}}})
+    r = await client.get("/v1/lps/jobs")
+    j = r.json()
+    assert r.status_code == 200 and j["total"] == 2
+
+    r = await client.get("/v1/lps/jobs", params={"status": "DONE"})
+    j = r.json()
+    assert j["total"] == 1
+    assert j["items"][0]["outcome"] == "found" and j["items"][0]["final_lowest"] == 12000
+    assert j["items"][0]["cost_usd"] == 0.01
+
+    r = await client.get("/v1/lps/jobs", params={"q": "맥심"})
+    assert r.json()["total"] == 1
+
+
+async def test_jobs_list_unknown_status(client, q):
+    r = await client.get("/v1/lps/jobs", params={"status": "NOPE"})
+    assert r.json()["result"]["success"] is False
+
+
+# ---- 재큐 ------------------------------------------------------------------
+
+async def test_requeue_dead_job(client, q):
+    jid = await q.enqueue(JobType.SEARCH.value, {"product_code": "A1"}, max_attempts=1, dedupe_key="search-A1")
+    await q.claim("w1")
+    await q.fail(jid, "w1", "boom", backoff_sec=0)          # 1/1 → DEAD
+    r = await client.post(f"/v1/lps/jobs/{jid}/requeue")
+    assert r.json()["requeued"] is True
+    assert (await q.counts())["PENDING"] == 1               # DEAD → PENDING
+
+
+async def test_requeue_rejects_non_dead_and_missing(client, q):
+    jid = await q.enqueue(JobType.SEARCH.value, {"product_code": "A1"})
+    r = await client.post(f"/v1/lps/jobs/{jid}/requeue")    # PENDING — 재큐 대상 아님
+    assert r.json()["result"]["success"] is False
+    r = await client.post("/v1/lps/jobs/not-a-uuid/requeue")
+    assert r.json()["result"]["success"] is False
+
+
+async def test_requeue_blocked_by_active_duplicate(client, q):
+    dead = await q.enqueue(JobType.SEARCH.value, {"product_code": "A1"}, max_attempts=1, dedupe_key="search-A1")
+    await q.claim("w1")
+    await q.fail(dead, "w1", "boom", backoff_sec=0)
+    await q.enqueue(JobType.SEARCH.value, {"product_code": "A1"}, dedupe_key="search-A1")  # 활성 중복 생성
+    r = await client.post(f"/v1/lps/jobs/{dead}/requeue")
+    assert r.json()["requeued"] is False
+    assert r.json()["result"]["success"] is False           # DB_ALREADY_SAME_KEY
+
+
+# ---- 상품 목록 --------------------------------------------------------------
+
+async def test_products_list_latest_snapshot(client, clean_all):
+    async with clean_all.begin() as conn:
+        await conn.execute(text("""
+            INSERT INTO price_history (product_code, outcome, naver_lowest, final_lowest, naver_name, triggered_at)
+            VALUES ('P1', 'found', 1000, 900, '커피 320개입', now() - interval '2 hour'),
+                   ('P1', 'found', 1100, 950, '커피 320개입', now() - interval '1 hour'),
+                   ('P2', 'not_found', NULL, NULL, NULL, now())
+        """))
+    r = await client.get("/v1/lps/products")
+    items = r.json()["items"]
+    assert [i["product_code"] for i in items] == ["P2", "P1"]   # 최근 검색순
+    p1 = items[1]
+    assert p1["searches"] == 2 and p1["final_lowest"] == 950    # 최신 스냅샷 + 누적 횟수
+    r = await client.get("/v1/lps/products", params={"q": "P1"})
+    assert len(r.json()["items"]) == 1
+
+
+# ---- 통계 3종 ---------------------------------------------------------------
+
+async def test_ip_session_stats(client, clean_all):
+    async with clean_all.begin() as conn:
+        await conn.execute(text("""
+            INSERT INTO ip_session (source, proxy_port, requests, ok_count, blocked_count, end_reason)
+            VALUES ('coupang', 10001, 3, 3, 0, 'budget'),
+                   ('coupang', 10002, 3, 3, 0, 'budget'),
+                   ('coupang', 10003, 5, 4, 1, 'block')
+        """))
+    r = await client.get("/v1/lps/stats/ip-sessions")
+    j = r.json()
+    assert j["by_reason"] == {"budget": 2, "block": 1}
+    assert j["block_min_requests"] == 5                         # 예산 튜닝 기준선
+    assert {"requests": 3, "count": 2} in j["histogram"]
+    assert len(j["sessions"]) == 3
+
+
+async def test_bot_stats(client, clean_all):
+    async with clean_all.begin() as conn:
+        await conn.execute(text("""
+            INSERT INTO bot_detection (source, query, ip_request_no, proxy_port, marker)
+            VALUES ('coupang', '생수', 4, 10001, '/akam/')
+        """))
+    r = await client.get("/v1/lps/stats/bot")
+    j = r.json()
+    assert len(j["items"]) == 1 and j["items"][0]["marker"] == "/akam/"
+    assert sum(h["count"] for h in j["hourly"]) == 1
+
+
+async def test_cost_stats(client, q):
+    for cost in (0.01, 0.02):
+        jid = await q.enqueue(JobType.SEARCH.value, {"product_code": f"C{cost}"})
+        job = await q.claim("w1")
+        await q.complete(job["job_id"], "w1",
+                         {"metrics": {"cost": {"ai_usd": cost / 2, "proxy_usd": cost / 2, "total_usd": cost},
+                                      "duration_ms": 15000}})
+    r = await client.get("/v1/lps/stats/cost")
+    buckets = r.json()["buckets"]
+    assert sum(b["total_usd"] for b in buckets) == 0.03
+    assert sum(b["jobs"] for b in buckets) == 2
+    assert buckets[0]["avg_ms"] == 15000