fix(downloads): Retry and Re-download take a cookie added after the failure

The login-wall copy tells the user to add their cookies and press Retry, but
Retry kept the job on the shared asset the worker never attaches a cookie to,
so it failed on the same wall. Resume now moves a pending/errored job onto the
asset enqueue would pick with the cookie (never off a file or a download in
flight).

Re-download salted its fresh asset with the job's previous asset id, which the
cookie gate cannot recompute, so even a cookie download rebuilt without the
cookie. The rebuild salt is now the new asset's own id (plus the user's cookie
salt when one applies), and is_authed_asset recognises it.

backend/app/downloads/service.py

@@ -7,6 +7,7 @@ handles everything else. State transitions used by the routes (pause/resume/canc
 the worker live in `worker.py` / the routes; this module owns enqueue + asset bookkeeping.
 """
 
+import uuid
 from datetime import datetime, timezone
 
 from sqlalchemy import func, select, update
@@ -341,7 +342,11 @@ def pause(db: Session, job: DownloadJob) -> None:
 def resume(db: Session, job: DownloadJob) -> None:
     """Requeue a paused/errored job for a fresh attempt, clearing its error. If the SHARED asset itself
     failed, reset it to pending so the worker actually re-downloads — otherwise the requeued job would
-    instantly re-inherit the asset's stale error (the worker short-circuits an errored asset)."""
+    instantly re-inherit the asset's stale error (the worker short-circuits an errored asset).
+
+    A cookie the user added AFTER the job was queued moves the job onto their private asset first
+    (`_follow_the_cookie`): the worker only ever attaches a cookie to that identity, so without the move
+    the Retry a login-wall message asks for would fail on the same wall."""
     if job.status in (JobStatus.PAUSED, JobStatus.ERROR):
         job.progress = 0
         job.error = None
@@ -349,6 +354,8 @@ def resume(db: Session, job: DownloadJob) -> None:
         job.error_detail = None
         if job.asset_id:
             asset = db.get(MediaAsset, job.asset_id)
+            if asset is not None:
+                asset = _follow_the_cookie(db, job, asset)
             if asset and asset.status == "error":  # MediaAsset.status (no enum), not a JobStatus
                 asset.status = "pending"
                 asset.error = None
@@ -356,6 +363,32 @@ def resume(db: Session, job: DownloadJob) -> None:
         db.commit()
 
 
+def _follow_the_cookie(db: Session, job: DownloadJob, asset: MediaAsset) -> MediaAsset:
+    """Re-point a requeued job at the asset enqueue would pick NOW, when the user has a cookie for the
+    link's host but the job still holds an asset the worker won't attach it to (queued before the
+    cookie existed). Returns the asset the job holds afterwards.
+
+    Only a `pending` / `error` asset moves — neither has a file nor a download in flight. A `ready` one
+    already has the file, and a `downloading` one may be this job's own child still being stopped by
+    the pause; the next Retry after it settles moves the job then. Removing a cookie never moves a job
+    back: the worker attaches nothing without a credential, so the private asset is as good as any."""
+    if asset.status not in ("pending", "error"):
+        return asset
+    url = source_url(job.source_kind, job.source_ref)
+    salt = credentials.salt_for_enqueue(db, job.user_id, url)
+    spec = formats.normalize(job.profile_snapshot)
+    if salt is None or credentials.is_authed_asset(asset, spec, job.user_id, url):
+        return asset
+    target = get_or_create_asset(
+        db, job.source_kind, job.source_ref, formats.format_sig(spec, salt=salt)
+    )
+    populate_from_catalog(db, target)
+    release_asset(db, job)  # drop the old hold BEFORE re-pointing — it keys off job.asset_id
+    incr_ref_count(db, target.id)
+    job.asset_id = target.id
+    return target
+
+
 def cancel(db: Session, job: DownloadJob) -> None:
     """Cancel a job unless it's already done/canceled, releasing its asset hold WHILE it still counts
     as holding — `canceled` ∉ HOLDING, so a later delete won't double-release."""
@@ -374,13 +407,31 @@ def delete(db: Session, job: DownloadJob) -> None:
 
 def redownload(db: Session, job: DownloadJob) -> None:
     """Rebuild a finished/failed job's file into a FRESH, private asset (the cache sig is salted with
-    the job id + current asset id, so it never reuses the dedup pool) and requeue the SAME job row —
+    the new asset's own id, so it never reuses the dedup pool) and requeue the SAME job row —
     so any public share link (DownloadLink -> job_id) keeps resolving to the newly-downloaded file.
     The old hold is dropped first (freeing the old file iff this job was its last holder). The caller
-    has already rejected edit clips + still-in-progress jobs (HTTP 409)."""
+    has already rejected edit clips + still-in-progress jobs (HTTP 409).
+
+    When the user has a cookie for the link's host, the salt carries it (`credentials.rebuild_salt`),
+    so the worker recognises the rebuild as their private identity and attaches the cookie — a cookie
+    download, or one the user added after a failure, rebuilds with the cookie."""
     spec = formats.normalize(job.profile_snapshot)
-    sig = formats.format_sig(spec, salt=f"{job.id}:{job.asset_id}")
-    asset = get_or_create_asset(db, job.source_kind, job.source_ref, sig)
+    cookie = credentials.salt_for_enqueue(
+        db, job.user_id, source_url(job.source_kind, job.source_ref)
+    )
+    # The salt names the row's own id, which exists only once the row does: insert it under a unique
+    # placeholder, then write the real signature in the same transaction (nobody sees the placeholder).
+    asset = MediaAsset(
+        source_kind=job.source_kind,
+        source_ref=job.source_ref,
+        format_sig=f"rebuild:{uuid.uuid4().hex}",
+        status="pending",
+        ref_count=0,
+    )
+    db.add(asset)
+    db.flush()
+    asset.format_sig = formats.format_sig(spec, salt=credentials.rebuild_salt(asset.id, cookie))
+    db.flush()
     populate_from_catalog(db, asset)
     # Drop the old hold BEFORE re-pointing the job — release_asset keys off the job's current asset.
     release_asset(db, job)