diff --git a/core/automation/blocks.py b/core/automation/blocks.py
index 7f33301e..1656ee29 100644
--- a/core/automation/blocks.py
+++ b/core/automation/blocks.py
@@ -261,7 +261,10 @@ ACTIONS: list[dict] = [
     {"type": "video_deep_scan_movies", "label": "Deep Scan Movie Library", "icon": "search", "scope": "video",
      "description": "Full reconcile of the Movie library: re-read every movie from the server and drop ones it no longer has (a read, NOT a Plex disk-scan; never touches TV)", "available": True},
     {"type": "video_add_airing_episodes", "label": "Wishlist Today's Airings", "icon": "calendar", "scope": "video",
-     "description": "Sonarr-style: add every episode airing today (for shows you follow) to the wishlist, skipping ones you already own", "available": True},
+     "description": "Sonarr-style: add every episode airing today (for shows you follow) to the wishlist, skipping ones you already own. Also tidies the watchlist by dropping shows that have ended/been canceled.", "available": True,
+     "config_fields": [
+         {"key": "prune_ended", "type": "checkbox", "label": "Also remove ended/canceled shows from the watchlist", "default": True}
+     ]},
 ]
 
 
diff --git a/core/automation/handlers/video_auto_wishlist_airing.py b/core/automation/handlers/video_auto_wishlist_airing.py
index 33c98af0..db142d2d 100644
--- a/core/automation/handlers/video_auto_wishlist_airing.py
+++ b/core/automation/handlers/video_auto_wishlist_airing.py
@@ -52,6 +52,67 @@ def _default_season_meta(tmdb_id: Any, season_number: Any):
     return get_video_enrichment_engine().tmdb_season(tmdb_id, season_number)
 
 
+# ── watchlist hygiene: drop shows that have ended/been canceled ───────────────
+_TERMINAL = ('ended', 'canceled', 'cancelled', 'completed')
+
+
+def _is_terminal_status(status) -> bool:
+    """A show that won't air again — pointless to keep on the (watch-for-new) list."""
+    return str(status or '').strip().lower() in _TERMINAL
+
+
+def _default_fetch_follows() -> List[Dict[str, Any]]:
+    from api.video import get_video_db
+    return get_video_db().followed_shows()
+
+
+def _default_show_status(tmdb_id: Any) -> Optional[str]:
+    """TMDB status for a follow with no local status (a tmdb-only follow). Cached by
+    the engine; returns None on any hiccup so we never prune on uncertainty."""
+    from core.video.enrichment.engine import get_video_enrichment_engine
+    d = get_video_enrichment_engine().tmdb_full_detail('show', tmdb_id) or {}
+    return d.get('status')
+
+
+def _default_remove_show(tmdb_id: Any) -> None:
+    from api.video import get_video_db
+    get_video_db().remove_from_watchlist('show', tmdb_id)
+
+
+def prune_ended_show_follows(deps, automation_id=None, *, fetch_follows=None,
+                             show_status=None, remove_show=None) -> int:
+    """Remove explicitly-followed shows that are no longer airing.
+
+    Auto-airing LIBRARY shows are already excluded by status, so this targets explicit
+    eye-button follows (which persist regardless). For follows with no local status (a
+    tmdb-only follow), look the status up on TMDB. Only prunes on a DEFINITIVE terminal
+    status — unknown status is left alone. Pure: all I/O injected. Returns count removed."""
+    fetch_follows = fetch_follows or _default_fetch_follows
+    show_status = show_status or _default_show_status
+    remove_show = remove_show or _default_remove_show
+    removed = 0
+    for f in (fetch_follows() or []):
+        tid = f.get('tmdb_id')
+        if not tid:
+            continue
+        status = f.get('status')
+        if not status:                       # tmdb-only follow — ask TMDB
+            try:
+                status = show_status(tid)
+            except Exception:   # noqa: BLE001 - never prune on a lookup failure
+                status = None
+        if status and _is_terminal_status(status):
+            try:
+                remove_show(tid)
+                removed += 1
+                deps.update_progress(
+                    automation_id, log_line="Removed ended show '%s' from the watchlist"
+                    % (f.get('title') or tid), log_type='info')
+            except Exception:   # noqa: BLE001
+                pass
+    return removed
+
+
 def _season_lookup(season_meta, tmdb_id, season_number, cache):
     """(season_poster_url, {episode_number: tmdb_episode}) for a show+season, fetched
     once and cached. A TMDB hiccup degrades to empty (DB values fill in)."""
@@ -74,17 +135,32 @@ def auto_video_add_airing_episodes(
     add_episodes: Optional[Callable[[Any, Any, List[Dict[str, Any]]], int]] = None,
     today_fn: Optional[Callable[[], str]] = None,
     season_meta: Optional[Callable[[Any, Any], Any]] = None,
+    prune_follows: Optional[Callable[[], List[Dict[str, Any]]]] = None,
+    show_status: Optional[Callable[[Any], Any]] = None,
+    remove_show: Optional[Callable[[Any], None]] = None,
 ) -> Dict[str, Any]:
-    """Add today's airing (unowned, followed-show) episodes to the video wishlist.
+    """Add today's airing (unowned, followed-show) episodes to the video wishlist, and
+    first tidy the watchlist by dropping shows that have ended / been canceled.
 
-    Returns ``{'status': 'completed', 'episodes_added': int, 'shows': int, ...}``."""
+    Returns ``{'status': 'completed', 'episodes_added': int, 'shows': int,
+    'shows_pruned': int, ...}``."""
     fetch_airing = fetch_airing or _default_fetch_airing
     add_episodes = add_episodes or _default_add_episodes
     season_meta = season_meta or _default_season_meta
     today_fn = today_fn or (lambda: date.today().isoformat())
     automation_id = config.get('_automation_id')
+    prune_ended = config.get('prune_ended', True)
     try:
         today = today_fn()
+        # Watchlist hygiene first: a followed show that has since ended/been canceled
+        # won't air again, so drop it (ended LIBRARY shows are already auto-excluded;
+        # this catches explicit eye-button follows).
+        pruned = 0
+        if prune_ended:
+            deps.update_progress(automation_id, phase='Tidying the watchlist…', progress=10,
+                                 log_line='Removing shows that have ended or been canceled', log_type='info')
+            pruned = prune_ended_show_follows(deps, automation_id, fetch_follows=prune_follows,
+                                              show_status=show_status, remove_show=remove_show)
         deps.update_progress(automation_id, phase="Checking today's airings…", progress=25,
                              log_line='Reading the calendar for episodes airing today', log_type='info')
         rows = fetch_airing(today) or []
@@ -126,13 +202,15 @@ def auto_video_add_airing_episodes(
                                       _show_poster_url(grp['library_id'])) or 0)
         shows = len(by_show)
 
+        done = ('Added %d airing episode(s) across %d show(s) to the wishlist'
+                % (added, shows)) if added else 'No new airing episodes to wishlist today'
+        if pruned:
+            done += ' · pruned %d ended show(s)' % pruned
         deps.update_progress(
             automation_id, status='finished', progress=100, phase='Complete',
-            log_line=('Added %d airing episode(s) across %d show(s) to the wishlist'
-                      % (added, shows)) if added else 'No new airing episodes to wishlist today',
-            log_type='success')
+            log_line=done, log_type='success')
         return {'status': 'completed', 'episodes_added': added, 'shows': shows,
-                '_manages_own_progress': True}
+                'shows_pruned': pruned, '_manages_own_progress': True}
     except Exception as e:  # noqa: BLE001
         deps.update_progress(automation_id, status='error', phase='Error', log_line=str(e), log_type='error')
         return {'status': 'error', 'error': str(e), '_manages_own_progress': True}
diff --git a/database/video_database.py b/database/video_database.py
index 88b4dcf7..faf31b5a 100644
--- a/database/video_database.py
+++ b/database/video_database.py
@@ -2055,6 +2055,19 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    def followed_shows(self) -> list[dict]:
+        """Explicitly-followed shows (state='follow') with their library status when
+        owned (NULL for tmdb-only follows). Used by the watchlist-prune pass to drop
+        shows that have since ended/been canceled."""
+        conn = self._get_connection()
+        try:
+            return [dict(r) for r in conn.execute(
+                "SELECT w.tmdb_id, w.title, w.library_id, s.status "
+                "FROM video_watchlist w LEFT JOIN shows s ON s.id = w.library_id "
+                "WHERE w.kind='show' AND w.state='follow'")]
+        finally:
+            conn.close()
+
     def remove_from_watchlist(self, kind: str, tmdb_id: int) -> bool:
         """Un-follow. Stored as a 'mute' tombstone (not a delete) so an
         actively-airing library show — watched by default — is not silently
diff --git a/tests/test_video_auto_wishlist_airing.py b/tests/test_video_auto_wishlist_airing.py
index b945f5a1..2f85def8 100644
--- a/tests/test_video_auto_wishlist_airing.py
+++ b/tests/test_video_auto_wishlist_airing.py
@@ -34,7 +34,7 @@ def test_adds_unowned_airings_grouped_by_show():
         return len(eps)
 
     res = auto_video_add_airing_episodes(
-        {"_automation_id": "a1"}, _Deps(),
+        {"_automation_id": "a1", "prune_ended": False}, _Deps(),
         fetch_airing=lambda today: rows, add_episodes=add, today_fn=lambda: "2026-06-21",
         season_meta=lambda *a: None)
 
@@ -66,7 +66,7 @@ def test_uses_tmdb_season_metadata_like_a_manual_add():
         captured["eps"] = eps
         return len(eps)
 
-    auto_video_add_airing_episodes({"_automation_id": "a"}, _Deps(),
+    auto_video_add_airing_episodes({"_automation_id": "a", "prune_ended": False}, _Deps(),
                                    fetch_airing=lambda t: rows, add_episodes=add,
                                    today_fn=lambda: "2026-06-21", season_meta=season_meta)
     ep = captured["eps"][0]
@@ -85,7 +85,7 @@ def test_falls_back_to_db_values_when_tmdb_unavailable():
         captured["eps"] = eps
         return len(eps)
 
-    auto_video_add_airing_episodes({"_automation_id": "a"}, _Deps(),
+    auto_video_add_airing_episodes({"_automation_id": "a", "prune_ended": False}, _Deps(),
                                    fetch_airing=lambda t: rows, add_episodes=add,
                                    today_fn=lambda: "2026-06-21", season_meta=lambda *a: None)
     ep = captured["eps"][0]
@@ -100,14 +100,14 @@ def test_queries_the_calendar_for_today():
         seen["today"] = today
         return []
 
-    auto_video_add_airing_episodes({"_automation_id": "a"}, _Deps(),
+    auto_video_add_airing_episodes({"_automation_id": "a", "prune_ended": False}, _Deps(),
                                    fetch_airing=fetch, add_episodes=lambda *a: 0,
                                    today_fn=lambda: "2026-06-21")
     assert seen["today"] == "2026-06-21"     # start == end == today
 
 
 def test_nothing_airing_is_a_clean_noop():
-    res = auto_video_add_airing_episodes({"_automation_id": "a"}, _Deps(),
+    res = auto_video_add_airing_episodes({"_automation_id": "a", "prune_ended": False}, _Deps(),
                                          fetch_airing=lambda t: [], add_episodes=lambda *a: 0,
                                          today_fn=lambda: "2026-06-21")
     assert res["status"] == "completed" and res["episodes_added"] == 0
@@ -118,7 +118,64 @@ def test_error_is_caught_and_reported():
         raise RuntimeError("calendar down")
 
     deps = _Deps()
-    res = auto_video_add_airing_episodes({"_automation_id": "a"}, deps,
+    res = auto_video_add_airing_episodes({"_automation_id": "a", "prune_ended": False}, deps,
                                          fetch_airing=boom, add_episodes=lambda *a: 0)
     assert res["status"] == "error" and "calendar down" in res["error"]
     assert any(p.get("status") == "error" for p in deps.progress)
+
+
+# ── watchlist hygiene: prune ended/canceled follows ─────────────────────────
+from core.automation.handlers.video_auto_wishlist_airing import prune_ended_show_follows  # noqa: E402
+
+
+def test_prune_removes_ended_and_canceled_follows():
+    follows = [
+        {"tmdb_id": 1, "title": "Returning Show", "status": "Returning Series"},
+        {"tmdb_id": 2, "title": "Done Show", "status": "Ended"},
+        {"tmdb_id": 3, "title": "Axed Show", "status": "Canceled"},
+        {"tmdb_id": 4, "title": "Live Show", "status": "In Production"},
+    ]
+    removed = []
+    n = prune_ended_show_follows(_Deps(), "a", fetch_follows=lambda: follows,
+                                 show_status=lambda t: None, remove_show=removed.append)
+    assert n == 2 and removed == [2, 3]            # only Ended + Canceled
+
+
+def test_prune_looks_up_status_for_tmdb_only_follows():
+    # no local status → fetch from TMDB; ended → prune
+    follows = [{"tmdb_id": 9, "title": "TMDB-only", "status": None}]
+    removed = []
+    prune_ended_show_follows(_Deps(), "a", fetch_follows=lambda: follows,
+                             show_status=lambda t: "Ended", remove_show=removed.append)
+    assert removed == [9]
+
+
+def test_prune_never_removes_on_unknown_status_or_lookup_error():
+    def _boom(t):
+        raise RuntimeError("tmdb down")
+    follows = [{"tmdb_id": 1, "title": "Unknown", "status": None},   # lookup returns None
+               {"tmdb_id": 2, "title": "Errored", "status": None}]   # lookup raises
+    removed = []
+    n = prune_ended_show_follows(
+        _Deps(), "a", fetch_follows=lambda: follows,
+        show_status=lambda t: (_boom(t) if t == 2 else None), remove_show=removed.append)
+    assert n == 0 and removed == []                # uncertainty → keep
+
+
+def test_airing_handler_runs_the_prune_pass():
+    removed = []
+    res = auto_video_add_airing_episodes(
+        {"_automation_id": "a"}, _Deps(),
+        fetch_airing=lambda t: [], add_episodes=lambda *a, **k: 0, today_fn=lambda: "2026-06-21",
+        prune_follows=lambda: [{"tmdb_id": 7, "title": "Old", "status": "Ended"}],
+        show_status=lambda t: None, remove_show=removed.append)
+    assert res["status"] == "completed" and res["shows_pruned"] == 1 and removed == [7]
+
+
+def test_airing_handler_can_disable_the_prune():
+    called = {"n": 0}
+    auto_video_add_airing_episodes(
+        {"_automation_id": "a", "prune_ended": False}, _Deps(),
+        fetch_airing=lambda t: [], add_episodes=lambda *a, **k: 0, today_fn=lambda: "2026-06-21",
+        prune_follows=lambda: called.update(n=called["n"] + 1) or [])
+    assert called["n"] == 0                          # prune skipped entirely
diff --git a/tests/test_video_followed_shows.py b/tests/test_video_followed_shows.py
new file mode 100644
index 00000000..10cae453
--- /dev/null
+++ b/tests/test_video_followed_shows.py
@@ -0,0 +1,35 @@
+"""followed_shows(): explicit show follows + their library status — feeds the
+watchlist-prune pass that drops ended/canceled shows."""
+
+from __future__ import annotations
+
+import pytest
+
+from database.video_database import VideoDatabase
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+
+
+def test_lists_explicit_follows_with_library_status(db):
+    # a tmdb-only follow (no library row → status None)
+    db.add_to_watchlist("show", 100, "TMDB Only")
+    # a follow backed by a library show carrying a status
+    conn = db._get_connection()
+    conn.execute("INSERT INTO shows (id, server_source, title, tmdb_id, status) "
+                 "VALUES (5, 'plex', 'Owned Show', 200, 'Ended')")
+    conn.commit(); conn.close()
+    db.add_to_watchlist("show", 200, "Owned Show", library_id=5)
+
+    rows = {r["tmdb_id"]: r for r in db.followed_shows()}
+    assert rows[100]["status"] is None                 # tmdb-only → no local status
+    assert rows[200]["status"] == "Ended"              # owned → carries the status
+
+
+def test_excludes_muted_and_people(db):
+    db.add_to_watchlist("show", 1, "A")
+    db.remove_from_watchlist("show", 1)                 # mute (tombstone)
+    db.add_to_watchlist("person", 2, "Someone")
+    assert db.followed_shows() == []                    # neither muted shows nor people