fix(dl): mux all requested ranges and select highest DV alongside hybrid

When -r included HYBRID, the dl pipeline only muxed the hybrid output and dropped every other downloaded range. DV was also treated purely as a hybrid ingredient pool, so an explicitly requested DV range never produced a standalone deliverable - only the lowest DV (the RPU ingredient) was selected.

- Select the best DV per resolution as a standalone deliverable when DV is in -r, while still using the lowest DV as the hybrid ingredient (dv_is_deliverable).
- Flag the ingredient-only DV (hybrid_base_only) so it is downloaded for the hybrid build but skipped during standalone muxing.
- Mux every requested range standalone after hybrid processing; build hybrids from deepcopied ingredients so the originals stay muxable.
- Add Tracks.merge_video_selections to de-dup the ingredient/deliverable overlap so a shared DV track is not downloaded or muxed twice.

tests/tracks/test_hybrid_selection.py

@@ -0,0 +1,171 @@
+"""Track-selection tests for HYBRID + DV behaviour.
+
+Covers the selection primitives that back `-r ...,DV,HYBRID` downloads:
+
+- ``Tracks.select_hybrid`` — picks the HDR base layer per resolution and the
+  single lowest DV track used as a hybrid ingredient.
+- ``Tracks.merge_video_selections`` — de-duplicates the ingredient/deliverable
+  overlap so a DV track that is chosen as both is not muxed/downloaded twice.
+
+The full ``dl`` selection pipeline (range filtering, the ``dv_is_deliverable``
+partition, the ``hybrid_base_only`` ingredient flag and the standalone mux loop)
+is orchestration glue inside the Click command; these tests lock down the pure
+units it relies on plus the documented end-state of a realistic ATV-style ladder.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from unshackle.core.tracks import Tracks, Video
+
+
+def make_video(track_id: str, *, range_: Video.Range, height: int, bitrate: int, codec: Video.Codec) -> Video:
+    return Video(
+        id_=track_id,
+        url=f"https://example.test/{track_id}.m3u8",
+        language="en",
+        codec=codec,
+        range_=range_,
+        width=int(height * 16 / 9),
+        height=height,
+        bitrate=bitrate,
+    )
+
+
+@pytest.fixture
+def ladder() -> list[Video]:
+    """Mirrors the reported ATV ladder: HDR10+, DV and SDR at multiple resolutions."""
+    H = Video.Codec.HEVC
+    A = Video.Codec.AVC
+    return [
+        make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=25_516_000, codec=H),
+        make_video("hdr10p-1080", range_=Video.Range.HDR10P, height=1080, bitrate=9_096_000, codec=H),
+        make_video("dv-2160", range_=Video.Range.DV, height=2160, bitrate=25_511_000, codec=H),
+        make_video("dv-1080", range_=Video.Range.DV, height=1080, bitrate=9_152_000, codec=H),
+        make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_328_000, codec=H),
+        make_video("sdr-2160", range_=Video.Range.SDR, height=2160, bitrate=21_501_000, codec=H),
+        make_video("sdr-1080-avc", range_=Video.Range.SDR, height=1080, bitrate=10_793_000, codec=A),
+        make_video("sdr-1080-hevc", range_=Video.Range.SDR, height=1080, bitrate=5_768_000, codec=H),
+    ]
+
+
+def ids(tracks: list[Video]) -> set[str]:
+    return {t.id for t in tracks}
+
+
+# ---------------------------------------------------------------------------
+# select_hybrid
+# ---------------------------------------------------------------------------
+
+
+def test_select_hybrid_picks_base_per_resolution_and_lowest_dv(ladder: list[Video]) -> None:
+    chosen = list(filter(Tracks().select_hybrid(ladder, [2160, 1080]), ladder))
+    # HDR10+ base at each requested resolution, plus the single lowest DV ingredient.
+    assert ids(chosen) == {"hdr10p-2160", "hdr10p-1080", "dv-360"}
+
+
+def test_select_hybrid_ingredient_is_lowest_dv_regardless_of_quality(ladder: list[Video]) -> None:
+    # Even when only 2160 is requested, the ingredient is the globally lowest DV.
+    chosen = list(filter(Tracks().select_hybrid(ladder, [2160]), ladder))
+    assert ids(chosen) == {"hdr10p-2160", "dv-360"}
+
+
+def test_select_hybrid_prefers_hdr10p_over_hdr10() -> None:
+    H = Video.Codec.HEVC
+    tracks = [
+        make_video("hdr10-2160", range_=Video.Range.HDR10, height=2160, bitrate=20_000_000, codec=H),
+        make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=H),
+        make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_000_000, codec=H),
+    ]
+    chosen = list(filter(Tracks().select_hybrid(tracks, [2160]), tracks))
+    assert ids(chosen) == {"hdr10p-2160", "dv-360"}
+
+
+def test_select_hybrid_base_picks_highest_bitrate_then_worst_flips() -> None:
+    H = Video.Codec.HEVC
+    tracks = [
+        make_video("hdr10p-2160-lo", range_=Video.Range.HDR10P, height=2160, bitrate=18_000_000, codec=H),
+        make_video("hdr10p-2160-hi", range_=Video.Range.HDR10P, height=2160, bitrate=25_000_000, codec=H),
+        make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_000_000, codec=H),
+    ]
+    best = list(filter(Tracks().select_hybrid(tracks, [2160]), tracks))
+    assert ids(best) == {"hdr10p-2160-hi", "dv-360"}
+
+    worst = list(filter(Tracks().select_hybrid(tracks, [2160], worst=True), tracks))
+    assert ids(worst) == {"hdr10p-2160-lo", "dv-360"}
+
+
+def test_select_hybrid_no_dv_selects_only_base() -> None:
+    H = Video.Codec.HEVC
+    tracks = [make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=H)]
+    chosen = list(filter(Tracks().select_hybrid(tracks, [2160]), tracks))
+    assert ids(chosen) == {"hdr10p-2160"}
+
+
+# ---------------------------------------------------------------------------
+# merge_video_selections (the dedup fix)
+# ---------------------------------------------------------------------------
+
+
+def test_merge_dedups_shared_ingredient_and_deliverable() -> None:
+    H = Video.Codec.HEVC
+    dv = make_video("dv-1080", range_=Video.Range.DV, height=1080, bitrate=9_000_000, codec=H)
+    hdr = make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=H)
+    sdr = make_video("sdr-2160", range_=Video.Range.SDR, height=2160, bitrate=21_000_000, codec=H)
+
+    # dv is both the hybrid ingredient and an explicit DV deliverable (same object).
+    merged = Tracks.merge_video_selections([hdr, dv], [sdr, dv])
+    assert [t.id for t in merged] == ["hdr10p-2160", "dv-1080", "sdr-2160"]
+
+
+def test_merge_preserves_order_and_keeps_distinct_dv() -> None:
+    H = Video.Codec.HEVC
+    ingredient = make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_000_000, codec=H)
+    deliverable = make_video("dv-2160", range_=Video.Range.DV, height=2160, bitrate=25_000_000, codec=H)
+
+    merged = Tracks.merge_video_selections([ingredient], [deliverable])
+    assert [t.id for t in merged] == ["dv-360", "dv-2160"]
+
+
+def test_merge_dedup_uses_track_identity_by_id() -> None:
+    # Tracks compare equal by id; merge must treat same-id tracks as one.
+    H = Video.Codec.HEVC
+    a = make_video("same", range_=Video.Range.DV, height=1080, bitrate=9_000_000, codec=H)
+    b = make_video("same", range_=Video.Range.DV, height=1080, bitrate=9_000_000, codec=H)
+    assert a == b
+    assert len(Tracks.merge_video_selections([a], [b])) == 1
+
+
+# ---------------------------------------------------------------------------
+# documented end-state for the reported command
+# ---------------------------------------------------------------------------
+
+
+def test_hybrid_plus_dv_deliverable_end_state(ladder: list[Video]) -> None:
+    """`-r SDR,HDR10P,DV,HYBRID -q 2160,1080`: hybrid base + lowest DV ingredient,
+    de-duplicated against the best-DV-per-resolution deliverables and SDR."""
+    quality = [2160, 1080]
+
+    hybrid_selected = list(filter(Tracks().select_hybrid(ladder, quality), ladder))
+
+    # Deliverables: best DV and SDR per requested resolution (what the dl Cartesian picks).
+    def best_per_res(range_: Video.Range, codec: Video.Codec | None = None) -> list[Video]:
+        out = []
+        for res in quality:
+            cands = [t for t in ladder if t.range == range_ and t.height == res and (codec is None or t.codec == codec)]
+            if cands:
+                out.append(max(cands, key=lambda t: t.bitrate))
+        return out
+
+    dv_deliverable = best_per_res(Video.Range.DV)
+    sdr_deliverable = [t for t in ladder if t.range == Video.Range.SDR]
+
+    final = Tracks.merge_video_selections(hybrid_selected, dv_deliverable + sdr_deliverable)
+
+    # No duplicates, and the lowest DV ingredient coexists with the DV deliverables.
+    assert len(final) == len({t.id for t in final})
+    assert "dv-360" in ids(final)  # ingredient retained for hybrid build
+    assert {"dv-2160", "dv-1080"} <= ids(final)  # standalone DV deliverables
+    assert {"hdr10p-2160", "hdr10p-1080"} <= ids(final)
+    assert {"sdr-2160", "sdr-1080-avc", "sdr-1080-hevc"} <= ids(final)

unshackle/commands/dl.py

@@ -1703,11 +1703,13 @@ class dl:
 
                     has_hybrid = any(r == Video.Range.HYBRID for r in range_)
                     non_hybrid_ranges = [r for r in range_ if r != Video.Range.HYBRID]
+                    # DV is both a hybrid ingredient (lowest track) and, when explicitly
+                    # requested, a standalone deliverable (best track per resolution).
+                    dv_is_deliverable = Video.Range.DV in non_hybrid_ranges
 
                     if quality:
                         missing_resolutions = []
                         if has_hybrid:
-                            # Split tracks: hybrid candidates vs non-hybrid
                             hybrid_candidate_tracks = [
                                 v
                                 for v in title.tracks.videos
@@ -1717,20 +1719,19 @@ class dl:
                                 v
                                 for v in title.tracks.videos
                                 if v.range not in (Video.Range.HDR10, Video.Range.HDR10P, Video.Range.DV)
+                                or (dv_is_deliverable and v.range == Video.Range.DV)
                             ]
 
-                            # Apply hybrid selection to HDR10+DV tracks
                             hybrid_filter = title.tracks.select_hybrid(hybrid_candidate_tracks, quality, worst=worst)
                             hybrid_selected = list(filter(hybrid_filter, hybrid_candidate_tracks))
 
                             if non_hybrid_ranges and non_hybrid_tracks:
-                                # Also filter non-hybrid tracks by resolution
                                 non_hybrid_selected = [
                                     v
                                     for v in non_hybrid_tracks
                                     if any(v.height == res or int(v.width * (9 / 16)) == res for res in quality)
                                 ]
-                                title.tracks.videos = hybrid_selected + non_hybrid_selected
+                                title.tracks.videos = Tracks.merge_video_selections(hybrid_selected, non_hybrid_selected)
                             else:
                                 title.tracks.videos = hybrid_selected
                         else:
@@ -1771,6 +1772,7 @@ class dl:
                             v
                             for v in title.tracks.videos
                             if v.range not in (Video.Range.HDR10, Video.Range.HDR10P, Video.Range.DV)
+                            or (dv_is_deliverable and v.range == Video.Range.DV)
                         ]
 
                         if not quality:
@@ -1813,7 +1815,16 @@ class dl:
                                 if match and match not in non_hybrid_selected:
                                     non_hybrid_selected.append(match)
 
-                        title.tracks.videos = hybrid_selected + non_hybrid_selected
+                        title.tracks.videos = Tracks.merge_video_selections(hybrid_selected, non_hybrid_selected)
+
+                        # Flag the lowest DV track as ingredient-only so mux skips it standalone,
+                        # unless it is itself the chosen DV deliverable (single DV rendition).
+                        selected_dv = [v for v in title.tracks.videos if v.range == Video.Range.DV]
+                        if selected_dv:
+                            ingredient_dv = min(selected_dv, key=lambda v: v.height)
+                            deliverable_dv = [v for v in non_hybrid_selected if v.range == Video.Range.DV]
+                            if not (dv_is_deliverable and ingredient_dv in deliverable_dv):
+                                ingredient_dv.hybrid_base_only = True
                     else:
                         selected_videos: list[Video] = []
                         if video_multi_lang:
@@ -2473,12 +2484,14 @@ class dl:
                             task_tracks = clone_tracks_for_audio(base_tracks, codec_audio_tracks)
                             multiplex_tasks.append((task_id, task_tracks, audio_codec))
 
-                    # Check if we're in hybrid mode
                     if any(r == Video.Range.HYBRID for r in range_) and title.tracks.videos:
-                        # Hybrid mode: process DV and HDR10 tracks separately for each resolution
                         self.log.info("Processing Hybrid HDR10+DV tracks...")
 
-                        # Group video tracks by resolution (prefer HDR10+ over HDR10 as base)
+                        # Snapshot videos before hybrid tracks are added so the originals
+                        # can still be muxed standalone afterwards.
+                        original_videos = list(title.tracks.videos)
+
+                        # Prefer HDR10+ over HDR10 as the hybrid base layer.
                         resolutions_processed = set()
                         base_tracks_list = [
                             v for v in title.tracks.videos if v.range in (Video.Range.HDR10P, Video.Range.HDR10)
@@ -2489,38 +2502,33 @@ class dl:
                             resolution = hdr10_track.height
                             if resolution in resolutions_processed:
                                 continue
-                            resolutions_processed.add(resolution)
 
-                            # Find matching DV track for this resolution (use the lowest DV resolution)
+                            # DV layer only supplies RPU metadata, so the lowest resolution suffices.
                             matching_dv = min(dv_tracks, key=lambda v: v.height) if dv_tracks else None
 
                             if matching_dv:
-                                # Create track pair for this resolution
+                                resolutions_processed.add(resolution)
-                                resolution_tracks = [hdr10_track, matching_dv]
 
+                                # Operate on copies so the originals stay muxable standalone.
+                                resolution_tracks = [deepcopy(hdr10_track), deepcopy(matching_dv)]
                                 for track in resolution_tracks:
                                     track.needs_duration_fix = True
 
-                                # Run the hybrid processing for this resolution
                                 Hybrid(resolution_tracks, self.service)
 
-                                # Create unique output filename for this resolution
                                 hybrid_filename = f"HDR10-DV-{resolution}p.hevc"
                                 hybrid_output_path = config.directories.temp / hybrid_filename
                                 hybrid_temp_paths.append(hybrid_output_path)
 
-                                # The Hybrid class creates HDR10-DV.hevc, rename it for this resolution
+                                # Hybrid always writes HDR10-DV.hevc; rename it per resolution.
                                 default_output = config.directories.temp / "HDR10-DV.hevc"
                                 if default_output.exists():
-                                    # If a previous run left this behind, replace it to avoid move() failures.
                                     hybrid_output_path.unlink(missing_ok=True)
                                     shutil.move(str(default_output), str(hybrid_output_path))
 
-                                # Create tracks with the hybrid video output for this resolution
                                 task_description = f"Multiplexing Hybrid HDR10+DV {resolution}p"
                                 task_tracks = Tracks(title.tracks) + title.tracks.chapters + title.tracks.attachments
 
-                                # Create a new video track for the hybrid output
                                 hybrid_track = deepcopy(hdr10_track)
                                 hybrid_track.id = f"hybrid_{hdr10_track.id}_{resolution}"
                                 hybrid_track.path = hybrid_output_path
@@ -2531,6 +2539,26 @@ class dl:
 
                                 enqueue_mux_tasks(task_description, task_tracks)
 
+                        # Mux every requested range standalone, skipping the ingredient-only DV.
+                        for video_track in original_videos:
+                            if getattr(video_track, "hybrid_base_only", False):
+                                continue
+                            if getattr(video_track, "dv_compatible_bitstream", False):
+                                apply_dv_fixup(video_track)
+
+                            task_description = "Multiplexing"
+                            if len(quality) > 1:
+                                task_description += f" {video_track.height}p"
+                            if len(range_) > 1:
+                                task_description += f" {video_track.range.name}"
+                            if len(vcodec) > 1:
+                                task_description += f" {video_track.codec.name}"
+
+                            task_tracks = Tracks(title.tracks) + title.tracks.chapters + title.tracks.attachments
+                            task_tracks.videos = [video_track]
+
+                            enqueue_mux_tasks(task_description, task_tracks)
+
                         console.print()
                     else:
                         # Normal mode: process each video track separately

unshackle/core/tracks/tracks.py

@@ -325,6 +325,20 @@ class Tracks:
         new_tracks.attachments = list(self.attachments)
         return new_tracks
 
+    @staticmethod
+    def merge_video_selections(*groups: list[Video]) -> list[Video]:
+        """Concatenate video selections, dropping duplicates (by track id, order-preserving).
+
+        A DV track can be chosen as both the hybrid ingredient (lowest) and an explicit
+        deliverable; without dedup the same track would be muxed/downloaded twice.
+        """
+        merged: list[Video] = []
+        for group in groups:
+            for video in group:
+                if video not in merged:
+                    merged.append(video)
+        return merged
+
     def select_hybrid(self, tracks, quality, worst: bool = False):
         # Prefer HDR10+ over HDR10 as the base layer (preserves dynamic metadata)
         base_ranges = (Video.Range.HDR10P, Video.Range.HDR10)