Merge #124: match RSS results against wanted episodes

Closes #124

.sqlx/query-3fc7d7e39520dc60fa81da21a4cae0c4a718e95f9718629f0aba9f3d00d179e1.json

@@ -0,0 +1,26 @@
+{
+  "db_name": "SQLite",
+  "query": "\n        SELECT   e.air_date,\n                 EXISTS (\n                     SELECT 1 FROM media_files f\n                     WHERE  f.owner_kind = 'episode' AND f.owner_id = e.id\n                 ) AS \"on_disk!: bool\"\n        FROM     episodes e\n        WHERE    e.season_id = ?\n        ORDER BY e.number\n        ",
+  "describe": {
+    "columns": [
+      {
+        "name": "air_date",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "on_disk!: bool",
+        "ordinal": 1,
+        "type_info": "Null"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true,
+      null
+    ]
+  },
+  "hash": "3fc7d7e39520dc60fa81da21a4cae0c4a718e95f9718629f0aba9f3d00d179e1"
+}

.sqlx/query-90799bcbc272af8890296ca4f2d90843a8bc71c820b56847e0a86a9497f03952.json

@@ -0,0 +1,50 @@
+{
+  "db_name": "SQLite",
+  "query": "\n        SELECT   e.id     AS \"id!: i64\",\n                 e.number AS \"episode!: i64\",\n                 se.id     AS \"season_id!: i64\",\n                 se.number AS \"season!: i64\",\n                 s.tmdb_id AS \"series_tmdb_id!: i64\",\n                 s.title   AS \"series_title!: String\"\n        FROM     episodes e\n        JOIN     seasons se ON se.id = e.season_id\n        JOIN     series s ON s.id = se.series_id\n        WHERE    e.wanted = 1\n          AND    NOT EXISTS (\n                      SELECT 1 FROM media_files f\n                      WHERE  f.owner_kind = 'episode' AND f.owner_id = e.id\n                  )\n          AND    NOT EXISTS (\n                      SELECT 1 FROM grabs g\n                      WHERE  g.target_kind = 'episode' AND g.target_id = e.id\n                        AND  g.state IN ('sent', 'downloaded', 'imported')\n                  )\n          AND    NOT EXISTS (\n                      SELECT 1 FROM grabs g\n                      WHERE  g.target_kind = 'season' AND g.target_id = se.id\n                        AND  g.state IN ('sent', 'downloaded', 'imported')\n                  )\n        ORDER BY e.id\n        ",
+  "describe": {
+    "columns": [
+      {
+        "name": "id!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      },
+      {
+        "name": "episode!: i64",
+        "ordinal": 1,
+        "type_info": "Integer"
+      },
+      {
+        "name": "season_id!: i64",
+        "ordinal": 2,
+        "type_info": "Integer"
+      },
+      {
+        "name": "season!: i64",
+        "ordinal": 3,
+        "type_info": "Integer"
+      },
+      {
+        "name": "series_tmdb_id!: i64",
+        "ordinal": 4,
+        "type_info": "Integer"
+      },
+      {
+        "name": "series_title!: String",
+        "ordinal": 5,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "90799bcbc272af8890296ca4f2d90843a8bc71c820b56847e0a86a9497f03952"
+}

.sqlx/query-d6d54d747b699ed28ebf5eb1ba5e73285778a186fd1da64a91ad18cd1c5c14d5.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "\n        SELECT   s.original_language\n        FROM     seasons se\n        JOIN     series s ON s.id = se.series_id\n        WHERE    se.id = ?\n        ",
+  "describe": {
+    "columns": [
+      {
+        "name": "original_language",
+        "ordinal": 0,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": [
+      true
+    ]
+  },
+  "hash": "d6d54d747b699ed28ebf5eb1ba5e73285778a186fd1da64a91ad18cd1c5c14d5"
+}

crates/arr-daemon/src/rss.rs

@@ -9,19 +9,31 @@
 //!
 //! `blocked` is honoured the other way round from targeted search (§6.3): it
 //! stops a title being searched for, and leaves it matching RSS.
+//!
+//! One pass serves films and episodes alike: each feed item goes through the
+//! movie matcher first, then the episode one (`arr_core::matching`). A
+//! single-episode release grabs its wanted, missing episode directly; a
+//! season pack only grabs when [`season_grab_mode`] allows packs for that
+//! season — the same guard §14 gives targeted search, so a season behaves
+//! the same however a release is found (§6.2). Like everything else here,
+//! none of it backs off or counts attempts.
 
 use std::collections::hash_map::Entry;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
 
-use arr_core::matching::{match_movie, MatchKind, ReleaseIds, WantedMovie};
+use arr_core::grabbing::{season_grab_mode, SeasonGrabFacts, SeasonGrabMode};
-use arr_core::{Language, MovieId};
+use arr_core::matching::{
-use arr_db::{Blacklist, Db, MoviePolicy};
+    match_episode, match_movie, MatchKind, MatchShape, ReleaseIds, WantedEpisode, WantedMovie,
+};
+use arr_core::{EpisodeId, Language, MovieId};
+use arr_db::{Blacklist, Db, MoviePolicy, TitlePolicy};
 use arr_dl::TransmissionClient;
 use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest};
 
 use crate::grab::{
-    store_release, Eligible, GrabError, GrabScope, GrabTarget, Grabber, SeedingRules,
+    store_episode_release, store_release, Eligible, GrabError, GrabScope, GrabTarget, Grabber,
+    SeedingRules,
 };
 use crate::indexers::IndexerDirectory;
 use crate::reconcile::{Action, ActionFuture, Outcome};
@@ -51,7 +63,8 @@ impl RssAction {
 
     async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, GrabError> {
         let wanted = wanted_movies(database).await?;
-        if wanted.is_empty() {
+        let episodes = wanted_episodes(database).await?;
+        if wanted.is_empty() && episodes.is_empty() {
             return Ok(Vec::new());
         }
         let indexers = self.indexers.searchable().await?;
@@ -62,12 +75,12 @@ impl RssAction {
 
         let releases = self.feeds(&indexers).await;
         let blacklist = Blacklist::load(database.pool()).await?;
-        let winners = self
+        let (movies, tv) = self
-            .match_feeds(database, &wanted, releases, &blacklist)
+            .match_feeds(database, &wanted, &episodes, releases, &blacklist)
             .await?;
 
         let mut outcomes = Vec::new();
-        for (movie, winner) in winners {
+        for (movie, winner) in movies {
             let outcome = self
                 .grabber
                 .send_winner(
@@ -86,6 +99,29 @@ impl RssAction {
                 .await?;
             outcomes.extend(outcome);
         }
+        for TvWinner {
+            grabbable,
+            scope,
+            candidate: winner,
+        } in tv
+        {
+            let outcome = self
+                .grabber
+                .send_winner(
+                    database,
+                    &GrabTarget {
+                        scope,
+                        title: &grabbable.series_title,
+                        // Same property as the movie lane above.
+                        counts_as_attempt: false,
+                    },
+                    &grabbable.policy,
+                    &blacklist,
+                    winner,
+                )
+                .await?;
+            outcomes.extend(outcome);
+        }
         Ok(outcomes)
     }
 
@@ -116,11 +152,13 @@ impl RssAction {
         &self,
         database: &Db,
         wanted: &[WantedMovie],
+        episodes: &[WantedEpisodeRow],
         releases: Vec<SearchRelease>,
         blacklist: &Blacklist,
-    ) -> Result<Vec<(Grabbable, Eligible)>, GrabError> {
+    ) -> Result<(Vec<(Grabbable, Eligible)>, Vec<TvWinner>), GrabError> {
         let mut policies: HashMap<i64, Option<Grabbable>> = HashMap::new();
         let mut best: HashMap<i64, Eligible> = HashMap::new();
+        let mut tv = TvCandidates::default();
 
         for release in releases {
             let claims = arr_parse::parse(&release.name);
@@ -128,9 +166,7 @@ impl RssAction {
                 tmdb_id: release.tmdb_id,
                 imdb_id: release.imdb_id.clone(),
             };
-            let Some(matched) = match_movie(wanted, &ids, &claims) else {
+            if let Some(matched) = match_movie(wanted, &ids, &claims) {
-                continue;
-            };
                 let movie_id = matched.movie.0;
                 let grabbable = match policies.entry(movie_id) {
                     Entry::Occupied(entry) => entry.into_mut(),
@@ -151,7 +187,7 @@ impl RssAction {
                     },
                     "RSS result matched a wanted title"
                 );
-            let stored = store_release(
+                if let Some(candidate) = store_release(
                     database,
                     movie_id,
                     &release,
@@ -160,24 +196,209 @@ impl RssAction {
                     &grabbable.language,
                     blacklist,
                 )
-            .await?;
+                .await?
-            let Some(candidate) = stored else {
+                {
-                continue;
-            };
                     let incumbent = best.get(&movie_id);
                     if incumbent.is_none_or(|incumbent| beats(&candidate, incumbent)) {
                         best.insert(movie_id, candidate);
                     }
                 }
+                continue;
+            }
+            tv.offer(database, episodes, &release, &ids, &claims, blacklist)
+                .await?;
+        }
 
-        let mut winners: Vec<(Grabbable, Eligible)> = best
+        let tv_winners = tv.winners(database, episodes).await?;
+        let mut movies: Vec<(Grabbable, Eligible)> = best
             .into_iter()
             .filter_map(|(movie_id, candidate)| {
                 let grabbable = policies.get(&movie_id).and_then(Clone::clone)?;
                 Some((grabbable, candidate))
             })
             .collect();
-        winners.sort_by_key(|(movie, _)| movie.id);
+        movies.sort_by_key(|(movie, _)| movie.id);
+        Ok((movies, tv_winners))
+    }
+}
+
+/// The episode lane's state over one feed pass: the grab context per season
+/// and the best eligible candidate per wanted episode and per season's pack.
+#[derive(Default)]
+struct TvCandidates {
+    policies: HashMap<i64, Option<TvGrabbable>>,
+    best_single: HashMap<i64, Eligible>,
+    /// Per season: the best pack alongside the open episodes it covers.
+    best_pack: HashMap<i64, (Eligible, Vec<i64>)>,
+}
+
+impl TvCandidates {
+    /// Offer one feed item to every wanted episode. A release holds either
+    /// exactly one episode or whole seasons, so each row it matches lands in
+    /// exactly one bucket.
+    async fn offer(
+        &mut self,
+        database: &Db,
+        episodes: &[WantedEpisodeRow],
+        release: &SearchRelease,
+        ids: &ReleaseIds,
+        claims: &arr_parse::NameClaims,
+        blacklist: &Blacklist,
+    ) -> Result<(), GrabError> {
+        let mut packs: HashMap<i64, Vec<i64>> = HashMap::new();
+        let mut single: Option<(i64, MatchKind)> = None;
+        for row in episodes {
+            let Some(matched) = match_episode(&row.wanted, ids, claims) else {
+                continue;
+            };
+            match matched.shape {
+                MatchShape::SeasonPack => {
+                    packs
+                        .entry(row.season_id)
+                        .or_default()
+                        .push(row.wanted.id.0);
+                }
+                MatchShape::SingleEpisode => single = Some((row.wanted.id.0, matched.kind)),
+            }
+        }
+        for (season_id, covered) in packs {
+            let Some(candidate) = self
+                .store_candidate(database, episodes, season_id, &covered, release, blacklist)
+                .await?
+            else {
+                continue;
+            };
+            let incumbent = self.best_pack.get(&season_id);
+            if incumbent.is_none_or(|(incumbent, _)| beats(&candidate, incumbent)) {
+                self.best_pack.insert(season_id, (candidate, covered));
+            }
+        }
+        if let Some((episode_id, kind)) = single {
+            tracing::info!(
+                episode_id,
+                release = release.name,
+                by = tv_match_kind(kind),
+                "RSS result matched a wanted episode"
+            );
+            let Some(row) = episodes.iter().find(|row| row.wanted.id.0 == episode_id) else {
+                return Ok(());
+            };
+            let Some(candidate) = self
+                .store_candidate(
+                    database,
+                    episodes,
+                    row.season_id,
+                    &[episode_id],
+                    release,
+                    blacklist,
+                )
+                .await?
+            else {
+                return Ok(());
+            };
+            let incumbent = self.best_single.get(&episode_id);
+            if incumbent.is_none_or(|incumbent| beats(&candidate, incumbent)) {
+                self.best_single.insert(episode_id, candidate);
+            }
+        }
+        Ok(())
+    }
+
+    /// Classify and store a feed item against the open episodes it covers,
+    /// loading the season's grab context on first use.
+    async fn store_candidate(
+        &mut self,
+        database: &Db,
+        episodes: &[WantedEpisodeRow],
+        season_id: i64,
+        covered: &[i64],
+        release: &SearchRelease,
+        blacklist: &Blacklist,
+    ) -> Result<Option<Eligible>, GrabError> {
+        let grabbable = match self.policies.entry(season_id) {
+            Entry::Occupied(entry) => entry.into_mut(),
+            Entry::Vacant(entry) => {
+                entry.insert(load_tv_grabbable(database, episodes, season_id).await?)
+            }
+        };
+        let Some(grabbable) = grabbable.as_ref() else {
+            return Ok(None);
+        };
+        let (_, stored) = store_episode_release(
+            database,
+            covered,
+            release,
+            &grabbable.policy.policy,
+            &grabbable.policy.overrides,
+            &grabbable.language,
+            blacklist,
+        )
+        .await?;
+        Ok(stored)
+    }
+
+    /// Turn the per-season and per-episode winners into grabs. A season with
+    /// an eligible pack only takes it when [`season_grab_mode`] allows packs
+    /// there — the guard targeted search obeys (§6.2, §14) — and when it
+    /// does, the season's singles stand down. Otherwise each open episode
+    /// takes its own winner.
+    async fn winners(
+        &self,
+        database: &Db,
+        episodes: &[WantedEpisodeRow],
+    ) -> Result<Vec<TvWinner>, GrabError> {
+        let mut winners = Vec::new();
+        let mut covered_by_packs: HashSet<i64> = HashSet::new();
+
+        let mut seasons: Vec<i64> = self.best_pack.keys().copied().collect();
+        seasons.sort_unstable();
+        for season_id in seasons {
+            let Some((candidate, episode_ids)) = self.best_pack.get(&season_id) else {
+                continue;
+            };
+            let Some(grabbable) = self.policies.get(&season_id).and_then(Clone::clone) else {
+                continue;
+            };
+            if !pack_allowed(database, season_id).await? {
+                tracing::info!(
+                    season_id,
+                    series = grabbable.series_title,
+                    "RSS skips a season pack the season's grab mode refuses"
+                );
+                continue;
+            }
+            covered_by_packs.extend(episode_ids.iter().copied());
+            winners.push(TvWinner {
+                grabbable,
+                scope: GrabScope::Season {
+                    season_id,
+                    episode_ids: episode_ids.clone(),
+                },
+                candidate: candidate.clone(),
+            });
+        }
+
+        let mut singles: Vec<i64> = self.best_single.keys().copied().collect();
+        singles.sort_unstable();
+        for episode_id in singles {
+            if covered_by_packs.contains(&episode_id) {
+                continue;
+            }
+            let Some(row) = episodes.iter().find(|row| row.wanted.id.0 == episode_id) else {
+                continue;
+            };
+            let Some(grabbable) = self.policies.get(&row.season_id).and_then(Clone::clone) else {
+                continue;
+            };
+            let Some(candidate) = self.best_single.get(&episode_id) else {
+                continue;
+            };
+            winners.push(TvWinner {
+                grabbable,
+                scope: GrabScope::Episode { episode_id },
+                candidate: candidate.clone(),
+            });
+        }
         Ok(winners)
     }
 }
@@ -255,6 +476,196 @@ async fn wanted_movies(database: &Db) -> Result<Vec<WantedMovie>, GrabError> {
         .collect())
 }
 
+/// A matched series season with everything a grab needs loaded.
+#[derive(Clone, Debug)]
+struct TvGrabbable {
+    series_title: String,
+    policy: TitlePolicy,
+    language: Language,
+}
+
+/// One winner of the episode lane: what to grab, for whom, and which
+/// release won.
+struct TvWinner {
+    grabbable: TvGrabbable,
+    scope: GrabScope,
+    candidate: Eligible,
+}
+
+/// A wanted episode with the season context matching and grabbing need.
+struct WantedEpisodeRow {
+    wanted: WantedEpisode,
+    season_id: i64,
+}
+
+fn tv_match_kind(kind: MatchKind) -> &'static str {
+    match kind {
+        MatchKind::TmdbId => "tmdb id",
+        MatchKind::ImdbId => "imdb id",
+        MatchKind::TitleAndYear => "title",
+    }
+}
+
+/// Every wanted episode still open — wanted, no file on disk, no live grab
+/// of its own or of its season. Same shape as the movie list: unlimited,
+/// unordered by recency, blocked series included on purpose (§6.3).
+async fn wanted_episodes(database: &Db) -> Result<Vec<WantedEpisodeRow>, GrabError> {
+    let rows = sqlx::query!(
+        r#"
+        SELECT   e.id     AS "id!: i64",
+                 e.number AS "episode!: i64",
+                 se.id     AS "season_id!: i64",
+                 se.number AS "season!: i64",
+                 s.tmdb_id AS "series_tmdb_id!: i64",
+                 s.title   AS "series_title!: String"
+        FROM     episodes e
+        JOIN     seasons se ON se.id = e.season_id
+        JOIN     series s ON s.id = se.series_id
+        WHERE    e.wanted = 1
+          AND    NOT EXISTS (
+                      SELECT 1 FROM media_files f
+                      WHERE  f.owner_kind = 'episode' AND f.owner_id = e.id
+                  )
+          AND    NOT EXISTS (
+                      SELECT 1 FROM grabs g
+                      WHERE  g.target_kind = 'episode' AND g.target_id = e.id
+                        AND  g.state IN ('sent', 'downloaded', 'imported')
+                  )
+          AND    NOT EXISTS (
+                      SELECT 1 FROM grabs g
+                      WHERE  g.target_kind = 'season' AND g.target_id = se.id
+                        AND  g.state IN ('sent', 'downloaded', 'imported')
+                  )
+        ORDER BY e.id
+        "#
+    )
+    .fetch_all(database.pool())
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|row| WantedEpisodeRow {
+            wanted: WantedEpisode {
+                id: EpisodeId(row.id),
+                tmdb_id: u32::try_from(row.series_tmdb_id).ok(),
+                // The series table carries no `IMDb` id yet, so this lane
+                // matches on TMDB id or title until one is backfilled.
+                imdb_id: None,
+                title: row.series_title,
+                season: u32::try_from(row.season).unwrap_or_default(),
+                episode: u32::try_from(row.episode).unwrap_or_default(),
+            },
+            season_id: row.season_id,
+        })
+        .collect())
+}
+
+/// The policy and original language a matched season needs, or `None` when
+/// it cannot be evaluated yet. Same refusals as [`load_grabbable`].
+async fn load_tv_grabbable(
+    database: &Db,
+    episodes: &[WantedEpisodeRow],
+    season_id: i64,
+) -> Result<Option<TvGrabbable>, GrabError> {
+    let Some(series_title) = episodes
+        .iter()
+        .find(|row| row.season_id == season_id)
+        .map(|row| row.wanted.title.clone())
+    else {
+        return Ok(None);
+    };
+    let Some(policy) = database.season_policy(season_id).await? else {
+        return Ok(None);
+    };
+    // §5.2, same as films.
+    let language: Option<String> = sqlx::query_scalar!(
+        r#"
+        SELECT   s.original_language
+        FROM     seasons se
+        JOIN     series s ON s.id = se.series_id
+        WHERE    se.id = ?
+        "#,
+        season_id
+    )
+    .fetch_optional(database.pool())
+    .await?
+    .flatten();
+    let Some(language) = language else {
+        tracing::warn!(
+            season_id,
+            series = series_title,
+            "no original language yet; not grabbing from RSS"
+        );
+        return Ok(None);
+    };
+
+    Ok(Some(TvGrabbable {
+        series_title,
+        policy,
+        language: arr_db::policy::language(&language),
+    }))
+}
+
+/// Whether §6.2's guard lets a season pack be grabbed here: exactly what
+/// [`season_grab_mode`] requires of targeted search, computed over every
+/// episode the season is known to hold (§14, amended in #117).
+async fn pack_allowed(database: &Db, season_id: i64) -> Result<bool, GrabError> {
+    let episodes = sqlx::query!(
+        r#"
+        SELECT   e.air_date,
+                 EXISTS (
+                     SELECT 1 FROM media_files f
+                     WHERE  f.owner_kind = 'episode' AND f.owner_id = e.id
+                 ) AS "on_disk!: bool"
+        FROM     episodes e
+        WHERE    e.season_id = ?
+        ORDER BY e.number
+        "#,
+        season_id
+    )
+    .fetch_all(database.pool())
+    .await?;
+    let pack_hard_failed = sqlx::query_scalar!(
+        r#"SELECT EXISTS (
+               SELECT 1 FROM grabs
+               WHERE target_kind = 'season' AND target_id = ? AND state = 'failed'
+           ) AS "failed!: bool""#,
+        season_id
+    )
+    .fetch_one(database.pool())
+    .await?;
+
+    Ok(season_grab_mode(&SeasonGrabFacts {
+        air_dates: &episodes
+            .iter()
+            .map(|episode| air_date_time(episode.air_date.as_deref()))
+            .collect::<Vec<_>>(),
+        now: std::time::SystemTime::now(),
+        any_episode_on_disk: episodes.iter().any(|episode| episode.on_disk),
+        pack_hard_failed,
+    }) == SeasonGrabMode::SeasonPack)
+}
+
+/// An `air_date` as TMDB writes it (`YYYY-MM-DD`), or a full timestamp if
+/// one ever arrives that way. Unknown is unaired.
+fn air_date_time(value: Option<&str>) -> Option<std::time::SystemTime> {
+    let value = value?;
+    let timestamp = if let Ok(date) = value.parse::<chrono::NaiveDate>() {
+        date.and_time(chrono::NaiveTime::MIN).and_utc().timestamp()
+    } else {
+        value
+            .parse::<chrono::DateTime<chrono::Utc>>()
+            .ok()?
+            .timestamp()
+    };
+    let seconds = u64::try_from(timestamp.abs()).ok()?;
+    if timestamp < 0 {
+        std::time::UNIX_EPOCH.checked_sub(std::time::Duration::from_secs(seconds))
+    } else {
+        std::time::UNIX_EPOCH.checked_add(std::time::Duration::from_secs(seconds))
+    }
+}
+
 /// The policy and original language a matched title needs, or `None` when
 /// the title cannot be evaluated yet.
 async fn load_grabbable(
@@ -312,6 +723,48 @@ mod tests {
     /// asked for.
     const FEED: &str = include_str!("../tests/fixtures/rss.xml");
 
+    /// A season pack and its three episodes, all eligible under the seeded
+    /// TV main policy (§5.5 bands: 2160p floor 3 GiB, per #95).
+    const TV_FEED: &str = r#"<rss version="2.0" xmlns:torznab="http://torznab.com/schemas/2015/feed">
+      <channel>
+        <item>
+          <title>Fallout.S01.2160p.WEB-DL.DDP5.1.Atmos.H.265-GROUP</title>
+          <guid>tv-pack</guid><link>https://indexer.invalid/download/tv-pack</link>
+          <enclosure url="https://indexer.invalid/download/tv-pack" length="85899345920"
+                     type="application/x-bittorrent"/>
+          <torznab:attr name="seeders" value="120"/>
+        </item>
+        <item>
+          <title>Fallout.S01E01.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>
+          <guid>tv-e01</guid><link>https://indexer.invalid/download/tv-e01</link>
+          <enclosure url="https://indexer.invalid/download/tv-e01" length="10737418240"
+                     type="application/x-bittorrent"/>
+          <torznab:attr name="seeders" value="80"/>
+        </item>
+        <item>
+          <title>Fallout.S01E02.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>
+          <guid>tv-e02</guid><link>https://indexer.invalid/download/tv-e02</link>
+          <enclosure url="https://indexer.invalid/download/tv-e02" length="10737418240"
+                     type="application/x-bittorrent"/>
+          <torznab:attr name="seeders" value="80"/>
+        </item>
+        <item>
+          <title>Fallout.S01E03.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>
+          <guid>tv-e03</guid><link>https://indexer.invalid/download/tv-e03</link>
+          <enclosure url="https://indexer.invalid/download/tv-e03" length="10737418240"
+                     type="application/x-bittorrent"/>
+          <torznab:attr name="seeders" value="80"/>
+        </item>
+        <item>
+          <title>The.Fallout.2021.S01E03.1080p.WEB-DL-GROUP</title>
+          <guid>tv-near-miss</guid><link>https://indexer.invalid/download/tv-near-miss</link>
+          <enclosure url="https://indexer.invalid/download/tv-near-miss" length="10737418240"
+                     type="application/x-bittorrent"/>
+          <torznab:attr name="seeders" value="80"/>
+        </item>
+      </channel>
+    </rss>"#;
+
     const INDEXERS: [i64; 2] = [7, 9];
 
     /// Enough of Transmission to add a torrent and list nothing back.
@@ -350,9 +803,9 @@ mod tests {
 
     /// Two indexers, both advertising a text search, both serving the same
     /// feed to an empty query.
-    async fn prowlarr() -> MockServer {
+    async fn prowlarr(feed: &str) -> MockServer {
         let server = MockServer::start().await;
-        let feed = test_downloads::rewrite(FEED, "https://indexer.invalid/download/", &server);
+        let feed = test_downloads::rewrite(feed, "https://indexer.invalid/download/", &server);
         test_downloads::mount(&server).await;
         Mock::given(method("GET"))
             .and(path("/api/v1/indexer"))
@@ -430,6 +883,59 @@ mod tests {
         )
     }
 
+    /// Fallout S01 with wanted episodes airing at the given dates. Returns
+    /// the season id and the episode ids in episode order.
+    async fn wanted_series(database: &Db, air_dates: &[&str]) -> (i64, Vec<i64>) {
+        sqlx::query(
+            "INSERT INTO series (tmdb_id, title, year, original_language, root_id)
+             SELECT 106379, 'Fallout', 2024, 'en', id
+             FROM roots WHERE kind = 'tv' AND audience = 'main'",
+        )
+        .execute(database.pool())
+        .await
+        .unwrap();
+        let season_id: i64 = sqlx::query_scalar(
+            "INSERT INTO seasons (series_id, number) VALUES (1, 1) RETURNING id",
+        )
+        .fetch_one(database.pool())
+        .await
+        .unwrap();
+        let mut episodes = Vec::new();
+        for (index, air_date) in air_dates.iter().enumerate() {
+            let id: i64 = sqlx::query_scalar(
+                "INSERT INTO episodes (season_id, number, title, air_date, wanted)
+                 VALUES (?, ?, ?, ?, 1) RETURNING id",
+            )
+            .bind(season_id)
+            .bind(i64::try_from(index).unwrap() + 1)
+            .bind(format!("Episode {}", index + 1))
+            .bind(air_date)
+            .fetch_one(database.pool())
+            .await
+            .unwrap();
+            episodes.push(id);
+        }
+        (season_id, episodes)
+    }
+
+    /// `(target_kind, target_id, state)` for every TV grab that was made.
+    async fn tv_grabs(database: &Db) -> Vec<(String, i64, String)> {
+        sqlx::query_as::<_, (String, i64, String)>(
+            "SELECT target_kind, target_id, state FROM grabs
+             WHERE target_kind IN ('episode', 'season') ORDER BY target_kind, target_id",
+        )
+        .fetch_all(database.pool())
+        .await
+        .unwrap()
+    }
+
+    async fn episode_states(database: &Db) -> Vec<String> {
+        sqlx::query_scalar::<_, String>("SELECT state FROM episodes ORDER BY number")
+            .fetch_all(database.pool())
+            .await
+            .unwrap()
+    }
+
     /// `(movie title, grabbed release name)` for every grab that was made.
     async fn grabbed(database: &Db) -> Vec<(String, String)> {
         sqlx::query_as::<_, (String, String)>(
@@ -462,7 +968,7 @@ mod tests {
             (438_631, "Dune", 2021, false),
         ])
         .await;
-        let indexer = prowlarr().await;
+        let indexer = prowlarr(FEED).await;
         let (downloader, fake) = transmission().await;
 
         let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
@@ -506,7 +1012,7 @@ mod tests {
     #[tokio::test]
     async fn a_near_miss_is_not_grabbed() {
         let (_dir, database) = wanted(&[(9_999_999, "Dune: Part Three", 2024, false)]).await;
-        let indexer = prowlarr().await;
+        let indexer = prowlarr(FEED).await;
         let (downloader, fake) = transmission().await;
 
         let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
@@ -525,7 +1031,7 @@ mod tests {
             (9_999_999, "Dune: Part Three", 2024, false),
         ])
         .await;
-        let indexer = prowlarr().await;
+        let indexer = prowlarr(FEED).await;
         let (downloader, _fake) = transmission().await;
 
         action(&indexer, &downloader).tick(&database).await.unwrap();
@@ -542,7 +1048,7 @@ mod tests {
     #[tokio::test]
     async fn a_blocked_title_still_matches_rss() {
         let (_dir, database) = wanted(&[(693_134, "Dune: Part Two", 2024, true)]).await;
-        let indexer = prowlarr().await;
+        let indexer = prowlarr(FEED).await;
         let (downloader, _fake) = transmission().await;
 
         let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
@@ -560,7 +1066,7 @@ mod tests {
             (438_631, "Dune", 2021, false),
         ])
         .await;
-        let indexer = prowlarr().await;
+        let indexer = prowlarr(FEED).await;
         let (downloader, _fake) = transmission().await;
 
         action(&indexer, &downloader).tick(&database).await.unwrap();
@@ -582,4 +1088,135 @@ mod tests {
             assert_eq!(feeds, 1);
         }
     }
+
+    /// A single-episode release in the feed closes a wanted, missing
+    /// episode's gap, with no targeted-search attempt spent (§6.2). The
+    /// season here is airing, so its pack is not eligible ([`season_grab_mode`]).
+    #[tokio::test]
+    async fn a_missing_episode_is_grabbed_from_rss() {
+        let (_dir, database) = wanted(&[]).await;
+        let (_season_id, episodes) = wanted_series(&database, &["2024-04-11", "2099-01-01"]).await;
+        let indexer = prowlarr(TV_FEED).await;
+        let (downloader, fake) = transmission().await;
+
+        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 2);
+        assert_eq!(
+            tv_grabs(&database).await,
+            episodes
+                .iter()
+                .map(|id| ("episode".to_owned(), *id, "sent".to_owned()))
+                .collect::<Vec<_>>(),
+            "each gap takes its own single; the pack and the near miss stay out"
+        );
+        assert_eq!(fake.added.lock().unwrap().len(), 2);
+        let attempts: Vec<(i64, Option<String>)> =
+            sqlx::query_as("SELECT search_attempts, last_searched_at FROM episodes ORDER BY id")
+                .fetch_all(database.pool())
+                .await
+                .unwrap();
+        assert!(attempts
+            .iter()
+            .all(|(count, at)| *count == 0 && at.is_none()));
+    }
+
+    /// The acceptance case: one feed pass over a fully aired season with
+    /// nothing on disk takes the pack alone, and the singles stand down.
+    #[tokio::test]
+    async fn a_completed_season_takes_one_pack_from_rss() {
+        let (_dir, database) = wanted(&[]).await;
+        let (season_id, _episodes) =
+            wanted_series(&database, &["2024-04-11", "2024-04-18", "2024-04-25"]).await;
+        // Only the pack is wanted here: strip the singles so the pass has to
+        // pick between shapes on coverage, not on availability.
+        let pack_only = TV_FEED.replace(
+            "Fallout.S01E01.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>",
+            "Unrelated.S01E01.2160p.WEB-DL-GROUP</title>",
+        );
+        let pack_only = pack_only.replace(
+            "Fallout.S01E02.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>",
+            "Unrelated.S01E02.2160p.WEB-DL-GROUP</title>",
+        );
+        let pack_only = pack_only.replace(
+            "Fallout.S01E03.2160p.WEB-DL.DDP5.1.H.265-GROUP</title>",
+            "Unrelated.S01E03.2160p.WEB-DL-GROUP</title>",
+        );
+        let indexer = prowlarr(&pack_only).await;
+        let (downloader, fake) = transmission().await;
+
+        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        assert_eq!(fake.added.lock().unwrap().len(), 1);
+        assert_eq!(
+            tv_grabs(&database).await,
+            vec![("season".to_owned(), season_id, "sent".to_owned())]
+        );
+        assert_eq!(
+            episode_states(&database).await,
+            vec!["downloading"; 3],
+            "the pack flips every episode it covers"
+        );
+    }
+
+    /// §6.2 with #117's guard: an episode already on disk keeps the season
+    /// per-episode here too — the pack is skipped and the open gaps take
+    /// their singles.
+    #[tokio::test]
+    async fn a_pack_for_a_season_with_a_file_on_disk_is_skipped() {
+        let (_dir, database) = wanted(&[]).await;
+        let (_season_id, episodes) =
+            wanted_series(&database, &["2024-04-11", "2024-04-18", "2024-04-25"]).await;
+        sqlx::query(
+            "INSERT INTO media_files (owner_kind, owner_id, path, size)
+             VALUES ('episode', ?, '/tv/fallout/s01e01.mkv', 1)",
+        )
+        .bind(episodes[0])
+        .execute(database.pool())
+        .await
+        .unwrap();
+        let indexer = prowlarr(TV_FEED).await;
+        let (downloader, _fake) = transmission().await;
+
+        action(&indexer, &downloader).tick(&database).await.unwrap();
+
+        let grabs = tv_grabs(&database).await;
+        assert!(
+            !grabs.iter().any(|(kind, _, _)| kind == "season"),
+            "no pack over an episode already on disk"
+        );
+        // Episode 1 is not wanted any more — it has a file — and the near
+        // miss never matched, so exactly the two open gaps are filled.
+        assert_eq!(
+            grabs,
+            vec![
+                ("episode".to_owned(), episodes[1], "sent".to_owned()),
+                ("episode".to_owned(), episodes[2], "sent".to_owned()),
+            ]
+        );
+    }
+
+    /// §6.3: `blocked` stops targeted search for a series and leaves RSS
+    /// matching on. The one-episode season here is fully aired, so what
+    /// matches is its pack — the same release targeted search would take.
+    #[tokio::test]
+    async fn a_blocked_series_still_matches_rss() {
+        let (_dir, database) = wanted(&[]).await;
+        let (season_id, _episodes) = wanted_series(&database, &["2024-04-11"]).await;
+        sqlx::query("UPDATE series SET blocked = 1")
+            .execute(database.pool())
+            .await
+            .unwrap();
+        let indexer = prowlarr(TV_FEED).await;
+        let (downloader, _fake) = transmission().await;
+
+        let outcomes = action(&indexer, &downloader).tick(&database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        assert_eq!(
+            tv_grabs(&database).await,
+            vec![("season".to_owned(), season_id, "sent".to_owned())]
+        );
+    }
 }