fix(download): repair a corrupt video and discard unresumable partials

Two cases the resume plan already routed to the download stage, but which the
stage could not actually fix.

yt-dlp's overwrites default is off (existing_video_file passes
default_overwrite=False), so an existing target is reported as "already
downloaded" and skipped. A truncated video.mkv was therefore planned for
re-download, never re-downloaded, and re-probed as the same corrupt file --
a job that reports success on media it could not read.

A .part without its .aria2 control file is a sparse file, not a valid prefix,
because aria2 writes its segments out of order. continuedl would resume from
its length and produce a video that silently stops being the source partway
through. Re-downloading is the cheaper mistake, so orphans go unconditionally;
a .part with its control file is left for aria2 to resume, and the sweep runs
once before the retry loop rather than between attempts.

find_video now matches only the merged output. Per-format streams from a
failed merge sort ahead of video.mkv and hold one stream each, so it could
return a video-only file to the audio stage.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/ccn_transcribe/jobs/verify.py

@@ -44,8 +44,8 @@ def _resolve(job: JobPaths, artifact: Artifact) -> Path | None:
 
 def _part_files(job: JobPaths) -> Iterator[Path]:
     for directory in (job.media_dir, job.tmp_dir):
-        if directory.is_dir():
-            yield from directory.glob("*.part")
+        # glob yields nothing for a directory that does not exist; no guard needed.
+        yield from directory.glob("*.part")
 
 
 def _resumable_partial_exists(job: JobPaths) -> bool:

src/ccn_transcribe/stages/download.py

@@ -17,7 +17,7 @@ from ccn_transcribe import errors, paths
 from ccn_transcribe.media import ytdlp_opts
 
 if TYPE_CHECKING:
-    from collections.abc import Sequence
+    from collections.abc import Iterator, Sequence
     from pathlib import Path
 
     from ccn_transcribe.paths import JobPaths
@@ -155,13 +155,59 @@ def identify(entry: dict[str, Any], url: str) -> str:
 
 
 def find_video(job: JobPaths) -> Path | None:
-    """The downloaded media file, whatever extension yt-dlp settled on."""
+    """The merged media file, whatever extension yt-dlp settled on.
+
+    Only the file yt-dlp's output template names directly. A failed merge leaves
+    per-format streams like video.f251.webm behind, which sort first and hold one
+    stream each; returning one would silently transcribe half the source.
+    """
     for candidate in sorted(job.media_dir.glob("video.*")):
-        if candidate.suffix not in (".part", ".aria2", ".json", ".ytdl"):
+        if candidate.stem == "video" and candidate.suffix not in (".part", ".aria2", ".ytdl"):
             return candidate
     return None
 
 
+def _part_files(job: JobPaths) -> Iterator[Path]:
+    for directory in (job.media_dir, job.tmp_dir):
+        # glob yields nothing for a directory that does not exist; no guard needed.
+        yield from directory.glob("*.part")
+
+
+def clear_stale_parts(job: JobPaths) -> list[Path]:
+    """Delete .part files that cannot be resumed, and say which.
+
+    aria2 writes its 16 segments out of order, so without the .aria2 control file
+    a .part is a sparse file full of holes rather than a valid prefix. yt-dlp's
+    continuedl would resume from its length and produce a video that plays for a
+    while and then isn't the source any more -- a corrupt transcript, not an error.
+    Re-downloading is the cheaper mistake, so an orphan goes even when the native
+    downloader (which does write sequentially) could have resumed it.
+    """
+    removed: list[Path] = []
+    for part in sorted(_part_files(job)):
+        if part.with_suffix(part.suffix + ".aria2").exists():
+            continue
+        part.unlink()
+        removed.append(part)
+        log.warning("discarded unresumable partial download %s", part.name)
+    return removed
+
+
+def discard_video(job: JobPaths) -> Path | None:
+    """Delete the merged file so yt-dlp will actually fetch it again.
+
+    yt-dlp's overwrites default is off, so it reports an existing target as
+    "already downloaded" and skips it; a truncated video.mkv would otherwise never
+    be repaired. Per-format streams are left alone -- yt-dlp re-merges complete
+    ones without re-downloading, so clearing them is a pessimization.
+    """
+    video = find_video(job)
+    if video is not None:
+        video.unlink()
+        log.info("discarded %s so it can be downloaded again", video.name)
+    return video
+
+
 def download_options(job: JobPaths, config: ytdlp_opts.DownloadConfig) -> dict[str, Any]:
     return ytdlp_opts.ydl_opts(
         str(job.media_dir / "video.%(ext)s"),
@@ -188,6 +234,10 @@ def download(
     without this loop that stall would become a job failure instead of a resume.
     """
     job.ensure()
+    # Before the loop, not inside it: a failed attempt leaves a .part with a live
+    # control file, which is exactly the progress the next attempt resumes from.
+    clear_stale_parts(job)
+    discard_video(job)
     options = download_options(job, config)
     make = factory or default_factory
 

tests/test_stages_download.py

@@ -214,6 +214,115 @@ class TestFindVideo:
     def test_none_when_nothing_downloaded(self, job: paths.JobPaths) -> None:
         assert download.find_video(job) is None
 
+    def test_ignores_unmerged_per_format_files(self, job: paths.JobPaths) -> None:
+        # A failed merge leaves video.f251.webm behind; it sorts before video.mkv
+        # and is a single stream, so returning it would transcribe half the source.
+        (job.media_dir / "video.f251.webm").write_bytes(b"audio only")
+        assert download.find_video(job) is None
+
+
+class TestClearStaleParts:
+    def test_removes_a_part_with_no_control_file(self, job: paths.JobPaths) -> None:
+        stale = job.media_dir / "video.mkv.part"
+        stale.write_bytes(b"sparse")
+        assert download.clear_stale_parts(job) == [stale]
+        assert not stale.exists()
+
+    def test_keeps_a_part_that_aria2_can_resume(self, job: paths.JobPaths) -> None:
+        part = job.media_dir / "video.mkv.part"
+        part.write_bytes(b"sparse")
+        (job.media_dir / "video.mkv.part.aria2").write_bytes(b"ctrl")
+        assert download.clear_stale_parts(job) == []
+        assert part.exists()
+
+    def test_sweeps_the_tmp_directory_too(self, job: paths.JobPaths) -> None:
+        stale = job.tmp_dir / "video.mkv.part"
+        stale.write_bytes(b"sparse")
+        assert download.clear_stale_parts(job) == [stale]
+
+    def test_nothing_to_do_is_not_an_error(self, job: paths.JobPaths) -> None:
+        assert download.clear_stale_parts(job) == []
+
+
+class TestDiscardVideo:
+    def test_removes_the_merged_file(self, job: paths.JobPaths) -> None:
+        video = job.media_dir / "video.mkv"
+        video.write_bytes(b"truncated")
+        assert download.discard_video(job) == video
+        assert not video.exists()
+
+    def test_leaves_per_format_files_alone(self, job: paths.JobPaths) -> None:
+        # yt-dlp re-merges complete per-format files without re-downloading them;
+        # deleting them turns a cheap remux into a full re-download.
+        intermediate = job.media_dir / "video.f137.mp4"
+        intermediate.write_bytes(b"video stream")
+        assert download.discard_video(job) is None
+        assert intermediate.exists()
+
+    def test_nothing_to_discard(self, job: paths.JobPaths) -> None:
+        assert download.discard_video(job) is None
+
+
+class TestDownloadClearsStaleState:
+    def test_a_corrupt_video_is_unlinked_before_yt_dlp_runs(self, job: paths.JobPaths) -> None:
+        # yt-dlp's overwrites default is off: it reports an existing target as
+        # "already downloaded" and skips, so a truncated file is never repaired.
+        video = job.media_dir / "video.mkv"
+        video.write_bytes(b"truncated")
+        seen: list[bool] = []
+
+        def on_download(_n: int) -> None:
+            seen.append(video.exists())
+            video.write_bytes(b"complete")
+
+        download.download(job, "u", factory=factory_for(on_download=on_download))
+        assert seen == [False]
+        assert video.read_bytes() == b"complete"
+
+    def test_an_orphan_part_is_unlinked_before_yt_dlp_runs(self, job: paths.JobPaths) -> None:
+        stale = job.media_dir / "video.mkv.part"
+        stale.write_bytes(b"sparse")
+        seen: list[bool] = []
+
+        def on_download(_n: int) -> None:
+            seen.append(stale.exists())
+            (job.media_dir / "video.mkv").write_bytes(b"data")
+
+        download.download(job, "u", factory=factory_for(on_download=on_download))
+        assert seen == [False]
+
+    def test_a_resumable_part_survives(self, job: paths.JobPaths) -> None:
+        part = job.media_dir / "video.mkv.part"
+        part.write_bytes(b"sparse")
+        (job.media_dir / "video.mkv.part.aria2").write_bytes(b"ctrl")
+        seen: list[bool] = []
+
+        def on_download(_n: int) -> None:
+            seen.append(part.exists())
+            (job.media_dir / "video.mkv").write_bytes(b"data")
+
+        download.download(job, "u", factory=factory_for(on_download=on_download))
+        assert seen == [True]
+
+    def test_the_sweep_happens_once_not_per_attempt(self, job: paths.JobPaths) -> None:
+        # aria2 writes its control file during the failed attempt; sweeping again
+        # between retries would delete the very progress the retry exists to keep.
+        part = job.media_dir / "video.mkv.part"
+        calls: list[int] = []
+
+        def on_download(_n: int) -> None:
+            calls.append(len(calls) + 1)
+            if len(calls) < 2:
+                part.write_bytes(b"sparse")
+                raise RuntimeError("aria2c exited with code 5")
+            assert part.exists()
+            (job.media_dir / "video.mkv").write_bytes(b"data")
+
+        download.download(
+            job, "u", factory=factory_for(on_download=on_download), sleep=lambda _s: None
+        )
+        assert len(calls) == 2
+
 
 class TestDownload:
     def _succeed(self, job: paths.JobPaths) -> Any: