Add live transcode progress with a dedicated now-processing panel

- ffmpeg progress streamed via -progress pipe:1, persisted on the job as a percent
- retry/retry-all now mark the old row as superseded instead of deleting it,
  preserving the failed/cancelled audit trail
- frontend shows the running job in its own panel above the list with a live
  progress bar instead of burying it as just another row

backend/models.py

@@ -366,6 +366,8 @@ class TranscodeJob(BaseModel):
     status: str = "pending"
     error: str = ""
     triggered_by: str = "manual"
+    progress: float = 0.0
+    superseded_by: Optional[str] = None
     created_at: str = Field(default_factory=_now_iso)
     started_at: Optional[str] = None
     finished_at: Optional[str] = None

backend/server.py

@@ -583,11 +583,13 @@ async def _process_job(job: dict):
 
     last_error = {"msg": ""}
 
-    async def cb(status, entry: Optional[str] = None, error: Optional[str] = None):
+    async def cb(status, entry: Optional[str] = None, error: Optional[str] = None, progress: Optional[float] = None):
         update = {"hls_status": status}
         if status == "done" and entry: update["hls_path"] = entry
         await coll.update_one({"id": job["movie_id"]}, {"$set": update})
         if error: last_error["msg"] = error
+        if progress is not None:
+            await db.transcode_queue.update_one({"id": job["id"]}, {"$set": {"progress": progress}})
 
     try:
         if job["quality"] == "abr":
@@ -684,7 +686,7 @@ async def list_queue(status: Optional[str] = None, user: dict = Depends(require_
 @api.get("/transcode/queue/stats")
 async def queue_stats(user: dict = Depends(require_admin)):
     pipeline = [{"$group": {"_id": "$status", "n": {"$sum": 1}}}]
-    out = {"pending": 0, "running": 0, "done": 0, "failed": 0, "cancelled": 0}
+    out = {"pending": 0, "running": 0, "done": 0, "failed": 0, "cancelled": 0, "superseded": 0}
     async for row in db.transcode_queue.aggregate(pipeline):
         out[row["_id"]] = row["n"]
     s = await get_settings()
@@ -715,7 +717,9 @@ async def retry_job(job_id: str, user: dict = Depends(require_admin)):
         raise HTTPException(status_code=400, detail="Only failed/cancelled jobs can be retried")
     new_job = await _enqueue_transcode(job["movie_id"], job["quality"], triggered_by=job.get("triggered_by", "manual"), content_type=job.get("content_type", "movie"))
     if new_job["id"] != job["id"]:
-        await db.transcode_queue.delete_one({"id": job["id"]})
+        # Keep the original row for the audit trail — mark it superseded rather than deleting,
+        # so cancelled/failed history never silently disappears from the list.
+        await db.transcode_queue.update_one({"id": job["id"]}, {"$set": {"status": "superseded", "superseded_by": new_job["id"]}})
     return {"ok": True, "job_id": new_job["id"]}
 
 
@@ -726,7 +730,7 @@ async def retry_all_failed(user: dict = Depends(require_admin)):
     for job in jobs:
         new_job = await _enqueue_transcode(job["movie_id"], job["quality"], triggered_by=job.get("triggered_by", "manual"), content_type=job.get("content_type", "movie"))
         if new_job["id"] != job["id"]:
-            await db.transcode_queue.delete_one({"id": job["id"]})
+            await db.transcode_queue.update_one({"id": job["id"]}, {"$set": {"status": "superseded", "superseded_by": new_job["id"]}})
         retried += 1
     return {"ok": True, "retried": retried}
 

backend/transcode.py

@@ -37,6 +37,23 @@ async def probe_video(source: Path) -> Tuple[int, int]:
         return (0, 0)
 
 
+async def probe_duration(source: Path) -> float:
+    """Return source duration in seconds using ffprobe; 0 on failure."""
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            "ffprobe", "-v", "error", "-show_entries", "format=duration",
+            "-of", "json", str(source),
+            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, _ = await proc.communicate()
+        if proc.returncode != 0:
+            return 0.0
+        data = json.loads(stdout.decode("utf-8", errors="ignore") or "{}")
+        return float((data.get("format") or {}).get("duration") or 0.0)
+    except Exception:
+        return 0.0
+
+
 # (height, video_bitrate, max_bitrate, buffer, audio_bitrate)
 ALL_VARIANTS = [
     (1080, "5000k", "5500k", "7500k", "192k"),
@@ -75,7 +92,7 @@ async def transcode_audio_fix(source: Path, out_dir: Path, on_status: Callable[.
         "-hls_segment_filename", str(out_dir / "seg_%04d.ts"),
         str(out_dir / "playlist.m3u8"),
     ]
-    await _run(cmd, on_status, out_dir, "playlist.m3u8")
+    await _run(cmd, on_status, out_dir, "playlist.m3u8", source)
 
 
 async def transcode_quick(source: Path, out_dir: Path, on_status: Callable[..., Awaitable]) -> None:
@@ -94,7 +111,7 @@ async def transcode_quick(source: Path, out_dir: Path, on_status: Callable[...,
         "-hls_segment_filename", str(out_dir / "seg_%04d.ts"),
         str(out_dir / "playlist.m3u8"),
     ]
-    await _run(cmd, on_status, out_dir, "playlist.m3u8")
+    await _run(cmd, on_status, out_dir, "playlist.m3u8", source)
 
 
 async def transcode_abr(source: Path, out_dir: Path, on_status: Callable[..., Awaitable]) -> None:
@@ -158,23 +175,61 @@ async def transcode_abr(source: Path, out_dir: Path, on_status: Callable[..., Aw
     for i in range(n):
         (out_dir / f"v{i}").mkdir(exist_ok=True)
 
-    await _run(cmd, on_status, out_dir, "master.m3u8")
+    await _run(cmd, on_status, out_dir, "master.m3u8", source)
 
 
-async def _run(cmd: List[str], on_status, out_dir: Path, entry_filename: str) -> None:
-    await on_status("running")
+def _parse_out_time_seconds(line: str) -> Optional[float]:
+    """Parse an `out_time=HH:MM:SS.microseconds` line from ffmpeg's -progress output."""
+    _, _, value = line.partition("=")
+    value = value.strip()
+    if not value or value == "N/A":
+        return None
+    try:
+        h, m, s = value.split(":")
+        return int(h) * 3600 + int(m) * 60 + float(s)
+    except ValueError:
+        return None
+
+
+async def _run(cmd: List[str], on_status, out_dir: Path, entry_filename: str, source: Optional[Path] = None) -> None:
+    duration = await probe_duration(source) if source else 0.0
+    await on_status("running", progress=0.0)
+    # Machine-readable progress on stdout, separate from ffmpeg's normal stderr logging.
+    cmd = [cmd[0], "-progress", "pipe:1", "-nostats"] + cmd[1:]
     try:
         proc = await asyncio.create_subprocess_exec(
             *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
         )
-        _stdout, stderr = await proc.communicate()
+
+        stderr_chunks: List[bytes] = []
+
+        async def _drain_stderr():
+            async for line in proc.stderr:
+                stderr_chunks.append(line)
+
+        stderr_task = asyncio.create_task(_drain_stderr())
+
+        last_reported = -1.0
+        async for raw in proc.stdout:
+            line = raw.decode("utf-8", errors="ignore").strip()
+            if line.startswith("out_time=") and duration > 0:
+                seconds = _parse_out_time_seconds(line)
+                if seconds is not None:
+                    pct = round(min(99.0, seconds / duration * 100), 1)
+                    if pct != last_reported:
+                        last_reported = pct
+                        await on_status("running", progress=pct)
+
+        await proc.wait()
+        await stderr_task
+        stderr = b"".join(stderr_chunks)
         if proc.returncode != 0:
             err = stderr.decode("utf-8", errors="ignore")[-500:]
             logger.error(f"ffmpeg failed: {err}")
             await on_status("failed", error=err[:200])
             shutil.rmtree(out_dir, ignore_errors=True)
             return
-        await on_status("done", entry=entry_filename)
+        await on_status("done", entry=entry_filename, progress=100.0)
     except FileNotFoundError:
         await on_status("failed", error="ffmpeg not installed")
     except Exception as e: