diff --git a/agent-office/app/agents/stock.py b/agent-office/app/agents/stock.py
index 629440a..8cf5f3d 100644
--- a/agent-office/app/agents/stock.py
+++ b/agent-office/app/agents/stock.py
@@ -233,6 +233,106 @@ class StockAgent(BaseAgent):
 
             await self.transition("idle", f"스크리너 오류: {err_msg[:80]}")
 
+    async def on_ai_news_schedule(self) -> None:
+        """AI 뉴스 sentiment 분석 자동 잡 (평일 08:00 KST).
+
+        흐름:
+          1) stock-lab /snapshot/refresh-news-sentiment 호출
+          2) status='skipped_weekend'/'skipped_holiday' → 종료 (텔레그램 미발신)
+          3) updated=0 → 운영자 알림 (HTML)
+          4) failures > 30% → 경고 알림 후 메인 메시지 발송
+          5) 정상 → Top 5 호재/악재 메시지 발송 (MarkdownV2)
+        """
+        if self.state not in ("idle", "break"):
+            return
+
+        task_id = create_task(self.agent_id, "ai_news_sentiment", {})
+        await self.transition("working", "AI 뉴스 분석 중...", task_id)
+
+        try:
+            result = await service_proxy.refresh_ai_news_sentiment()
+        except Exception as e:
+            err_msg = str(e)
+            add_log(self.agent_id, f"AI 뉴스 분석 실패: {err_msg}", "error", task_id)
+            update_task_status(task_id, "failed", {"error": err_msg})
+            try:
+                from ..telegram.messaging import send_raw
+                await send_raw(
+                    f"⚠️ <b>AI 뉴스 분석 실패</b>\n"
+                    f"<code>{html.escape(err_msg)[:500]}</code>"
+                )
+            except Exception as notify_err:
+                add_log(
+                    self.agent_id,
+                    f"operator notify failed: {notify_err}",
+                    "warning", task_id,
+                )
+            await self.transition("idle", f"AI 뉴스 오류: {err_msg[:80]}")
+            return
+
+        status = result.get("status")
+        if status in ("skipped_weekend", "skipped_holiday"):
+            update_task_status(task_id, "succeeded", {"status": status})
+            add_log(self.agent_id, f"AI 뉴스 건너뜀: {status}", "info", task_id)
+            await self.transition("idle", "휴일/주말 — 건너뜀")
+            return
+
+        updated = int(result.get("updated", 0))
+        failures = result.get("failures", []) or []
+        if updated == 0:
+            update_task_status(task_id, "failed", {"reason": "0 tickers updated"})
+            try:
+                from ..telegram.messaging import send_raw
+                await send_raw(
+                    "⚠️ <b>AI 뉴스 분석 0종목</b>\n"
+                    "스크래핑/LLM 전체 실패 — 어제 데이터 사용"
+                )
+            except Exception:
+                pass
+            await self.transition("idle", "AI 뉴스 0건")
+            return
+
+        # 실패율 경고 (별도 알림, 본 메시지는 계속 발송)
+        failure_rate = len(failures) / max(1, updated + len(failures))
+        if failure_rate > 0.3:
+            try:
+                from ..telegram.messaging import send_raw
+                await send_raw(
+                    f"⚠️ <b>AI 뉴스 실패율 {failure_rate:.0%}</b>\n"
+                    f"updated={updated}, failures={len(failures)}"
+                )
+            except Exception:
+                pass
+
+        # 정상 — Top 5 메시지 (stock-lab이 빌드해서 응답에 telegram_text 동봉)
+        text = result.get("telegram_text") or ""
+        if not text:
+            raise RuntimeError("telegram_text 누락")
+
+        await self.transition("reporting", "AI 뉴스 알림 전송 중...")
+        from ..telegram.messaging import send_raw
+        tg = await send_raw(text, parse_mode="MarkdownV2")
+
+        update_task_status(task_id, "succeeded", {
+            "asof": result["asof"],
+            "updated": updated,
+            "failures": len(failures),
+            "tokens_input": int(result.get("tokens_input", 0)),
+            "tokens_output": int(result.get("tokens_output", 0)),
+            "telegram_sent": tg.get("ok", False),
+        })
+
+        if not tg.get("ok"):
+            desc = tg.get("description") or "unknown"
+            code = tg.get("error_code")
+            add_log(
+                self.agent_id,
+                f"AI news telegram send failed: [{code}] {desc}",
+                "warning", task_id,
+            )
+
+        await self.transition("idle", "AI 뉴스 완료")
+
     async def on_command(self, command: str, params: dict) -> dict:
         if command == "run_screener":
             await self.on_screener_schedule()