diff --git a/backend/src/module/api/bangumi.py b/backend/src/module/api/bangumi.py
index 514908ba..2785376d 100644
--- a/backend/src/module/api/bangumi.py
+++ b/backend/src/module/api/bangumi.py
@@ -6,7 +6,8 @@ from pydantic import BaseModel
 
 from module.conf import settings
 from module.database import Database, get_db
-from module.manager import TorrentManager
+from module.downloader import DownloadClient
+from module.manager import Renamer, TorrentManager
 from module.models import APIResponse, Bangumi, BangumiUpdate, ResponseModel
 from module.parser.analyser.offset_detector import (
     OffsetSuggestion as DetectorSuggestion,
@@ -67,12 +68,6 @@ class DetectOffsetResponse(BaseModel):
 router = APIRouter(prefix="/bangumi", tags=["bangumi"])
 
 
-def str_to_list(data: Bangumi):
-    data.filter = data.filter.split(",")
-    data.rss_link = data.rss_link.split(",")
-    return data
-
-
 @router.get(
     "/get/all", response_model=list[Bangumi], dependencies=[Depends(get_current_user)]
 )
@@ -337,7 +332,8 @@ async def detect_offset(request: DetectOffsetRequest):
             has_mismatch=True,
             suggestion=OffsetSuggestionDetail(
                 season_offset=suggestion.season_offset,
-                episode_offset=suggestion.episode_offset,
+                # None means "no episode offset needed" (see offset_detector).
+                episode_offset=suggestion.episode_offset or 0,
                 reason=suggestion.reason,
                 confidence=suggestion.confidence,
             ),
@@ -381,6 +377,73 @@ async def dismiss_review(bangumi_id: int):
         )
 
 
+async def _trigger_rename() -> None:
+    """Run a rename pass so applied offsets take effect immediately."""
+    async with DownloadClient() as client:
+        renamer = Renamer(client)
+        await renamer.rename()
+
+
+# Registered before /apply-offset/{bangumi_id} so "many" is never captured as an id.
+@router.post(
+    path="/apply-offset/many",
+    response_model=APIResponse,
+    dependencies=[Depends(get_current_user)],
+)
+async def apply_offset_many(bangumi_id: list[int]):
+    """Apply suggested offsets for multiple bangumi and trigger a rename pass."""
+    if not bangumi_id:
+        return _empty_id_list_response()
+
+    async with Database() as db:
+        results = [await db.bangumi.apply_offset(i) for i in bangumi_id]
+
+    succeeded = sum(1 for r in results if r)
+    if succeeded:
+        await _trigger_rename()
+
+    all_ok = succeeded == len(results)
+    return u_response(
+        ResponseModel(
+            status_code=200 if all_ok else 500,
+            status=all_ok,
+            msg_en=f"Applied offset for {succeeded}/{len(results)} bangumi.",
+            msg_zh=f"已为 {succeeded}/{len(results)} 部番剧应用偏移量。",
+        )
+    )
+
+
+@router.post(
+    path="/apply-offset/{bangumi_id}",
+    response_model=APIResponse,
+    dependencies=[Depends(get_current_user)],
+)
+async def apply_offset(bangumi_id: int):
+    """Apply the suggested season/episode offset and trigger a rename pass."""
+    async with Database() as db:
+        success = await db.bangumi.apply_offset(bangumi_id)
+
+    if success:
+        await _trigger_rename()
+        return JSONResponse(
+            status_code=200,
+            content={
+                "status": True,
+                "msg_en": "Offset applied.",
+                "msg_zh": "已应用偏移量。",
+            },
+        )
+    else:
+        return JSONResponse(
+            status_code=404,
+            content={
+                "status": False,
+                "msg_en": f"Bangumi {bangumi_id} not found.",
+                "msg_zh": f"未找到番剧 {bangumi_id}。",
+            },
+        )
+
+
 @router.get(
     path="/needs-review",
     response_model=list[Bangumi],
diff --git a/backend/src/module/core/offset_scanner.py b/backend/src/module/core/offset_scanner.py
index 9d3e6933..d2d3b859 100644
--- a/backend/src/module/core/offset_scanner.py
+++ b/backend/src/module/core/offset_scanner.py
@@ -5,7 +5,9 @@ import logging
 from module.conf import settings
 from module.database import Database
 from module.models import Bangumi
+from module.notification.events import OffsetReviewEvent
 from module.parser.analyser.offset_detector import detect_offset_mismatch
+from module.parser.analyser.raw_parser import raw_parser
 from module.parser.analyser.tmdb_parser import tmdb_parser
 
 logger = logging.getLogger(__name__)
@@ -14,11 +16,11 @@ logger = logging.getLogger(__name__)
 class OffsetScanner:
     """Periodically scan bangumi for season/episode mismatches with TMDB."""
 
-    async def scan_all(self) -> int:
+    async def scan_all(self) -> list[OffsetReviewEvent]:
         """Scan all active bangumi for offset mismatches.
 
         Returns:
-            Number of bangumi flagged for review.
+            One event per bangumi flagged for review.
         """
         logger.info("[OffsetScanner] Starting offset scan...")
 
@@ -27,45 +29,47 @@ class OffsetScanner:
 
         if not bangumi_list:
             logger.debug("[OffsetScanner] No active bangumi to scan.")
-            return 0
+            return []
 
-        flagged_count = 0
+        events: list[OffsetReviewEvent] = []
         for bangumi in bangumi_list:
             try:
-                if await self._check_bangumi(bangumi):
-                    flagged_count += 1
+                event = await self._check_bangumi(bangumi)
+                if event is not None:
+                    events.append(event)
             except Exception as e:
                 logger.warning(
                     f"[OffsetScanner] Error checking {bangumi.official_title}: {e}"
                 )
 
         logger.info(
-            f"[OffsetScanner] Scan complete. Flagged {flagged_count} bangumi for review."
+            f"[OffsetScanner] Scan complete. Flagged {len(events)} bangumi for review."
         )
-        return flagged_count
+        return events
 
-    async def _check_bangumi(self, bangumi: Bangumi) -> bool:
+    async def _check_bangumi(self, bangumi: Bangumi) -> OffsetReviewEvent | None:
         """Check a single bangumi for offset mismatch.
 
         Args:
             bangumi: The bangumi to check.
 
         Returns:
-            True if flagged for review, False otherwise.
+            An event describing the flag if the bangumi was flagged for
+            review, None otherwise.
         """
         # Skip if already needs review
         if bangumi.needs_review:
             logger.debug(
                 f"[OffsetScanner] Skipping {bangumi.official_title}: already needs review"
             )
-            return False
+            return None
 
         # Skip if user has already configured offsets
         if bangumi.season_offset != 0 or bangumi.episode_offset != 0:
             logger.debug(
                 f"[OffsetScanner] Skipping {bangumi.official_title}: has configured offsets"
             )
-            return False
+            return None
 
         # Get TMDB info
         language = settings.rss_parser.language
@@ -75,11 +79,16 @@ class OffsetScanner:
             logger.debug(
                 f"[OffsetScanner] Skipping {bangumi.official_title}: no TMDB info"
             )
-            return False
+            return None
 
-        # Get latest episode for this bangumi (use season as proxy since we don't track episodes)
-        # For now, we'll check based on the bangumi's season
-        parsed_episode = 1  # Default to episode 1 for season-based detection
+        # Get the real latest parsed episode from this bangumi's torrent records,
+        # instead of guessing. No torrents parsed yet means no signal to act on.
+        parsed_episode = await self._get_latest_parsed_episode(bangumi.id)
+        if parsed_episode is None:
+            logger.debug(
+                f"[OffsetScanner] Skipping {bangumi.official_title}: no parsed episode data"
+            )
+            return None
 
         # Detect mismatch
         suggestion = detect_offset_mismatch(
@@ -100,9 +109,29 @@ class OffsetScanner:
                 f"[OffsetScanner] Flagged {bangumi.official_title} for review: {suggestion.reason} "
                 f"(suggested: season={suggestion.season_offset}, episode={suggestion.episode_offset})"
             )
-            return True
+            return OffsetReviewEvent(
+                official_title=bangumi.official_title, reason=suggestion.reason
+            )
 
-        return False
+        return None
+
+    async def _get_latest_parsed_episode(self, bangumi_id: int) -> int | None:
+        """从 Torrent 表解析该番剧已知的最新集数，作为真实信号使用。
+
+        Returns:
+            解析到的最大集数；若没有种子记录或均无法解析出集数则返回 None。
+        """
+        async with Database() as db:
+            torrents = await db.torrent.search_by_bangumi_id(bangumi_id)
+
+        latest: int | None = None
+        for torrent in torrents:
+            episode = raw_parser(torrent.name)
+            if episode is None or not episode.episode:
+                continue
+            if latest is None or episode.episode > latest:
+                latest = episode.episode
+        return latest
 
     async def check_single(self, bangumi_id: int) -> bool:
         """Check a single bangumi by ID.
@@ -120,4 +149,4 @@ class OffsetScanner:
             logger.warning(f"[OffsetScanner] Bangumi {bangumi_id} not found")
             return False
 
-        return await self._check_bangumi(bangumi)
+        return await self._check_bangumi(bangumi) is not None
diff --git a/backend/src/module/database/bangumi.py b/backend/src/module/database/bangumi.py
index 0dc807ee..a2545ac3 100644
--- a/backend/src/module/database/bangumi.py
+++ b/backend/src/module/database/bangumi.py
@@ -173,7 +173,7 @@ class BangumiDatabase:
             )
         )
         result = await self.session.execute(statement)
-        candidates = result.scalars().all()
+        candidates = list(result.scalars().all())
 
         match = _find_semantic_match(data, candidates)
         if match:
@@ -187,7 +187,7 @@ class BangumiDatabase:
         return match
 
     async def add_title_alias(
-        self, bangumi_id: int, new_title_raw: str, auto_commit: bool = True
+        self, bangumi_id: int, new_title_raw: str | None, auto_commit: bool = True
     ) -> bool:
         """
         Add a new title_raw alias to an existing bangumi.
@@ -298,13 +298,15 @@ class BangumiDatabase:
         official_titles = {d.official_title for d in to_add}
         candidates_by_title: dict[str, list[Bangumi]] = {}
         if official_titles:
-            statement = select(Bangumi).where(
+            # SQLModel 类属性在 mypy 看来是普通字段类型而非 InstrumentedAttribute，
+            # 无法识别 .in_() 等查询方法（无官方 mypy 插件支持）。
+            dup_statement = select(Bangumi).where(
                 and_(
-                    Bangumi.official_title.in_(official_titles),
+                    Bangumi.official_title.in_(official_titles),  # type: ignore[attr-defined]
                     Bangumi.deleted == false(),
                 )
             )
-            result = await self.session.execute(statement)
+            result = await self.session.execute(dup_statement)
             for candidate in result.scalars().all():
                 candidates_by_title.setdefault(candidate.official_title, []).append(
                     candidate
@@ -371,7 +373,9 @@ class BangumiDatabase:
             )
         return len(unique_to_add)
 
-    async def update(self, data: Bangumi | BangumiUpdate, _id: int = None) -> bool:
+    async def update(
+        self, data: Bangumi | BangumiUpdate, _id: int | None = None
+    ) -> bool:
         if _id and isinstance(data, BangumiUpdate):
             db_data = await self.session.get(Bangumi, _id)
         elif isinstance(data, Bangumi):
@@ -454,7 +458,7 @@ class BangumiDatabase:
         """Batch lookup multiple bangumi by their IDs."""
         if not ids:
             return []
-        statement = select(Bangumi).where(Bangumi.id.in_(ids))
+        statement = select(Bangumi).where(Bangumi.id.in_(ids))  # type: ignore[attr-defined]
         result = await self.session.execute(statement)
         return list(result.scalars().all())
 
@@ -464,7 +468,7 @@ class BangumiDatabase:
         )
         result = await self.session.execute(statement)
         data = result.scalar_one_or_none()
-        return data.poster_link if data else ""
+        return (data.poster_link or "") if data else ""
 
     async def match_list(self, torrent_list: list, rss_link: str) -> list:
         match_datas = await self.search_all()
@@ -534,11 +538,13 @@ class BangumiDatabase:
         return list(result.scalars().all())
 
     async def not_added(self) -> list[Bangumi]:
+        # SQLModel 类属性在 mypy 看来是普通字段类型而非 InstrumentedAttribute，
+        # 无法识别 .is_() 等查询方法（无官方 mypy 插件支持）。
         conditions = select(Bangumi).where(
             or_(
                 Bangumi.added == 0,
-                Bangumi.rule_name.is_(None),
-                Bangumi.save_path.is_(None),
+                Bangumi.rule_name.is_(None),  # type: ignore[union-attr]
+                Bangumi.save_path.is_(None),  # type: ignore[union-attr]
             )
         )
         result = await self.session.execute(conditions)
@@ -688,6 +694,33 @@ class BangumiDatabase:
         )
         return True
 
+    async def apply_offset(self, _id: int) -> bool:
+        """将建议的季度/集数偏移写入正式的 season_offset/episode_offset，并清除检查标记。
+
+        Args:
+            _id: The bangumi ID
+        """
+        bangumi = await self.session.get(Bangumi, _id)
+        if not bangumi:
+            return False
+        if bangumi.suggested_season_offset is not None:
+            bangumi.season_offset = bangumi.suggested_season_offset
+        if bangumi.suggested_episode_offset is not None:
+            bangumi.episode_offset = bangumi.suggested_episode_offset
+        bangumi.needs_review = False
+        bangumi.needs_review_reason = None
+        bangumi.suggested_season_offset = None
+        bangumi.suggested_episode_offset = None
+        self.session.add(bangumi)
+        await self.session.commit()
+        logger.debug(
+            "[Database] Applied offset for bangumi id %s: season=%s, episode=%s",
+            _id,
+            bangumi.season_offset,
+            bangumi.episode_offset,
+        )
+        return True
+
     async def clear_needs_review(self, _id: int) -> bool:
         """Clear the needs_review flag and suggested offsets for a bangumi."""
         bangumi = await self.session.get(Bangumi, _id)
diff --git a/backend/src/module/database/torrent.py b/backend/src/module/database/torrent.py
index 188d33da..21d6228f 100644
--- a/backend/src/module/database/torrent.py
+++ b/backend/src/module/database/torrent.py
@@ -54,7 +54,9 @@ class TorrentDatabase:
         if not torrents_list:
             return []
         urls = [t.url for t in torrents_list]
-        statement = select(Torrent.url).where(Torrent.url.in_(urls))
+        # SQLModel 类属性在 mypy 看来是普通字段类型而非 InstrumentedAttribute，
+        # 无法识别 .in_()/.is_() 等查询方法（无官方 mypy 插件支持）。
+        statement = select(Torrent.url).where(Torrent.url.in_(urls))  # type: ignore[attr-defined]
         result = await self.session.execute(statement)
         existing_urls = set(result.scalars().all())
         return [t for t in torrents_list if t.url not in existing_urls]
@@ -71,7 +73,14 @@ class TorrentDatabase:
         if not qb_hashes:
             return []
         result = await self.session.execute(
-            select(Torrent).where(Torrent.qb_hash.in_(qb_hashes))
+            select(Torrent).where(Torrent.qb_hash.in_(qb_hashes))  # type: ignore[union-attr]
+        )
+        return list(result.scalars().all())
+
+    async def search_by_bangumi_id(self, bangumi_id: int) -> list[Torrent]:
+        """Find all torrent records associated with a bangumi."""
+        result = await self.session.execute(
+            select(Torrent).where(Torrent.bangumi_id == bangumi_id)
         )
         return list(result.scalars().all())
 
diff --git a/backend/src/test/test_api_bangumi_extended.py b/backend/src/test/test_api_bangumi_extended.py
index 1efb9561..9b1d302c 100644
--- a/backend/src/test/test_api_bangumi_extended.py
+++ b/backend/src/test/test_api_bangumi_extended.py
@@ -287,6 +287,97 @@ class TestNeedsReview:
         assert response.status_code == 404
 
 
+# ---------------------------------------------------------------------------
+# Apply offset endpoints
+# ---------------------------------------------------------------------------
+
+
+class TestApplyOffset:
+    def test_apply_offset_success_triggers_rename(self, authed_client):
+        """POST /bangumi/apply-offset/{id} applies offset and reruns the renamer."""
+        mock_db = MagicMock()
+        mock_db.bangumi.apply_offset = AsyncMock(return_value=True)
+        mock_renamer = AsyncMock()
+        with patch("module.api.bangumi.Database", _async_db_mock(mock_db)):
+            with patch("module.api.bangumi.Renamer", return_value=mock_renamer):
+                with patch("module.api.bangumi.DownloadClient") as MockClient:
+                    MockClient.return_value.__aenter__ = AsyncMock(
+                        return_value=MagicMock()
+                    )
+                    MockClient.return_value.__aexit__ = AsyncMock(return_value=False)
+                    response = authed_client.post("/api/v1/bangumi/apply-offset/1")
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] is True
+        mock_db.bangumi.apply_offset.assert_awaited_once_with(1)
+        mock_renamer.rename.assert_awaited_once()
+
+    def test_apply_offset_not_found_skips_rename(self, authed_client):
+        """POST /bangumi/apply-offset/{id} with a non-existent bangumi does not rename."""
+        mock_db = MagicMock()
+        mock_db.bangumi.apply_offset = AsyncMock(return_value=False)
+        with patch("module.api.bangumi.Database", _async_db_mock(mock_db)):
+            with patch("module.api.bangumi.Renamer") as MockRenamer:
+                response = authed_client.post("/api/v1/bangumi/apply-offset/999")
+
+        assert response.status_code == 404
+        MockRenamer.assert_not_called()
+
+    def test_apply_offset_many_success(self, authed_client):
+        """POST /bangumi/apply-offset/many applies offsets to every id and renames once."""
+        mock_db = MagicMock()
+        mock_db.bangumi.apply_offset = AsyncMock(side_effect=[True, True])
+        mock_renamer = AsyncMock()
+        with patch("module.api.bangumi.Database", _async_db_mock(mock_db)):
+            with patch("module.api.bangumi.Renamer", return_value=mock_renamer):
+                with patch("module.api.bangumi.DownloadClient") as MockClient:
+                    MockClient.return_value.__aenter__ = AsyncMock(
+                        return_value=MagicMock()
+                    )
+                    MockClient.return_value.__aexit__ = AsyncMock(return_value=False)
+                    response = authed_client.post(
+                        "/api/v1/bangumi/apply-offset/many", json=[1, 2]
+                    )
+
+        assert response.status_code == 200
+        assert mock_db.bangumi.apply_offset.await_count == 2
+        mock_renamer.rename.assert_awaited_once()
+
+    def test_apply_offset_many_partial_failure(self, authed_client):
+        """POST /bangumi/apply-offset/many reports a partial failure and still renames."""
+        mock_db = MagicMock()
+        mock_db.bangumi.apply_offset = AsyncMock(side_effect=[True, False])
+        mock_renamer = AsyncMock()
+        with patch("module.api.bangumi.Database", _async_db_mock(mock_db)):
+            with patch("module.api.bangumi.Renamer", return_value=mock_renamer):
+                with patch("module.api.bangumi.DownloadClient") as MockClient:
+                    MockClient.return_value.__aenter__ = AsyncMock(
+                        return_value=MagicMock()
+                    )
+                    MockClient.return_value.__aexit__ = AsyncMock(return_value=False)
+                    response = authed_client.post(
+                        "/api/v1/bangumi/apply-offset/many", json=[1, 2]
+                    )
+
+        assert response.status_code == 500
+        mock_renamer.rename.assert_awaited_once()
+
+    def test_apply_offset_many_empty_list(self, authed_client):
+        """POST /bangumi/apply-offset/many with an empty list is a 400, no DB/rename calls."""
+        with patch("module.api.bangumi.Renamer") as MockRenamer:
+            response = authed_client.post("/api/v1/bangumi/apply-offset/many", json=[])
+
+        assert response.status_code == 400
+        MockRenamer.assert_not_called()
+
+    def test_apply_offset_auth_required(self, unauthed_client):
+        """POST /bangumi/apply-offset/{id} requires authentication."""
+        with patch("module.security.api.DEV_AUTH_BYPASS", False):
+            response = unauthed_client.post("/api/v1/bangumi/apply-offset/1")
+        assert response.status_code == 401
+
+
 # ---------------------------------------------------------------------------
 # Batch operations
 # ---------------------------------------------------------------------------
@@ -308,9 +399,7 @@ class TestSetWeekday:
 
     def test_set_weekday_out_of_range(self, authed_client):
         """PATCH /bangumi/{id}/weekday rejects weekday outside 0-6."""
-        response = authed_client.patch(
-            "/api/v1/bangumi/1/weekday", json={"weekday": 9}
-        )
+        response = authed_client.patch("/api/v1/bangumi/1/weekday", json={"weekday": 9})
         assert response.status_code == 400
 
 
diff --git a/backend/src/test/test_database.py b/backend/src/test/test_database.py
index 57106181..f7a14ace 100644
--- a/backend/src/test/test_database.py
+++ b/backend/src/test/test_database.py
@@ -50,12 +50,14 @@ async def test_bangumi_database(db_session):
     # insert
     await db.add(test_data)
     result = await db.search_id(1)
+    assert result is not None
     assert result.official_title == test_data.official_title
 
     # update
     test_data.official_title = "无职转生，到了异世界就拿出真本事II"
     await db.update(test_data)
     result = await db.search_id(1)
+    assert result is not None
     assert result.official_title == test_data.official_title
 
     # search poster
@@ -66,6 +68,7 @@ async def test_bangumi_database(db_session):
     result = await db.match_torrent(
         "[Lilith-Raws] 无职转生，到了异世界就拿出真本事 / Mushoku Tensei - 11 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4]"
     )
+    assert result is not None
     assert result.official_title == "无职转生，到了异世界就拿出真本事II"
 
     # delete
@@ -84,12 +87,14 @@ async def test_torrent_database(db_session):
     # insert
     await db.add(test_data)
     result = await db.search(1)
+    assert result is not None
     assert result.name == test_data.name
 
     # update
     test_data.downloaded = True
     await db.update(test_data)
     result = await db.search(1)
+    assert result is not None
     assert result.downloaded is True
 
 
@@ -99,6 +104,7 @@ async def test_rss_database(db_session):
 
     await db.add(RSSItem(url=rss_url, name="Test RSS"))
     result = await db.search_id(1)
+    assert result is not None
     assert result.url == rss_url
 
 
@@ -178,6 +184,7 @@ async def test_torrent_update_qb_hash(db_session):
 
     # Verify update
     result = await db.search(torrent.id)
+    assert result is not None
     assert result.qb_hash == "new_hash_value"
 
 
@@ -265,6 +272,7 @@ async def test_add_title_alias(db_session):
 
     # Verify alias was added
     updated = await db.search_id(bangumi_id)
+    assert updated is not None
     assert updated.title_aliases is not None
     aliases = json.loads(updated.title_aliases)
     assert "Test Anime Season 1" in aliases
@@ -486,6 +494,7 @@ async def test_add_all_semantic_duplicates_across_batch_all_merged_as_aliases(
     assert len(await db.search_all()) == 5
     for i in range(5):
         original = await db.search_official_title(f"Anime {i}")
+        assert original is not None
         aliases = json.loads(original.title_aliases) if original.title_aliases else []
         assert f"Raw {i} New" in aliases
 
@@ -623,6 +632,103 @@ class TestDeleteByBangumiId:
         assert new[0].url == "https://mikan.me/t/001"
 
 
+class TestSearchByBangumiId:
+    """Tests for TorrentDatabase.search_by_bangumi_id."""
+
+    async def test_returns_matching_torrents(self, db_session):
+        db = TorrentDatabase(db_session)
+        await _ensure_bangumi(db_session, 11)
+        await db.add(
+            Torrent(name="ep01", url="https://example.com/ep01", bangumi_id=11)
+        )
+        await db.add(
+            Torrent(name="ep02", url="https://example.com/ep02", bangumi_id=11)
+        )
+
+        torrents = await db.search_by_bangumi_id(11)
+        assert {t.name for t in torrents} == {"ep01", "ep02"}
+
+    async def test_excludes_other_bangumi_torrents(self, db_session):
+        db = TorrentDatabase(db_session)
+        await _ensure_bangumi(db_session, 21)
+        await _ensure_bangumi(db_session, 22)
+        await db.add(
+            Torrent(name="keep", url="https://example.com/keep", bangumi_id=21)
+        )
+        await db.add(
+            Torrent(name="other", url="https://example.com/other", bangumi_id=22)
+        )
+
+        torrents = await db.search_by_bangumi_id(21)
+        assert [t.name for t in torrents] == ["keep"]
+
+    async def test_no_match_returns_empty_list(self, db_session):
+        db = TorrentDatabase(db_session)
+        assert await db.search_by_bangumi_id(999) == []
+
+
+class TestApplyOffset:
+    """Tests for BangumiDatabase.apply_offset."""
+
+    async def test_copies_suggested_offsets_into_live_offsets_and_clears_review(
+        self, db_session
+    ):
+        db = BangumiDatabase(db_session)
+        bangumi = Bangumi(
+            official_title="Apply Offset Anime",
+            title_raw="Apply Offset Anime Raw",
+            group_name="TestGroup",
+            rss_link="r",
+            needs_review=True,
+            needs_review_reason="mismatch detected",
+            suggested_season_offset=-1,
+            suggested_episode_offset=12,
+        )
+        db_session.add(bangumi)
+        await db_session.commit()
+
+        result = await db.apply_offset(bangumi.id)
+        assert result is True
+
+        updated = await db.search_id(bangumi.id)
+        assert updated is not None
+        assert updated.season_offset == -1
+        assert updated.episode_offset == 12
+        assert updated.needs_review is False
+        assert updated.needs_review_reason is None
+        assert updated.suggested_season_offset is None
+        assert updated.suggested_episode_offset is None
+
+    async def test_leaves_episode_offset_untouched_when_no_episode_suggestion(
+        self, db_session
+    ):
+        """episode_offset suggestion of None means "no change needed", not "reset to 0"."""
+        db = BangumiDatabase(db_session)
+        bangumi = Bangumi(
+            official_title="Season Only Anime",
+            title_raw="Season Only Anime Raw",
+            group_name="TestGroup",
+            rss_link="r",
+            episode_offset=3,
+            needs_review=True,
+            suggested_season_offset=-1,
+            suggested_episode_offset=None,
+        )
+        db_session.add(bangumi)
+        await db_session.commit()
+
+        await db.apply_offset(bangumi.id)
+
+        updated = await db.search_id(bangumi.id)
+        assert updated is not None
+        assert updated.season_offset == -1
+        assert updated.episode_offset == 3
+
+    async def test_nonexistent_bangumi_returns_false(self, db_session):
+        db = BangumiDatabase(db_session)
+        assert await db.apply_offset(999) is False
+
+
 def test_groups_are_similar():
     """Test group name similarity detection."""
     from module.database.bangumi import _groups_are_similar
@@ -666,6 +772,7 @@ async def test_get_all_title_patterns(db_session):
 
     # Get all patterns
     updated = await db.search_id(bangumi_id)
+    assert updated is not None
     patterns = db.get_all_title_patterns(updated)
 
     assert len(patterns) == 3
diff --git a/backend/src/test/test_offset_scanner.py b/backend/src/test/test_offset_scanner.py
new file mode 100644
index 00000000..48d1d270
--- /dev/null
+++ b/backend/src/test/test_offset_scanner.py
@@ -0,0 +1,105 @@
+"""Tests for OffsetScanner: real parsed-episode signal instead of a hardcoded guess."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from module.core.offset_scanner import OffsetScanner
+from module.database import Database
+from module.models import Torrent
+from module.notification.events import OffsetReviewEvent
+from test.factories import make_bangumi
+
+
+class TestGetLatestParsedEpisode:
+    """Tests for OffsetScanner._get_latest_parsed_episode."""
+
+    async def test_returns_none_when_no_torrents(self):
+        scanner = OffsetScanner()
+        assert await scanner._get_latest_parsed_episode(999) is None
+
+    async def test_returns_max_parsed_episode_across_torrents(self):
+        scanner = OffsetScanner()
+        async with Database() as db:
+            bangumi = make_bangumi(title_raw="Signal Anime Raw")
+            await db.bangumi.add(bangumi)
+            for ep in (1, 5, 3):
+                await db.torrent.add(
+                    Torrent(
+                        name=f"[TestGroup] Signal Anime Raw - {ep:02d} [1080p].mkv",
+                        url=f"https://example.com/{ep}",
+                        bangumi_id=bangumi.id,
+                    )
+                )
+
+        latest = await scanner._get_latest_parsed_episode(bangumi.id)
+        assert latest == 5
+
+    async def test_ignores_torrents_that_fail_to_parse(self):
+        scanner = OffsetScanner()
+        async with Database() as db:
+            bangumi = make_bangumi(title_raw="Unparseable Anime Raw")
+            await db.bangumi.add(bangumi)
+            await db.torrent.add(
+                Torrent(
+                    name="this has no episode markers at all",
+                    url="https://example.com/unparseable",
+                    bangumi_id=bangumi.id,
+                )
+            )
+
+        assert await scanner._get_latest_parsed_episode(bangumi.id) is None
+
+
+class TestCheckBangumiUsesRealSignal:
+    """Tests for OffsetScanner._check_bangumi consuming the real torrent signal."""
+
+    async def test_skips_when_no_torrent_signal(self):
+        """No torrent records for the bangumi means no real signal: never flag."""
+        scanner = OffsetScanner()
+        bangumi = make_bangumi(id=1, season=2, season_offset=0, episode_offset=0)
+
+        with patch(
+            "module.core.offset_scanner.tmdb_parser",
+            new=AsyncMock(return_value=MagicMock(last_season=1)),
+        ):
+            event = await scanner._check_bangumi(bangumi)
+
+        assert event is None
+
+    async def test_flags_using_real_latest_episode_from_torrent_table(self):
+        """A mismatch is only suggested from the actual parsed episode, not a constant."""
+        scanner = OffsetScanner()
+        async with Database() as db:
+            bangumi = make_bangumi(
+                title_raw="Signal Anime Raw 2",
+                season=2,
+                season_offset=0,
+                episode_offset=0,
+            )
+            await db.bangumi.add(bangumi)
+            await db.torrent.add(
+                Torrent(
+                    name="[TestGroup] Signal Anime Raw 2 - 25 [1080p].mkv",
+                    url="https://example.com/25",
+                    bangumi_id=bangumi.id,
+                )
+            )
+
+        fake_tmdb_info = MagicMock()
+        fake_tmdb_info.last_season = 1
+        fake_tmdb_info.season_episode_counts = {1: 24}
+        fake_tmdb_info.series_status = "Ended"
+        fake_tmdb_info.virtual_season_starts = None
+
+        with patch(
+            "module.core.offset_scanner.tmdb_parser",
+            new=AsyncMock(return_value=fake_tmdb_info),
+        ):
+            event = await scanner._check_bangumi(bangumi)
+
+        assert isinstance(event, OffsetReviewEvent)
+        assert event.official_title == bangumi.official_title
+
+        async with Database() as db:
+            updated = await db.bangumi.search_id(bangumi.id)
+        assert updated.needs_review is True
+        assert updated.suggested_season_offset == -1