Import pipeline: probe, hardlink, rename, layout (#82)

.sqlx/query-0f53855c99d83e4d291e24643f035bc7d0f24dc115e2515c101f8478464b75e0.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "UPDATE movies\n             SET    state = 'missing',\n                    updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n             WHERE  id = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "0f53855c99d83e4d291e24643f035bc7d0f24dc115e2515c101f8478464b75e0"
+}

.sqlx/query-48c46aa1908add3d6abff142dbb22c2bce92c6483564c159731e64fdd5f7f8bf.json

@@ -0,0 +1,20 @@
+{
+  "db_name": "SQLite",
+  "query": "SELECT normalised_name AS \"normalised_name!: String\" FROM blacklist",
+  "describe": {
+    "columns": [
+      {
+        "name": "normalised_name!: String",
+        "ordinal": 0,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "48c46aa1908add3d6abff142dbb22c2bce92c6483564c159731e64fdd5f7f8bf"
+}

.sqlx/query-93d05b3370c722a59239b486f5ca68123fb59610ba8ab8e8f6c1a1189610e377.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "UPDATE movies\n         SET    state = 'imported',\n                updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n         WHERE  id = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "93d05b3370c722a59239b486f5ca68123fb59610ba8ab8e8f6c1a1189610e377"
+}

.sqlx/query-b4b63449fdfb0bf5fb7095577d2e4759784a92ad87cb151150689fc355473328.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO blacklist (infohash, normalised_name, reason)\n             VALUES (?, ?, ?)\n             ON CONFLICT (infohash) DO NOTHING",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 3
+    },
+    "nullable": []
+  },
+  "hash": "b4b63449fdfb0bf5fb7095577d2e4759784a92ad87cb151150689fc355473328"
+}

.sqlx/query-c249ca062f44f32ee4a554f36a0749cf7a3e1a53a8324e5c816c7745418ae34d.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "UPDATE grabs\n         SET    state = 'imported',\n                imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')\n         WHERE  id = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "c249ca062f44f32ee4a554f36a0749cf7a3e1a53a8324e5c816c7745418ae34d"
+}

.sqlx/query-c602f6e08e0a8d828dfff1d821cbd08215204fadfb92b7f676e857e0b3400408.json

@@ -0,0 +1,62 @@
+{
+  "db_name": "SQLite",
+  "query": "\n        SELECT   g.id                AS \"grab_id!: i64\",\n                 g.infohash          AS \"infohash!: String\",\n                 m.id                AS \"movie_id!: i64\",\n                 m.tmdb_id           AS \"tmdb_id!: i64\",\n                 m.title             AS \"title!: String\",\n                 m.year,\n                 m.original_language,\n                 r.name              AS \"release_name!: String\"\n        FROM     grabs g\n        JOIN     movies m   ON m.id = g.target_id\n        JOIN     releases r ON r.id = g.release_id\n        WHERE    g.state = 'downloaded' AND g.target_kind = 'movie'\n        ORDER BY g.id\n        ",
+  "describe": {
+    "columns": [
+      {
+        "name": "grab_id!: i64",
+        "ordinal": 0,
+        "type_info": "Integer"
+      },
+      {
+        "name": "infohash!: String",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "movie_id!: i64",
+        "ordinal": 2,
+        "type_info": "Integer"
+      },
+      {
+        "name": "tmdb_id!: i64",
+        "ordinal": 3,
+        "type_info": "Integer"
+      },
+      {
+        "name": "title!: String",
+        "ordinal": 4,
+        "type_info": "Text"
+      },
+      {
+        "name": "year",
+        "ordinal": 5,
+        "type_info": "Integer"
+      },
+      {
+        "name": "original_language",
+        "ordinal": 6,
+        "type_info": "Text"
+      },
+      {
+        "name": "release_name!: String",
+        "ordinal": 7,
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      true,
+      false,
+      false,
+      false,
+      false,
+      true,
+      true,
+      false
+    ]
+  },
+  "hash": "c602f6e08e0a8d828dfff1d821cbd08215204fadfb92b7f676e857e0b3400408"
+}

.sqlx/query-c66f29b6edc4bca3a751e85e133fa379d232b045498310debab5ddd97f039294.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "UPDATE grabs SET state = 'failed' WHERE id = ?",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 1
+    },
+    "nullable": []
+  },
+  "hash": "c66f29b6edc4bca3a751e85e133fa379d232b045498310debab5ddd97f039294"
+}

.sqlx/query-ff1f65308ec991007ba9e0d7c22a4f31ebed4321365caceb557c7ea23ba69f7d.json

@@ -0,0 +1,12 @@
+{
+  "db_name": "SQLite",
+  "query": "INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver)\n         VALUES ('movie', ?, ?, ?, ?, ?)\n         ON CONFLICT (path) DO UPDATE SET\n             size       = excluded.size,\n             probed     = excluded.probed,\n             waiver     = excluded.waiver,\n             updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Right": 5
+    },
+    "nullable": []
+  },
+  "hash": "ff1f65308ec991007ba9e0d7c22a4f31ebed4321365caceb557c7ea23ba69f7d"
+}

Cargo.lock

@@ -86,6 +86,7 @@ dependencies = [
  "arr-indexer",
  "arr-meta",
  "arr-parse",
+ "arr-probe",
  "axum",
  "chrono",
  "include_dir",

crates/arr-core/src/layout.rs

@@ -0,0 +1,198 @@
+//! Library layout and naming (`DESIGN.md` §7.4). Pure string building; the
+//! import pipeline joins these onto a root path and does the IO.
+//!
+//! ```text
+//! Dune Part Two (2024) [tmdbid-693134]/
+//!     Dune Part Two (2024) [tmdbid-693134] - [2160p][WEB-DL][HDR10].mkv
+//! ```
+//!
+//! The provider ID in the folder name turns Jellyfin matching into exact
+//! lookup, and the attribute tags come from `ffprobe` so they are true — with
+//! one exception: no file knows whether it came off a disc or a streaming
+//! service (§5.6), so the source tag stays the release name's claim. Release
+//! group is deliberately absent.
+
+use crate::{HdrFormat, Language, ProbedMedia, Source};
+
+/// The per-title folder: `Dune Part Two (2024) [tmdbid-693134]`.
+///
+/// One folder per title even for a single file, so sidecars stay contained
+/// and deletes are atomic.
+#[must_use]
+pub fn movie_folder(title: &str, year: Option<i64>, tmdb_id: i64) -> String {
+    let title = sanitise(title);
+    match year {
+        Some(year) => format!("{title} ({year}) [tmdbid-{tmdb_id}]"),
+        None => format!("{title} [tmdbid-{tmdb_id}]"),
+    }
+}
+
+/// The feature's filename: the folder name, the attribute tags, and the
+/// source file's extension.
+#[must_use]
+pub fn movie_file_name(
+    title: &str,
+    year: Option<i64>,
+    tmdb_id: i64,
+    tags: &[String],
+    extension: Option<&str>,
+) -> String {
+    let stem = movie_folder(title, year, tmdb_id);
+    let tags = tags.iter().fold(String::new(), |mut out, tag| {
+        out.push('[');
+        out.push_str(tag);
+        out.push(']');
+        out
+    });
+    match extension {
+        Some(extension) => format!("{stem} - {tags}.{extension}"),
+        None => format!("{stem} - {tags}"),
+    }
+}
+
+/// The §7.4 attribute tags, in a fixed order: resolution, source, HDR,
+/// Portuguese audio.
+///
+/// Everything but the source comes from the probe. HDR is tagged only when
+/// present, and audio only for Portuguese variants — that is the audit `ls`
+/// answers: which files are DV, which of the kids' files are still
+/// English-only.
+#[must_use]
+pub fn attribute_tags(media: &ProbedMedia, claimed_source: Option<Source>) -> Vec<String> {
+    let mut tags = vec![media.resolution.to_string()];
+
+    if let Some(source) = claimed_source {
+        if source != Source::Other {
+            tags.push(source.to_string());
+        }
+    }
+
+    if media.hdr != HdrFormat::Sdr {
+        tags.push(media.hdr.to_string());
+    }
+
+    let mut portuguese: Vec<&Language> = Vec::new();
+    for track in &media.audio_tracks {
+        if matches!(
+            track.language,
+            Language::PortuguesePortugal
+                | Language::PortugueseBrazil
+                | Language::PortugueseUnverified
+        ) && !portuguese.contains(&&track.language)
+        {
+            portuguese.push(&track.language);
+        }
+    }
+    tags.extend(portuguese.into_iter().map(ToString::to_string));
+
+    tags
+}
+
+/// A title as a filesystem name: filesystem-hostile characters dropped,
+/// whitespace collapsed. TMDB's `Dune: Part Two` becomes `Dune Part Two`.
+fn sanitise(title: &str) -> String {
+    let cleaned: String = title
+        .chars()
+        .filter(|c| {
+            !matches!(c, '<' | '>' | ':' | '"' | '/' | '\\' | '|' | '?' | '*') && !c.is_control()
+        })
+        .collect();
+    cleaned
+        .split_whitespace()
+        .collect::<Vec<_>>()
+        .join(" ")
+        .trim_matches(|c| c == '.' || c == ' ')
+        .to_owned()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{AudioTrack, DolbyVisionProfile, Resolution, SubtitleTrack};
+
+    fn probed(resolution: Resolution, hdr: HdrFormat, audio: Vec<Language>) -> ProbedMedia {
+        ProbedMedia {
+            resolution,
+            source: None,
+            hdr,
+            audio_tracks: audio
+                .into_iter()
+                .map(|language| AudioTrack {
+                    language,
+                    title: None,
+                    handler_name: None,
+                })
+                .collect(),
+            subtitle_tracks: Vec::<SubtitleTrack>::new(),
+        }
+    }
+
+    /// The exact §7.4 movie example.
+    #[test]
+    fn the_design_document_movie_example() {
+        let media = probed(
+            Resolution::R2160p,
+            HdrFormat::Hdr10,
+            vec![Language::Other("en".into())],
+        );
+        let tags = attribute_tags(&media, Some(Source::WebDl));
+
+        assert_eq!(
+            movie_folder("Dune: Part Two", Some(2024), 693_134),
+            "Dune Part Two (2024) [tmdbid-693134]"
+        );
+        assert_eq!(
+            movie_file_name("Dune: Part Two", Some(2024), 693_134, &tags, Some("mkv")),
+            "Dune Part Two (2024) [tmdbid-693134] - [2160p][WEB-DL][HDR10].mkv"
+        );
+    }
+
+    /// The kids audit surface: a pt-PT track is tagged, SDR is not.
+    #[test]
+    fn portuguese_audio_is_tagged_and_sdr_is_not() {
+        let media = probed(
+            Resolution::R1080p,
+            HdrFormat::Sdr,
+            vec![
+                Language::Other("en".into()),
+                Language::PortuguesePortugal,
+                Language::PortuguesePortugal,
+            ],
+        );
+        assert_eq!(
+            attribute_tags(&media, Some(Source::WebDl)),
+            vec!["1080p", "WEB-DL", "pt-PT"]
+        );
+    }
+
+    /// `ls` shows which files are DV, profile included (§5.3, §7.4).
+    #[test]
+    fn dolby_vision_tags_carry_the_profile() {
+        let media = probed(
+            Resolution::R2160p,
+            HdrFormat::DolbyVision(DolbyVisionProfile {
+                profile: 8,
+                compatibility_id: Some(1),
+            }),
+            vec![],
+        );
+        assert_eq!(attribute_tags(&media, None), vec!["2160p", "DV8.1"]);
+    }
+
+    /// An unknown source claim is no tag rather than a lie.
+    #[test]
+    fn unknown_sources_are_not_tagged() {
+        let media = probed(Resolution::R1080p, HdrFormat::Sdr, vec![]);
+        assert_eq!(attribute_tags(&media, Some(Source::Other)), vec!["1080p"]);
+        assert_eq!(attribute_tags(&media, None), vec!["1080p"]);
+    }
+
+    #[test]
+    fn titles_are_sanitised_for_the_filesystem() {
+        assert_eq!(
+            movie_folder("What / If: A* Story?", Some(2020), 1),
+            "What If A Story (2020) [tmdbid-1]"
+        );
+        assert_eq!(movie_folder("Untitled", None, 2), "Untitled [tmdbid-2]");
+    }
+}

crates/arr-core/src/lib.rs

@@ -2,9 +2,10 @@
 //!
 //! This crate must never depend on `axum`, `sqlx` or `reqwest`.
 
-use std::{collections::BTreeMap, path::PathBuf, time::SystemTime};
+use std::{collections::BTreeMap, fmt, path::PathBuf, time::SystemTime};
 
 pub mod lang;
+pub mod layout;
 pub mod policy;
 pub mod score;
 pub mod status;
@@ -54,6 +55,19 @@ pub enum Language {
     Other(String),
 }
 
+impl fmt::Display for Language {
+    /// The tag the policy columns and §7.4 filenames spell it as:
+    /// `pt-PT`, `pt-BR`, `por-unverified`, anything else verbatim.
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::PortuguesePortugal => formatter.write_str("pt-PT"),
+            Self::PortugueseBrazil => formatter.write_str("pt-BR"),
+            Self::PortugueseUnverified => formatter.write_str("por-unverified"),
+            Self::Other(tag) => formatter.write_str(tag),
+        }
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub enum Resolution {
     R2160p,
@@ -62,6 +76,18 @@ pub enum Resolution {
     Other(u16),
 }
 
+impl fmt::Display for Resolution {
+    /// The spelling `resolution_pref` and §7.4 filename tags use.
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::R2160p => formatter.write_str("2160p"),
+            Self::R1080p => formatter.write_str("1080p"),
+            Self::R720p => formatter.write_str("720p"),
+            Self::Other(height) => write!(formatter, "{height}p"),
+        }
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub enum Source {
     Remux,
@@ -77,6 +103,25 @@ pub enum Source {
     Other,
 }
 
+impl fmt::Display for Source {
+    /// The spelling `source_weights` and §7.4 filename tags use.
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str(match self {
+            Self::Remux => "Remux",
+            Self::BluRay => "BluRay",
+            Self::WebDl => "WEB-DL",
+            Self::WebRip => "WEBRip",
+            Self::Hdtv => "HDTV",
+            Self::Dvd => "DVD",
+            Self::Telecine => "Telecine",
+            Self::Telesync => "Telesync",
+            Self::Cam => "CAM",
+            Self::Screener => "Screener",
+            Self::Other => "Unknown",
+        })
+    }
+}
+
 impl From<arr_parse::Resolution> for Resolution {
     fn from(value: arr_parse::Resolution) -> Self {
         match value {
@@ -121,6 +166,24 @@ pub enum HdrFormat {
     DolbyVision(DolbyVisionProfile),
 }
 
+impl fmt::Display for HdrFormat {
+    /// The §7.4 filename tag: `HDR10`, `DV8.1`, and so on. Dolby Vision
+    /// carries its profile because that is the whole point of probing it
+    /// (§5.3) — `ls` shows which files are DV and which profile.
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Sdr => formatter.write_str("SDR"),
+            Self::Hdr10 => formatter.write_str("HDR10"),
+            Self::Hdr10Plus => formatter.write_str("HDR10+"),
+            Self::Hlg => formatter.write_str("HLG"),
+            Self::DolbyVision(dv) => match dv.compatibility_id {
+                Some(compatibility_id) => write!(formatter, "DV{}.{compatibility_id}", dv.profile),
+                None => write!(formatter, "DV{}", dv.profile),
+            },
+        }
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub enum RequiredAudio {
     OriginalLanguage,
@@ -258,6 +321,24 @@ pub enum Rule {
     Other(String),
 }
 
+impl Rule {
+    /// The stable name shared by `releases.rejected_rule`, blacklist reasons
+    /// and `media_files.waiver`, so one rule reads the same everywhere.
+    #[must_use]
+    pub fn name(&self) -> String {
+        match self {
+            Self::RequiredAudio => "required_audio".into(),
+            Self::DubBlacklist(_) => "dub_blacklist".into(),
+            Self::PortugueseUnverified => "portuguese_unverified".into(),
+            Self::DolbyVisionProfile(_) => "dolby_vision_profile".into(),
+            Self::Resolution(_) => "resolution".into(),
+            Self::Source(_) => "source".into(),
+            Self::Size => "size".into(),
+            Self::Other(name) => name.clone(),
+        }
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub enum Verdict {
     Eligible,

crates/arr-daemon/Cargo.toml

@@ -19,6 +19,7 @@ arr-dl = { workspace = true }
 arr-indexer = { workspace = true }
 arr-meta = { workspace = true }
 arr-parse = { workspace = true }
+arr-probe = { workspace = true }
 axum = { workspace = true }
 chrono = { workspace = true }
 include_dir = { workspace = true }

crates/arr-daemon/src/grab.rs

@@ -13,12 +13,12 @@
 //! - the `grabs` row is written from that response, so a crash between the add
 //!   and the insert heals on the next tick instead of leaving an orphan.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use arr_core::policy::{evaluate, Candidate};
-use arr_core::{score::score, Language, Policy, Rule, TitleOverrides, Verdict};
+use arr_core::{score::score, Language, Policy, TitleOverrides, Verdict};
 use arr_db::{Db, MoviePolicy};
 use arr_dl::{AddTorrent, TorrentSource, TransmissionClient};
 use arr_indexer::{ProwlarrClient, SearchRelease, SearchRequest};
@@ -317,7 +317,14 @@ impl GrabAction {
         let candidates = self
             .search(database, movie, indexers, &loaded, &original_language)
             .await?;
-        let Some(winner) = candidates.into_iter().next() else {
+        // §6.3: anything that hard-failed post-ffprobe is never grabbed
+        // again. The same release reappears under new infohashes, so the key
+        // is the normalised name.
+        let blacklisted = blacklisted_names(database).await?;
+        let Some(winner) = candidates
+            .into_iter()
+            .find(|candidate| !blacklisted.contains(&arr_parse::normalise(&candidate.name)))
+        else {
             tracing::info!(
                 movie_id = movie.id,
                 title = movie.title,
@@ -558,6 +565,17 @@ async fn store_release(
     }))
 }
 
+/// Every blacklist key (§6.3), for filtering candidates. Household scale: a
+/// handful of rows, refetched per title rather than cached.
+async fn blacklisted_names(database: &Db) -> Result<HashSet<String>, GrabError> {
+    let names = sqlx::query_scalar!(
+        r#"SELECT normalised_name AS "normalised_name!: String" FROM blacklist"#
+    )
+    .fetch_all(database.pool())
+    .await?;
+    Ok(names.into_iter().collect())
+}
+
 /// Record that the title was searched, so the next tick takes a different one.
 async fn record_search(database: &Db, movie_id: i64) -> Result<(), GrabError> {
     sqlx::query!(
@@ -605,20 +623,7 @@ fn verdict_columns(verdict: &Verdict) -> (&'static str, Option<String>) {
     match verdict {
         Verdict::Eligible => ("eligible", None),
         Verdict::Waived(_) => ("waived", None),
-        Verdict::Rejected(rule) => ("rejected", Some(rule_name(rule))),
+        Verdict::Rejected(rule) => ("rejected", Some(rule.name())),
-    }
-}
-
-fn rule_name(rule: &Rule) -> String {
-    match rule {
-        Rule::RequiredAudio => "required_audio".into(),
-        Rule::DubBlacklist(_) => "dub_blacklist".into(),
-        Rule::PortugueseUnverified => "portuguese_unverified".into(),
-        Rule::DolbyVisionProfile(_) => "dolby_vision_profile".into(),
-        Rule::Resolution(_) => "resolution".into(),
-        Rule::Source(_) => "source".into(),
-        Rule::Size => "size".into(),
-        Rule::Other(name) => name.clone(),
     }
 }
 
@@ -990,6 +995,33 @@ mod tests {
         assert!(fake.torrents().is_empty());
     }
 
+    /// §6.3: a hard-failed release is never grabbed again, even when it is
+    /// still the best-scoring candidate — the next one down wins instead.
+    #[tokio::test]
+    async fn a_blacklisted_release_is_passed_over() {
+        let (_dir, database) = wanted_movie().await;
+        sqlx::query("INSERT INTO blacklist (infohash, normalised_name, reason) VALUES (?, ?, ?)")
+            .bind("ffff")
+            .bind(arr_parse::normalise(
+                "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos",
+            ))
+            .bind("dolby_vision_profile")
+            .execute(database.pool())
+            .await
+            .unwrap();
+        let indexer = prowlarr().await;
+        let (downloader, fake) = transmission().await;
+
+        action(&indexer, &downloader).tick(&database).await.unwrap();
+
+        assert_eq!(fake.torrents().len(), 1);
+        assert!(
+            fake.torrents()[0].source.ends_with("huge.torrent"),
+            "the remux wins once the WEB-DL is blacklisted: {}",
+            fake.torrents()[0].source
+        );
+    }
+
     /// §6.3: `blocked` stops targeted search for a title.
     #[tokio::test]
     async fn a_blocked_title_is_not_searched() {

crates/arr-daemon/src/import.rs

@@ -0,0 +1,991 @@
+//! The import pipeline: probe a completed download, judge it against the
+//! policy a second time with real evidence, and link it into the library. See
+//! DESIGN.md §5.7, §7.2, §7.3 and §7.4.
+//!
+//! The torrent's own files are never moved, renamed or deleted — the torrent
+//! and the library entry are separate lifecycles (§7.3). A hard-failed
+//! release is blacklisted and its grab marked failed, but the torrent keeps
+//! seeding until Transmission's own limits clear it.
+//!
+//! Everything here is idempotent from domain rows (§8): a grab in
+//! `downloaded` with no import recorded is the gap, and re-running any prefix
+//! of the pipeline after a crash converges — the hardlink call tolerates the
+//! destination already existing, and the `media_files` insert upserts on
+//! path.
+
+use std::collections::HashMap;
+use std::ffi::OsString;
+use std::io;
+use std::path::{Component, Path, PathBuf};
+
+use arr_core::layout;
+use arr_core::policy::{evaluate, Candidate};
+use arr_core::{ProbedMedia, Rule, Source, Verdict};
+use arr_db::Db;
+use arr_dl::TransmissionClient;
+use arr_probe::Prober;
+
+use crate::reconcile::{Action, ActionFuture, Outcome};
+
+/// A failure during one import tick.
+#[derive(Debug, thiserror::Error)]
+pub enum ImportError {
+    #[error("database: {0}")]
+    Database(#[from] sqlx::Error),
+    #[error("policy: {0}")]
+    Policy(#[from] arr_db::PolicyError),
+    #[error("transmission: {0}")]
+    Transmission(#[from] arr_dl::Error),
+    #[error("probe: {0}")]
+    Probe(#[from] arr_probe::Error),
+    #[error("blocking task: {0}")]
+    Join(#[from] tokio::task::JoinError),
+    #[error("{action} {path}: {source}")]
+    Io {
+        action: &'static str,
+        path: PathBuf,
+        source: io::Error,
+    },
+}
+
+/// How the feature reached the library.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Placement {
+    Linked,
+    /// `link()` raised `EXDEV`; the file was copied instead (§7.2).
+    Copied,
+    /// The destination already existed — an earlier attempt placed it before
+    /// the process died. Never a partial file: copies land via rename.
+    AlreadyPlaced,
+}
+
+/// What one probe attempt settled about one path.
+#[derive(Debug, Clone)]
+enum ProbeOutcome {
+    Media(Box<arr_probe::ProbedFile>),
+    /// A fact about the file — `.nfo`, artwork, corrupt — not the prober.
+    NotMedia,
+}
+
+/// Imports every downloaded grab: probe, second policy pass, hardlink into
+/// the §7.4 layout.
+#[derive(Debug)]
+pub struct ImportAction {
+    transmission: TransmissionClient,
+    prober: Prober,
+    /// Probe results by path, kept across ticks. The reconcile lane cancels
+    /// the whole action after its 25 s budget while one probe alone may take
+    /// up to 60 s, so without this a large multi-file torrent would restart
+    /// from the first file every tick and never finish. Shared with the
+    /// detached probe tasks, which is what lets a probe outlive a cancelled
+    /// tick and still deposit its result. Transient state, rebuilt by
+    /// re-probing after a restart (§8); entries are dropped once their grab
+    /// settles.
+    probed: std::sync::Arc<tokio::sync::Mutex<HashMap<PathBuf, ProbeOutcome>>>,
+}
+
+/// A movie grab Transmission finished downloading, not yet imported.
+#[derive(Debug, Clone)]
+struct PendingImport {
+    grab_id: i64,
+    infohash: String,
+    movie_id: i64,
+    tmdb_id: i64,
+    title: String,
+    year: Option<i64>,
+    original_language: Option<String>,
+    release_name: String,
+}
+
+impl ImportAction {
+    #[must_use]
+    pub fn new(transmission: TransmissionClient, prober: Prober) -> Self {
+        Self {
+            transmission,
+            prober,
+            probed: std::sync::Arc::new(tokio::sync::Mutex::new(HashMap::new())),
+        }
+    }
+
+    /// Probe every path, reusing results settled on earlier ticks, and
+    /// return the readable video files.
+    ///
+    /// Each probe runs as a detached task that writes the cache itself, so a
+    /// tick cancelled by the lane's 25 s budget mid-probe loses nothing: the
+    /// probe finishes in the background (its own 60 s limit still applies)
+    /// and the next tick reads the deposited result. A tick that re-requests
+    /// a path an orphaned probe is still on starts a second probe of the same
+    /// file — bounded overlap, identical result, chosen over tracking
+    /// in-flight probes.
+    ///
+    /// Only facts about a file are cached; a prober failure — missing
+    /// binary, timeout, unparseable output — is returned so the tick retries.
+    async fn probe_all(
+        &self,
+        paths: &[PathBuf],
+    ) -> Result<Vec<arr_probe::ProbedFile>, ImportError> {
+        let mut files = Vec::new();
+        for path in paths {
+            let cached = self.probed.lock().await.get(path).cloned();
+            let outcome = if let Some(outcome) = cached {
+                outcome
+            } else {
+                let prober = self.prober.clone();
+                let cache = std::sync::Arc::clone(&self.probed);
+                let target = path.clone();
+                tokio::spawn(async move {
+                    let outcome = match prober.probe(target.clone()).await {
+                        Ok(file) => ProbeOutcome::Media(Box::new(file)),
+                        Err(error) if error.is_about_the_file() => {
+                            tracing::debug!(path = %target.display(), %error, "not a video file, skipping");
+                            ProbeOutcome::NotMedia
+                        }
+                        Err(error) => return Err(error),
+                    };
+                    cache.lock().await.insert(target, outcome.clone());
+                    Ok(outcome)
+                })
+                .await??
+            };
+            if let ProbeOutcome::Media(file) = outcome {
+                files.push(*file);
+            }
+        }
+        Ok(files)
+    }
+
+    /// Drop a settled grab's probe results — imported or blacklisted, they
+    /// will not be needed again.
+    async fn forget_probes(&self, paths: &[PathBuf]) {
+        let mut probed = self.probed.lock().await;
+        for path in paths {
+            probed.remove(path);
+        }
+    }
+
+    async fn tick(&self, database: &Db) -> Result<Vec<Outcome>, ImportError> {
+        let mut outcomes = Vec::new();
+        for pending in pending_imports(database).await? {
+            match self.import_one(database, &pending).await {
+                Ok(Some(outcome)) => outcomes.push(outcome),
+                Ok(None) => {}
+                // One grab's failure must not cost the rest of the tick.
+                Err(error) => tracing::error!(
+                    grab_id = pending.grab_id,
+                    movie_id = pending.movie_id,
+                    title = pending.title,
+                    %error,
+                    "import failed"
+                ),
+            }
+        }
+        Ok(outcomes)
+    }
+
+    async fn import_one(
+        &self,
+        database: &Db,
+        pending: &PendingImport,
+    ) -> Result<Option<Outcome>, ImportError> {
+        let Some(loaded) = database.movie_policy(pending.movie_id).await? else {
+            return Ok(None);
+        };
+        // §5.2: no original language, nothing to judge audio against.
+        let Some(original_language) = pending.original_language.as_deref() else {
+            tracing::warn!(
+                movie_id = pending.movie_id,
+                title = pending.title,
+                "no original language yet; not importing"
+            );
+            return Ok(None);
+        };
+        let original_language = arr_db::policy::language(original_language);
+
+        let Some(content) = self.transmission.torrent_content(&pending.infohash).await? else {
+            // Gone from Transmission. Whether that is a failure or a manual
+            // removal is issue #24's call; leave the grab alone.
+            tracing::warn!(
+                grab_id = pending.grab_id,
+                infohash = pending.infohash,
+                "downloaded grab has no torrent in Transmission; not importing"
+            );
+            return Ok(None);
+        };
+        // Torrent-declared names are untrusted input: an absolute or
+        // `..`-carrying entry would escape the download root and get probed —
+        // and possibly hardlinked — from anywhere on disk.
+        let paths: Vec<PathBuf> = content
+            .files
+            .iter()
+            .filter_map(|file| {
+                let path = safe_join(&content.download_dir, &file.path);
+                if path.is_none() {
+                    tracing::warn!(
+                        grab_id = pending.grab_id,
+                        path = %file.path.display(),
+                        "torrent file path escapes the download root; skipping"
+                    );
+                }
+                path
+            })
+            .collect();
+
+        // No expected runtime yet: the movies table carries no TMDB runtime,
+        // so feature selection is by size alone (largest readable video).
+        let mut candidates = self.probe_all(&paths).await?;
+        candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.size));
+        let Some(feature) = candidates.into_iter().next() else {
+            // §5.7 "corrupt, wrong content": nothing in the torrent is a
+            // readable video file.
+            self.forget_probes(&paths).await;
+            return self
+                .hard_fail(database, pending, "no readable video file")
+                .await
+                .map(Some);
+        };
+
+        // §5.6 second phase of truth: same policy, real evidence.
+        let evaluation = evaluate(
+            &loaded.policy,
+            &loaded.overrides,
+            &original_language,
+            Candidate::PostDownload(&feature.media),
+            Some(feature.size),
+        );
+        let waiver: Option<Rule> = match evaluation.verdict {
+            Verdict::Rejected(rule) => {
+                self.forget_probes(&paths).await;
+                return self
+                    .hard_fail(database, pending, &rule.name())
+                    .await
+                    .map(Some);
+            }
+            Verdict::Waived(rule) => Some(rule),
+            Verdict::Eligible => None,
+        };
+
+        // The source tag is the one claim a file cannot verify (§5.6); every
+        // other tag comes from the probe.
+        let claimed_source = arr_parse::parse(&pending.release_name)
+            .source
+            .map(Source::from);
+        let tags = layout::attribute_tags(&feature.media, claimed_source);
+        let extension = feature.path.extension().and_then(|ext| ext.to_str());
+        let folder = layout::movie_folder(&pending.title, pending.year, pending.tmdb_id);
+        let file_name = layout::movie_file_name(
+            &pending.title,
+            pending.year,
+            pending.tmdb_id,
+            &tags,
+            extension,
+        );
+        let destination = Path::new(&loaded.root_path).join(folder).join(file_name);
+
+        let source_path = feature.path.clone();
+        let link_target = destination.clone();
+        let placement =
+            tokio::task::spawn_blocking(move || place(&source_path, &link_target)).await??;
+
+        record_import(database, pending, &feature, waiver.as_ref(), &destination).await?;
+        self.forget_probes(&paths).await;
+        let path_text = destination.to_string_lossy().into_owned();
+        tracing::info!(
+            grab_id = pending.grab_id,
+            movie_id = pending.movie_id,
+            title = pending.title,
+            path = path_text,
+            placement = ?placement,
+            waived = waiver.is_some(),
+            "imported"
+        );
+        Ok(Some(Outcome::new(
+            format!("grab {} downloaded, not imported", pending.grab_id),
+            format!("imported {path_text}"),
+        )))
+    }
+
+    /// §5.7 hard fail: blacklist the release, fail the grab, reopen the gap.
+    /// The torrent is deliberately untouched (§7.3).
+    async fn hard_fail(
+        &self,
+        database: &Db,
+        pending: &PendingImport,
+        reason: &str,
+    ) -> Result<Outcome, ImportError> {
+        let normalised = arr_parse::normalise(&pending.release_name);
+        sqlx::query!(
+            "INSERT INTO blacklist (infohash, normalised_name, reason)
+             VALUES (?, ?, ?)
+             ON CONFLICT (infohash) DO NOTHING",
+            pending.infohash,
+            normalised,
+            reason
+        )
+        .execute(database.pool())
+        .await?;
+        sqlx::query!(
+            "UPDATE grabs SET state = 'failed' WHERE id = ?",
+            pending.grab_id
+        )
+        .execute(database.pool())
+        .await?;
+        sqlx::query!(
+            "UPDATE movies
+             SET    state = 'missing',
+                    updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+             WHERE  id = ?",
+            pending.movie_id
+        )
+        .execute(database.pool())
+        .await?;
+
+        tracing::warn!(
+            grab_id = pending.grab_id,
+            movie_id = pending.movie_id,
+            title = pending.title,
+            release = pending.release_name,
+            reason,
+            "hard fail post-probe; blacklisted, torrent left seeding"
+        );
+        Ok(Outcome::new(
+            format!("grab {} hard-failed post-probe: {reason}", pending.grab_id),
+            format!("blacklisted {}", pending.release_name),
+        ))
+    }
+}
+
+impl Action for ImportAction {
+    fn name(&self) -> &'static str {
+        "import"
+    }
+
+    fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
+        Box::pin(async move { self.tick(database).await.map_err(Into::into) })
+    }
+}
+
+/// Settle a placed file into the rows: the `media_files` record (§4), the
+/// grab and the movie. The upsert on path is the crash seam — a re-run after
+/// a death between the link and here converges instead of erroring.
+async fn record_import(
+    database: &Db,
+    pending: &PendingImport,
+    feature: &arr_probe::ProbedFile,
+    waiver: Option<&Rule>,
+    destination: &Path,
+) -> Result<(), ImportError> {
+    let probed = probed_json(&feature.media).to_string();
+    let waiver_json = waiver.map(|rule| serde_json::json!({ "rule": rule.name() }).to_string());
+    let size = i64::try_from(feature.size).unwrap_or(i64::MAX);
+    let path_text = destination.to_string_lossy().into_owned();
+    sqlx::query!(
+        "INSERT INTO media_files (owner_kind, owner_id, path, size, probed, waiver)
+         VALUES ('movie', ?, ?, ?, ?, ?)
+         ON CONFLICT (path) DO UPDATE SET
+             size       = excluded.size,
+             probed     = excluded.probed,
+             waiver     = excluded.waiver,
+             updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
+        pending.movie_id,
+        path_text,
+        size,
+        probed,
+        waiver_json
+    )
+    .execute(database.pool())
+    .await?;
+    sqlx::query!(
+        "UPDATE grabs
+         SET    state = 'imported',
+                imported_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+         WHERE  id = ?",
+        pending.grab_id
+    )
+    .execute(database.pool())
+    .await?;
+    sqlx::query!(
+        "UPDATE movies
+         SET    state = 'imported',
+                updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
+         WHERE  id = ?",
+        pending.movie_id
+    )
+    .execute(database.pool())
+    .await?;
+    Ok(())
+}
+
+/// The gap, straight out of the domain rows (§8): a movie grab Transmission
+/// finished that no import has settled.
+async fn pending_imports(database: &Db) -> Result<Vec<PendingImport>, ImportError> {
+    let rows = sqlx::query!(
+        r#"
+        SELECT   g.id                AS "grab_id!: i64",
+                 g.infohash          AS "infohash!: String",
+                 m.id                AS "movie_id!: i64",
+                 m.tmdb_id           AS "tmdb_id!: i64",
+                 m.title             AS "title!: String",
+                 m.year,
+                 m.original_language,
+                 r.name              AS "release_name!: String"
+        FROM     grabs g
+        JOIN     movies m   ON m.id = g.target_id
+        JOIN     releases r ON r.id = g.release_id
+        WHERE    g.state = 'downloaded' AND g.target_kind = 'movie'
+        ORDER BY g.id
+        "#
+    )
+    .fetch_all(database.pool())
+    .await?;
+
+    Ok(rows
+        .into_iter()
+        .map(|row| PendingImport {
+            grab_id: row.grab_id,
+            infohash: row.infohash,
+            movie_id: row.movie_id,
+            tmdb_id: row.tmdb_id,
+            title: row.title,
+            year: row.year,
+            original_language: row.original_language,
+            release_name: row.release_name,
+        })
+        .collect())
+}
+
+/// The `probed` column (§4, §5.6): what `ffprobe` found, in the spellings the
+/// policy columns use.
+fn probed_json(media: &ProbedMedia) -> serde_json::Value {
+    serde_json::json!({
+        "resolution": media.resolution.to_string(),
+        "source": media.source.map(|source| source.to_string()),
+        "hdr": media.hdr.to_string(),
+        "audio_tracks": media
+            .audio_tracks
+            .iter()
+            .map(|track| serde_json::json!({
+                "language": track.language.to_string(),
+                "title": track.title,
+                "handler_name": track.handler_name,
+            }))
+            .collect::<Vec<_>>(),
+        "sub_tracks": media
+            .subtitle_tracks
+            .iter()
+            .map(|track| serde_json::json!({ "language": track.language.to_string() }))
+            .collect::<Vec<_>>(),
+    })
+}
+
+/// Join a torrent-declared file path onto the download root, refusing
+/// anything that could land outside it: absolute paths, drive prefixes and
+/// `..` components. `None` means the entry is hostile or malformed.
+fn safe_join(root: &Path, declared: &Path) -> Option<PathBuf> {
+    let mut clean = PathBuf::new();
+    for component in declared.components() {
+        match component {
+            Component::Normal(part) => clean.push(part),
+            Component::CurDir => {}
+            Component::RootDir | Component::Prefix(_) | Component::ParentDir => return None,
+        }
+    }
+    if clean.as_os_str().is_empty() {
+        return None;
+    }
+    Some(root.join(clean))
+}
+
+/// Hardlink `source` to `destination`, falling back to copy on `EXDEV` only
+/// (§7.2). No configuration flag.
+fn place(source: &Path, destination: &Path) -> Result<Placement, ImportError> {
+    if let Some(parent) = destination.parent() {
+        std::fs::create_dir_all(parent).map_err(|error| ImportError::Io {
+            action: "create library folder",
+            path: parent.to_path_buf(),
+            source: error,
+        })?;
+    }
+
+    match std::fs::hard_link(source, destination) {
+        Ok(()) => Ok(Placement::Linked),
+        // A completed earlier attempt: links and copies both land whole
+        // (copies via rename), so an existing destination is a finished one.
+        Err(error) if error.kind() == io::ErrorKind::AlreadyExists => Ok(Placement::AlreadyPlaced),
+        Err(error) if error.kind() == io::ErrorKind::CrossesDevices => {
+            copy_into_place(source, destination)
+        }
+        Err(error) => Err(ImportError::Io {
+            action: "hardlink into",
+            path: destination.to_path_buf(),
+            source: error,
+        }),
+    }
+}
+
+/// Copy through a dot-name in the destination folder, then rename, so the
+/// library never shows a partial file.
+fn copy_into_place(source: &Path, destination: &Path) -> Result<Placement, ImportError> {
+    let mut temp_name = OsString::from(".");
+    temp_name.push(destination.file_name().unwrap_or_default());
+    temp_name.push(".partial");
+    let temp = destination.with_file_name(temp_name);
+
+    let copied = std::fs::copy(source, &temp).and_then(|_| std::fs::rename(&temp, destination));
+    if let Err(error) = copied {
+        let _ = std::fs::remove_file(&temp);
+        return Err(ImportError::Io {
+            action: "copy into",
+            path: destination.to_path_buf(),
+            source: error,
+        });
+    }
+    Ok(Placement::Copied)
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod tests {
+    use std::os::unix::fs::{MetadataExt, PermissionsExt};
+    use std::path::PathBuf;
+
+    use serde_json::json;
+    use wiremock::matchers::method;
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    use super::*;
+
+    const INFOHASH: &str = "0123456789abcdef0123456789abcdef01234567";
+    const RELEASE_NAME: &str = "Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos";
+
+    /// A 2160p HDR10 file with an English track — what the seeded main-movies
+    /// policy accepts. The size is the container's claim, matching §5.5's
+    /// band; the bytes on disk are tiny.
+    const HDR10_PROBE: &str = r#"{
+        "format": {"format_name": "matroska,webm", "duration": "9060.0", "size": "23622320128"},
+        "streams": [
+            {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600,
+             "color_transfer": "smpte2084"},
+            {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}}
+        ]
+    }"#;
+
+    /// The same file as a Dolby Vision Profile 5 stream — §5.3's hard reject.
+    const DV5_PROBE: &str = r#"{
+        "format": {"format_name": "matroska,webm", "duration": "9060.0", "size": "23622320128"},
+        "streams": [
+            {"codec_type": "video", "codec_name": "hevc", "width": 3840, "height": 1600,
+             "color_transfer": "smpte2084",
+             "side_data_list": [{"side_data_type": "DOVI configuration record", "dv_profile": 5}]},
+            {"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}}
+        ]
+    }"#;
+
+    struct Harness {
+        _dir: tempfile::TempDir,
+        database: Db,
+        downloads: PathBuf,
+        library: PathBuf,
+        action: ImportAction,
+        _server: MockServer,
+    }
+
+    /// An `ffprobe` stand-in: canned JSON for media, a `tty` document for the
+    /// `.nfo`, so feature selection sees what the real binary would report.
+    fn fake_ffprobe(directory: &Path, media_json: &str) -> PathBuf {
+        let path = directory.join("ffprobe");
+        let script = format!(
+            "#!/bin/sh\nfor arg; do last=\"$arg\"; done\ncase \"$last\" in\n  *.nfo) printf '%s' '{{\"format\":{{\"format_name\":\"tty\"}}}}' ;;\n  *) cat <<'PROBE_EOF'\n{media_json}\nPROBE_EOF\n;;\nesac\n"
+        );
+        std::fs::write(&path, script).unwrap();
+        let mut permissions = std::fs::metadata(&path).unwrap().permissions();
+        permissions.set_mode(0o755);
+        std::fs::set_permissions(&path, permissions).unwrap();
+        path
+    }
+
+    async fn harness(media_json: &str) -> Harness {
+        harness_with(
+            media_json,
+            json!([
+                {"name": "Dune/Dune.mkv", "length": 13, "bytesCompleted": 13},
+                {"name": "Dune/Dune.nfo", "length": 10, "bytesCompleted": 10}
+            ]),
+        )
+        .await
+    }
+
+    async fn harness_with(media_json: &str, files: serde_json::Value) -> Harness {
+        let dir = tempfile::tempdir().unwrap();
+        let downloads = dir.path().join("downloads");
+        let library = dir.path().join("library");
+        std::fs::create_dir_all(downloads.join("Dune")).unwrap();
+        std::fs::create_dir_all(&library).unwrap();
+        std::fs::write(downloads.join("Dune/Dune.mkv"), b"feature bytes").unwrap();
+        std::fs::write(downloads.join("Dune/Dune.nfo"), b"not a film").unwrap();
+
+        let database = Db::connect(dir.path().join("arr.db")).await.unwrap();
+        database.migrate().await.unwrap();
+        let library_text = library.to_string_lossy().into_owned();
+        sqlx::query("UPDATE roots SET path = ? WHERE kind = 'movie' AND audience = 'main'")
+            .bind(&library_text)
+            .execute(database.pool())
+            .await
+            .unwrap();
+        sqlx::query(
+            "INSERT INTO movies (tmdb_id, title, year, original_language, root_id, state)
+             SELECT 693134, 'Dune: Part Two', 2024, 'en', id, 'grabbed'
+             FROM roots WHERE kind = 'movie' AND audience = 'main'",
+        )
+        .execute(database.pool())
+        .await
+        .unwrap();
+        let release_id = sqlx::query(
+            "INSERT INTO releases (indexer_id, guid, name, size, download_url, parsed, verdict)
+             VALUES (7, 'good', ?, 23622320128, 'magnet:x', '{}', 'eligible')",
+        )
+        .bind(RELEASE_NAME)
+        .execute(database.pool())
+        .await
+        .unwrap()
+        .last_insert_rowid();
+        sqlx::query(
+            "INSERT INTO grabs (release_id, target_kind, target_id, infohash, state)
+             VALUES (?, 'movie', 1, ?, 'downloaded')",
+        )
+        .bind(release_id)
+        .bind(INFOHASH)
+        .execute(database.pool())
+        .await
+        .unwrap();
+
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+                "result": "success",
+                "arguments": {"torrents": [{
+                    "hashString": INFOHASH,
+                    "downloadDir": downloads.to_string_lossy(),
+                    "files": files
+                }]}
+            })))
+            .mount(&server)
+            .await;
+
+        let prober = Prober::new().with_binary(fake_ffprobe(dir.path(), media_json));
+        let action = ImportAction::new(TransmissionClient::new(&server.uri()).unwrap(), prober);
+
+        Harness {
+            _dir: dir,
+            database,
+            downloads,
+            library,
+            action,
+            _server: server,
+        }
+    }
+
+    fn expected_library_file(library: &Path) -> PathBuf {
+        library
+            .join("Dune Part Two (2024) [tmdbid-693134]")
+            .join("Dune Part Two (2024) [tmdbid-693134] - [2160p][WEB-DL][HDR10].mkv")
+    }
+
+    /// The issue's acceptance case: the exact §7.4 path exists, and the
+    /// torrent's own file still exists with a link count of two.
+    #[tokio::test]
+    async fn the_feature_lands_on_the_design_layout_and_keeps_seeding() {
+        let h = harness(HDR10_PROBE).await;
+
+        let outcomes = h.action.tick(&h.database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let library_file = expected_library_file(&h.library);
+        assert!(library_file.is_file(), "missing {}", library_file.display());
+
+        let seeding_file = h.downloads.join("Dune/Dune.mkv");
+        let metadata = std::fs::metadata(&seeding_file).unwrap();
+        assert_eq!(metadata.nlink(), 2, "§7.2: hardlinked, not moved or copied");
+
+        let (path, probed, waiver): (String, String, Option<String>) =
+            sqlx::query_as("SELECT path, probed, waiver FROM media_files WHERE owner_kind = 'movie' AND owner_id = 1")
+                .fetch_one(h.database.pool())
+                .await
+                .unwrap();
+        assert_eq!(path, library_file.to_string_lossy());
+        assert!(probed.contains("\"2160p\""), "{probed}");
+        assert!(probed.contains("HDR10"), "{probed}");
+        assert_eq!(waiver, None);
+
+        let (grab_state, imported_at): (String, Option<String>) =
+            sqlx::query_as("SELECT state, imported_at FROM grabs WHERE infohash = ?")
+                .bind(INFOHASH)
+                .fetch_one(h.database.pool())
+                .await
+                .unwrap();
+        assert_eq!(grab_state, "imported");
+        assert!(imported_at.is_some());
+        let movie_state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
+            .fetch_one(h.database.pool())
+            .await
+            .unwrap();
+        assert_eq!(movie_state, "imported");
+    }
+
+    /// §5.3 through §5.7: Profile 5 is a hard fail — blacklisted, grab
+    /// failed, gap reopened, and the torrent's files untouched.
+    #[tokio::test]
+    async fn a_dolby_vision_profile_5_file_hard_fails() {
+        let h = harness(DV5_PROBE).await;
+
+        let outcomes = h.action.tick(&h.database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        assert!(
+            std::fs::read_dir(&h.library).unwrap().next().is_none(),
+            "nothing may reach the library"
+        );
+        assert!(h.downloads.join("Dune/Dune.mkv").is_file(), "§7.3");
+
+        let (infohash, normalised, reason): (String, String, String) =
+            sqlx::query_as("SELECT infohash, normalised_name, reason FROM blacklist")
+                .fetch_one(h.database.pool())
+                .await
+                .unwrap();
+        assert_eq!(infohash, INFOHASH);
+        assert_eq!(normalised, arr_parse::normalise(RELEASE_NAME));
+        assert_eq!(reason, "dolby_vision_profile");
+
+        let grab_state: String = sqlx::query_scalar("SELECT state FROM grabs")
+            .fetch_one(h.database.pool())
+            .await
+            .unwrap();
+        assert_eq!(grab_state, "failed");
+        let movie_state: String = sqlx::query_scalar("SELECT state FROM movies WHERE id = 1")
+            .fetch_one(h.database.pool())
+            .await
+            .unwrap();
+        assert_eq!(
+            movie_state, "missing",
+            "the gap reopens for the next candidate"
+        );
+    }
+
+    /// §8: killed between the hardlink and the bookkeeping, a restart
+    /// converges instead of failing on the existing destination.
+    #[tokio::test]
+    async fn a_restart_after_the_link_converges() {
+        let h = harness(HDR10_PROBE).await;
+        h.action.tick(&h.database).await.unwrap();
+
+        // The crash: the file is placed, the database never heard.
+        sqlx::query("DELETE FROM media_files")
+            .execute(h.database.pool())
+            .await
+            .unwrap();
+        sqlx::query("UPDATE grabs SET state = 'downloaded', imported_at = NULL")
+            .execute(h.database.pool())
+            .await
+            .unwrap();
+        sqlx::query("UPDATE movies SET state = 'grabbed'")
+            .execute(h.database.pool())
+            .await
+            .unwrap();
+
+        let outcomes = h.action.tick(&h.database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let files: i64 = sqlx::query_scalar("SELECT count(*) FROM media_files")
+            .fetch_one(h.database.pool())
+            .await
+            .unwrap();
+        assert_eq!(files, 1);
+        let metadata = std::fs::metadata(h.downloads.join("Dune/Dune.mkv")).unwrap();
+        assert_eq!(metadata.nlink(), 2, "no second link, no copy");
+    }
+
+    /// A settled tick is idle — an imported grab is not a gap.
+    #[tokio::test]
+    async fn a_second_tick_imports_nothing_new() {
+        let h = harness(HDR10_PROBE).await;
+        h.action.tick(&h.database).await.unwrap();
+
+        let outcomes = h.action.tick(&h.database).await.unwrap();
+
+        assert!(outcomes.is_empty());
+    }
+
+    /// Torrent-declared names are untrusted: absolute and `..`-carrying
+    /// entries are skipped, and the import proceeds from what remains.
+    #[tokio::test]
+    async fn hostile_torrent_paths_never_leave_the_download_root() {
+        let h = harness_with(
+            HDR10_PROBE,
+            json!([
+                {"name": "../outside.mkv", "length": 13, "bytesCompleted": 13},
+                {"name": "/tmp/absolute.mkv", "length": 13, "bytesCompleted": 13},
+                {"name": "Dune/Dune.mkv", "length": 13, "bytesCompleted": 13}
+            ]),
+        )
+        .await;
+
+        let outcomes = h.action.tick(&h.database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        assert!(expected_library_file(&h.library).is_file());
+        let files: i64 = sqlx::query_scalar("SELECT count(*) FROM media_files")
+            .fetch_one(h.database.pool())
+            .await
+            .unwrap();
+        assert_eq!(files, 1, "only the safe path is imported");
+    }
+
+    /// A torrent whose every entry escapes the root has nothing importable:
+    /// hard fail, not an escape.
+    #[tokio::test]
+    async fn a_torrent_of_only_hostile_paths_hard_fails() {
+        let h = harness_with(
+            HDR10_PROBE,
+            json!([{"name": "../../etc/passwd", "length": 13, "bytesCompleted": 13}]),
+        )
+        .await;
+
+        let outcomes = h.action.tick(&h.database).await.unwrap();
+
+        assert_eq!(outcomes.len(), 1);
+        let reason: String = sqlx::query_scalar("SELECT reason FROM blacklist")
+            .fetch_one(h.database.pool())
+            .await
+            .unwrap();
+        assert_eq!(reason, "no readable video file");
+        assert!(std::fs::read_dir(&h.library).unwrap().next().is_none());
+    }
+
+    #[test]
+    fn safe_join_refuses_escapes_and_keeps_normal_paths() {
+        let root = Path::new("/downloads");
+        assert_eq!(
+            safe_join(root, Path::new("Dune/./Dune.mkv")),
+            Some(PathBuf::from("/downloads/Dune/Dune.mkv"))
+        );
+        assert_eq!(safe_join(root, Path::new("../outside.mkv")), None);
+        assert_eq!(safe_join(root, Path::new("Dune/../../outside.mkv")), None);
+        assert_eq!(safe_join(root, Path::new("/etc/passwd")), None);
+        assert_eq!(safe_join(root, Path::new("")), None);
+    }
+
+    /// The reconcile lane cancels the action after 25 s while one probe may
+    /// take 60 s, so results settled on one tick must survive to the next —
+    /// otherwise a large torrent restarts from file one forever.
+    #[tokio::test]
+    async fn probe_results_are_reused_across_calls() {
+        let dir = tempfile::tempdir().unwrap();
+        let media = dir.path().join("film.mkv");
+        std::fs::write(&media, b"bytes").unwrap();
+        let counter = dir.path().join("count");
+        let script = dir.path().join("ffprobe");
+        std::fs::write(
+            &script,
+            format!(
+                "#!/bin/sh\necho x >> {}\ncat <<'PROBE_EOF'\n{HDR10_PROBE}\nPROBE_EOF\n",
+                counter.display()
+            ),
+        )
+        .unwrap();
+        let mut permissions = std::fs::metadata(&script).unwrap().permissions();
+        permissions.set_mode(0o755);
+        std::fs::set_permissions(&script, permissions).unwrap();
+
+        let action = ImportAction::new(
+            TransmissionClient::new("http://127.0.0.1:1").unwrap(),
+            Prober::new().with_binary(&script),
+        );
+        let paths = vec![media];
+
+        action.probe_all(&paths).await.unwrap();
+        action.probe_all(&paths).await.unwrap();
+        assert_eq!(
+            std::fs::read_to_string(&counter).unwrap().lines().count(),
+            1,
+            "the second call reuses the first call's result"
+        );
+
+        action.forget_probes(&paths).await;
+        action.probe_all(&paths).await.unwrap();
+        assert_eq!(
+            std::fs::read_to_string(&counter).unwrap().lines().count(),
+            2,
+            "a settled grab's entries are dropped"
+        );
+    }
+
+    /// The probe outlives a cancelled tick: the detached task deposits its
+    /// result after the caller's future is dropped, and the next tick reads
+    /// it instead of restarting the same probe forever.
+    #[tokio::test]
+    async fn a_cancelled_probe_still_deposits_its_result() {
+        let dir = tempfile::tempdir().unwrap();
+        let media = dir.path().join("film.mkv");
+        std::fs::write(&media, b"bytes").unwrap();
+        let counter = dir.path().join("count");
+        let script = dir.path().join("ffprobe");
+        std::fs::write(
+            &script,
+            format!(
+                "#!/bin/sh\nsleep 0.3\necho x >> {}\ncat <<'PROBE_EOF'\n{HDR10_PROBE}\nPROBE_EOF\n",
+                counter.display()
+            ),
+        )
+        .unwrap();
+        let mut permissions = std::fs::metadata(&script).unwrap().permissions();
+        permissions.set_mode(0o755);
+        std::fs::set_permissions(&script, permissions).unwrap();
+        let action = ImportAction::new(
+            TransmissionClient::new("http://127.0.0.1:1").unwrap(),
+            Prober::new().with_binary(&script),
+        );
+        let paths = vec![media];
+
+        // The lane's budget, in miniature: the tick is cancelled mid-probe.
+        let cancelled = tokio::time::timeout(
+            std::time::Duration::from_millis(50),
+            action.probe_all(&paths),
+        )
+        .await;
+        assert!(cancelled.is_err());
+
+        // The detached probe finishes on its own and deposits the result.
+        tokio::time::sleep(std::time::Duration::from_millis(600)).await;
+        let files = action.probe_all(&paths).await.unwrap();
+        assert_eq!(files.len(), 1);
+        assert_eq!(
+            std::fs::read_to_string(&counter).unwrap().lines().count(),
+            1,
+            "the next tick reused the deposited result instead of re-probing"
+        );
+    }
+
+    /// The `EXDEV` fallback path lands whole files via rename (§7.2).
+    #[test]
+    fn the_copy_fallback_lands_a_whole_file_and_cleans_up() {
+        let dir = tempfile::tempdir().unwrap();
+        let source = dir.path().join("source.mkv");
+        std::fs::write(&source, b"feature bytes").unwrap();
+        let destination = dir.path().join("library").join("feature.mkv");
+        std::fs::create_dir_all(destination.parent().unwrap()).unwrap();
+
+        let placement = copy_into_place(&source, &destination).unwrap();
+
+        assert_eq!(placement, Placement::Copied);
+        assert_eq!(std::fs::read(&destination).unwrap(), b"feature bytes");
+        assert!(
+            std::fs::read_dir(destination.parent().unwrap())
+                .unwrap()
+                .all(|entry| !entry
+                    .unwrap()
+                    .file_name()
+                    .to_string_lossy()
+                    .contains("partial")),
+            "no partial file left behind"
+        );
+    }
+}

crates/arr-daemon/src/main.rs

@@ -2,6 +2,7 @@
 
 mod config;
 mod grab;
+mod import;
 pub mod reconcile;
 mod web;
 
@@ -14,6 +15,7 @@ use arr_db::Db;
 use arr_meta::TmdbClient;
 use config::Config;
 use grab::{GrabAction, SeedingLimits};
+use import::ImportAction;
 use reconcile::{ReconcileLoop, Tick};
 use tower_http::trace::TraceLayer;
 
@@ -89,17 +91,17 @@ async fn run() -> Result<(), Error> {
     let database = Db::connect(&config.database_path).await?;
     database.migrate().await?;
 
+    let transmission = arr_dl::TransmissionClient::new(&config.transmission_url)?;
     let mut reconcile = ReconcileLoop::new(database.clone());
     // Without a Prowlarr key nothing can be searched, so the grab lane stays
     // unregistered rather than failing a tick every 30 seconds.
     if let Some(key) = config.prowlarr_api_key.clone() {
         let prowlarr = arr_indexer::ProwlarrClient::new(config.prowlarr_url.clone(), key)?;
-        let transmission = arr_dl::TransmissionClient::new(&config.transmission_url)?;
         reconcile = reconcile.register(
             Tick::Reconcile,
             GrabAction::new(
                 prowlarr,
-                transmission,
+                transmission.clone(),
                 config.download_dir.clone(),
                 SeedingLimits {
                     ratio: config.seed_ratio_limit,
@@ -110,6 +112,12 @@ async fn run() -> Result<(), Error> {
     } else {
         tracing::warn!("no Prowlarr API key: nothing will be grabbed");
     }
+    // Grab before import, so a download that completes on this tick is
+    // imported on this tick.
+    reconcile = reconcile.register(
+        Tick::Reconcile,
+        ImportAction::new(transmission, arr_probe::Prober::new()),
+    );
 
     // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and
     // needs its own TMDB client for `movie/lookup`.

crates/arr-dl/src/lib.rs

@@ -70,6 +70,24 @@ pub struct Torrent {
     pub labels: Vec<String>,
 }
 
+/// One file inside a torrent, as Transmission reports it.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct TorrentFile {
+    /// Path relative to the torrent's download directory, torrent folder
+    /// included.
+    pub path: PathBuf,
+    /// Size in bytes.
+    pub size: u64,
+}
+
+/// Where one torrent's data lives on disk.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct TorrentContent {
+    pub hash: String,
+    pub download_dir: PathBuf,
+    pub files: Vec<TorrentFile>,
+}
+
 /// A Transmission RPC failure.
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
@@ -218,6 +236,46 @@ impl TransmissionClient {
         Ok(response.torrents.into_iter().map(Torrent::from).collect())
     }
 
+    /// The file list and download directory of one torrent, by infohash.
+    ///
+    /// `None` when Transmission no longer knows the hash — the operator may
+    /// have removed the torrent by hand.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error for transport failures or rejected/malformed RPC
+    /// responses.
+    pub async fn torrent_content(&self, hash: &str) -> Result<Option<TorrentContent>, Error> {
+        let arguments = self
+            .call(
+                "torrent-get",
+                json!({
+                    "ids": [hash],
+                    "fields": ["hashString", "downloadDir", "files"]
+                }),
+            )
+            .await?;
+        let response: RpcContentList = serde_json::from_value(arguments)
+            .map_err(|error| Error::InvalidResponse(error.to_string()))?;
+
+        Ok(response
+            .torrents
+            .into_iter()
+            .next()
+            .map(|torrent| TorrentContent {
+                hash: torrent.hash,
+                download_dir: torrent.download_dir,
+                files: torrent
+                    .files
+                    .into_iter()
+                    .map(|file| TorrentFile {
+                        path: file.name,
+                        size: file.length,
+                    })
+                    .collect(),
+            }))
+    }
+
     /// Remove one torrent, optionally deleting its downloaded data.
     ///
     /// # Errors
@@ -291,6 +349,27 @@ struct RpcTorrentList {
     torrents: Vec<RpcTorrent>,
 }
 
+#[derive(Debug, Deserialize)]
+struct RpcContentList {
+    torrents: Vec<RpcContent>,
+}
+
+#[derive(Debug, Deserialize)]
+struct RpcContent {
+    #[serde(rename = "hashString")]
+    hash: String,
+    #[serde(rename = "downloadDir")]
+    download_dir: PathBuf,
+    #[serde(default)]
+    files: Vec<RpcFile>,
+}
+
+#[derive(Debug, Deserialize)]
+struct RpcFile {
+    name: PathBuf,
+    length: u64,
+}
+
 #[derive(Debug, Deserialize)]
 struct RpcTorrent {
     id: i64,
@@ -459,6 +538,61 @@ mod tests {
         assert_eq!(torrents[0].state, TorrentState::Unknown(99));
     }
 
+    #[tokio::test]
+    async fn torrent_content_lists_files_and_download_dir() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(body_partial_json(json!({
+                "method": "torrent-get",
+                "arguments": {"ids": ["abc"]}
+            })))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+                "result": "success",
+                "arguments": {"torrents": [{
+                    "hashString": "abc",
+                    "downloadDir": "/downloads",
+                    "files": [
+                        {"name": "Movie/Movie.mkv", "length": 100, "bytesCompleted": 100},
+                        {"name": "Movie/Movie.nfo", "length": 5, "bytesCompleted": 5}
+                    ]
+                }]}
+            })))
+            .expect(1)
+            .mount(&server)
+            .await;
+
+        let client = TransmissionClient::new(&server.uri()).expect("client");
+        let content = client
+            .torrent_content("abc")
+            .await
+            .expect("content")
+            .expect("torrent known");
+
+        assert_eq!(content.download_dir, PathBuf::from("/downloads"));
+        assert_eq!(content.files.len(), 2);
+        assert_eq!(content.files[0].path, PathBuf::from("Movie/Movie.mkv"));
+        assert_eq!(content.files[0].size, 100);
+    }
+
+    #[tokio::test]
+    async fn a_torrent_transmission_forgot_is_none_not_an_error() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+                "result": "success",
+                "arguments": {"torrents": []}
+            })))
+            .mount(&server)
+            .await;
+
+        let client = TransmissionClient::new(&server.uri()).expect("client");
+        assert!(client
+            .torrent_content("gone")
+            .await
+            .expect("call succeeds")
+            .is_none());
+    }
+
     fn success() -> ResponseTemplate {
         ResponseTemplate::new(200).set_body_json(json!({"result": "success", "arguments": {}}))
     }

crates/arr-parse/src/lib.rs

@@ -332,6 +332,27 @@ pub fn parse(name: &str) -> NameClaims {
     claims
 }
 
+/// Collapse a release name to the blacklist key (`DESIGN.md` §6.3): lowercase,
+/// every run of non-alphanumerics as one dot. The same release reappearing
+/// with different separators or a different infohash still matches.
+#[must_use]
+pub fn normalise(name: &str) -> String {
+    let mut out = String::with_capacity(name.len());
+    let mut pending_gap = false;
+    for character in name.chars() {
+        if character.is_alphanumeric() {
+            if pending_gap && !out.is_empty() {
+                out.push('.');
+            }
+            pending_gap = false;
+            out.extend(character.to_lowercase());
+        } else {
+            pending_gap = true;
+        }
+    }
+    out
+}
+
 fn apply_marker(claims: &mut NameClaims, sources: &mut Vec<Source>, m: Marker) {
     match m {
         Marker::Resolution(r) => {

crates/arr-parse/tests/parse.rs

@@ -423,3 +423,15 @@ fn non_tv_names_claim_no_episode() {
         assert_eq!(parse(name).episode, None, "{name}");
     }
 }
+
+#[test]
+fn normalisation_is_separator_and_case_blind() {
+    let key = arr_parse::normalise("Dune.Part.Two.2024.2160p.WEB-DL.DDP5.1.Atmos");
+    assert_eq!(key, "dune.part.two.2024.2160p.web.dl.ddp5.1.atmos");
+    assert_eq!(
+        arr_parse::normalise("dune part two [2024] 2160p WEB DL DDP5 1 atmos "),
+        key,
+        "the same release under different separators is the same blacklist key"
+    );
+    assert_eq!(arr_parse::normalise("!!!"), "");
+}