feat(subs): translate from an image-only release

A release whose only subtitle is a PGS or VobSub track was stuck both ways: the track satisfied English so no English SRT was ever fetched, and bitmaps can never feed a translator. §15 is amended to separate satisfying viewing from providing a translation source, and to carve a fetch made to obtain a source out of the no-upgrade rule. Timings come from the disc rather than from alass guessing at the audio. At import, each non-forced image track's packet timestamps are paired show-to-clear into a cue skeleton and stored; the fetched source is then aligned against that skeleton, and the translation made from it skips the post-translation pass, which could only move disc-exact timings off. Pairing is validated before it is trusted — even packet count, plausible durations, sane density for the runtime — because PGS allows several composition segments per subtitle and a slipped pairing is quietly half a second out. A track that fails validation gets no skeleton and falls back to aligning against the video, as does any file imported before this: there is no backfill. Closes #268 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-30 11:53:26 +01:00
parent 015b90a458
commit 6cd2902341
18 changed files with 1514 additions and 97 deletions
@@ -1,6 +1,6 @@
 {
  "db_name": "SQLite",
-  "query": "INSERT INTO subtitle_files\n               (media_file_id, language, origin, provider, candidate_id, engine,\n                forced, sdh, synced, sync_rejected, path)\n           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)\n           ON CONFLICT (path) DO NOTHING\n           ON CONFLICT (media_file_id, language, forced, sdh)\n               WHERE origin = 'embedded' DO NOTHING\n           RETURNING id AS \"id!: i64\"",
+  "query": "INSERT INTO subtitle_files\n               (media_file_id, language, origin, provider, candidate_id, engine,\n                forced, sdh, synced, sync_rejected, skeleton_aligned, path)\n           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)\n           ON CONFLICT (path) DO NOTHING\n           ON CONFLICT (media_file_id, language, forced, sdh)\n               WHERE origin = 'embedded' DO NOTHING\n           RETURNING id AS \"id!: i64\"",
  "describe": {
    "columns": [
      {
@@ -16,11 +16,11 @@
      }
    ],
    "parameters": {
-      "Right": 11
+      "Right": 12
    },
    "nullable": [
      null
    ]
  },
-  "hash": "3d48b78768bf8dc2e337d0935889a5654089f38bb7ee2a15b7e19f4ed5540262"
+  "hash": "266c8e63e7f81cb07921de180f32b47ac8c4e9d47a31af786742404f8e54f517"
 }
@@ -0,0 +1,26 @@
 {
  "db_name": "SQLite",
  "query": "SELECT id AS \"id!: i64\" FROM media_files WHERE path = ?",
  "describe": {
    "columns": [
      {
        "name": "id!: i64",
        "ordinal": 0,
        "type_info": "Integer",
        "origin": {
          "Table": {
            "table": "media_files",
            "name": "id"
          }
        }
      }
    ],
    "parameters": {
      "Right": 1
    },
    "nullable": [
      false
    ]
  },
  "hash": "8651688f9f999629b4996c0451f87f5a84fabe748849b4b80063af1c18f73917"
 }
@@ -1,6 +1,6 @@
 {
  "db_name": "SQLite",
-  "query": "SELECT id            AS \"id!: i64\",\n                  media_file_id AS \"media_file_id!: i64\",\n                  language      AS \"language!: String\",\n                  origin        AS \"origin!: SubtitleOrigin\",\n                  provider,\n                  candidate_id,\n                  engine,\n                  forced        AS \"forced!: bool\",\n                  sdh           AS \"sdh!: bool\",\n                  synced        AS \"synced!: bool\",\n                  sync_rejected AS \"sync_rejected!: bool\",\n                  path\n           FROM   subtitle_files\n           WHERE  media_file_id = ?\n           ORDER BY language, id",
+  "query": "SELECT id            AS \"id!: i64\",\n                  media_file_id AS \"media_file_id!: i64\",\n                  language      AS \"language!: String\",\n                  origin        AS \"origin!: SubtitleOrigin\",\n                  provider,\n                  candidate_id,\n                  engine,\n                  forced        AS \"forced!: bool\",\n                  sdh           AS \"sdh!: bool\",\n                  synced        AS \"synced!: bool\",\n                  sync_rejected AS \"sync_rejected!: bool\",\n                  skeleton_aligned AS \"skeleton_aligned!: bool\",\n                  path\n           FROM   subtitle_files\n           WHERE  media_file_id = ?\n           ORDER BY language, id",
  "describe": {
    "columns": [
      {
@@ -125,8 +125,19 @@
        }
      },
      {
-        "name": "path",
+        "name": "skeleton_aligned!: bool",
        "ordinal": 11,
        "type_info": "Integer",
        "origin": {
          "Table": {
            "table": "subtitle_files",
            "name": "skeleton_aligned"
          }
        }
      },
      {
        "name": "path",
        "ordinal": 12,
        "type_info": "Text",
        "origin": {
          "Table": {
@@ -151,8 +162,9 @@
      false,
      false,
      false,
      false,
      true
    ]
  },
-  "hash": "62d625c322c8c64b48317cdbf466d74043708d8e12997e0efec23a18eaaaaace"
+  "hash": "8abb1e992e0972a19e76c7788c8c350951a1d14766088b1cf3fb886fb658d575"
 }
@@ -0,0 +1,12 @@
 {
  "db_name": "SQLite",
  "query": "INSERT INTO subtitle_skeletons (media_file_id, stream_index, language, cues)\n         VALUES (?, ?, ?, ?)\n         ON CONFLICT (media_file_id, stream_index) DO UPDATE SET\n             language = excluded.language,\n             cues     = excluded.cues",
  "describe": {
    "columns": [],
    "parameters": {
      "Right": 4
    },
    "nullable": []
  },
  "hash": "8e3c23ed488bc82ff3dd4f6da02d00435525d321b6002b9f31979a39281369ce"
 }
@@ -0,0 +1,50 @@
 {
  "db_name": "SQLite",
  "query": "SELECT stream_index AS \"stream_index!: i64\",\n                  language     AS \"language!: String\",\n                  cues         AS \"cues!: String\"\n           FROM   subtitle_skeletons\n           WHERE  media_file_id = ?\n           ORDER BY stream_index",
  "describe": {
    "columns": [
      {
        "name": "stream_index!: i64",
        "ordinal": 0,
        "type_info": "Integer",
        "origin": {
          "Table": {
            "table": "subtitle_skeletons",
            "name": "stream_index"
          }
        }
      },
      {
        "name": "language!: String",
        "ordinal": 1,
        "type_info": "Text",
        "origin": {
          "Table": {
            "table": "subtitle_skeletons",
            "name": "language"
          }
        }
      },
      {
        "name": "cues!: String",
        "ordinal": 2,
        "type_info": "Text",
        "origin": {
          "Table": {
            "table": "subtitle_skeletons",
            "name": "cues"
          }
        }
      }
    ],
    "parameters": {
      "Right": 1
    },
    "nullable": [
      false,
      false,
      false
    ]
  },
  "hash": "eb7df52055a60a00b721e3eef152e1b04a8512a85e8ad3618545c634c23abf31"
 }
@@ -122,6 +122,7 @@ dependencies = [
 "tempfile",
 "thiserror",
 "tokio",
 "tracing",
 ]
 [[package]]
@@ -940,7 +940,9 @@ Replaces Bazarr. Phase 9 in §13; the `arr-subs` crate in §11.
 **Wanted set.** Global, not per root. Two languages are separately wanted for
 every media file: Portuguese — pt-PT preferred, pt-BR accepted — and English.
 A file is satisfied for a language when a subtitle in it exists, embedded or as
-a sidecar. This is deliberately unlike §5.2's audio rules, which attach to a
+a sidecar. Satisfying a language is a statement about *viewing* it, and not
 about having text in it — the distinction matters where an image track is all
 there is, and **Translation** below is where it bites. This is deliberately unlike §5.2's audio rules, which attach to a
 root: subtitles carry no blacklist there and none here. pt-BR subtitles are
 always fine.
@@ -952,6 +954,22 @@ they satisfy viewing but can never feed a translator, and arr does not OCR
 them. `arr-probe` already reports subtitle tracks with resolved languages; the
 format is the new fact it must carry.
 **Cue skeletons.** What an image track *does* have is exact timings, because
 its packet timestamps are the disc's own cue structure. At import — while the
 file is being read and hardlinked anyway — arr derives a **cue skeleton** from
 each non-forced image track: `ffprobe -show_packets` on the track, packets
 paired show-to-clear, no pixel read. The skeleton has timings and no text, and
 that is enough, because `alass` matches on interval structure rather than on
 words. A sparse skeleton is still a strong reference; a downloaded subtitle and
 a retail disc's track never carry the same cues anyway.
 The pairing is the whole of it, so it is validated before it is trusted: an
 even packet count, every implied duration plausible, and a cue density that
 fits the runtime. PGS permits several composition segments per subtitle, and a
 track built that way pairs into something that is quietly half a second out —
 worse than no skeleton. A track that fails validation gets none, and alignment
 falls back to the video.
 **Providers.** OpenSubtitles.com, behind one trait.
 **Ranking.** A `moviehash` match wins outright. Then an exact release-name
@@ -975,6 +993,18 @@ a downloaded one, or one extracted from a text-format embedded track. Being
 able to translate from an embedded track is a deliberate improvement on
 Bazarr, which cannot.
 **Fetching a source.** A release whose only subtitle is an image track has no
 text source and never will: the track satisfies its language, so nothing is
 ever fetched in it, and the track itself cannot be translated from. That is a
 deadlock, and it is broken by a narrow carve-out — when a wanted language needs
 translating and no text source exists, arr fetches a text subtitle in the image
 track's language *even though that language reads as satisfied*. The fetch
 obtains a source; it settles no want of its own, and the no-upgrade rule below
 does not apply to it. Where that track has a cue skeleton, the fetched subtitle
 is aligned against the skeleton rather than the video, which puts it on the
 disc's own timings instead of on a heuristic read of the audio. OCR remains a
 non-goal: this reaches the same place with real text.
 **Translation backends.** Pluggable, each behind its own cargo feature: an
 OpenAI-compatible HTTP endpoint, DeepL, Google Translate, and a generic remote
 command driven by a configured template (`ssh box claude -p` is one instance
@@ -988,6 +1018,11 @@ too — arr stops working on it. A real subtitle appearing later does not
 replace anything. Replacement is a manual action from the UI. This is §5.4's
 rule applied to subtitles.
 The one exception is the source fetch above. It is not an upgrade: the language
 it downloads is already satisfied and stays satisfied by the same track it was
 before, and what the download settles is a *different* language's gap. Nothing
 is replaced, so nothing about the rule changes.
 **On disk.** Sidecars live next to the video inside the §7.4 title folder,
 named `<video basename>.<lang>.srt`, e.g.
 `… - [2160p][WEB-DL][HDR10].pt-PT.srt`. A machine translation carries an extra
@@ -1006,6 +1041,12 @@ reports no confidence value, so its output is accepted unless it is
 implausible — a shift beyond 60 seconds, or cues lost — in which case the
 unsynced original is kept and the file is flagged.
 The reference is the video, except where a cue skeleton exists, and then it is
 the skeleton. A subtitle a skeleton accepted is already on disc-exact timings,
 and translation copies those timings over untouched, so the pass that would
 otherwise run on the translation is skipped: a second alignment has nothing
 left to find and can only move them off.
 **Configuration.** Provider credentials and translator API keys are bootstrap
 config or environment, per §10 — a secret never becomes a database row. Wanted
 languages, chosen engine, per-provider enable and the daily budgets are
@@ -365,6 +365,14 @@ impl SubtitleCodec {
        matches!(self, Self::SubRip | Self::Ass | Self::MovText)
    }
    /// §15's bitmap tracks. They carry no text, but their packet timings are
    /// exact for the release they came off, which is what a cue skeleton is
    /// derived from (#268).
    #[must_use]
    pub const fn is_image(self) -> bool {
        matches!(self, Self::Pgs | Self::VobSub)
    }
    /// The inverse of [`Display`](fmt::Display): read back a codec that was
    /// written out under its `ffprobe` name. `ffprobe`'s own aliases are
    /// accepted alongside, so a column written by an older probe still
@@ -23,7 +23,7 @@ use arr_core::policy::{evaluate, Candidate};
 use arr_core::{ProbedMedia, Rule, Source, Verdict};
 use arr_db::Db;
 use arr_dl::QbitClient;
-use arr_probe::Prober;
+use arr_probe::{Prober, Skeletons};
 use crate::notify::Notifier;
 use crate::reconcile::{Action, ActionFuture, Outcome};
@@ -75,6 +75,11 @@ enum ProbeOutcome {
 pub struct ImportAction {
    qbit: QbitClient,
    prober: Prober,
    /// §15, #268. Reads the packet timings of image-format subtitle tracks
    /// into a cue skeleton. Here rather than in the subtitle lane because
    /// deriving one costs a full demux, and import is where the file is
    /// being read and hardlinked anyway.
    skeletons: Skeletons,
    jellyfin: JellyfinClient,
    notifier: Notifier,
    /// The operator's ntfy topic (DESIGN.md §9.5), for the *broken*
@@ -121,6 +126,7 @@ impl ImportAction {
        Self {
            qbit,
            prober,
            skeletons: Skeletons::new(),
            jellyfin,
            notifier,
            operator_topic,
@@ -176,6 +182,94 @@ impl ImportAction {
        Ok(files)
    }
    /// Derive the cue skeleton of every non-forced image-format subtitle
    /// track in a file that has just been placed (§15, #268).
    ///
    /// Detached, like the probes above: one derivation is a full demux and
    /// the reconcile lane's tick budget is 25 s. Nothing downstream waits on
    /// it — a skeleton that never lands, because the process died or because
    /// the packets did not pair, reads exactly like a track that has none,
    /// and the subtitle lane aligns against the video instead.
    async fn spawn_skeletons(&self, database: &Db, placed: &Path, feature: &arr_probe::ProbedFile) {
        let tracks: Vec<(usize, String)> = feature
            .media
            .subtitle_tracks
            .iter()
            .enumerate()
            .filter(|(_, track)| !track.forced && track.codec.is_image())
            .map(|(index, track)| (index, track.language.to_string()))
            .collect();
        if tracks.is_empty() {
            return;
        }
        let path_text = placed.to_string_lossy().into_owned();
        let media_file_id = match sqlx::query_scalar!(
            r#"SELECT id AS "id!: i64" FROM media_files WHERE path = ?"#,
            path_text
        )
        .fetch_optional(database.pool())
        .await
        {
            Ok(Some(id)) => id,
            Ok(None) => return,
            Err(error) => {
                tracing::warn!(path = %placed.display(), %error, "no media file row to hang a cue skeleton on");
                return;
            }
        };
        let skeletons = self.skeletons.clone();
        let database = database.clone();
        let video = placed.to_path_buf();
        let runtime = feature.duration;
        tokio::spawn(async move {
            for (stream_index, language) in tracks {
                let outcome = skeletons.derive(&video, stream_index, Some(runtime)).await;
                let index = i64::try_from(stream_index).unwrap_or(i64::MAX);
                match outcome {
                    Ok(arr_probe::SkeletonOutcome::Derived(skeleton)) => {
                        let cues: Vec<(i64, i64)> = skeleton
                            .cues
                            .iter()
                            .map(|cue| (millis(cue.start), millis(cue.end)))
                            .collect();
                        if let Err(error) = arr_db::skeletons::record(
                            database.pool(),
                            media_file_id,
                            index,
                            &language,
                            &cues,
                        )
                        .await
                        {
                            tracing::warn!(%error, media_file_id, stream_index, "cue skeleton not recorded");
                        } else {
                            tracing::info!(
                                media_file_id,
                                stream_index,
                                cues = cues.len(),
                                "cue skeleton derived"
                            );
                        }
                    }
                    Ok(arr_probe::SkeletonOutcome::Implausible(reason)) => tracing::info!(
                        media_file_id,
                        stream_index,
                        %reason,
                        "cue skeleton refused, alignment falls back to the video"
                    ),
                    Err(error) => tracing::warn!(
                        media_file_id,
                        stream_index,
                        %error,
                        "cue skeleton could not be read"
                    ),
                }
            }
        });
    }
    /// Drop a settled grab's probe results — imported or blacklisted, they
    /// will not be needed again.
    async fn forget_probes(&self, paths: &[PathBuf]) {
@@ -345,6 +439,7 @@ impl ImportAction {
            tokio::task::spawn_blocking(move || place(&source_path, &link_target)).await??;
        record_import(database, pending, &feature, waiver.as_ref(), &destination).await?;
        self.spawn_skeletons(database, &destination, &feature).await;
        self.forget_probes(&paths).await;
        self.refresh_jellyfin().await;
        let path_text = destination.to_string_lossy().into_owned();
@@ -591,6 +686,7 @@ impl ImportAction {
            record_episode_import(database, episode.id, feature, waiver.as_ref(), &destination)
                .await?;
            self.spawn_skeletons(database, &destination, feature).await;
            imported += 1;
            tracing::info!(
                grab_id = pending.grab_id,
@@ -1200,6 +1296,11 @@ async fn record_episode_import(
    Ok(())
 }
 /// Milliseconds, saturating. A cue past 292 million years does not exist.
 fn millis(value: std::time::Duration) -> i64 {
    i64::try_from(value.as_millis()).unwrap_or(i64::MAX)
 }
 /// The `probed` column (§4, §5.6): what `ffprobe` found, in the spellings the
 /// policy columns use.
 fn probed_json(media: &ProbedMedia) -> serde_json::Value {
@@ -151,6 +151,14 @@ impl EmbeddedTrack {
    fn is_text(&self) -> bool {
        matches!(self.codec.as_deref(), Some("subrip" | "ass" | "mov_text"))
    }
    /// §15's bitmap tracks. They feed no translator, but their timings are
    /// exact for the release, which is what a cue skeleton is made of (#268).
    fn is_image(&self) -> bool {
        self.codec
            .as_deref()
            .is_some_and(|codec| arr_core::SubtitleCodec::from_probe_name(codec).is_image())
    }
 }
 fn embedded_tracks(probed: &str) -> Vec<EmbeddedTrack> {
@@ -403,6 +411,73 @@ struct Worker {
    in_flight: Arc<Mutex<BTreeSet<(i64, String)>>>,
 }
 /// A finished translation on its way to disk.
 struct Translation<'a> {
    engine: String,
    cues: &'a [arr_subs::Cue],
    /// Its source came off a cue skeleton, so §15's post-translation `alass`
    /// pass is skipped (#268).
    skeleton_aligned: bool,
 }
 /// A subtitle to translate from (§15).
 struct Source {
    path: String,
    language: Language,
    /// Its timings came off a cue skeleton, so they are exact for this
    /// release and no further `alass` pass should touch them (#268).
    skeleton_aligned: bool,
 }
 /// Whether a translation source could be had, and at whose expense.
 ///
 /// Distinct from `Option` because looking for one may now spend a provider
 /// download (#268), so "none" and "not today" are different answers and land
 /// the gap in different states.
 enum SourceOutcome {
    Found(Source),
    /// Nothing on this file can feed a translator.
    None,
    /// A source exists but the provider is at its daily allowance.
    Capped,
    /// The fetch was tried and broke.
    Failed(String),
 }
 impl SourceOutcome {
    /// The source, or the state and sentence the attempt row records instead.
    fn settled(self, wanted_tag: &str) -> Result<Source, (SubtitleState, String)> {
        match self {
            Self::Found(source) => Ok(source),
            Self::None => Err((
                SubtitleState::Unavailable,
                format!(
                    "no provider has {wanted_tag} and there is no text source to translate from"
                ),
            )),
            Self::Capped => Err((
                SubtitleState::Capped,
                "provider daily budget exhausted fetching a translation source".to_owned(),
            )),
            Self::Failed(reason) => Err((SubtitleState::Failed, reason)),
        }
    }
 }
 /// Read a translation source off disk. The error is the sentence the attempt
 /// row records.
 async fn read_cues(path: &str) -> Result<Vec<arr_subs::Cue>, String> {
    let raw = tokio::fs::read_to_string(path)
        .await
        .map_err(|error| format!("{path}: {error}"))?;
    arr_subs::srt::parse(&raw).map_err(|error| format!("{path}: not SRT: {error}"))
 }
 /// Milliseconds as stored in a cue skeleton.
 fn duration_from_millis(value: i64) -> std::time::Duration {
    std::time::Duration::from_millis(u64::try_from(value).unwrap_or(0))
 }
 /// What one close settled to, for the log.
 enum Closed {
    Fetched {
@@ -756,17 +831,19 @@ impl Worker {
            Ok(Closed::Recorded { state, reason })
        };
-        let Some((source_path, source_language)) =
+        let source = match self
-            self.translation_source(database, target).await?
+            .translation_source(database, target, wanted, settings)
-        else {
+            .await?
-            return record(
+            .settled(wanted_tag)
-                SubtitleState::Unavailable,
+        {
-                format!(
+            Ok(source) => source,
-                    "no provider has {wanted_tag} and there is no text source to translate from"
+            Err((state, reason)) => return record(state, reason).await,
                ),
            )
            .await;
        };
        let Source {
            path: source_path,
            language: source_language,
            skeleton_aligned,
        } = source;
        let Some(engine) = settings.translation_engine.clone() else {
            return record(
@@ -788,41 +865,14 @@ impl Worker {
            .await;
        };
-        let raw = match tokio::fs::read_to_string(&source_path).await {
+        let cues = match read_cues(&source_path).await {
            Ok(raw) => raw,
            Err(error) => {
                return record(SubtitleState::Failed, format!("{source_path}: {error}")).await
            }
        };
        let cues = match arr_subs::srt::parse(&raw) {
            Ok(cues) => cues,
-            Err(error) => {
+            Err(reason) => return record(SubtitleState::Failed, reason).await,
                return record(
                    SubtitleState::Failed,
                    format!("{source_path}: not SRT: {error}"),
                )
                .await
            }
        };
-        // Translators bill per character, not per call (§15), and the text
+        if !self
-        // going out is known before any of it is sent — no need to ask the
+            .claim_translator_budget(database, &engine, &cues, settings)
-        // backend afterwards (#197).
+            .await?
        let characters = i64::try_from(
            cues.iter()
                .map(|cue| cue.text.chars().count())
                .sum::<usize>(),
        )
        .unwrap_or(i64::MAX);
        let allowance = settings.translator_allowance(&engine);
        if !budget::try_spend(
            database.pool(),
            BudgetKind::Translator,
            &engine,
            characters,
            allowance,
        )
        .await?
        {
            return record(
                SubtitleState::Capped,
@@ -846,8 +896,45 @@ impl Worker {
            };
        self.clear_broken(&broken_name).await;
-        self.write_translation(database, target, wanted, wanted_tag, engine, &translated)
+        self.write_translation(
-            .await
+            database,
            target,
            wanted,
            wanted_tag,
            Translation {
                engine,
                cues: &translated,
                skeleton_aligned,
            },
        )
        .await
    }
    /// Claim the translator's daily character allowance for `cues` (§15,
    /// #197). Translators bill per character, not per call, and the text
    /// going out is known before any of it is sent — no need to ask the
    /// backend afterwards.
    async fn claim_translator_budget(
        &self,
        database: &Db,
        engine: &str,
        cues: &[arr_subs::Cue],
        settings: &Settings,
    ) -> Result<bool, SubtitleError> {
        let characters = i64::try_from(
            cues.iter()
                .map(|cue| cue.text.chars().count())
                .sum::<usize>(),
        )
        .unwrap_or(i64::MAX);
        Ok(budget::try_spend(
            database.pool(),
            BudgetKind::Translator,
            engine,
            characters,
            settings.translator_allowance(engine),
        )
        .await?)
    }
    /// Sync and write a finished translation, record it, settle the want.
@@ -857,9 +944,13 @@ impl Worker {
        target: &Target,
        wanted: &Language,
        wanted_tag: &str,
-        engine: String,
+        translation: Translation<'_>,
        translated: &[arr_subs::Cue],
    ) -> Result<Closed, SubtitleError> {
        let Translation {
            engine,
            cues: translated,
            skeleton_aligned,
        } = translation;
        let media_file_id = target.media_file_id;
        let record = |state: SubtitleState, reason: String| async move {
            db::record_attempt(
@@ -884,37 +975,58 @@ impl Worker {
            return record(SubtitleState::Failed, error).await;
        }
-        let sync = self.syncer.settle(&target.path, &destination).await;
+        // §15 as amended by #268: a source aligned against a cue skeleton
        // already carries the disc's own timings, and translation copies them
        // over unchanged. A second `alass` pass has nothing left to find and
        // can only move them off, so it is skipped.
        let sync = if skeleton_aligned {
            arr_subs::Settled {
                content: None,
                state: arr_subs::SyncState::NotRun,
            }
        } else {
            self.syncer.settle(&target.path, &destination).await
        };
        if let Some(synced) = &sync.content {
            if let Err(error) = write_sidecar(&destination, synced).await {
                return record(SubtitleState::Failed, error).await;
            }
        }
-        db::record_file(
+        let mut record_row = arr_db::NewSubtitleFile::translated(
-            database.pool(),
+            media_file_id,
-            &arr_db::NewSubtitleFile::translated(
+            wanted_tag,
-                media_file_id,
+            &engine,
-                wanted_tag,
+            &destination.to_string_lossy(),
                &engine,
                &destination.to_string_lossy(),
            )
            .sync(db_sync_state(sync.state)),
        )
-        .await?;
+        .sync(db_sync_state(sync.state));
        if skeleton_aligned {
            record_row = record_row.skeleton_aligned();
        }
        db::record_file(database.pool(), &record_row).await?;
        db::mark_satisfied(database.pool(), media_file_id, wanted_tag).await?;
        self.refresh_jellyfin().await;
        Ok(Closed::Translated { engine })
    }
    /// The subtitle to translate from: any non-forced sidecar arr wrote — a
-    /// real one before a machine translation — or, failing that, a
+    /// real one before a machine translation — then a text-format embedded
-    /// text-format embedded track extracted now (§15).
+    /// track extracted now, and failing both a text subtitle downloaded for
    /// the purpose (§15, #268).
    ///
    /// That last step is the one §15 had to be amended for. A release whose
    /// only subtitle is an image track reads as *satisfied* in that language
    /// and so is never fetched again, while the track itself can never feed a
    /// translator — the file is stuck both ways. The carve-out is narrow: a
    /// fetch made to obtain a source, in a language the file already
    /// satisfies, which settles no want of its own.
    async fn translation_source(
        &self,
        database: &Db,
        target: &Target,
-    ) -> Result<Option<(String, Language)>, SubtitleError> {
+        wanted: &Language,
        settings: &Settings,
    ) -> Result<SourceOutcome, SubtitleError> {
        let files = db::files_for(database.pool(), target.media_file_id).await?;
        let sidecar = files
            .iter()
@@ -922,7 +1034,11 @@ impl Worker {
            .min_by_key(|file| matches!(file.origin, SubtitleOrigin::Translated));
        if let Some(file) = sidecar {
            if let Some(path) = file.path.clone() {
-                return Ok(Some((path, arr_db::policy::language(&file.language))));
+                return Ok(SourceOutcome::Found(Source {
                    path,
                    language: arr_db::policy::language(&file.language),
                    skeleton_aligned: file.skeleton_aligned,
                }));
            }
        }
@@ -934,42 +1050,209 @@ impl Worker {
        .await?
        .flatten();
        let Some(probed) = probed else {
-            return Ok(None);
+            return Ok(SourceOutcome::None);
        };
-        let Some(track) = embedded_tracks(&probed)
+        let tracks = embedded_tracks(&probed);
-            .into_iter()
+
-            .find(|track| !track.forced && track.is_text())
+        if let Some(track) = tracks.iter().find(|track| !track.forced && track.is_text()) {
-        else {
+            let language = arr_db::policy::language(&track.language);
-            return Ok(None);
+            let Some(destination) = target.sidecar(&language, false) else {
                return Ok(SourceOutcome::None);
            };
            if let Err(error) = self
                .extractor
                .extract_srt(&target.path, track.index, &destination)
                .await
            {
                tracing::warn!(
                    path = %target.path.display(),
                    stream = track.index,
                    %error,
                    "embedded track extraction failed"
                );
                return Ok(SourceOutcome::None);
            }
            let mut record = arr_db::NewSubtitleFile::extracted(
                target.media_file_id,
                &track.language,
                &destination.to_string_lossy(),
            );
            if track.sdh {
                record = record.sdh();
            }
            db::record_file(database.pool(), &record).await?;
            return Ok(SourceOutcome::Found(Source {
                path: destination.to_string_lossy().into_owned(),
                language,
                skeleton_aligned: false,
            }));
        }
        self.fetch_source(database, target, wanted, &tracks, settings)
            .await
    }
    /// Every candidate the enabled providers offer in `languages`.
    ///
    /// Best-effort, unlike the search a gap opens with: this one runs after
    /// that one has already reported, so a provider that errors here is
    /// logged and skipped rather than backing the gap off a second time.
    async fn offered_in(
        &self,
        target: &Target,
        languages: &[Language],
        settings: &Settings,
    ) -> Vec<Candidate> {
        let request = target.search_request(languages.to_vec());
        let mut offered: Vec<Candidate> = Vec::new();
        for provider in self
            .providers
            .iter()
            .filter(|provider| settings.providers_enabled.contains(provider.id().as_str()))
        {
            match provider.search(&request).await {
                Ok(candidates) => offered.extend(
                    candidates
                        .into_iter()
                        .filter(|candidate| languages.contains(&candidate.language)),
                ),
                Err(error) => tracing::debug!(
                    %error,
                    provider = %provider.id(),
                    "no translation source from this provider"
                ),
            }
        }
        offered
    }
    /// Download a text subtitle to translate from, in a language an image
    /// track already satisfies (§15 as amended, #268).
    ///
    /// The image track decides the language: a skeleton derived from it is
    /// the alignment reference, so the source has to be the subtitle that
    /// track is a bitmap rendering of. Where no skeleton was derived — the
    /// packets did not pair, or the file predates #268 — the fetch still
    /// happens and `alass` aligns against the video as it always has.
    async fn fetch_source(
        &self,
        database: &Db,
        target: &Target,
        wanted: &Language,
        tracks: &[EmbeddedTrack],
        settings: &Settings,
    ) -> Result<SourceOutcome, SubtitleError> {
        let skeletons =
            arr_db::skeletons::for_media_file(database.pool(), target.media_file_id).await?;
        // A forced track covers signs alone, and one whose language already
        // answers the want is the wrong direction to translate in.
        let wanted_tag = wanted.to_string();
        let Some(track) = tracks.iter().find(|track| {
            !track.forced && track.is_image() && !satisfies(&wanted_tag, &track.language)
        }) else {
            return Ok(SourceOutcome::None);
        };
        let language = arr_db::policy::language(&track.language);
        let languages = vec![language.clone()];
        let Some(destination) = target.sidecar(&language, false) else {
-            return Ok(None);
+            return Ok(SourceOutcome::None);
        };
-        if let Err(error) = self
+
-            .extractor
+        let offered = self.offered_in(target, &languages, settings).await;
-            .extract_srt(&target.path, track.index, &destination)
+        let hash = moviehash(&target.path, target.size).await;
-            .await
+        let (winner, budget_capped) = self
-        {
+            .claim_within_budget(
-            tracing::warn!(
+                database,
-                path = %target.path.display(),
+                target,
-                stream = track.index,
+                hash.as_deref(),
-                %error,
+                offered,
-                "embedded track extraction failed"
+                &languages,
-            );
+                settings,
-            return Ok(None);
+            )
            .await?;
        let Some(winner) = winner else {
            return Ok(if budget_capped {
                SourceOutcome::Capped
            } else {
                SourceOutcome::None
            });
        };
        let provider_name = winner.provider.to_string();
        let Some(provider) = self.provider(&provider_name) else {
            return Ok(SourceOutcome::Failed(format!(
                "provider {provider_name} vanished mid-close"
            )));
        };
        let fetched = match provider.download(&winner.id).await {
            Ok(fetched) => fetched,
            Err(error) => return Ok(SourceOutcome::Failed(error.to_string())),
        };
        let text = match fetched.to_srt() {
            Ok(text) => text,
            Err(error) => return Ok(SourceOutcome::Failed(error.to_string())),
        };
        if let Err(error) = write_sidecar(&destination, &text).await {
            return Ok(SourceOutcome::Failed(error));
        }
-        let mut record = arr_db::NewSubtitleFile::extracted(
+
        let skeleton = skeletons
            .iter()
            .find(|skeleton| usize::try_from(skeleton.stream_index) == Ok(track.index));
        let sync = match skeleton {
            Some(skeleton) => {
                let spans: Vec<(std::time::Duration, std::time::Duration)> = skeleton
                    .cues
                    .iter()
                    .map(|(start, end)| (duration_from_millis(*start), duration_from_millis(*end)))
                    .collect();
                self.syncer.settle_against_cues(&spans, &destination).await
            }
            None => self.syncer.settle(&target.path, &destination).await,
        };
        if let Some(synced) = &sync.content {
            if let Err(error) = write_sidecar(&destination, synced).await {
                return Ok(SourceOutcome::Failed(error));
            }
        }
        // Only an accepted alignment against the skeleton leaves the file on
        // disc-exact timings. A rejected or unrun one leaves the provider's
        // own, which the post-translation pass should still get a look at.
        let skeleton_aligned =
            skeleton.is_some() && matches!(sync.state, arr_subs::SyncState::Synced);
        let mut record = arr_db::NewSubtitleFile::fetched(
            target.media_file_id,
-            &track.language,
+            &winner.language.to_string(),
            &provider_name,
            winner.id.as_str(),
            &destination.to_string_lossy(),
-        );
+        )
-        if track.sdh {
+        .sync(db_sync_state(sync.state));
        if winner.sdh {
            record = record.sdh();
        }
        if skeleton_aligned {
            record = record.skeleton_aligned();
        }
        db::record_file(database.pool(), &record).await?;
-        Ok(Some((destination.to_string_lossy().into_owned(), language)))
+        self.refresh_jellyfin().await;
        tracing::info!(
            media_file_id = target.media_file_id,
            provider = %provider_name,
            language = %winner.language,
            skeleton_aligned,
            "translation source fetched for a language an image track satisfies"
        );
        Ok(SourceOutcome::Found(Source {
            path: destination.to_string_lossy().into_owned(),
            language: winner.language,
            skeleton_aligned,
        }))
    }
    /// The same single rescan import makes (§7.5): Jellyfin's watcher misses
@@ -1439,10 +1722,228 @@ mod tests {
        .inline()
    }
    fn action_with_syncer(
        providers: Vec<Arc<dyn Provider>>,
        backends: Vec<Arc<dyn Backend>>,
        syncer: Syncer,
    ) -> SubtitleAction {
        SubtitleAction::new(
            providers,
            backends,
            syncer,
            Extractor::new().with_binary("ffmpeg-not-installed"),
            JellyfinClient::new("http://127.0.0.1:1", None).unwrap(),
        )
        .inline()
    }
    fn no_tracks() -> serde_json::Value {
        serde_json::json!({ "sub_tracks": [] })
    }
    /// One non-forced PGS track, the shape #268 was found on: it satisfies
    /// English for viewing and can never feed a translator.
    fn one_image_track() -> serde_json::Value {
        serde_json::json!({ "sub_tracks": [
            { "language": "en", "codec": "hdmv_pgs_subtitle", "forced": false, "sdh": false },
        ] })
    }
    /// A fake `alass` that copies its input to its output — every run
    /// accepted — and appends a line to `log` so the runs can be counted.
    async fn counting_alass(directory: &std::path::Path, log: &std::path::Path) -> PathBuf {
        use std::os::unix::fs::PermissionsExt;
        let binary = directory.join("alass");
        tokio::fs::write(
            &binary,
            format!(
                "#!/bin/sh\necho run >> {}\ncp \"$2\" \"$3\"\n",
                log.display()
            ),
        )
        .await
        .unwrap();
        tokio::fs::set_permissions(&binary, std::fs::Permissions::from_mode(0o755))
            .await
            .unwrap();
        binary
    }
    async fn alass_runs(log: &std::path::Path) -> usize {
        tokio::fs::read_to_string(log)
            .await
            .map_or(0, |text| text.lines().count())
    }
    /// #268's deadlock. The release carries one English PGS track, so English
    /// reads as satisfied and is never fetched, while the track itself is
    /// bitmaps and can never be translated from. §15 as amended lets the
    /// translation lane fetch a text English subtitle anyway.
    #[tokio::test]
    async fn an_image_track_does_not_block_the_language_it_satisfies_from_being_fetched() {
        let fixture = Fixture::new(&one_image_track()).await;
        fixture.configure(r#"["pt-PT"]"#, Some("openai")).await;
        let provider = StubProvider::new(
            "opensubtitles",
            vec![candidate(
                "opensubtitles",
                "en-1",
                Language::Other("en".to_owned()),
            )],
        );
        let action = action(vec![Arc::new(provider)], vec![Arc::new(StubBackend)]);
        action.run(&fixture.database).await.unwrap();
        let translated = fixture
            .directory
            .path()
            .join("Movie (2024) - [1080p].pt-PT.mt.srt");
        assert!(
            tokio::fs::read_to_string(&translated)
                .await
                .unwrap()
                .contains("HELLO THERE"),
            "the gap closed by translating the fetched English source"
        );
        let files = fixture.files().await;
        let source = files
            .iter()
            .find(|file| file.origin == SubtitleOrigin::Provider)
            .expect("the English source was fetched despite the image track satisfying English");
        assert_eq!(source.language, "en");
        assert_eq!(
            fixture.attempt("pt-PT").await.unwrap().state,
            SubtitleState::Satisfied
        );
    }
    /// The point of a skeleton: the fetched source is aligned against the
    /// disc's own cue structure, and the translation made from it is then
    /// left alone. A second `alass` pass can only move disc-exact timings
    /// off.
    #[tokio::test]
    async fn a_source_aligned_to_a_skeleton_skips_the_post_translation_pass() {
        let fixture = Fixture::new(&one_image_track()).await;
        fixture.configure(r#"["pt-PT"]"#, Some("openai")).await;
        arr_db::skeletons::record(
            fixture.database.pool(),
            fixture.media_file_id,
            0,
            "en",
            &[(1_000, 2_500), (3_000, 4_000)],
        )
        .await
        .unwrap();
        let log = fixture.directory.path().join("alass.log");
        let binary = counting_alass(fixture.directory.path(), &log).await;
        let action = action_with_syncer(
            vec![Arc::new(StubProvider::new(
                "opensubtitles",
                vec![candidate(
                    "opensubtitles",
                    "en-1",
                    Language::Other("en".to_owned()),
                )],
            ))],
            vec![Arc::new(StubBackend)],
            Syncer::new().with_binary(&binary),
        );
        action.run(&fixture.database).await.unwrap();
        assert_eq!(
            alass_runs(&log).await,
            1,
            "alass aligned the source against the skeleton and was not run again"
        );
        let files = fixture.files().await;
        assert!(
            files
                .iter()
                .find(|file| file.origin == SubtitleOrigin::Provider)
                .unwrap()
                .skeleton_aligned,
            "the fetched source is recorded as disc-exact"
        );
        assert!(
            files
                .iter()
                .find(|file| file.origin == SubtitleOrigin::Translated)
                .unwrap()
                .skeleton_aligned,
            "and so is what was translated from it"
        );
    }
    /// Without a skeleton — the packets did not pair, or the file predates
    /// #268 — the fetch still happens and both passes align against the
    /// video, exactly as they always did.
    #[tokio::test]
    async fn without_a_skeleton_both_passes_still_align_against_the_video() {
        let fixture = Fixture::new(&one_image_track()).await;
        fixture.configure(r#"["pt-PT"]"#, Some("openai")).await;
        let log = fixture.directory.path().join("alass.log");
        let binary = counting_alass(fixture.directory.path(), &log).await;
        let action = action_with_syncer(
            vec![Arc::new(StubProvider::new(
                "opensubtitles",
                vec![candidate(
                    "opensubtitles",
                    "en-1",
                    Language::Other("en".to_owned()),
                )],
            ))],
            vec![Arc::new(StubBackend)],
            Syncer::new().with_binary(&binary),
        );
        action.run(&fixture.database).await.unwrap();
        assert_eq!(
            alass_runs(&log).await,
            2,
            "source and translation both synced"
        );
        assert!(fixture
            .files()
            .await
            .iter()
            .all(|file| !file.skeleton_aligned));
    }
    /// A forced image track covers signs alone (§15), so it is not a reason
    /// to go looking for a source subtitle either.
    #[tokio::test]
    async fn a_forced_image_track_is_not_a_translation_source_to_fetch() {
        let fixture = Fixture::new(&serde_json::json!({ "sub_tracks": [
            { "language": "en", "codec": "hdmv_pgs_subtitle", "forced": true, "sdh": false },
        ] }))
        .await;
        fixture.configure(r#"["pt-PT"]"#, Some("openai")).await;
        let action = action(
            vec![Arc::new(StubProvider::new(
                "opensubtitles",
                vec![candidate(
                    "opensubtitles",
                    "en-1",
                    Language::Other("en".to_owned()),
                )],
            ))],
            vec![Arc::new(StubBackend)],
        );
        action.run(&fixture.database).await.unwrap();
        let attempt = fixture.attempt("pt-PT").await.unwrap();
        assert_eq!(attempt.state, SubtitleState::Unavailable);
        assert!(fixture.files().await.iter().all(|file| file.path.is_none()));
    }
    #[tokio::test]
    async fn an_embedded_track_satisfies_without_asking_any_provider() {
        let fixture = Fixture::new(&serde_json::json!({ "sub_tracks": [
@@ -13,6 +13,7 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 sqlx = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
 [dev-dependencies]
 tempfile = { workspace = true }
@@ -0,0 +1,34 @@
 -- #268. The cue structure of an image-format subtitle track (DESIGN.md §15).
 --
 -- A PGS or VobSub track carries bitmaps, so it can never feed a translator.
 -- Its packet timings are still exact for the release it came off, and that is
 -- what `alass` matches on: aligning a downloaded English subtitle against this
 -- skeleton lands it on the disc's own cue structure instead of on a heuristic
 -- read of the video's audio.
 --
 -- Derived once at import, where the file is being read and hardlinked anyway.
 -- A track whose packets do not pair cleanly gets no row at all — the absence
 -- is the fallback signal, and the lane aligns against the video instead.
 CREATE TABLE subtitle_skeletons (
    media_file_id INTEGER NOT NULL REFERENCES media_files (id) ON DELETE CASCADE,
    -- The track's position among the file's subtitle streams, the same index
    -- `ffmpeg -map 0:s:N` takes.
    stream_index  INTEGER NOT NULL CHECK (stream_index >= 0),
    -- As `arr_core::Language` spells it. The language a source subtitle is
    -- fetched in when this skeleton is the alignment reference.
    language      TEXT NOT NULL,
    -- The cues, as a JSON array of `[start_ms, end_ms]` pairs in file order.
    -- No text: there is none to read, and `alass` does not want any.
    cues          TEXT NOT NULL,
    created_at    TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    PRIMARY KEY (media_file_id, stream_index)
 ) STRICT;
 -- §15, as amended by #268: a subtitle aligned against a cue skeleton already
 -- carries disc-exact timings, so the post-translation `alass` pass is skipped
 -- for it and for anything translated from it. A second pass can only move
 -- those timings off. This says which rows that applies to, durably — a crash
 -- between the fetch and the translation must not lose the fact.
 ALTER TABLE subtitle_files
    ADD COLUMN skeleton_aligned INTEGER NOT NULL DEFAULT 0
    CHECK (skeleton_aligned IN (0, 1));
@@ -7,11 +7,13 @@ use std::path::Path;
 pub mod blacklist;
 pub mod policy;
 pub mod skeletons;
 pub mod subtitle_budget;
 pub mod subtitles;
 pub use blacklist::Blacklist;
 pub use policy::{MoviePolicy, PolicyColumns, PolicyError, TitlePolicy};
 pub use skeletons::Skeleton;
 pub use subtitle_budget::BudgetKind;
 pub use subtitles::{
    NewSubtitleFile, PendingSubtitle, SubtitleAttempt, SubtitleFile, SubtitleOrigin, SubtitleState,
@@ -0,0 +1,104 @@
 //! Cue skeletons of image-format subtitle tracks (`DESIGN.md` §15, #268).
 //!
 //! A skeleton is the cue structure of a PGS or `VobSub` track, derived from
 //! its packet timings at import. It carries no text — there is none to read —
 //! and exists for one purpose: to be `alass`'s alignment reference instead of
 //! the video, so a downloaded subtitle lands on the disc's own timings.
 //!
 //! A row means the track's packets paired cleanly. Its absence means they did
 //! not, or that nothing has looked: both cases fall back to aligning against
 //! the video, so the reader never has to tell them apart.
 use sqlx::SqlitePool;
 /// The cue structure of one image-format track.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Skeleton {
    /// The track's position among the file's subtitle streams.
    pub stream_index: i64,
    /// As `arr_core::Language` spells it: the language a source subtitle is
    /// worth fetching in when this skeleton is the reference.
    pub language: String,
    /// `(start, end)` in milliseconds, in file order.
    pub cues: Vec<(i64, i64)>,
 }
 /// Store the skeleton of one track, replacing whatever a previous import
 /// derived for it.
 ///
 /// A re-import of the same file re-derives from the same packets, so the
 /// replace is a convergence rather than an upgrade.
 ///
 /// # Errors
 ///
 /// If the insert fails, or the cues cannot be serialised.
 pub async fn record(
    pool: &SqlitePool,
    media_file_id: i64,
    stream_index: i64,
    language: &str,
    cues: &[(i64, i64)],
 ) -> Result<(), sqlx::Error> {
    let encoded = serde_json::to_string(cues).map_err(|error| sqlx::Error::Encode(error.into()))?;
    sqlx::query!(
        "INSERT INTO subtitle_skeletons (media_file_id, stream_index, language, cues)
         VALUES (?, ?, ?, ?)
         ON CONFLICT (media_file_id, stream_index) DO UPDATE SET
             language = excluded.language,
             cues     = excluded.cues",
        media_file_id,
        stream_index,
        language,
        encoded
    )
    .execute(pool)
    .await?;
    Ok(())
 }
 /// Every skeleton known for one media file, by stream index.
 ///
 /// A row whose cues do not decode is dropped rather than returned: the caller
 /// treats a missing skeleton as "align against the video", which is the right
 /// answer for an unreadable one too.
 ///
 /// # Errors
 ///
 /// If the query fails.
 pub async fn for_media_file(
    pool: &SqlitePool,
    media_file_id: i64,
 ) -> Result<Vec<Skeleton>, sqlx::Error> {
    let rows = sqlx::query!(
        r#"SELECT stream_index AS "stream_index!: i64",
                  language     AS "language!: String",
                  cues         AS "cues!: String"
           FROM   subtitle_skeletons
           WHERE  media_file_id = ?
           ORDER BY stream_index"#,
        media_file_id
    )
    .fetch_all(pool)
    .await?;
    Ok(rows
        .into_iter()
        .filter_map(|row| {
            let cues: Vec<(i64, i64)> = serde_json::from_str(&row.cues)
                .inspect_err(|error| {
                    tracing::warn!(
                        media_file_id,
                        stream_index = row.stream_index,
                        %error,
                        "subtitle skeleton does not decode, ignoring it"
                    );
                })
                .ok()?;
            Some(Skeleton {
                stream_index: row.stream_index,
                language: row.language,
                cues,
            })
        })
        .collect())
 }
@@ -101,6 +101,10 @@ pub struct SubtitleFile {
    pub forced: bool,
    pub sdh: bool,
    pub sync: SubtitleSync,
    /// Aligned against a cue skeleton rather than the video (§15, #268), so
    /// its timings are exact for this release and a second `alass` pass —
    /// here or on anything translated from it — can only move them off.
    pub skeleton_aligned: bool,
    /// `None` only for [`SubtitleOrigin::Embedded`].
    pub path: Option<String>,
 }
@@ -122,6 +126,7 @@ pub struct NewSubtitleFile {
    forced: bool,
    sdh: bool,
    sync: SubtitleSync,
    skeleton_aligned: bool,
    path: Option<String>,
 }
@@ -137,6 +142,7 @@ impl NewSubtitleFile {
            forced: false,
            sdh: false,
            sync: SubtitleSync::NotRun,
            skeleton_aligned: false,
            path: None,
        }
    }
@@ -206,6 +212,15 @@ impl NewSubtitleFile {
        self.sync = sync;
        self
    }
    /// Aligned against a cue skeleton rather than the video (§15, #268).
    /// Timings are disc-exact from here on, so the post-translation `alass`
    /// pass is skipped for this file and for what is translated from it.
    #[must_use]
    pub fn skeleton_aligned(mut self) -> Self {
        self.skeleton_aligned = true;
        self
    }
 }
 /// Record a subtitle, returning its row id.
@@ -235,8 +250,8 @@ pub async fn record_file(
    let inserted = sqlx::query_scalar!(
        r#"INSERT INTO subtitle_files
               (media_file_id, language, origin, provider, candidate_id, engine,
-                forced, sdh, synced, sync_rejected, path)
+                forced, sdh, synced, sync_rejected, skeleton_aligned, path)
-           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
           ON CONFLICT (path) DO NOTHING
           ON CONFLICT (media_file_id, language, forced, sdh)
               WHERE origin = 'embedded' DO NOTHING
@@ -251,6 +266,7 @@ pub async fn record_file(
        subtitle.sdh,
        synced,
        sync_rejected,
        subtitle.skeleton_aligned,
        subtitle.path,
    )
    .fetch_optional(pool)
@@ -306,6 +322,7 @@ pub async fn files_for(
                  sdh           AS "sdh!: bool",
                  synced        AS "synced!: bool",
                  sync_rejected AS "sync_rejected!: bool",
                  skeleton_aligned AS "skeleton_aligned!: bool",
                  path
           FROM   subtitle_files
           WHERE  media_file_id = ?
@@ -328,6 +345,7 @@ pub async fn files_for(
            forced: row.forced,
            sdh: row.sdh,
            sync: SubtitleSync::from_columns(row.synced, row.sync_rejected),
            skeleton_aligned: row.skeleton_aligned,
            path: row.path,
        })
        .collect())
@@ -27,6 +27,7 @@ mod extract;
 mod ffprobe;
 mod language;
 mod model;
 mod skeleton;
 // `unused_crate_dependencies` is a per-target lint and the library's own test
 // target links the dev-dependencies without using them. The real uses are in
@@ -37,6 +38,10 @@ use tempfile as _;
 pub use error::{Error, Result};
 pub use extract::{Extractor, DEFAULT_EXTRACTION_TIMEOUT};
 pub use model::{FeatureSelection, ProbedFile, RuntimeMatch};
 pub use skeleton::{
    Implausible, Outcome as SkeletonOutcome, Skeleton, SkeletonCue, Skeletons,
    DEFAULT_SKELETON_TIMEOUT,
 };
 /// The binary invoked when nothing else is configured.
 pub const DEFAULT_BINARY: &str = "ffprobe";
@@ -0,0 +1,448 @@
 //! Deriving a cue skeleton from an image-format subtitle track (§15, #268).
 //!
 //! A PGS or `VobSub` track carries bitmaps, so nothing here reads a pixel.
 //! What it reads is packet metadata: `ffprobe -show_packets` returns the
 //! presentation timestamps of a subtitle track directly, and on a retail disc
 //! those alternate show, clear, show, clear. Pairing them yields the cue
 //! structure of the release — the disc's own timings, exact for that file.
 //!
 //! That structure is what `alass` matches on. It compares interval shapes and
 //! not words, so a skeleton with no text in it is still a strong alignment
 //! reference for a downloaded subtitle whose cues never line up one-for-one
 //! with the disc's.
 //!
 //! The pairing is load-bearing. Treating every packet as a cue start instead
 //! of pairing them puts the whole reference out by roughly half a cue —
 //! better than no alignment, and visibly wrong. So a track that does not pair
 //! cleanly is rejected here rather than handed on: PGS permits several
 //! composition segments per subtitle, and the caller must fall back to
 //! aligning against the video rather than trust a skeleton that is quietly
 //! half a second out.
 use std::{ffi::OsString, path::Path, process::Stdio, time::Duration};
 use serde::Deserialize;
 use tokio::process::Command;
 use crate::error::{Error, Result};
 /// The binary invoked when nothing else is configured.
 pub const DEFAULT_BINARY: &str = "ffprobe";
 /// How long one derivation may take. Generous: reading every subtitle packet
 /// out of a 60 GB remux is a full demux of the container.
 pub const DEFAULT_SKELETON_TIMEOUT: Duration = Duration::from_secs(300);
 /// The shortest a cue may last and still be a cue.
 pub const MIN_CUE: Duration = Duration::from_millis(300);
 /// The longest a cue may last and still be a cue. A pairing that has slipped
 /// spans the gap between two subtitles, which on any real track is longer
 /// than this.
 pub const MAX_CUE: Duration = Duration::from_secs(10);
 /// The sparsest a real subtitle track gets, in cues per minute of runtime.
 pub const MIN_DENSITY: f64 = 0.2;
 /// The densest a real subtitle track gets, in cues per minute of runtime.
 pub const MAX_DENSITY: f64 = 60.0;
 /// How much of `ffprobe`'s stderr is kept in an error.
 const STDERR_LIMIT: usize = 512;
 /// One cue of a skeleton: when the bitmap appeared, and when it was cleared.
 /// No text — there is none to read, and `alass` does not want any.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct SkeletonCue {
    /// When the subtitle appears.
    pub start: Duration,
    /// When it disappears.
    pub end: Duration,
 }
 impl SkeletonCue {
    /// How long the subtitle is on screen.
    #[must_use]
    pub fn duration(&self) -> Duration {
        self.end.saturating_sub(self.start)
    }
 }
 /// The cue structure of one image-format track.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Skeleton {
    /// The track's position among the file's subtitle streams — the same
    /// index [`crate::Extractor::extract_srt`] maps.
    pub stream_index: usize,
    /// The cues, in file order.
    pub cues: Vec<SkeletonCue>,
 }
 /// Why a track's packets could not be trusted as a cue structure (§15).
 #[derive(Clone, Copy, Debug, PartialEq)]
 pub enum Implausible {
    /// Fewer than two packets: nothing to pair.
    TooFewPackets {
        /// How many there were.
        packets: usize,
    },
    /// An odd number of packets, so at least one show has no clear. The
    /// track does not follow the show/clear alternation pairing assumes.
    OddPacketCount {
        /// How many there were.
        packets: usize,
    },
    /// A paired cue lasts an implausible length of time, which means the
    /// pairing slipped somewhere at or before it.
    CueDuration {
        /// The cue's position in the track, 0-based.
        index: usize,
        /// How long it came out.
        duration: Duration,
    },
    /// The cue count does not fit the runtime — too sparse to be a subtitle
    /// track, or too dense to be one cue per subtitle.
    Density {
        /// How many cues were paired.
        cues: usize,
        /// Per minute of runtime.
        per_minute: f64,
    },
 }
 impl std::fmt::Display for Implausible {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::TooFewPackets { packets } => {
                write!(formatter, "{packets} packets is too few to pair")
            }
            Self::OddPacketCount { packets } => {
                write!(formatter, "{packets} packets do not pair show to clear")
            }
            Self::CueDuration { index, duration } => {
                write!(formatter, "cue {index} lasts {duration:.1?}")
            }
            Self::Density { cues, per_minute } => {
                write!(formatter, "{cues} cues is {per_minute:.1} per minute")
            }
        }
    }
 }
 /// What one derivation decided.
 #[derive(Clone, Debug, PartialEq)]
 pub enum Outcome {
    /// The packets paired into a trustworthy cue structure.
    Derived(Skeleton),
    /// They did not. The caller aligns against the video instead.
    Implausible(Implausible),
 }
 /// Reads the packet timings of image-format subtitle tracks.
 ///
 /// The same shape as [`crate::Prober`] and [`crate::Extractor`]: one small
 /// binary, invoked and discarded, with a configurable path and its own
 /// timeout. It is a separate handle rather than a method on `Prober` because
 /// the timeout is a different order of magnitude — a metadata read against a
 /// full demux.
 #[derive(Clone, Debug)]
 pub struct Skeletons {
    binary: OsString,
    timeout: Duration,
 }
 impl Default for Skeletons {
    fn default() -> Self {
        Self {
            binary: DEFAULT_BINARY.into(),
            timeout: DEFAULT_SKELETON_TIMEOUT,
        }
    }
 }
 impl Skeletons {
    /// A reader that runs `ffprobe` from `PATH`.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }
    /// Run a specific binary instead of whatever `PATH` resolves to.
    #[must_use]
    pub fn with_binary(mut self, binary: impl Into<OsString>) -> Self {
        self.binary = binary.into();
        self
    }
    /// Change how long one derivation may take.
    #[must_use]
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }
    /// Derive the cue skeleton of one subtitle track.
    ///
    /// `stream_index` is the track's position among the file's subtitle
    /// streams, not the container's absolute stream id — the same index
    /// [`crate::Extractor::extract_srt`] takes. `runtime` is the file's
    /// playing time, which the density check needs; without it that check is
    /// skipped and the rest still applies.
    ///
    /// # Errors
    ///
    /// [`Error::Spawn`] when `ffprobe` is not installed, [`Error::Timeout`],
    /// [`Error::Rejected`] when it exits non-zero — which includes selecting
    /// a stream that does not exist — and [`Error::Decode`] on output that is
    /// not the JSON this asked for.
    pub async fn derive(
        &self,
        video: &Path,
        stream_index: usize,
        runtime: Option<Duration>,
    ) -> Result<Outcome> {
        let output = self.run(video, stream_index).await?;
        let document: PacketList =
            serde_json::from_slice(&output).map_err(|source| Error::Decode {
                path: video.to_path_buf(),
                source,
            })?;
        let stamps: Vec<Duration> = document
            .packets
            .iter()
            .filter_map(Packet::timestamp)
            .collect();
        Ok(pair(stream_index, &stamps, runtime))
    }
    /// Spawn `ffprobe` and collect its stdout.
    ///
    /// Only `pts_time` is asked for: a feature-length track is a few thousand
    /// packets, and the rest of what `-show_packets` prints per packet would
    /// be megabytes of JSON for facts nothing here reads.
    async fn run(&self, video: &Path, stream_index: usize) -> Result<Vec<u8>> {
        let select = format!("s:{stream_index}");
        let mut command = Command::new(&self.binary);
        command
            .args([
                "-v",
                "error",
                "-hide_banner",
                "-print_format",
                "json",
                "-select_streams",
                &select,
                "-show_entries",
                "packet=pts_time",
                "-i",
            ])
            .arg(video)
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        let child = command.spawn().map_err(|source| Error::Spawn {
            binary: self.binary.to_string_lossy().into_owned(),
            source,
        })?;
        let output = match tokio::time::timeout(self.timeout, child.wait_with_output()).await {
            Ok(output) => output.map_err(|source| Error::Io {
                path: video.to_path_buf(),
                source,
            })?,
            Err(_) => {
                return Err(Error::Timeout {
                    path: video.to_path_buf(),
                })
            }
        };
        if output.status.success() {
            return Ok(output.stdout);
        }
        let stderr = String::from_utf8_lossy(&output.stderr)
            .trim()
            .chars()
            .take(STDERR_LIMIT)
            .collect();
        Err(Error::Rejected {
            path: video.to_path_buf(),
            status: output.status.code(),
            stderr,
        })
    }
 }
 /// Pair packet timestamps into cues, and judge the result.
 ///
 /// Split out from the spawn so the rule is testable without `ffprobe`: this
 /// is the whole of what makes a skeleton trustworthy.
 fn pair(stream_index: usize, stamps: &[Duration], runtime: Option<Duration>) -> Outcome {
    if stamps.len() < 2 {
        return Outcome::Implausible(Implausible::TooFewPackets {
            packets: stamps.len(),
        });
    }
    if !stamps.len().is_multiple_of(2) {
        return Outcome::Implausible(Implausible::OddPacketCount {
            packets: stamps.len(),
        });
    }
    let mut cues = Vec::with_capacity(stamps.len() / 2);
    for (index, [start, end]) in stamps.as_chunks::<2>().0.iter().enumerate() {
        let cue = SkeletonCue {
            start: *start,
            end: *end,
        };
        let duration = cue.duration();
        if duration < MIN_CUE || duration > MAX_CUE {
            return Outcome::Implausible(Implausible::CueDuration { index, duration });
        }
        cues.push(cue);
    }
    if let Some(runtime) = runtime {
        let minutes = runtime.as_secs_f64() / 60.0;
        if minutes > 0.0 {
            #[allow(clippy::cast_precision_loss)]
            let per_minute = cues.len() as f64 / minutes;
            if per_minute < MIN_DENSITY || per_minute > MAX_DENSITY {
                return Outcome::Implausible(Implausible::Density {
                    cues: cues.len(),
                    per_minute,
                });
            }
        }
    }
    Outcome::Derived(Skeleton { stream_index, cues })
 }
 /// The slice of `ffprobe -show_entries packet=pts_time` this reads.
 #[derive(Debug, Deserialize)]
 struct PacketList {
    #[serde(default)]
    packets: Vec<Packet>,
 }
 #[derive(Debug, Deserialize)]
 struct Packet {
    /// `"12.345000"`, or absent — `ffprobe` omits a field rather than
    /// nulling it, and prints the string `"N/A"` where it has no value.
    pts_time: Option<String>,
 }
 impl Packet {
    fn timestamp(&self) -> Option<Duration> {
        let seconds: f64 = self.pts_time.as_deref()?.trim().parse().ok()?;
        if seconds.is_finite() && seconds >= 0.0 {
            Duration::try_from_secs_f64(seconds).ok()
        } else {
            None
        }
    }
 }
 #[cfg(test)]
 mod tests {
    use std::time::Duration;
    use super::{pair, Implausible, Outcome, SkeletonCue};
    fn secs(value: f64) -> Duration {
        Duration::from_secs_f64(value)
    }
    /// The real shape: packets strictly alternate show and clear, so every
    /// other one closes the cue the one before it opened.
    #[test]
    fn packets_pair_show_to_clear() {
        let stamps = [secs(1.0), secs(3.0), secs(5.0), secs(7.5)];
        let Outcome::Derived(skeleton) = pair(0, &stamps, Some(Duration::from_secs(600))) else {
            panic!("the pairing must be derived");
        };
        assert_eq!(
            skeleton.cues,
            vec![
                SkeletonCue {
                    start: secs(1.0),
                    end: secs(3.0)
                },
                SkeletonCue {
                    start: secs(5.0),
                    end: secs(7.5)
                },
            ]
        );
    }
    /// #268's measurement: reading every packet as a cue start is out by
    /// roughly half a cue over the whole film. An odd count is the visible
    /// symptom of a track that does not alternate, and it is refused rather
    /// than paired anyway.
    #[test]
    fn an_odd_packet_count_is_refused() {
        let stamps = [secs(1.0), secs(3.0), secs(5.0)];
        assert_eq!(
            pair(0, &stamps, None),
            Outcome::Implausible(Implausible::OddPacketCount { packets: 3 })
        );
    }
    #[test]
    fn fewer_than_two_packets_pair_into_nothing() {
        assert_eq!(
            pair(0, &[secs(1.0)], None),
            Outcome::Implausible(Implausible::TooFewPackets { packets: 1 })
        );
    }
    /// PGS permits several composition segments per subtitle. A track built
    /// that way still has an even packet count, and pairing it spans the gap
    /// between two subtitles — which is what the duration bound catches.
    #[test]
    fn a_slipped_pairing_shows_up_as_an_impossible_duration() {
        let stamps = [secs(1.0), secs(2.0), secs(3.0), secs(40.0)];
        assert_eq!(
            pair(0, &stamps, None),
            Outcome::Implausible(Implausible::CueDuration {
                index: 1,
                duration: secs(37.0)
            })
        );
    }
    #[test]
    fn a_cue_shorter_than_a_glance_is_refused() {
        let stamps = [secs(1.0), secs(1.1)];
        assert!(matches!(
            pair(0, &stamps, None),
            Outcome::Implausible(Implausible::CueDuration { index: 0, .. })
        ));
    }
    /// Two cues over a two-hour film is a signs-only oddity, not the
    /// structure of the film's dialogue.
    #[test]
    fn a_track_too_sparse_for_the_runtime_is_refused() {
        let stamps = [secs(1.0), secs(3.0), secs(5.0), secs(7.0)];
        assert!(matches!(
            pair(0, &stamps, Some(Duration::from_secs(7200))),
            Outcome::Implausible(Implausible::Density { cues: 2, .. })
        ));
    }
    /// Without a runtime there is nothing to judge density against, and the
    /// rest of the checks still stand.
    #[test]
    fn density_is_skipped_when_the_runtime_is_unknown() {
        let stamps = [secs(1.0), secs(3.0)];
        assert!(matches!(pair(0, &stamps, None), Outcome::Derived(_)));
    }
 }
@@ -11,8 +11,9 @@
 //! went missing between input and output, make the whole result implausible —
 //! and an implausible result is not an error but a [`Outcome::Rejected`], so
 //! the caller keeps the unsynced original and flags the row (#186). The
-//! reference is always the media file itself; subtitle and video are both on
+//! reference is the media file itself, or — for a release whose only text
-//! disk by the time this runs.
+//! source has to be downloaded (#268) — the cue skeleton of one of its image
 //! subtitle tracks, which carries the disc's own timings.
 use std::{ffi::OsStr, ffi::OsString, path::Path, process::Stdio, time::Duration};
@@ -37,6 +38,10 @@ pub const MAX_SHIFT: Duration = Duration::from_secs(60);
 /// How much of `alass`'s stderr is kept in an error.
 const STDERR_LIMIT: usize = 512;
 /// What a cue skeleton's cues say. `alass` reads only their timings, and SRT
 /// has no way to spell a cue with no text at all.
 const SKELETON_TEXT: &str = ".";
 /// Whether a configured external binary resolves to an executable.
 ///
 /// A name with any path component (`/usr/local/bin/alass`, `./alass`) must
@@ -175,6 +180,54 @@ impl Syncer {
        })
    }
    /// Settle `subtitle` against a cue skeleton instead of the video (§15,
    /// #268).
    ///
    /// `reference` is the `(start, end)` spans of an image-format embedded
    /// track, whose timings are exact for the release the file came off.
    /// `alass` matches on interval structure and not on words, so the
    /// placeholder text written here is never read — a sparse skeleton whose
    /// cues do not correspond one-for-one with the subtitle's is still a
    /// strong signal.
    ///
    /// Degrades the same way [`Self::settle`] does: a reference that cannot
    /// be written is a [`SyncState::NotRun`], not a failure of the caller.
    pub async fn settle_against_cues(
        &self,
        reference: &[(Duration, Duration)],
        subtitle: impl AsRef<Path>,
    ) -> Settled {
        let cues: Vec<srt::Cue> = reference
            .iter()
            .map(|(start, end)| srt::Cue {
                start: *start,
                end: *end,
                text: SKELETON_TEXT.to_owned(),
            })
            .collect();
        let directory = match tempfile::tempdir() {
            Ok(directory) => directory,
            Err(error) => {
                tracing::warn!(%error, "no temporary directory for the cue skeleton");
                return Settled {
                    content: None,
                    state: SyncState::NotRun,
                };
            }
        };
        let path = directory.path().join("skeleton.srt");
        if let Err(error) = tokio::fs::write(&path, srt::render(&cues)).await {
            tracing::warn!(%error, "the cue skeleton could not be written");
            return Settled {
                content: None,
                state: SyncState::NotRun,
            };
        }
        self.settle(&path, subtitle).await
    }
    /// Run [`Self::sync`] and settle the result into a form neither caller
    /// has to branch on twice.
    ///