fix(downloads): a conversion's commit, a stopped upload, the fast join and a clip's copy hold up

- The fast join gives each ffconcat entry its own media-only format whitelist, so a library file
  can no longer be opened as a nested concat list (the list's whitelist was handed down to it).
- A file the guarded probe cannot read is a download-only file, not a guess from yt-dlp's info.
- A conversion's or an ingest's commit that raises is re-read in a fresh session before any file
  is removed: a commit that landed keeps the file its row now names.
- An original that cannot be removed after conversion is logged; the upload stays ready.
- An upload whose conversion is cancelled or deleted midway is freed once nothing holds it; a
  paused one is kept.
- A shared link says "preparing" only while its video's job is queued or running.
- The admin storage view counts every file on disk ("Files on disk"), not only ready ones.
- A clip's failure title, unknown failure and cancel dialog, and a clip's and an upload's delete
  dialog speak of what the job made; an upload's quota refusal explains that the original is kept.
- /watch announces "preparing" as an alert and moves focus to the new Retry after an action.

backend/app/downloads/edit.py

@@ -338,13 +338,16 @@ def build_concat_plan(src: Path, dest: Path, spec: dict, out_ext: str, staging:
         ]
     else:
         listp = staging / "concat.txt"
-        lines = []
+        lines = ["ffconcat version 1.0"]
         for s in segs:
             lines.append(f"file '{_concat_quote(src)}'")
+            # ffmpeg hands the list's whitelist (which must name `concat` to read the list) down to
+            # the file it names, so a library file whose bytes are a concat list would be read as one
+            # and open the paths inside it. Each entry's own whitelist replaces the inherited one.
+            lines.append(f"option format_whitelist {INPUT_GUARD[1]}")
             lines.append(f"inpoint {s['start_s']}")
             lines.append(f"outpoint {s['end_s']}")
         prep[listp] = "\n".join(lines) + "\n"
-        # The list is ours; the whitelist reaches the file it names too (ffmpeg hands it down).
         guard = ["-format_whitelist", "concat," + INPUT_GUARD[1]]
         cmd = head + ["-f", "concat", "-safe", "0", *guard, "-i", str(listp), "-c", "copy"]
 

backend/app/downloads/pipeline.py

@@ -914,14 +914,59 @@ def _unplace(placed: Path | None) -> None:
         placed.unlink(missing_ok=True)
 
 
+def _conversion_landed(asset_id: int, new_rel: str) -> bool | None:
+    """After the commit that makes a converted upload ready RAISED: did it land anyway? A commit can
+    reach the server and lose its answer (the connection drops after COMMIT). True/False by what the
+    row now says; None when the row cannot be read either."""
+    try:
+        with SessionLocal() as db:
+            asset = db.get(MediaAsset, asset_id)
+            return asset is not None and asset.status == "ready" and asset.rel_path == new_rel
+    except Exception:  # noqa: BLE001 — "cannot tell" is the answer, the caller acts on it
+        # silent-ok: the caller re-raises the commit's own error, which `_fail_edit` reports.
+        return None
+
+
+def _drop_original(job_id: int, src: Path) -> None:
+    """Remove an upload's original once the converted file is committed in its place. Best effort:
+    the job is done and its file is ready, so a failure here must not fail it — the original is
+    named by no row now, and the GC's `.uploads` sweep takes it."""
+    try:
+        src.unlink(missing_ok=True)
+    except OSError as exc:
+        log.warning(
+            "upload job %s: original not removed, left to the GC sweep: %s",
+            job_id,
+            exc,
+            extra=logs.event("upload.original.unlink_failed", job_id=job_id),
+        )
+
+
+def _free_unheld_upload(asset_id: int) -> bool:
+    """Free an upload no job holds any more — cancelled or deleted while its conversion ran, when
+    `release_asset` had to leave it because the conversion still read its file. Once that has
+    stopped, free it as the release would have; left alone, its file would sit on disk counted by no
+    owner's footprint until the GC's upload pass. False when a job still holds it (paused)."""
+    with SessionLocal() as db:
+        asset = db.get(MediaAsset, asset_id)
+        if asset is None or asset.ref_count != 0:
+            return False
+        if asset.rel_path:
+            storage.delete_asset_files(settings.download_root, asset.rel_path)
+        db.delete(asset)
+        db.commit()
+    return True
+
+
 def _convert_upload(job_id: int, asset_id: int) -> None:
     """Convert a queued upload into the phone-safe shape (`playable`) and make it ready.
 
     The original sits at the asset's `rel_path` (see `upload`); the converted file replaces it there
     under the new extension, and only then is the original removed — so a failed or stopped
     conversion leaves the original in place for Retry, and a delete still finds it by `rel_path`.
-    A converted file placed but never committed is removed again on the way out; a stop in the
-    instant between the commit and the original's unlink is left to the GC's `.uploads` sweep."""
+    A converted file placed but never committed is removed again on the way out. When the commit's
+    own outcome is unknown neither file is removed: the row names one of them, and the GC's `.uploads`
+    sweep takes the one it does not — as it does an original whose unlink after the commit failed."""
     with SessionLocal() as db:
         asset = db.get(MediaAsset, asset_id)
         old_rel = asset.rel_path if asset else None
@@ -965,12 +1010,25 @@ def _convert_upload(job_id: int, asset_id: int) -> None:
             finalized = service.finalize_if_running(
                 db, job_id, status=JobStatus.DONE, progress=100, phase=None
             )
-            db.commit()
+            try:
+                db.commit()
+            except Exception:
+                landed = _conversion_landed(asset_id, new_rel)
+                if landed is False:
+                    raise  # rolled back: the original is still the asset's file — fail as any step
+                # Landed, or cannot tell: the converted file may be the asset's file now, so it is
+                # never removed blind (the sweep takes whichever file no row names).
+                placed = None
+                if landed is None:
+                    raise
+                db.rollback()
             placed = None
             if finalized:
                 _notify_download_done(db, owner_id, title, job_id, kind="upload")
         if new_rel != old_rel:
-            src.unlink(missing_ok=True)
+            _drop_original(job_id, src)
+        if not finalized:
+            _free_unheld_upload(asset_id)  # finished after its job was cancelled or deleted
         if finalized:
             log.info(
                 "upload job %s converted: asset %s",
@@ -979,9 +1037,11 @@ def _convert_upload(job_id: int, asset_id: int) -> None:
                 extra=logs.event("upload.job.done", job_id=job_id, asset_id=asset_id, ext=ext),
             )
     except editmod.EditAborted:
-        # Paused/canceled mid-encode: the original is untouched; free the asset for a later retry.
+        # Paused/canceled mid-encode: the original is untouched. A paused job keeps it for its
+        # resume; one cancelled or deleted holds it no more, so it is freed now.
         _unplace(placed)
-        _reset_asset_pending(asset_id)
+        if not _free_unheld_upload(asset_id):
+            _reset_asset_pending(asset_id)
         log.info(
             "upload job %s conversion aborted (pause/cancel)",
             job_id,

backend/app/downloads/probe.py

@@ -121,18 +121,18 @@ def resolve_media_facts(
     is authoritative — INCLUDING a deliberate None (cover-art-only audio, a still image). A per-field
     `probe or info` fallback would let `info`'s stale vcodec resurrect a video track the file doesn't
     have (yt-dlp reports a progressive `vcodec` for an audio-mode download whose video it then strips),
-    mislabelling an audio download as a broken in-app 'video'. Only when ffprobe could NOT read the
-    file at all do we fall back to `info` (the pre-probe behaviour, a safety net); the cosmetic
-    dimensions and the duration still gap-fill from `info` in either case."""
+    mislabelling an audio download as a broken in-app 'video'. When ffprobe could NOT read the file
+    at all, no codec is reported: the probe runs with the media demuxer whitelist, so a file it
+    cannot open is one the player and the editor cannot open either — and taking `info`'s codecs
+    for it would make it an editable 'video' whose bytes are whatever the remote site served. The
+    cosmetic dimensions and the duration still gap-fill from `info` in either case."""
     data = editmod.run_ffprobe_json(path, show_streams=True, show_format=True)
-    info_v = info.get("vcodec") if info.get("vcodec") not in (None, "none") else None
-    info_a = info.get("acodec") if info.get("acodec") not in (None, "none") else None
     info_w = int(info["width"]) if info.get("width") else None
     info_h = int(info["height"]) if info.get("height") else None
     info_d = int(info["duration"]) if info.get("duration") else None
     if data is None:
-        # ffprobe could not read the file — fall back to the extractor's info entirely.
-        return info_v, info_a, info_w, info_h, info_d
+        # Unreadable to the guarded probe: a download-only 'file' (no codecs → media_kind FILE).
+        return None, None, info_w, info_h, info_d
     pv, pa, pw, ph, pd = _facts_from_probe(data)
     # Probe ran → its codecs win outright (a None means "no such track", not "unknown"); info only
     # gap-fills the width/height/duration the probe couldn't read.

backend/app/downloads/upload.py

@@ -83,9 +83,10 @@ def ingest_upload(
     rel = storage.upload_rel_path(user_id, asset.id, filename)
     storage.place_file(src_path, settings.download_root, rel)
     # From here the file is on disk under `.uploads/`. If the transaction fails before it commits, the
-    # asset/job rows roll back but the placed file would be ORPHANED — nothing references it, the
-    # per-user delete path keys off asset.rel_path (which is gone), and the GC orphan sweep skips the
-    # dot-prefixed `.uploads/` tree. So roll back the on-disk side too on any failure.
+    # asset/job rows roll back and nothing names the placed file: the per-user delete path keys off
+    # asset.rel_path (which is gone), and the GC's `.uploads` sweep reaches it only an hour later. So
+    # roll back the on-disk side too — unless the commit itself raised after it landed (see below).
+    asset_id = asset.id
     try:
         final = storage.abs_path(settings.download_root, rel)
         actual_size = final.stat().st_size if final.exists() else size
@@ -126,6 +127,11 @@ def ingest_upload(
         db.commit()
     except Exception:
         db.rollback()
+        if _row_landed(db, asset_id) is not False:
+            # The commit reached the server and lost its answer, or the row cannot be read: the file
+            # may be the asset's now, so it is never removed blind — the sweep takes it if no row
+            # names it.
+            raise
         # The rows are gone; remove the now-unreferenced file so no bytes leak. Clean the poster by its
         # DETERMINISTIC path (rel with a .jpg suffix, the same path ensure_poster writes to) rather than
         # ensure_poster's return value: it writes the frame directly to that path and returns None on an
@@ -137,3 +143,13 @@ def ingest_upload(
         raise
     db.refresh(job)
     return job
+
+
+def _row_landed(db: Session, asset_id: int) -> bool | None:
+    """After the ingest transaction raised: is the asset row there anyway? A commit can reach the
+    server and lose its answer (the connection drops after COMMIT). None when it cannot be read."""
+    try:
+        return db.get(MediaAsset, asset_id) is not None
+    except Exception:  # noqa: BLE001 — "cannot tell" is the answer, the caller acts on it
+        # silent-ok: the caller re-raises the ingest's own error.
+        return None

backend/app/routes/downloads.py

@@ -1144,7 +1144,9 @@ def admin_list_downloads(
 
 @admin_router.get("/storage")
 def admin_storage(admin: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
-    ready = db.execute(
+    # Every file on disk (`quota.ON_DISK`): ready ones, and uploads waiting for or failed in their
+    # conversion — the same disk the footprint and the GC cap count.
+    on_disk = db.execute(
         select(func.count(), func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(quota.ON_DISK)
     ).one()
     total_assets = db.execute(select(func.count()).select_from(MediaAsset)).scalar()
@@ -1158,8 +1160,8 @@ def admin_storage(admin: User = Depends(admin_user), db: Session = Depends(get_d
         for u in db.execute(select(User).where(User.id.in_([uid for uid, _ in per_user]))).scalars()
     }
     return {
-        "ready_files": ready[0],
-        "total_bytes": int(ready[1]),
+        "files_on_disk": on_disk[0],
+        "total_bytes": int(on_disk[1]),
         "total_assets": total_assets,
         "total_cap_bytes": sysconfig.get_int(db, "download_total_max_bytes"),
         "per_user": [

backend/app/routes/public.py

@@ -16,6 +16,7 @@ from app.config import settings
 from app.db import get_db
 from app.downloads import links as linksmod
 from app.downloads import storage
+from app.downloads.status import JobStatus
 from app.models import DownloadJob, DownloadLink, MediaAsset
 from app.ratelimit import RateLimiter, client_key
 from app.security import verify_password
@@ -46,9 +47,16 @@ def _link_or_404(db: Session, token: str) -> DownloadLink:
 def _ready_target(db: Session, link: DownloadLink) -> tuple[DownloadJob, MediaAsset]:
     job = db.get(DownloadJob, link.job_id)
     asset = db.get(MediaAsset, job.asset_id) if job and job.asset_id else None
-    if asset is not None and asset.status in ("pending", "downloading"):
+    if (
+        job is not None
+        and asset is not None
+        and asset.status in ("pending", "downloading")
+        and job.status in (JobStatus.QUEUED, JobStatus.RUNNING)
+    ):
         # Being (re)made — a re-download, or a conversion — so the link works again once it is ready:
         # its own reason, so the page can say "try again shortly" rather than "this link is gone".
+        # Only while the job is still going: a cancelled or paused rebuild leaves the asset pending
+        # with nothing working on it, and "try again in a few minutes" would never come true.
         raise refused(404, "video_preparing")
     if job is None or asset is None or asset.status != "ready" or not asset.rel_path:
         raise refused(404, "video_gone")

backend/tests/test_ffmpeg_input_guard.py

@@ -75,10 +75,45 @@ def test_a_join_is_guarded(accurate, tmp_path):
     segs = [{"start_s": 0, "end_s": 1}, {"start_s": 2, "end_s": 3}]
     spec = edit.normalize_edit_spec({"segments": segs, "accurate": accurate})
     cmd, _ = edit.build_concat_plan(Path("in.mp4"), Path("out.mp4"), spec, "mp4", tmp_path)
-    # The fast join reads a concat list WE write; the whitelist rides down to the file it names.
+    # The fast join reads a concat list WE write, so the list itself may be opened as concat.
     _assert_media_guarded(cmd, extra=set() if accurate else {"concat"})
 
 
+def test_the_fast_join_list_gives_each_entry_the_media_list_without_concat(tmp_path):
+    # ffmpeg hands the list's whitelist (with `concat`) down to the library file each entry names;
+    # the entry's own `option format_whitelist` must replace it, or that file may open as a list.
+    segs = [{"start_s": 0, "end_s": 1}, {"start_s": 2, "end_s": 3}]
+    spec = edit.normalize_edit_spec({"segments": segs, "accurate": False})
+    _, prep = edit.build_concat_plan(Path("in.mp4"), Path("out.mp4"), spec, "mp4", tmp_path)
+    (text,) = prep.values()
+    lines = text.splitlines()
+    entries = [i for i, ln in enumerate(lines) if ln.startswith("file ")]
+    assert len(entries) == 2
+    for i in entries:
+        key, _, allowed = lines[i + 1].partition(" format_whitelist ")
+        assert key == "option", lines[i + 1]
+        assert set(allowed.split(",")) == MEDIA
+
+
+@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")
+def test_the_fast_join_still_joins_a_real_file(tmp_path):
+    src = tmp_path / "in.mp4"
+    subprocess.run(
+        ["ffmpeg", "-y", "-loglevel", "error", "-f", "lavfi", "-i", "testsrc=size=64x48:rate=10",
+         "-f", "lavfi", "-i", "sine=frequency=440", "-t", "4", "-c:v", "libx264",
+         "-pix_fmt", "yuv420p", "-c:a", "aac", str(src)],
+        check=True,
+    )  # fmt: skip
+    segs = [{"start_s": 0, "end_s": 1}, {"start_s": 2, "end_s": 3}]
+    spec = edit.normalize_edit_spec({"segments": segs, "accurate": False})
+    out = tmp_path / "out.mp4"
+    cmd, prep = edit.build_concat_plan(src, out, spec, "mp4", tmp_path)
+    for p, content in prep.items():
+        p.write_text(content, encoding="utf-8")
+    subprocess.run(cmd, check=True, capture_output=True)
+    assert (edit.probe_duration(out) or 0) > 0
+
+
 def test_the_poster_and_the_filmstrip_are_guarded():
     _assert_media_guarded(edit.build_poster_cmd(Path("in.mp4"), Path("p.jpg"), 2.0))
     geom = edit.storyboard_geometry(60)

backend/tests/test_upload.py

@@ -803,10 +803,10 @@ def test_ingest_rejects_over_max_jobs(db, make_user, _root, monkeypatch):
 
 def test_ingest_cleans_up_placed_files_on_commit_failure(db, make_user, _root, monkeypatch):
     """A failure AFTER storage.place_file has moved the file onto disk must roll back AND unlink the
-    placed file + any (even partial) poster — otherwise the orphan is permanent (the per-user delete
-    keys off the rolled-back asset.rel_path, and the GC orphan sweep skips the dot-prefixed .uploads/
-    tree). Exercises the rollback/cleanup branch that IS the leak remediation, incl. the poster case
-    where ensure_poster left a partial file but returned None."""
+    placed file + any (even partial) poster — otherwise nothing names it (the per-user delete keys
+    off the rolled-back asset.rel_path) and it sits on disk until the GC's `.uploads` sweep. Exercises
+    the rollback/cleanup branch, incl. the poster case where ensure_poster left a partial file but
+    returned None."""
     _probe(monkeypatch, ("h264", "aac", 320, 240, 5))  # video → the poster path runs
     u = make_user()
 
@@ -837,6 +837,28 @@ def test_ingest_cleans_up_placed_files_on_commit_failure(db, make_user, _root, m
     assert db.query(MediaAsset).count() == 0  # the flushed asset rolled back — no dangling row
 
 
+def test_ingest_keeps_the_file_when_its_commit_landed_but_raised(db, make_user, _root, monkeypatch):
+    # The commit reached the server and lost its answer: the row names the file now, so removing it
+    # as "rolled back" would leave a ready upload pointing at nothing.
+    _probe(monkeypatch, ("h264", "aac", 320, 240, 5))
+    monkeypatch.setattr(uploadmod.editmod, "ensure_poster", lambda *_a: None)
+    u = make_user()
+    real_commit = db.commit
+
+    def _landed_then_lost():
+        real_commit()
+        raise RuntimeError("connection lost after COMMIT")
+
+    monkeypatch.setattr(db, "commit", _landed_then_lost)
+    src = _root / "clip.mp4"
+    src.write_bytes(b"fake-video-bytes")
+    with pytest.raises(RuntimeError):
+        uploadmod.ingest_upload(db, u.id, filename="clip.mp4", src_path=src, size=16)
+
+    (asset,) = db.query(MediaAsset).all()
+    assert storage.abs_path(str(_root), asset.rel_path).exists()
+
+
 # --- classifier: the REAL probe_media parsing (stubs ffprobe one level down) ------------------
 # The ingest/route tests above stub probe_media wholesale; these exercise its OWN logic — image
 # detection, the cover-art skip, dimension/duration parsing — by faking only run_ffprobe_json, so a
@@ -1022,22 +1044,16 @@ class TestResolveMediaFacts:
         )
         assert v is None and mediakind.derive(v, a) == mediakind.FILE
 
-    def test_info_fills_in_only_when_ffprobe_could_not_read_the_file(self, monkeypatch):
-        # ffprobe failed entirely (None) → fall back to info so we never regress the working path.
+    def test_a_file_the_guarded_probe_cannot_read_is_a_download_only_file(self, monkeypatch):
+        # ffprobe (media demuxers only) failed entirely: the player and the editor cannot open it
+        # either, so info's codecs must not make it an editable 'video'. Dimensions still gap-fill.
         self._ffprobe(monkeypatch, None)
         v, a, w, h, _d = probemod.resolve_media_facts(
-            Path("x"), {"vcodec": "vp9", "acodec": "opus", "width": 1920, "height": 1080}
+            Path("x"),
+            {"vcodec": "avc1.64001f", "acodec": "mp4a.40.2", "width": 1920, "height": 1080},
         )
-        assert (v, a, w, h) == ("vp9", "opus", 1920, 1080)
-
-    def test_info_none_codec_string_is_treated_as_absent(self, monkeypatch):
-        # yt-dlp writes the literal "none" for a missing stream — it must not become a codec name.
-        self._ffprobe(monkeypatch, None)  # probe failed → info path, where the "none" scrub applies
-        v, a, w, h, _d = probemod.resolve_media_facts(
-            Path("x"), {"vcodec": "none", "acodec": "aac"}
-        )
-        assert (v, a) == (None, "aac")
-        assert mediakind.derive(v, a) == mediakind.AUDIO
+        assert (v, a, w, h) == (None, None, 1920, 1080)
+        assert mediakind.derive(v, a) == mediakind.FILE
 
 
 # --- the phone-safe conversion (an iPhone's HEVC, a webm, an mkv) ------------------------------

backend/tests/test_upload_conversion.py

@@ -21,7 +21,9 @@ from app.downloads import probe as probemod
 from app.downloads import upload as uploadmod
 from app.downloads.stated import StatedFailure
 from app.models import DownloadJob, DownloadQuota, MediaAsset, Notification
-from sqlalchemy import select
+from sqlalchemy import select, update
+from sqlalchemy.exc import OperationalError
+from sqlalchemy.orm import Session
 
 
 @pytest.fixture
@@ -164,6 +166,33 @@ def test_two_dotted_bare_names_keep_a_file_each(db, make_user, _root, monkeypatc
     assert all(storage.abs_path(str(_root), r).exists() for r in finals)
 
 
+def test_a_bare_names_stem_holds_no_dot_so_a_swapped_suffix_keeps_its_id():
+    # Each of the two name rules is held on its own: a new name's stem has no dot, so a sibling
+    # derived by `with_suffix` (the poster, the converted file) keeps `_a{id}`.
+    rel = storage.upload_rel_path(1, 5, "Holiday.2024.ä")
+    assert rel == ".uploads/1/Holiday_2024_a5"
+    assert Path(rel).with_suffix(".jpg").name == "Holiday_2024_a5.jpg"
+
+
+def test_an_upload_placed_with_a_dotted_bare_name_keeps_its_id_when_converted(
+    db, make_user, _root, monkeypatch
+):
+    # A row from before the stems lost their dots: its stem's tail is no extension.
+    _probe(monkeypatch)
+    u = make_user()
+    _converts_to(monkeypatch, size=5)
+    job, asset = _ingest(db, _root, u.id, name="Holiday.2024.ä")
+    legacy = f".uploads/{u.id}/Holiday.2024_a{asset.id}"
+    storage.abs_path(str(_root), asset.rel_path).rename(storage.abs_path(str(_root), legacy))
+    asset.rel_path, asset.container = legacy, None
+    _claim(db, job, asset)
+
+    pipeline._convert_upload(job.id, asset.id)
+
+    db.expire_all()
+    assert db.get(MediaAsset, asset.id).rel_path == f"{legacy}.mp4"
+
+
 def test_a_conversion_failing_after_it_placed_its_file_removes_it(
     db, make_user, _root, monkeypatch
 ):
@@ -187,6 +216,94 @@ def test_a_conversion_failing_after_it_placed_its_file_removes_it(
     assert [p.name for p in _uploads_dir(_root, u.id).iterdir()] == [original.name]
 
 
+def _finishing_commit_raises(monkeypatch, *, lands: bool) -> None:
+    """The commit that makes the converted upload ready raises — after it reached the server
+    (`lands`: the connection dropped before its answer) or before (it rolled back)."""
+    real_finalize, real_commit = pipeline.service.finalize_if_running, Session.commit
+    state = {"session": None, "fired": False}
+
+    def _finalize(db, *a, **k):
+        if not state["fired"] and state["session"] is None:
+            state["session"] = db
+        return real_finalize(db, *a, **k)
+
+    def _commit(self):
+        if self is state["session"] and not state["fired"]:
+            state["fired"] = True
+            if lands:
+                real_commit(self)
+            raise OperationalError("COMMIT", {}, Exception("connection lost"))
+        return real_commit(self)
+
+    monkeypatch.setattr(pipeline.service, "finalize_if_running", _finalize)
+    monkeypatch.setattr(Session, "commit", _commit)
+
+
+def test_a_finishing_commit_that_landed_but_raised_keeps_the_converted_file(
+    db, make_user, _root, monkeypatch
+):
+    # The row names the converted file now: removing it as "never committed" would leave the asset
+    # pointing at nothing and its original named by no row, for the sweep to take — both copies gone.
+    _probe(monkeypatch)
+    u = make_user()
+    job, asset = _ingest(db, _root, u.id)
+    _claim(db, job, asset)
+    _converts_to(monkeypatch, size=5)
+    _finishing_commit_raises(monkeypatch, lands=True)
+
+    pipeline._convert_upload(job.id, asset.id)
+
+    db.expire_all()
+    asset = db.get(MediaAsset, asset.id)
+    assert (db.get(DownloadJob, job.id).status, asset.status) == ("done", "ready")
+    assert storage.abs_path(str(_root), asset.rel_path).exists()
+
+
+def test_a_finishing_commit_that_rolled_back_fails_and_keeps_the_original(
+    db, make_user, _root, monkeypatch
+):
+    _probe(monkeypatch)
+    u = make_user()
+    job, asset = _ingest(db, _root, u.id)
+    original = storage.abs_path(str(_root), asset.rel_path)
+    _claim(db, job, asset)
+    _converts_to(monkeypatch, size=5)
+    _finishing_commit_raises(monkeypatch, lands=False)
+
+    pipeline._convert_upload(job.id, asset.id)
+
+    db.expire_all()
+    assert db.get(DownloadJob, job.id).status == "error"
+    assert [p.name for p in _uploads_dir(_root, u.id).iterdir()] == [original.name]
+
+
+def test_an_original_that_cannot_be_removed_leaves_the_upload_ready(
+    db, make_user, _root, monkeypatch
+):
+    # The converted file is committed and the job done: the original's unlink failing must not flip
+    # a working asset to an error no Retry is offered for. The sweep takes the original later.
+    _probe(monkeypatch)
+    u = make_user()
+    job, asset = _ingest(db, _root, u.id)
+    original = storage.abs_path(str(_root), asset.rel_path)
+    _claim(db, job, asset)
+    _converts_to(monkeypatch, size=5)
+    real_unlink = Path.unlink
+
+    def _unlink(self, missing_ok=False):
+        if self == original:
+            raise PermissionError("read-only")
+        return real_unlink(self, missing_ok=missing_ok)
+
+    monkeypatch.setattr(Path, "unlink", _unlink)
+    pipeline._convert_upload(job.id, asset.id)
+
+    db.expire_all()
+    asset = db.get(MediaAsset, asset.id)
+    assert (db.get(DownloadJob, job.id).status, asset.status) == ("done", "ready")
+    assert original.exists()
+
+
 # --- what the owner is told ------------------------------------------------------------------
 
 
@@ -272,6 +389,116 @@ def test_the_sweep_takes_a_stranded_upload_file_and_keeps_a_waiting_original(
     assert not stranded.exists()  # no row names it
 
 
+@pytest.mark.parametrize(
+    ("job_status", "asset_status"),
+    [
+        ("queued", "pending"),  # waiting for the worker
+        ("running", "downloading"),  # being converted, for longer than the sweep's age floor
+        ("error", "error"),  # failed, kept for Retry
+        ("done", "ready"),  # converted
+    ],
+)
+def test_the_sweep_keeps_an_aged_upload_file_its_row_names_in_any_state(
+    job_status, asset_status, db, make_user, _root, monkeypatch
+):
+    _probe(monkeypatch)
+    job, asset = _ingest(db, _root, make_user().id)
+    path = storage.abs_path(str(_root), asset.rel_path)
+    job.status, asset.status = job_status, asset_status
+    db.commit()
+    _aged(path)
+
+    gcmod._sweep_orphan_files(db, str(_root))
+
+    assert path.exists()
+
+
+def _row(db, asset_id: int) -> MediaAsset | None:
+    """The asset as the database has it now (a `get` would refresh the session's own copy, and
+    raise for one deleted under it)."""
+    db.expire_all()
+    return db.execute(select(MediaAsset).where(MediaAsset.id == asset_id)).scalar_one_or_none()
+
+
+def _route_moved(db, job, asset, job_status: str, ref_count: int) -> None:
+    """What the cancel/pause/delete route leaves behind, written as SQL: the session's copy of
+    `ref_count` is stale (the routes count it in SQL), so an ORM assignment of the same value would
+    write nothing."""
+    db.execute(update(DownloadJob).where(DownloadJob.id == job.id).values(status=job_status))
+    db.execute(update(MediaAsset).where(MediaAsset.id == asset.id).values(ref_count=ref_count))
+    db.commit()
+
+
+def _stopped_by(monkeypatch, db, job, asset, job_status: str, ref_count: int) -> None:
+    """The conversion is stopped mid-encode after the route moved the job to `job_status` and its
+    hold on the asset to `ref_count` (cancel and delete release it; pause keeps it)."""
+
+    def _encode(*_a, **_k):
+        _route_moved(db, job, asset, job_status, ref_count)
+        raise pipeline.editmod.EditAborted()
+
+    monkeypatch.setattr(pipeline, "_ensure_phone_safe", _encode)
+
+
+def test_an_upload_cancelled_mid_conversion_is_freed_once_the_conversion_stops(
+    db, make_user, _root, monkeypatch
+):
+    # `release_asset` leaves it while the conversion reads the file; once it stopped, nothing holds it
+    # — kept, it would sit on disk outside every footprint until the GC's upload pass.
+    _probe(monkeypatch)
+    u = make_user()
+    job, asset = _ingest(db, _root, u.id, size=10)
+    original = storage.abs_path(str(_root), asset.rel_path)
+    _claim(db, job, asset)
+    _stopped_by(monkeypatch, db, job, asset, "canceled", 0)
+
+    pipeline._convert_upload(job.id, asset.id)
+
+    assert _row(db, asset.id) is None
+    assert not original.exists()
+    assert quota.footprint(db, u.id) == 0
+
+
+def test_an_upload_paused_mid_conversion_keeps_its_original_for_the_resume(
+    db, make_user, _root, monkeypatch
+):
+    _probe(monkeypatch)
+    u = make_user()
+    job, asset = _ingest(db, _root, u.id, size=10)
+    original = storage.abs_path(str(_root), asset.rel_path)
+    _claim(db, job, asset)
+    _stopped_by(monkeypatch, db, job, asset, "paused", 1)
+
+    pipeline._convert_upload(job.id, asset.id)
+
+    db.expire_all()
+    assert db.get(MediaAsset, asset.id).status == "pending"
+    assert original.exists()
+
+
+def test_an_upload_whose_conversion_finished_after_its_job_was_cancelled_is_freed(
+    db, make_user, _root, monkeypatch
+):
+    _probe(monkeypatch)
+    u = make_user()
+    job, asset = _ingest(db, _root, u.id, size=10)
+    _claim(db, job, asset)
+    _converts_to(monkeypatch, size=5)
+    real_encode = pipeline._ensure_phone_safe
+
+    def _encode_then_cancelled(*a, **k):
+        out = real_encode(*a, **k)
+        _route_moved(db, job, asset, "canceled", 0)
+        return out
+
+    monkeypatch.setattr(pipeline, "_ensure_phone_safe", _encode_then_cancelled)
+    pipeline._convert_upload(job.id, asset.id)
+
+    assert _row(db, asset.id) is None
+    left = _uploads_dir(_root, u.id)
+    assert not left.exists() or list(left.iterdir()) == []
+
+
 def test_the_cache_cap_never_evicts_an_upload(db, make_user, _root, monkeypatch):
     _probe(monkeypatch, vcodec="h264")
     u = make_user()
@@ -301,12 +528,21 @@ def test_the_cache_cap_never_evicts_an_upload(db, make_user, _root, monkeypatch)
 # --- a share link ----------------------------------------------------------------------------
 
 
-@pytest.mark.parametrize("state", ["pending", "downloading"])
-def test_a_link_whose_video_is_being_made_again_says_preparing_not_gone(
-    state, db, as_user, make_user, _root, monkeypatch, upload_headers
+@pytest.mark.parametrize(
+    ("job_status", "asset_status", "reason"),
+    [
+        ("queued", "pending", "video_preparing"),  # a re-download waiting for the worker
+        ("running", "downloading", "video_preparing"),  # being made now
+        ("canceled", "pending", "video_gone"),  # a rebuild cancelled: nothing will finish it
+        ("paused", "pending", "video_gone"),  # held until its owner resumes it, maybe never
+    ],
+)
+def test_a_link_says_preparing_only_while_its_video_is_being_made(
+    job_status, asset_status, reason, db, as_user, make_user, _root, monkeypatch, upload_headers
 ):
     # A link is made on a finished job; its asset can be (re)made under it later — a shared cache row
-    # re-fetched, a conversion retried. Until it is ready the page says "try again shortly".
+    # re-fetched, a conversion retried. While that runs the page says "try again shortly"; once
+    # nothing is making it, it says the video is gone rather than promising it forever.
     _probe(monkeypatch, vcodec="h264")
     c = as_user(make_user())
     job = c.post(
@@ -314,11 +550,12 @@ def test_a_link_whose_video_is_being_made_again_says_preparing_not_gone(
     ).json()
     link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": False}).json()
     assert c.get(f"/api/public/watch/{link['token']}").status_code == 200
-    db.get(MediaAsset, job["asset"]["id"]).status = state
+    db.get(MediaAsset, job["asset"]["id"]).status = asset_status
+    db.get(DownloadJob, job["id"]).status = job_status
     db.commit()
     r = c.get(f"/api/public/watch/{link['token']}")
     assert r.status_code == 404
-    assert r.json()["detail"]["reason"] == "video_preparing"
+    assert r.json()["detail"]["reason"] == reason
 
 
 def test_the_admin_storage_view_counts_a_waiting_upload(db, as_user, make_user, _root, monkeypatch):
@@ -327,7 +564,7 @@ def test_the_admin_storage_view_counts_a_waiting_upload(db, as_user, make_user,
     u = make_user()
     _ingest(db, _root, u.id, size=10)
     r = as_user(make_user(role="admin")).get("/api/admin/downloads/storage").json()
-    assert (r["ready_files"], r["total_bytes"]) == (1, 10)
+    assert (r["files_on_disk"], r["total_bytes"]) == (1, 10)
     assert {p["user_id"]: p["footprint_bytes"] for p in r["per_user"]} == {u.id: 10}
 
 
@@ -345,11 +582,11 @@ def test_the_upload_reap_leaves_one_still_held_or_still_read(
     _probe(monkeypatch)
     job, asset = _ingest(db, _root, make_user().id)
     path = storage.abs_path(str(_root), asset.rel_path)
-    job.status, asset.status, asset.ref_count = job_status, asset_status, ref_count
+    asset.status = asset_status
     db.commit()
+    _route_moved(db, job, asset, job_status, ref_count)
 
     gcmod.run_download_gc(db)
 
-    db.expire_all()
-    assert db.get(MediaAsset, asset.id) is not None
+    assert _row(db, asset.id) is not None
     assert path.exists()