diff --git a/backend/src/module/core/context.py b/backend/src/module/core/context.py
index 85c36ae2..41a56f31 100644
--- a/backend/src/module/core/context.py
+++ b/backend/src/module/core/context.py
@@ -104,7 +104,7 @@ class AppContext:
             [
                 PeriodicTask(
                     name="rss",
-                    run=lambda: rss_tick(analyser),
+                    run=lambda: rss_tick(analyser, notifier),
                     interval=lambda: settings_obj.program.rss_time,
                     enabled=Checker.check_analyser,
                 ),
@@ -116,7 +116,7 @@ class AppContext:
                 ),
                 PeriodicTask(
                     name="offset_scan",
-                    run=offset_scan_tick,
+                    run=lambda: offset_scan_tick(notifier),
                     interval=lambda: OFFSET_SCAN_INTERVAL,
                     initial_delay=OFFSET_SCAN_INITIAL_DELAY,
                 ),
diff --git a/backend/src/module/core/loops.py b/backend/src/module/core/loops.py
index 4f1f8c61..a60f845d 100644
--- a/backend/src/module/core/loops.py
+++ b/backend/src/module/core/loops.py
@@ -18,7 +18,7 @@ from .offset_scanner import OffsetScanner
 logger = logging.getLogger(__name__)
 
 
-async def rss_tick(analyser: RSSAnalyser) -> None:
+async def rss_tick(analyser: RSSAnalyser, notifier: NotificationManager) -> None:
     """Analyse aggregate RSS feeds and refresh the RSS engine once."""
     async with DownloadClient() as client:
         async with Database() as db:
@@ -44,7 +44,10 @@ async def rss_tick(analyser: RSSAnalyser) -> None:
                             exc_info=True,
                         )
             # Run RSS Engine
-            await engine.refresh_rss(client)
+            events = await engine.refresh_rss(client)
+    if settings.notification.enable:
+        for event in events:
+            await notifier.send_event(event)
     if settings.bangumi_manage.eps_complete:
         await eps_complete()
 
@@ -59,11 +62,14 @@ async def rename_tick(notifier: NotificationManager) -> None:
             await notifier.send_all(info)
 
 
-async def offset_scan_tick() -> None:
+async def offset_scan_tick(notifier: NotificationManager) -> None:
     """Scan all bangumi for season/episode offset mismatches."""
     scanner = OffsetScanner()
-    flagged = await scanner.scan_all()
-    logger.info("[OffsetScanThread] Scan complete, flagged %s bangumi", flagged)
+    events = await scanner.scan_all()
+    logger.info("[OffsetScanThread] Scan complete, flagged %s bangumi", len(events))
+    if settings.notification.enable:
+        for event in events:
+            await notifier.send_event(event)
 
 
 async def calendar_tick() -> None:
diff --git a/backend/src/module/notification/__init__.py b/backend/src/module/notification/__init__.py
index 23aa4937..049897e0 100644
--- a/backend/src/module/notification/__init__.py
+++ b/backend/src/module/notification/__init__.py
@@ -1,9 +1,19 @@
 from .base import NotificationProvider
+from .events import (
+    DownloadFailureEvent,
+    OffsetReviewEvent,
+    RssFailureEvent,
+    SystemEvent,
+)
 from .manager import NotificationManager
 from .providers import PROVIDER_REGISTRY
 
 __all__ = [
+    "DownloadFailureEvent",
     "NotificationManager",
     "NotificationProvider",
+    "OffsetReviewEvent",
     "PROVIDER_REGISTRY",
+    "RssFailureEvent",
+    "SystemEvent",
 ]
diff --git a/backend/src/module/notification/base.py b/backend/src/module/notification/base.py
index 249f34df..fa2c043b 100644
--- a/backend/src/module/notification/base.py
+++ b/backend/src/module/notification/base.py
@@ -1,20 +1,39 @@
 """Base class for notification providers."""
 
 from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
 
 from module.conf import settings
 from module.models.bangumi import Notification
 from module.network import RequestContent
-from module.network.request_url import RequestURL
+from module.notification.events import SystemEvent
+
+if TYPE_CHECKING:
+    from module.models.config import NotificationProvider as ProviderConfig
 
 
-class NotificationProvider(RequestContent, ABC):
+class NotificationProvider(ABC):
     """Abstract base class for notification providers.
 
     All notification providers must inherit from this class and implement
-    the send() and test() methods.
+    the send() and test() methods. HTTP delivery is composed via a
+    ``RequestContent`` instance (``self._http``) rather than inherited -- a
+    provider IS-A notification channel, not an HTTP client.
     """
 
+    def __init__(self, config: "ProviderConfig") -> None:
+        self._http = RequestContent()
+        # 单集通知模板（{{title}}/{{season}}/{{episode}}/{{poster_url}}）；
+        # 未设置时 ``_format_message`` 回退到默认中文文案。
+        self.template = config.template
+
+    async def __aenter__(self) -> "NotificationProvider":
+        await self._http.__aenter__()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
+        await self._http.__aexit__(exc_type, exc_val, exc_tb)
+
     @abstractmethod
     async def send(self, notification: Notification) -> bool:
         """Send a notification.
@@ -37,8 +56,43 @@ class NotificationProvider(RequestContent, ABC):
         """
         pass
 
+    async def send_event(self, event: SystemEvent) -> bool:
+        """Send a system event (RSS failure, download failure, offset review).
+
+        System events use each provider's default title/body delivery
+        (:meth:`_deliver_text`) -- the per-episode template only covers
+        {{title}}/{{season}}/{{episode}}/{{poster_url}}, which don't apply
+        here.
+
+        Args:
+            event: The system event to send.
+
+        Returns:
+            True if the event was delivered successfully, False otherwise.
+        """
+        title, body = event.describe()
+        return await self._deliver_text(title, body)
+
+    @abstractmethod
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a plain title + body text message. Backs :meth:`send_event`.
+
+        Args:
+            title: Short event title.
+            body: Event details.
+
+        Returns:
+            True if delivered successfully, False otherwise.
+        """
+        pass
+
     def _format_message(self, notify: Notification) -> str:
-        """Format the default notification message.
+        """Format the per-episode notification message.
+
+        Uses the provider's configured ``template`` when set, substituting
+        ``{{title}}``/``{{season}}``/``{{episode}}``/``{{poster_url}}``;
+        otherwise falls back to the default Chinese message (existing
+        configs without a template keep this exact text).
 
         Args:
             notify: The notification data.
@@ -46,11 +100,22 @@ class NotificationProvider(RequestContent, ABC):
         Returns:
             Formatted message string.
         """
-        return (
-            f"番剧名称：{notify.official_title}\n"
-            f"季度： 第{notify.season}季\n"
-            f"更新集数： 第{notify.episode}集"
-        )
+        if not self.template:
+            return (
+                f"番剧名称：{notify.official_title}\n"
+                f"季度： 第{notify.season}季\n"
+                f"更新集数： 第{notify.episode}集"
+            )
+        replacements = {
+            "{{title}}": notify.official_title,
+            "{{season}}": str(notify.season),
+            "{{episode}}": str(notify.episode),
+            "{{poster_url}}": self._poster_url(notify) or "",
+        }
+        rendered = self.template
+        for pattern, value in replacements.items():
+            rendered = rendered.replace(pattern, value)
+        return rendered
 
     def _poster_url(self, notify: Notification) -> str | None:
         """Build a public poster URL from a locally-stored poster path.
@@ -73,13 +138,17 @@ class NotificationProvider(RequestContent, ABC):
             return None
         return f"{base_url.rstrip('/')}/{notify.poster_path}"
 
+    async def post_data(self, url: str, data: dict):
+        """Form-encoded POST, delegating to the composed HTTP client."""
+        return await self._http.post_data(url, data)
+
+    async def post_files(self, url: str, data: dict, files: dict):
+        """Multipart POST, delegating to the composed HTTP client."""
+        return await self._http.post_files(url, data, files)
+
     async def _post_json(self, url: str, json_data: dict):
         """POST a JSON payload via ``RequestURL.post_json`` (httpx ``json=``).
 
-        Called explicitly on ``RequestURL`` rather than ``self.post_json``
-        because ``RequestContent`` (a parent of this class) defines its own
-        unrelated, form-encoded ``post_json`` that would otherwise shadow it.
-
         Args:
             url: The target URL.
             json_data: The JSON-serializable payload.
@@ -87,4 +156,4 @@ class NotificationProvider(RequestContent, ABC):
         Returns:
             The httpx response.
         """
-        return await RequestURL.post_json(self, url, json_data)
+        return await self._http.post_json(url, json_data)
diff --git a/backend/src/module/notification/events.py b/backend/src/module/notification/events.py
new file mode 100644
index 00000000..efd883e1
--- /dev/null
+++ b/backend/src/module/notification/events.py
@@ -0,0 +1,57 @@
+"""非「新集数」通知的类型化事件负载。
+
+刻意与 ``Notification``（module/models/bangumi.py，仅描述「新集数下载完成」）
+分开：RSS 订阅失败、下载添加失败、偏移量待确认三类事件携带的字段与集数通知完全
+无关，硬塞进 ``Notification`` 只会制造出没有意义的 season/episode 占位值。
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass(slots=True, frozen=True)
+class RssFailureEvent:
+    """RSS 订阅连接状态从正常变为异常（仅状态翻转时触发一次，而非每个 tick）。"""
+
+    rss_name: str
+    rss_url: str
+    error: str
+
+    def describe(self) -> tuple[str, str]:
+        """返回该事件的默认 (标题, 正文)。"""
+        return (
+            "RSS 订阅连接异常",
+            f"订阅：{self.rss_name}\n地址：{self.rss_url}\n错误：{self.error}",
+        )
+
+
+@dataclass(slots=True, frozen=True)
+class DownloadFailureEvent:
+    """匹配到的种子在（内置 HTTP 重试后）仍添加下载器失败。"""
+
+    official_title: str
+    torrent_name: str
+
+    def describe(self) -> tuple[str, str]:
+        return (
+            "种子添加失败",
+            f"番剧：{self.official_title}\n种子：{self.torrent_name}\n"
+            "重试后仍添加失败，请检查下载器连接。",
+        )
+
+
+@dataclass(slots=True, frozen=True)
+class OffsetReviewEvent:
+    """番剧被标记为需要人工确认季度/集数偏移。"""
+
+    official_title: str
+    reason: str
+
+    def describe(self) -> tuple[str, str]:
+        return (
+            "集数偏移待确认",
+            f"番剧：{self.official_title}\n原因：{self.reason}\n请前往设置页确认偏移量。",
+        )
+
+
+# 除「新集数」以外的通知事件的联合类型。
+SystemEvent = RssFailureEvent | DownloadFailureEvent | OffsetReviewEvent
diff --git a/backend/src/module/notification/manager.py b/backend/src/module/notification/manager.py
index f223a0e9..fac23105 100644
--- a/backend/src/module/notification/manager.py
+++ b/backend/src/module/notification/manager.py
@@ -2,11 +2,12 @@
 
 import asyncio
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Awaitable, Callable
 
 from module.conf import settings
 from module.database import Database
 from module.models.bangumi import Notification
+from module.notification.events import SystemEvent
 
 if TYPE_CHECKING:
     from module.models.config import NotificationProvider as ProviderConfig
@@ -60,6 +61,36 @@ class NotificationManager:
             if data:
                 notification.poster_path = data.poster_link
 
+    async def _broadcast(
+        self, label: str, send_one: Callable[["NotificationProvider"], Awaitable[None]]
+    ):
+        """Run ``send_one`` against every provider in parallel, logging failures.
+
+        Shared by :meth:`send_all` and :meth:`send_event` so both broadcast
+        the same way: one provider's exception never blocks the others.
+
+        Args:
+            label: Short description used in warning logs on failure.
+            send_one: Coroutine function that delivers to a single provider.
+        """
+        if not self.providers:
+            logger.debug("No notification providers configured")
+            return
+
+        async def guarded(provider: "NotificationProvider"):
+            try:
+                async with provider:
+                    await send_one(provider)
+            except Exception as e:
+                logger.warning(
+                    f"Failed to send {label} via {provider.__class__.__name__}: {e}"
+                )
+
+        await asyncio.gather(
+            *[guarded(p) for p in self.providers],
+            return_exceptions=True,
+        )
+
     async def send_all(self, notification: Notification):
         """Send notification to all enabled providers.
 
@@ -73,25 +104,31 @@ class NotificationManager:
         # Fetch poster if needed
         await self._get_poster(notification)
 
-        # Send to all providers in parallel
-        async def send_to_provider(provider: "NotificationProvider"):
-            try:
-                async with provider:
-                    await provider.send(notification)
-                logger.debug(
-                    "Sent notification via %s: %s",
-                    provider.__class__.__name__,
-                    notification.official_title,
-                )
-            except Exception as e:
-                logger.warning(
-                    f"Failed to send notification via {provider.__class__.__name__}: {e}"
-                )
+        async def send_one(provider: "NotificationProvider"):
+            await provider.send(notification)
+            logger.debug(
+                "Sent notification via %s: %s",
+                provider.__class__.__name__,
+                notification.official_title,
+            )
 
-        await asyncio.gather(
-            *[send_to_provider(p) for p in self.providers],
-            return_exceptions=True,
-        )
+        await self._broadcast("notification", send_one)
+
+    async def send_event(self, event: SystemEvent):
+        """Send a system event (RSS failure, download failure, offset review)
+        to all enabled providers.
+
+        Args:
+            event: The system event to send.
+        """
+
+        async def send_one(provider: "NotificationProvider"):
+            await provider.send_event(event)
+            logger.debug(
+                "Sent system event via %s: %s", provider.__class__.__name__, event
+            )
+
+        await self._broadcast("system event", send_one)
 
     async def test_provider(self, index: int) -> tuple[bool, str]:
         """Test a specific provider by index.
diff --git a/backend/src/module/notification/providers/bark.py b/backend/src/module/notification/providers/bark.py
index 005d25f0..dd4e9a12 100644
--- a/backend/src/module/notification/providers/bark.py
+++ b/backend/src/module/notification/providers/bark.py
@@ -18,7 +18,7 @@ class BarkProvider(NotificationProvider):
     DEFAULT_SERVER = "https://api.day.app"
 
     def __init__(self, config: "ProviderConfig"):
-        super().__init__()
+        super().__init__(config)
         # Support both legacy token field and new device_key field
         self.device_key = config.device_key or config.token
         server_url = config.server_url or self.DEFAULT_SERVER
@@ -55,3 +55,9 @@ class BarkProvider(NotificationProvider):
                 return False, f"Bark API returned status {resp.status_code}"
         except Exception as e:
             return False, f"Bark test failed: {e}"
+
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a system event via Bark."""
+        data = {"title": title, "body": body, "device_key": self.device_key}
+        resp = await self._post_json(self.notification_url, data)
+        return resp.status_code == 200
diff --git a/backend/src/module/notification/providers/discord.py b/backend/src/module/notification/providers/discord.py
index 649c6099..1c7cebda 100644
--- a/backend/src/module/notification/providers/discord.py
+++ b/backend/src/module/notification/providers/discord.py
@@ -16,17 +16,24 @@ class DiscordProvider(NotificationProvider):
     """Discord webhook notification provider."""
 
     def __init__(self, config: "ProviderConfig"):
-        super().__init__()
+        super().__init__(config)
         self.webhook_url = config.webhook_url
 
     async def send(self, notification: Notification) -> bool:
         """Send notification via Discord webhook."""
-        embed = {
-            "title": f"📺 {notification.official_title}",
-            "description": (
+        # A configured template overrides the description only; the default
+        # (no template) description is kept byte-for-byte for existing configs.
+        description = (
+            self._format_message(notification)
+            if self.template
+            else (
                 f"**季度:** 第{notification.season}季\n"
                 f"**集数:** 第{notification.episode}集"
-            ),
+            )
+        )
+        embed = {
+            "title": f"📺 {notification.official_title}",
+            "description": description,
             "color": 0x00BFFF,  # Deep Sky Blue
         }
 
@@ -60,3 +67,9 @@ class DiscordProvider(NotificationProvider):
                 return False, f"Discord API returned status {resp.status_code}"
         except Exception as e:
             return False, f"Discord test failed: {e}"
+
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a system event via Discord webhook."""
+        embed = {"title": title, "description": body, "color": 0xFFA500}  # Orange
+        resp = await self._post_json(self.webhook_url, {"embeds": [embed]})
+        return resp.status_code in (200, 204)
diff --git a/backend/src/module/notification/providers/gotify.py b/backend/src/module/notification/providers/gotify.py
index b5317269..39cff010 100644
--- a/backend/src/module/notification/providers/gotify.py
+++ b/backend/src/module/notification/providers/gotify.py
@@ -16,7 +16,7 @@ class GotifyProvider(NotificationProvider):
     """Gotify notification provider."""
 
     def __init__(self, config: "ProviderConfig"):
-        super().__init__()
+        super().__init__(config)
         server_url = config.server_url.rstrip("/")
         self.token = config.token
         self.notification_url = f"{server_url}/message?token={self.token}"
@@ -62,3 +62,9 @@ class GotifyProvider(NotificationProvider):
                 return False, f"Gotify API returned status {resp.status_code}"
         except Exception as e:
             return False, f"Gotify test failed: {e}"
+
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a system event via Gotify."""
+        data = {"title": title, "message": body, "priority": 5}
+        resp = await self._post_json(self.notification_url, data)
+        return resp.status_code == 200
diff --git a/backend/src/module/notification/providers/pushover.py b/backend/src/module/notification/providers/pushover.py
index 85871a38..c73c3f1b 100644
--- a/backend/src/module/notification/providers/pushover.py
+++ b/backend/src/module/notification/providers/pushover.py
@@ -18,7 +18,7 @@ class PushoverProvider(NotificationProvider):
     API_URL = "https://api.pushover.net/1/messages.json"
 
     def __init__(self, config: "ProviderConfig"):
-        super().__init__()
+        super().__init__(config)
         self.user_key = config.user_key
         self.api_token = config.api_token
 
@@ -35,7 +35,10 @@ class PushoverProvider(NotificationProvider):
         }
 
         # Add poster as supplementary URL if available
-        if notification.poster_path and notification.poster_path != "https://mikanani.me":
+        if (
+            notification.poster_path
+            and notification.poster_path != "https://mikanani.me"
+        ):
             data["url"] = notification.poster_path
             data["url_title"] = "查看海报"
 
@@ -67,3 +70,14 @@ class PushoverProvider(NotificationProvider):
                 return False, f"Pushover API returned status {resp.status_code}"
         except Exception as e:
             return False, f"Pushover test failed: {e}"
+
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a system event via Pushover."""
+        data = {
+            "token": self.api_token,
+            "user": self.user_key,
+            "title": title,
+            "message": body,
+        }
+        resp = await self.post_data(self.API_URL, data)
+        return resp.status_code == 200
diff --git a/backend/src/module/notification/providers/server_chan.py b/backend/src/module/notification/providers/server_chan.py
index 73774d0d..03f753d3 100644
--- a/backend/src/module/notification/providers/server_chan.py
+++ b/backend/src/module/notification/providers/server_chan.py
@@ -16,7 +16,7 @@ class ServerChanProvider(NotificationProvider):
     """Server Chan (Server酱) notification provider for WeChat."""
 
     def __init__(self, config: "ProviderConfig"):
-        super().__init__()
+        super().__init__(config)
         token = config.token
         self.notification_url = f"https://sctapi.ftqq.com/{token}.send"
 
@@ -46,3 +46,9 @@ class ServerChanProvider(NotificationProvider):
                 return False, f"Server Chan API returned status {resp.status_code}"
         except Exception as e:
             return False, f"Server Chan test failed: {e}"
+
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a system event via Server Chan."""
+        data = {"title": title, "desp": body}
+        resp = await self.post_data(self.notification_url, data)
+        return resp.status_code == 200
diff --git a/backend/src/module/notification/providers/telegram.py b/backend/src/module/notification/providers/telegram.py
index f9fdee86..eccef11e 100644
--- a/backend/src/module/notification/providers/telegram.py
+++ b/backend/src/module/notification/providers/telegram.py
@@ -17,7 +17,7 @@ class TelegramProvider(NotificationProvider):
     """Telegram Bot notification provider."""
 
     def __init__(self, config: "ProviderConfig"):
-        super().__init__()
+        super().__init__(config)
         token = config.token
         self.chat_id = config.chat_id
         self.photo_url = f"https://api.telegram.org/bot{token}/sendPhoto"
@@ -56,3 +56,13 @@ class TelegramProvider(NotificationProvider):
                 return False, f"Telegram API returned status {resp.status_code}"
         except Exception as e:
             return False, f"Telegram test failed: {e}"
+
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a system event via Telegram."""
+        data = {
+            "chat_id": self.chat_id,
+            "text": f"{title}\n{body}",
+            "disable_notification": True,
+        }
+        resp = await self.post_data(self.message_url, data)
+        return resp.status_code == 200
diff --git a/backend/src/module/notification/providers/webhook.py b/backend/src/module/notification/providers/webhook.py
index e5d1713b..7085cec1 100644
--- a/backend/src/module/notification/providers/webhook.py
+++ b/backend/src/module/notification/providers/webhook.py
@@ -29,9 +29,11 @@ class WebhookProvider(NotificationProvider):
     """Generic webhook notification provider with customizable templates."""
 
     def __init__(self, config: "ProviderConfig"):
-        super().__init__()
+        super().__init__(config)
         self.url = config.url
-        self.template = config.template or DEFAULT_TEMPLATE
+        # Narrower than the base class's `str | None` -- webhook always has a
+        # rendered template (falls back to DEFAULT_TEMPLATE).
+        self.template: str = config.template or DEFAULT_TEMPLATE
 
     def _render_template(self, notification: Notification) -> dict:
         """Render the template with notification data.
@@ -96,3 +98,14 @@ class WebhookProvider(NotificationProvider):
                 return False, f"Webhook returned status {resp.status_code}"
         except Exception as e:
             return False, f"Webhook test failed: {e}"
+
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a system event via webhook (fixed minimal JSON shape).
+
+        System events don't use the user-configured episode template -- it
+        expects {{title}}/{{season}}/{{episode}}/{{poster_url}}, which system
+        events don't have.
+        """
+        data = {"title": title, "message": body}
+        resp = await self._post_json(self.url, data)
+        return 200 <= resp.status_code < 300
diff --git a/backend/src/module/notification/providers/wecom.py b/backend/src/module/notification/providers/wecom.py
index dd46b757..b36875ea 100644
--- a/backend/src/module/notification/providers/wecom.py
+++ b/backend/src/module/notification/providers/wecom.py
@@ -16,7 +16,7 @@ class WecomProvider(NotificationProvider):
     """WeChat Work (企业微信) notification provider using news message format."""
 
     def __init__(self, config: "ProviderConfig"):
-        super().__init__()
+        super().__init__(config)
         # Support both webhook_url and legacy chat_id field
         self.notification_url = config.webhook_url or config.chat_id
         self.token = config.token
@@ -56,3 +56,9 @@ class WecomProvider(NotificationProvider):
                 return False, f"WeChat Work API returned status {resp.status_code}"
         except Exception as e:
             return False, f"WeChat Work test failed: {e}"
+
+    async def _deliver_text(self, title: str, body: str) -> bool:
+        """Deliver a system event via WeChat Work."""
+        data = {"key": self.token, "type": "news", "title": title, "msg": body}
+        resp = await self.post_data(self.notification_url, data)
+        return resp.status_code == 200
diff --git a/backend/src/module/rss/engine.py b/backend/src/module/rss/engine.py
index e98d78d2..44e56697 100644
--- a/backend/src/module/rss/engine.py
+++ b/backend/src/module/rss/engine.py
@@ -11,6 +11,11 @@ from module.database.bangumi import match_bangumi_in_list
 from module.downloader import DownloadClient
 from module.models import Bangumi, ResponseModel, RSSItem, Torrent
 from module.network import RequestContent
+from module.notification.events import (
+    DownloadFailureEvent,
+    RssFailureEvent,
+    SystemEvent,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -147,7 +152,9 @@ class RSSEngine:
                 return matched
         return None
 
-    async def refresh_rss(self, client: DownloadClient, rss_id: Optional[int] = None):
+    async def refresh_rss(
+        self, client: DownloadClient, rss_id: Optional[int] = None
+    ) -> list[SystemEvent]:
         # Get All RSS Items
         if not rss_id:
             rss_items: list[RSSItem] = await self.db.rss.search_active()
@@ -185,24 +192,46 @@ class RSSEngine:
         # torrent against it in memory (this is the job the old module-level
         # TTL cache existed to do).
         bangumi_list = await self.db.bangumi.search_all()
+        events: list[SystemEvent] = []
         # Process results sequentially (DB operations)
         for rss_item, (new_torrents, error) in item_results:
-            # Update connection status
+            # Update connection status. Only notify on the working->error
+            # transition (previous_status read before it's overwritten below),
+            # not on every tick a feed stays broken.
+            previous_status = rss_item.connection_status
             rss_item.connection_status = "error" if error else "healthy"
             rss_item.last_checked_at = now
             rss_item.last_error = error
             self.db.add(rss_item)
+            if error and previous_status != "error":
+                events.append(
+                    RssFailureEvent(
+                        rss_name=rss_item.name or rss_item.url,
+                        rss_url=rss_item.url,
+                        error=error,
+                    )
+                )
             for torrent in new_torrents:
                 matched_data = self.match_torrent(torrent, bangumi_list)
                 if matched_data:
                     if await client.add_torrent(torrent, matched_data):
                         logger.debug("[Engine] Add torrent %s to client", torrent.name)
                         torrent.downloaded = True
-                    # else: leave downloaded=False so a transient add failure
-                    # (e.g. downloader unreachable) gets retried next tick.
+                    else:
+                        # add_torrent() already retried the underlying fetch
+                        # internally (RequestURL.get_url, retry=3); leave
+                        # downloaded=False so a future tick can retry, and
+                        # surface the failure now.
+                        events.append(
+                            DownloadFailureEvent(
+                                official_title=matched_data.official_title,
+                                torrent_name=torrent.name,
+                            )
+                        )
             # Add all torrents to database
             await self.db.torrent.add_all(new_torrents)
         await self.db.commit()
+        return events
 
     async def download_bangumi(self, bangumi: Bangumi):
         async with RequestContent() as req:
diff --git a/backend/src/test/test_loops.py b/backend/src/test/test_loops.py
new file mode 100644
index 00000000..6adbadfb
--- /dev/null
+++ b/backend/src/test/test_loops.py
@@ -0,0 +1,125 @@
+"""Tests for core/loops.py: notifier wiring for system events.
+
+Covers the new call sites added for non-episode notifications (RSS failure,
+download failure, offset review) -- the engine/scanner logic that produces
+these events is already covered by test_rss_engine_new.py and
+test_offset_scanner.py; these tests exercise the tick-level glue that turns
+returned events into ``notifier.send_event`` calls, gated by
+``settings.notification.enable``.
+"""
+
+from unittest.mock import AsyncMock, patch
+
+from module.core.loops import offset_scan_tick, rss_tick
+from module.notification.events import OffsetReviewEvent, RssFailureEvent
+
+# ---------------------------------------------------------------------------
+# offset_scan_tick
+# ---------------------------------------------------------------------------
+
+
+class TestOffsetScanTick:
+    async def test_sends_event_per_flagged_bangumi_when_enabled(self):
+        """Each event scan_all() returns is forwarded to the notifier."""
+        event = OffsetReviewEvent(official_title="Test Anime", reason="mismatch")
+        notifier = AsyncMock()
+
+        with patch("module.core.loops.OffsetScanner") as mock_scanner_cls:
+            mock_scanner_cls.return_value.scan_all = AsyncMock(return_value=[event])
+            with patch("module.core.loops.settings") as mock_settings:
+                mock_settings.notification.enable = True
+                await offset_scan_tick(notifier)
+
+        notifier.send_event.assert_awaited_once_with(event)
+
+    async def test_no_events_sent_when_notifications_disabled(self):
+        """settings.notification.enable=False suppresses all sends."""
+        event = OffsetReviewEvent(official_title="Test Anime", reason="mismatch")
+        notifier = AsyncMock()
+
+        with patch("module.core.loops.OffsetScanner") as mock_scanner_cls:
+            mock_scanner_cls.return_value.scan_all = AsyncMock(return_value=[event])
+            with patch("module.core.loops.settings") as mock_settings:
+                mock_settings.notification.enable = False
+                await offset_scan_tick(notifier)
+
+        notifier.send_event.assert_not_awaited()
+
+    async def test_no_events_flagged_sends_nothing(self):
+        """An empty scan result sends no notifications."""
+        notifier = AsyncMock()
+
+        with patch("module.core.loops.OffsetScanner") as mock_scanner_cls:
+            mock_scanner_cls.return_value.scan_all = AsyncMock(return_value=[])
+            with patch("module.core.loops.settings") as mock_settings:
+                mock_settings.notification.enable = True
+                await offset_scan_tick(notifier)
+
+        notifier.send_event.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# rss_tick
+# ---------------------------------------------------------------------------
+
+
+def _async_cm(entered_value):
+    """Build a mock usable as ``async with X() as y:`` yielding entered_value."""
+    cm = AsyncMock()
+    cm.__aenter__ = AsyncMock(return_value=entered_value)
+    cm.__aexit__ = AsyncMock(return_value=False)
+    return cm
+
+
+class TestRssTick:
+    async def test_sends_events_returned_by_refresh_rss(self):
+        """Every event refresh_rss() returns is forwarded to the notifier."""
+        events = [RssFailureEvent(rss_name="Feed", rss_url="u", error="e")]
+        notifier = AsyncMock()
+        analyser = AsyncMock()
+
+        mock_db = AsyncMock()
+        mock_db.rss.search_aggregate = AsyncMock(return_value=[])
+        mock_engine = AsyncMock()
+        mock_engine.refresh_rss = AsyncMock(return_value=events)
+
+        with (
+            patch(
+                "module.core.loops.DownloadClient",
+                return_value=_async_cm(AsyncMock()),
+            ),
+            patch("module.core.loops.Database", return_value=_async_cm(mock_db)),
+            patch("module.core.loops.RSSEngine", return_value=mock_engine),
+            patch("module.core.loops.settings") as mock_settings,
+        ):
+            mock_settings.notification.enable = True
+            mock_settings.bangumi_manage.eps_complete = False
+            await rss_tick(analyser, notifier)
+
+        notifier.send_event.assert_awaited_once_with(events[0])
+
+    async def test_no_events_sent_when_notifications_disabled(self):
+        """settings.notification.enable=False suppresses all sends."""
+        events = [RssFailureEvent(rss_name="Feed", rss_url="u", error="e")]
+        notifier = AsyncMock()
+        analyser = AsyncMock()
+
+        mock_db = AsyncMock()
+        mock_db.rss.search_aggregate = AsyncMock(return_value=[])
+        mock_engine = AsyncMock()
+        mock_engine.refresh_rss = AsyncMock(return_value=events)
+
+        with (
+            patch(
+                "module.core.loops.DownloadClient",
+                return_value=_async_cm(AsyncMock()),
+            ),
+            patch("module.core.loops.Database", return_value=_async_cm(mock_db)),
+            patch("module.core.loops.RSSEngine", return_value=mock_engine),
+            patch("module.core.loops.settings") as mock_settings,
+        ):
+            mock_settings.notification.enable = False
+            mock_settings.bangumi_manage.eps_complete = False
+            await rss_tick(analyser, notifier)
+
+        notifier.send_event.assert_not_awaited()
diff --git a/backend/src/test/test_notification.py b/backend/src/test/test_notification.py
index b2766be3..d386f361 100644
--- a/backend/src/test/test_notification.py
+++ b/backend/src/test/test_notification.py
@@ -6,7 +6,9 @@ import pytest
 
 from module.models import Notification
 from module.models.config import NotificationProvider as ProviderConfig
+from module.network import RequestContent
 from module.notification import PROVIDER_REGISTRY, NotificationManager
+from module.notification.events import OffsetReviewEvent, RssFailureEvent
 from module.notification.providers import (
     BarkProvider,
     DiscordProvider,
@@ -83,7 +85,9 @@ class TestNotificationManager:
 
     def test_load_single_provider(self, mock_settings):
         """Manager loads a single enabled provider."""
-        config = ProviderConfig(type="telegram", enabled=True, token="test", chat_id="123")
+        config = ProviderConfig(
+            type="telegram", enabled=True, token="test", chat_id="123"
+        )
         mock_settings.notification.providers = [config]
 
         manager = NotificationManager()
@@ -92,7 +96,9 @@ class TestNotificationManager:
 
     def test_skip_disabled_provider(self, mock_settings):
         """Manager skips disabled providers."""
-        config = ProviderConfig(type="telegram", enabled=False, token="test", chat_id="123")
+        config = ProviderConfig(
+            type="telegram", enabled=False, token="test", chat_id="123"
+        )
         mock_settings.notification.providers = [config]
 
         manager = NotificationManager()
@@ -102,7 +108,9 @@ class TestNotificationManager:
         """Manager loads multiple enabled providers."""
         configs = [
             ProviderConfig(type="telegram", enabled=True, token="test", chat_id="123"),
-            ProviderConfig(type="discord", enabled=True, webhook_url="https://discord.com/webhook"),
+            ProviderConfig(
+                type="discord", enabled=True, webhook_url="https://discord.com/webhook"
+            ),
             ProviderConfig(type="bark", enabled=True, device_key="device123"),
         ]
         mock_settings.notification.providers = configs
@@ -128,7 +136,9 @@ class TestNotificationManager:
         """Manager sends to all providers."""
         configs = [
             ProviderConfig(type="telegram", enabled=True, token="test", chat_id="123"),
-            ProviderConfig(type="discord", enabled=True, webhook_url="https://discord.com/webhook"),
+            ProviderConfig(
+                type="discord", enabled=True, webhook_url="https://discord.com/webhook"
+            ),
         ]
         mock_settings.notification.providers = configs
 
@@ -146,11 +156,13 @@ class TestNotificationManager:
             await manager.send_all(notify)
 
         for provider in manager.providers:
-            provider.send.assert_called_once_with(notify)
+            provider.send.assert_called_once_with(notify)  # type: ignore[attr-defined]
 
     async def test_test_provider(self, mock_settings):
         """Manager can test a specific provider."""
-        config = ProviderConfig(type="telegram", enabled=True, token="test", chat_id="123")
+        config = ProviderConfig(
+            type="telegram", enabled=True, token="test", chat_id="123"
+        )
         mock_settings.notification.providers = [config]
 
         manager = NotificationManager()
@@ -182,18 +194,25 @@ class TestNotificationManager:
 class TestTelegramProvider:
     @pytest.fixture
     def provider(self):
-        config = ProviderConfig(type="telegram", enabled=True, token="test_token", chat_id="12345")
+        config = ProviderConfig(
+            type="telegram", enabled=True, token="test_token", chat_id="12345"
+        )
         return TelegramProvider(config)
 
     async def test_send_with_photo(self, provider):
         """Sends photo when poster available."""
         notify = Notification(
-            official_title="Test Anime", season=1, episode=5, poster_path="/path/to/poster.jpg"
+            official_title="Test Anime",
+            season=1,
+            episode=5,
+            poster_path="/path/to/poster.jpg",
         )
 
         with patch.object(provider, "post_files", new_callable=AsyncMock) as mock_post:
             mock_post.return_value = MagicMock(status_code=200)
-            with patch("module.notification.providers.telegram.load_image") as mock_load:
+            with patch(
+                "module.notification.providers.telegram.load_image"
+            ) as mock_load:
                 mock_load.return_value = b"image_data"
                 result = await provider.send(notify)
 
@@ -206,7 +225,9 @@ class TestTelegramProvider:
 
         with patch.object(provider, "post_data", new_callable=AsyncMock) as mock_post:
             mock_post.return_value = MagicMock(status_code=200)
-            with patch("module.notification.providers.telegram.load_image") as mock_load:
+            with patch(
+                "module.notification.providers.telegram.load_image"
+            ) as mock_load:
                 mock_load.return_value = None
                 result = await provider.send(notify)
 
@@ -227,14 +248,19 @@ class TestDiscordProvider:
     @pytest.fixture
     def provider(self):
         config = ProviderConfig(
-            type="discord", enabled=True, webhook_url="https://discord.com/api/webhooks/123"
+            type="discord",
+            enabled=True,
+            webhook_url="https://discord.com/api/webhooks/123",
         )
         return DiscordProvider(config)
 
     async def test_send(self, provider):
         """Sends embed message."""
         notify = Notification(
-            official_title="Test Anime", season=1, episode=5, poster_path="https://example.com/poster.jpg"
+            official_title="Test Anime",
+            season=1,
+            episode=5,
+            poster_path="https://example.com/poster.jpg",
         )
 
         with patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post:
@@ -295,6 +321,259 @@ class TestWebhookProvider:
         assert result is True
 
 
+# ---------------------------------------------------------------------------
+# Composition over inheritance (base.py no longer inherits RequestContent)
+# ---------------------------------------------------------------------------
+
+
+class TestProviderComposition:
+    def test_provider_composes_but_does_not_inherit_request_content(self):
+        """A provider holds a RequestContent instance rather than being one."""
+        config = ProviderConfig(type="bark", enabled=True, device_key="device123")
+        provider = BarkProvider(config)
+
+        assert isinstance(provider._http, RequestContent)
+        assert not isinstance(provider, RequestContent)
+
+
+# ---------------------------------------------------------------------------
+# Per-episode message templates (base._format_message)
+# ---------------------------------------------------------------------------
+
+
+class TestFormatMessageTemplate:
+    def test_format_message_no_template_uses_default_message(self):
+        """Without a template, existing configs keep the exact default message."""
+        config = ProviderConfig(
+            type="gotify", enabled=True, server_url="https://gotify.example", token="t"
+        )
+        provider = GotifyProvider(config)
+        notify = Notification(official_title="Test Anime", season=1, episode=5)
+
+        message = provider._format_message(notify)
+
+        assert message == ("番剧名称：Test Anime\n季度： 第1季\n更新集数： 第5集")
+
+    def test_format_message_with_template_substitutes_variables(self):
+        """A configured template replaces {{title}}/{{season}}/{{episode}}."""
+        config = ProviderConfig(
+            type="gotify",
+            enabled=True,
+            server_url="https://gotify.example",
+            token="t",
+            template="{{title}} S{{season}}E{{episode}}",
+        )
+        provider = GotifyProvider(config)
+        notify = Notification(official_title="Test Anime", season=1, episode=5)
+
+        assert provider._format_message(notify) == "Test Anime S1E5"
+
+
+class TestDiscordTemplate:
+    async def test_send_no_template_keeps_default_embed_description(self):
+        """No template configured: Discord's default embed text is unchanged."""
+        config = ProviderConfig(
+            type="discord", enabled=True, webhook_url="https://discord.com/webhook"
+        )
+        provider = DiscordProvider(config)
+        notify = Notification(official_title="Test Anime", season=1, episode=5)
+
+        with patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=204)
+            await provider.send(notify)
+
+        embed = mock_post.call_args[0][1]["embeds"][0]
+        assert embed["description"] == "**季度:** 第1季\n**集数:** 第5集"
+
+    async def test_send_with_template_uses_rendered_description(self):
+        """A configured template overrides the embed description."""
+        config = ProviderConfig(
+            type="discord",
+            enabled=True,
+            webhook_url="https://discord.com/webhook",
+            template="{{title}} ep {{episode}}",
+        )
+        provider = DiscordProvider(config)
+        notify = Notification(official_title="Test Anime", season=1, episode=5)
+
+        with patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=204)
+            await provider.send(notify)
+
+        embed = mock_post.call_args[0][1]["embeds"][0]
+        assert embed["description"] == "Test Anime ep 5"
+
+
+# ---------------------------------------------------------------------------
+# System events (RSS failure / download failure / offset review)
+# ---------------------------------------------------------------------------
+
+
+class TestSendEvent:
+    async def test_send_event_delivers_title_and_body_from_event(self):
+        """send_event() formats the event and hands (title, body) to _deliver_text."""
+        config = ProviderConfig(type="bark", enabled=True, device_key="device123")
+        provider = BarkProvider(config)
+        event = RssFailureEvent(
+            rss_name="My Feed", rss_url="https://x.example/rss", error="timeout"
+        )
+
+        with patch.object(
+            provider, "_deliver_text", new_callable=AsyncMock
+        ) as mock_deliver:
+            mock_deliver.return_value = True
+            result = await provider.send_event(event)
+
+        assert result is True
+        title, body = mock_deliver.call_args[0]
+        assert title == "RSS 订阅连接异常"
+        assert "My Feed" in body
+        assert "timeout" in body
+
+    async def test_manager_send_event_broadcasts_to_all_providers(self):
+        """NotificationManager.send_event fans out to every enabled provider."""
+        with patch("module.notification.manager.settings") as mock_settings:
+            configs = [
+                ProviderConfig(
+                    type="telegram", enabled=True, token="test", chat_id="123"
+                ),
+                ProviderConfig(type="bark", enabled=True, device_key="device123"),
+            ]
+            mock_settings.notification.providers = configs
+            manager = NotificationManager()
+
+        for provider in manager.providers:
+            provider.send_event = AsyncMock(return_value=True)
+            provider.__aenter__ = AsyncMock(return_value=provider)
+            provider.__aexit__ = AsyncMock(return_value=None)
+
+        event = OffsetReviewEvent(official_title="Test Anime", reason="mismatch")
+        await manager.send_event(event)
+
+        for provider in manager.providers:
+            provider.send_event.assert_called_once_with(  # type: ignore[attr-defined]
+                event
+            )
+
+
+class TestDeliverText:
+    """Each provider's own ``_deliver_text`` hook, exercised directly."""
+
+    async def test_bark_deliver_text_posts_title_and_body(self):
+        config = ProviderConfig(type="bark", enabled=True, device_key="device123")
+        provider = BarkProvider(config)
+
+        with patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=200)
+            result = await provider._deliver_text("标题", "正文")
+
+        assert result is True
+        data = mock_post.call_args[0][1]
+        assert data["title"] == "标题"
+        assert data["body"] == "正文"
+
+    async def test_discord_deliver_text_posts_embed(self):
+        config = ProviderConfig(
+            type="discord", enabled=True, webhook_url="https://discord.com/webhook"
+        )
+        provider = DiscordProvider(config)
+
+        with patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=204)
+            result = await provider._deliver_text("标题", "正文")
+
+        assert result is True
+        embed = mock_post.call_args[0][1]["embeds"][0]
+        assert embed["title"] == "标题"
+        assert embed["description"] == "正文"
+
+    async def test_gotify_deliver_text_posts_title_and_message(self):
+        config = ProviderConfig(
+            type="gotify", enabled=True, server_url="https://gotify.example", token="t"
+        )
+        provider = GotifyProvider(config)
+
+        with patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=200)
+            result = await provider._deliver_text("标题", "正文")
+
+        assert result is True
+        data = mock_post.call_args[0][1]
+        assert data["title"] == "标题"
+        assert data["message"] == "正文"
+
+    async def test_pushover_deliver_text_posts_title_and_message(self):
+        config = ProviderConfig(
+            type="pushover", enabled=True, user_key="u", api_token="a"
+        )
+        provider = PushoverProvider(config)
+
+        with patch.object(provider, "post_data", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=200)
+            result = await provider._deliver_text("标题", "正文")
+
+        assert result is True
+        data = mock_post.call_args[0][1]
+        assert data["title"] == "标题"
+        assert data["message"] == "正文"
+
+    async def test_server_chan_deliver_text_posts_title_and_desp(self):
+        config = ProviderConfig(type="server-chan", enabled=True, token="t")
+        provider = ServerChanProvider(config)
+
+        with patch.object(provider, "post_data", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=200)
+            result = await provider._deliver_text("标题", "正文")
+
+        assert result is True
+        data = mock_post.call_args[0][1]
+        assert data["title"] == "标题"
+        assert data["desp"] == "正文"
+
+    async def test_telegram_deliver_text_posts_combined_text(self):
+        config = ProviderConfig(
+            type="telegram", enabled=True, token="test_token", chat_id="12345"
+        )
+        provider = TelegramProvider(config)
+
+        with patch.object(provider, "post_data", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=200)
+            result = await provider._deliver_text("标题", "正文")
+
+        assert result is True
+        data = mock_post.call_args[0][1]
+        assert data["text"] == "标题\n正文"
+
+    async def test_webhook_deliver_text_posts_title_and_message(self):
+        config = ProviderConfig(
+            type="webhook", enabled=True, url="https://example.com/webhook"
+        )
+        provider = WebhookProvider(config)
+
+        with patch.object(provider, "_post_json", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=200)
+            result = await provider._deliver_text("标题", "正文")
+
+        assert result is True
+        data = mock_post.call_args[0][1]
+        assert data == {"title": "标题", "message": "正文"}
+
+    async def test_wecom_deliver_text_posts_title_and_msg(self):
+        config = ProviderConfig(
+            type="wecom", enabled=True, webhook_url="https://wecom.example/webhook"
+        )
+        provider = WecomProvider(config)
+
+        with patch.object(provider, "post_data", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = MagicMock(status_code=200)
+            result = await provider._deliver_text("标题", "正文")
+
+        assert result is True
+        data = mock_post.call_args[0][1]
+        assert data["title"] == "标题"
+        assert data["msg"] == "正文"
+
+
 # ---------------------------------------------------------------------------
 # Config Migration
 # ---------------------------------------------------------------------------
@@ -322,7 +601,9 @@ class TestConfigMigration:
         """New format with providers doesn't trigger migration."""
         from module.models.config import Notification as NotificationConfig
 
-        provider = ProviderConfig(type="discord", enabled=True, webhook_url="https://discord.com/webhook")
+        provider = ProviderConfig(
+            type="discord", enabled=True, webhook_url="https://discord.com/webhook"
+        )
         new_config = NotificationConfig(
             enable=True,
             providers=[provider],
diff --git a/backend/src/test/test_rss_engine_new.py b/backend/src/test/test_rss_engine_new.py
index 19bd4a94..0e364074 100644
--- a/backend/src/test/test_rss_engine_new.py
+++ b/backend/src/test/test_rss_engine_new.py
@@ -8,6 +8,7 @@ import pytest_asyncio
 
 from module.database import Database
 from module.models import Torrent
+from module.notification.events import DownloadFailureEvent, RssFailureEvent
 from module.rss.engine import RSSEngine
 from test.factories import make_bangumi, make_rss_item, make_torrent
 
@@ -272,6 +273,116 @@ class TestRefreshRss:
         mock_get.assert_not_called()
 
 
+# ---------------------------------------------------------------------------
+# refresh_rss notification events (RSS failure transition, download failure)
+# ---------------------------------------------------------------------------
+
+
+class TestRefreshRssEvents:
+    async def test_healthy_to_error_transition_returns_rss_failure_event(
+        self, rss_engine
+    ):
+        """A feed going from healthy to error is reported exactly once."""
+        rss_item = make_rss_item(
+            name="Feed 1", url="https://feed1.example/rss", enabled=True
+        )
+        await rss_engine.db.rss.add(rss_item)
+
+        with patch.object(
+            RSSEngine, "_pull_rss_with_status", new_callable=AsyncMock
+        ) as mock_pull:
+            mock_pull.return_value = ([], "connection refused")
+            client = AsyncMock()
+            events = await rss_engine.refresh_rss(client)
+
+        assert len(events) == 1
+        assert isinstance(events[0], RssFailureEvent)
+        assert events[0].rss_name == "Feed 1"
+        assert events[0].error == "connection refused"
+
+    async def test_error_stays_error_does_not_repeat_event(self, rss_engine):
+        """A feed already in 'error' does not re-fire on every subsequent tick."""
+        rss_item = make_rss_item(url="https://feed1.example/rss", enabled=True)
+        await rss_engine.db.rss.add(rss_item)
+
+        with patch.object(
+            RSSEngine, "_pull_rss_with_status", new_callable=AsyncMock
+        ) as mock_pull:
+            mock_pull.return_value = ([], "connection refused")
+            client = AsyncMock()
+            first_events = await rss_engine.refresh_rss(client)
+            second_events = await rss_engine.refresh_rss(client)
+
+        assert len(first_events) == 1
+        assert second_events == []
+
+    async def test_no_error_returns_no_rss_failure_event(self, rss_engine):
+        """A healthy feed produces no failure event."""
+        rss_item = make_rss_item(url="https://feed1.example/rss", enabled=True)
+        await rss_engine.db.rss.add(rss_item)
+
+        with patch.object(
+            RSSEngine, "_pull_rss_with_status", new_callable=AsyncMock
+        ) as mock_pull:
+            mock_pull.return_value = ([], None)
+            client = AsyncMock()
+            events = await rss_engine.refresh_rss(client)
+
+        assert events == []
+
+    async def test_add_torrent_failure_returns_download_failure_event(self, rss_engine):
+        """A matched torrent that fails to add is reported as a download failure."""
+        rss_item = make_rss_item(enabled=True)
+        await rss_engine.db.rss.add(rss_item)
+        bangumi = make_bangumi(title_raw="Mushoku Tensei", filter="")
+        await rss_engine.db.bangumi.add(bangumi)
+
+        new_torrent = Torrent(
+            name="[Sub] Mushoku Tensei - 12 [1080p].mkv",
+            url="https://example.com/ep12.torrent",
+        )
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
+            mock_get.return_value = [new_torrent]
+            client = AsyncMock()
+            client.add_torrent = AsyncMock(return_value=False)
+
+            events = await rss_engine.refresh_rss(client)
+
+        assert len(events) == 1
+        assert isinstance(events[0], DownloadFailureEvent)
+        assert events[0].official_title == bangumi.official_title
+        assert events[0].torrent_name == new_torrent.name
+
+        all_torrents = await rss_engine.db.torrent.search_all()
+        assert all_torrents[0].downloaded is False
+
+    async def test_add_torrent_success_returns_no_download_failure_event(
+        self, rss_engine, mock_qb_client
+    ):
+        """A torrent that adds successfully produces no failure event."""
+        rss_item = make_rss_item(enabled=True)
+        await rss_engine.db.rss.add(rss_item)
+        bangumi = make_bangumi(title_raw="Mushoku Tensei", filter="")
+        await rss_engine.db.bangumi.add(bangumi)
+
+        new_torrent = Torrent(
+            name="[Sub] Mushoku Tensei - 12 [1080p].mkv",
+            url="https://example.com/ep12.torrent",
+        )
+        with patch.object(
+            RSSEngine, "_get_torrents", new_callable=AsyncMock
+        ) as mock_get:
+            mock_get.return_value = [new_torrent]
+            client = AsyncMock()
+            client.add_torrent = AsyncMock(return_value=True)
+
+            events = await rss_engine.refresh_rss(client)
+
+        assert events == []
+
+
 # ---------------------------------------------------------------------------
 # add_rss
 # ---------------------------------------------------------------------------