From 0ed82fe6b6226cf715628ebbef714515bdab5ee7 Mon Sep 17 00:00:00 2001
From: npeter83 <npeter83+claude@gmail.com>
Date: Wed, 12 Aug 2026 22:39:17 +0200
Subject: [PATCH 1/5] feat(downloads): run each download/edit in a supervised
 child process with a hard-kill watchdog
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Defense-in-depth after the ReDoS incident. The heavy work (yt-dlp download / ffmpeg edit)
now runs in a child process (app.downloads.runner) supervised by the worker thread, not in
the thread itself. A thread can't be killed and a C-level spin can't even be cooperatively
cancelled, so one wedged download used to pin a core, hold the GIL and freeze the whole
queue; a child process can be SIGTERM/SIGKILLed.

The supervisor kills the child (and its process group, so ffmpeg grandchildren die too)
when: a route pauses/cancels the job, it makes no progress for download_stall_timeout_seconds
(150s), it runs past download_wall_clock_max_seconds (2h), the worker is shutting down, or
the child crashes without setting state. On a kill it frees the asset and fails the job with
a retryable network error (stalls are not auto-requeued). A clean exit is left alone — the
child already wrote the terminal state, including a cooperative pause via the progress hook.

- config.py: stall / wall-clock / supervisor-poll / kill-grace settings.
- worker.py: _run_supervised + _spawn_download_child / _job_liveness / _stop_child /
  _cleanup_staging; _process_job claims the asset, then hands the heavy work to the child.
- app/downloads/runner.py: the child entrypoint.
- docker-compose.localdev.yml: init:true on the worker so tini (PID 1) reaps orphaned
  ffmpeg grandchildren left by a hard-kill (the worker reaps its own direct children).
- tests: watchdog decision logic + SIGTERM->SIGKILL escalation (fake proc, no real subprocess).
---
 backend/app/config.py                   |  12 ++
 backend/app/downloads/runner.py         |  51 +++++++
 backend/app/worker.py                   | 174 +++++++++++++++++++++--
 backend/tests/test_worker_supervisor.py | 175 ++++++++++++++++++++++++
 docker-compose.localdev.yml             |   3 +
 5 files changed, 407 insertions(+), 8 deletions(-)
 create mode 100644 backend/app/downloads/runner.py
 create mode 100644 backend/tests/test_worker_supervisor.py

diff --git a/backend/app/config.py b/backend/app/config.py
index 78bb9a6f..e0f2707f 100644
--- a/backend/app/config.py
+++ b/backend/app/config.py
@@ -173,6 +173,18 @@ class Settings(BaseSettings):
     download_worker_concurrency: int = 2
     # How often (empty pending queue) the worker polls for a new job, in seconds.
     download_poll_seconds: int = 5
+    # --- download-subprocess watchdog (each download/edit runs in a supervised child process so a
+    # wedged yt-dlp/ffmpeg can be hard-killed; a thread cannot) ---
+    # Kill a running download if its job row hasn't advanced (no progress written) for this long.
+    # Generous: a live download writes progress ~1/s, so only a genuine 0-progress stall trips it.
+    download_stall_timeout_seconds: int = 150
+    # Absolute ceiling on a single download/edit regardless of progress (a legitimately huge file
+    # still finishes well under this; a runaway is capped).
+    download_wall_clock_max_seconds: int = 7200  # 2 hours
+    # How often the supervisor checks the child's liveness + the job row.
+    download_supervisor_poll_seconds: int = 3
+    # Grace between SIGTERM and SIGKILL when stopping a child (matches run_ffmpeg's escalation).
+    download_kill_grace_seconds: int = 5
     # Total cache size cap across ALL users (bytes). When >0 and exceeded, the GC evicts the
     # least-recently-accessed ready assets until under it. 0 = unlimited (disk is the limit).
     download_total_max_bytes: int = 0
diff --git a/backend/app/downloads/runner.py b/backend/app/downloads/runner.py
new file mode 100644
index 00000000..44e9f64d
--- /dev/null
+++ b/backend/app/downloads/runner.py
@@ -0,0 +1,51 @@
+"""Download/edit child process — runs ONE job's heavy work (yt-dlp download or ffmpeg edit) in its
+own process, launched and supervised by the worker (`app.worker._run_supervised`).
+
+Why a separate process: a worker THREAD can't be killed, a C-level spin (a catastrophic regex, a
+tight yt-dlp retry loop) can't even be cooperatively cancelled, and one such thread starves the
+others via the GIL. Running the work in a child means the supervisor can SIGTERM/SIGKILL it (and its
+process group, so ffmpeg grandchildren die too) on a stall / pause / wall-clock breach.
+
+Contract: the asset is already claimed ('downloading') by the parent. This process does the work and
+writes the terminal job/asset state itself (done / error, or a cooperative pause via the in-flight
+progress hook), then exits 0. An unexpected crash exits non-zero and the parent fails the job.
+"""
+import logging
+import sys
+
+from app.db import SessionLocal
+from app.models import DownloadJob
+
+log = logging.getLogger("siftlode.worker.runner")
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO)
+    if len(sys.argv) != 2:
+        print("usage: python -m app.downloads.runner <job_id>", file=sys.stderr)
+        raise SystemExit(2)
+    job_id = int(sys.argv[1])
+
+    with SessionLocal() as db:
+        job = db.get(DownloadJob, job_id)
+        if job is None:
+            return
+        asset_id = job.asset_id
+        job_kind = job.job_kind
+        spec = dict(job.profile_snapshot or {})
+        source_kind, source_ref = job.source_kind, job.source_ref
+    if asset_id is None:
+        return
+
+    # Imported here (not at module top) so the child pays the yt-dlp import cost only when it has
+    # real work, and to keep the import graph obvious: runner is a thin entry over worker's helpers.
+    from app.worker import _download, _process_edit
+
+    if job_kind == "edit":
+        _process_edit(job_id, asset_id)
+    else:
+        _download(job_id, asset_id, source_kind, source_ref, spec)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/backend/app/worker.py b/backend/app/worker.py
index 0364703e..57dce801 100644
--- a/backend/app/worker.py
+++ b/backend/app/worker.py
@@ -16,9 +16,12 @@ Design:
   * Crash-safe: on startup, orphaned `running` jobs are requeued and `downloading` assets reset.
 """
 import logging
+import os
 import re
 import shutil
 import signal
+import subprocess
+import sys
 import threading
 import time
 from datetime import date, datetime, timedelta, timezone
@@ -205,19 +208,176 @@ def _claim_asset(asset_id: int) -> bool:
         return row is not None
 
 
+# --- supervised child execution + watchdog --------------------------------------------------
+# The heavy work runs in a child process so a wedged yt-dlp/ffmpeg is HARD-killable. A worker thread
+# can't be killed, a C-level spin (e.g. a catastrophic regex) can't even be cooperatively cancelled,
+# and one such thread starves the rest via the GIL — the incident that motivated this. The parent
+# thread supervises: it kills the child (and its process group, so ffmpeg grandchildren die too) on
+# a route pause/cancel, on a no-progress stall, or past an absolute wall-clock cap.
+
+_CHILD_OK = 0
+
+
+def _spawn_download_child(job_id: int) -> subprocess.Popen:
+    """Start the child that does the actual download/edit for `job_id`, in its OWN session/process
+    group so stopping it also stops any ffmpeg / helper processes yt-dlp spawned."""
+    return subprocess.Popen(
+        [sys.executable, "-m", "app.downloads.runner", str(job_id)],
+        start_new_session=True,
+    )
+
+
+def _job_liveness(job_id: int) -> tuple[str | None, float | None]:
+    """(status, seconds-since-last-row-update) for the job, computed with the DB clock so there is
+    no parent↔DB skew. A live download rewrites its row ~1/s; a stalled one never does. Returns
+    (None, None) if the row is gone."""
+    with SessionLocal() as db:
+        row = db.execute(
+            text(
+                "SELECT status, EXTRACT(EPOCH FROM (now() - updated_at)) "
+                "FROM download_jobs WHERE id = :id"
+            ),
+            {"id": job_id},
+        ).one_or_none()
+    if row is None:
+        return None, None
+    return row[0], (float(row[1]) if row[1] is not None else None)
+
+
+def _signal_group(proc: subprocess.Popen, pgid: int | None, sig: int) -> None:
+    """Signal the child's whole process group (falling back to the child alone), swallowing the race
+    where it has already exited."""
+    try:
+        if pgid is not None:
+            os.killpg(pgid, sig)
+        else:
+            proc.send_signal(sig)
+    except (ProcessLookupError, OSError):
+        pass
+
+
+def _stop_child(proc: subprocess.Popen, grace_s: float) -> None:
+    """SIGTERM the child's process group, then SIGKILL if it outlives the grace period — the same
+    terminate→wait→kill escalation as run_ffmpeg, so a child that ignores SIGTERM is still reaped."""
+    if proc.poll() is not None:
+        return
+    try:
+        pgid: int | None = os.getpgid(proc.pid)
+    except OSError:
+        pgid = None
+    _signal_group(proc, pgid, signal.SIGTERM)
+    try:
+        proc.wait(timeout=grace_s)
+        return
+    except subprocess.TimeoutExpired:
+        pass
+    _signal_group(proc, pgid, signal.SIGKILL)
+    try:
+        proc.wait(timeout=grace_s)
+    except subprocess.TimeoutExpired:
+        log.error("download child pid %s survived SIGKILL", proc.pid)
+
+
+def _run_supervised(job_id: int, asset_id: int) -> None:
+    """Run the heavy download/edit for an already-claimed asset in a child process and watchdog it.
+
+    A clean child exit means the child already wrote the terminal job/asset state (done/error, or a
+    cooperative pause via the in-child progress hook) — we leave it. We step in only when the child
+    can't finish or self-abort: a route paused/cancelled it (kill, free the asset, keep the route's
+    status), it made no progress for `stall_timeout` or ran past `wall_clock_max` (kill, fail the
+    job), the worker is shutting down (kill, requeue), or it crashed without setting state."""
+    poll_s = max(1, settings.download_supervisor_poll_seconds)
+    stall_s = settings.download_stall_timeout_seconds
+    wall_s = settings.download_wall_clock_max_seconds
+    grace_s = settings.download_kill_grace_seconds
+
+    try:
+        proc = _spawn_download_child(job_id)
+    except Exception:  # noqa: BLE001 — spawn failed; don't leave the asset stuck 'downloading'
+        log.exception("failed to spawn download child for job %s", job_id)
+        _reset_asset_pending(asset_id)
+        _requeue_if_running(job_id, None)
+        return
+
+    started = time.monotonic()
+    reason: str | None = None
+    while True:
+        try:
+            proc.wait(timeout=poll_s)
+            break  # child finished on its own
+        except subprocess.TimeoutExpired:
+            pass
+        status, stale = _job_liveness(job_id)
+        if status is None or status != "running":
+            reason = "cancelled"  # a route paused/cancelled it (or the row vanished)
+            break
+        if stale is not None and stale > stall_s:
+            reason = "stalled"
+            break
+        if time.monotonic() - started > wall_s:
+            reason = "timeout"
+            break
+        if _stop.is_set():
+            reason = "shutdown"
+            break
+
+    if reason is None:
+        # Child exited. If it died WITHOUT setting a terminal state (non-zero exit / crash / OOM),
+        # the job is still 'running' — fail it here so nothing is left stuck.
+        if proc.returncode != _CHILD_OK:
+            status, _ = _job_liveness(job_id)
+            if status == "running":
+                _reset_asset_pending(asset_id)
+                _cleanup_staging(asset_id)
+                _set_job(
+                    job_id, status="error", error_code="unknown", phase=None,
+                    error=f"The download process exited unexpectedly (code {proc.returncode}).",
+                )
+                log.warning("job %s: child exited %s without finishing", job_id, proc.returncode)
+        return
+
+    _stop_child(proc, grace_s)
+    _reset_asset_pending(asset_id)
+    _cleanup_staging(asset_id)
+    if reason == "cancelled":
+        # The route already set paused/cancelled (or the job is gone); just free the asset.
+        log.info("job %s: child stopped (route pause/cancel)", job_id)
+        return
+    if reason == "shutdown":
+        _requeue_if_running(job_id, None)  # resume on the next boot
+        return
+    msg = (
+        f"The download stalled (no progress for {int(stall_s)}s) and was stopped."
+        if reason == "stalled"
+        else f"The download exceeded the {int(wall_s)}s time limit and was stopped."
+    )
+    _set_job(job_id, status="error", error=msg, error_code="network", phase=None)
+    log.warning("job %s: stopped by watchdog (%s)", job_id, reason)
+
+
+def _cleanup_staging(asset_id: int) -> None:
+    """Best-effort removal of a killed child's leftover staging dirs (its own `finally` never ran).
+    The next attempt on this asset also wipes staging on start, so this is just tidiness/disk."""
+    for path in (_staging_dir(asset_id), _edit_staging_dir(asset_id)):
+        shutil.rmtree(path, ignore_errors=True)
+
+
 # --- processing -----------------------------------------------------------------------------
 
 def _process_job(job_id: int) -> None:
+    """Dispatch a claimed job. The trivial outcomes (cache hit, dead/contended asset) are handled
+    inline and cheaply; the actual heavy work (yt-dlp download / ffmpeg edit) is claimed here and
+    then run in a supervised CHILD PROCESS (`_run_supervised`) so a wedged yt-dlp/ffmpeg can be
+    hard-killed — a worker THREAD cannot be killed, nor (when it spins in a C call) even
+    cooperatively cancelled, and one such thread starves the others via the GIL."""
     with SessionLocal() as db:
         job = db.get(DownloadJob, job_id)
         if job is None:
             return
         asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
-        spec = dict(job.profile_snapshot or {})
-        source_kind, source_ref = job.source_kind, job.source_ref
-        job_kind = job.job_kind
         asset_id = asset.id if asset else None
         asset_status = asset.status if asset else None
+        asset_error = asset.error if asset else None
 
     if asset is None:
         _set_job(job_id, status="error", error="No media asset", phase=None)
@@ -227,7 +387,7 @@ def _process_job(job_id: int) -> None:
         _finish_from_cache(job_id, asset_id)
         return
     if asset_status == "error":
-        _set_job(job_id, status="error", error=asset.error or "Download failed", phase=None)
+        _set_job(job_id, status="error", error=asset_error or "Download failed", phase=None)
         return
     if asset_status == "downloading":
         # Another worker owns this asset; wait and let the job be reclaimed later.
@@ -241,10 +401,8 @@ def _process_job(job_id: int) -> None:
         time.sleep(2)
         return
 
-    if job_kind == "edit":
-        _process_edit(job_id, asset_id)
-    else:
-        _download(job_id, asset_id, source_kind, source_ref, spec)
+    # The asset is now ours ('downloading'): run the heavy work in a supervised child process.
+    _run_supervised(job_id, asset_id)
 
 
 def _finish_from_cache(job_id: int, asset_id: int) -> None:
diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py
new file mode 100644
index 00000000..ce8373c6
--- /dev/null
+++ b/backend/tests/test_worker_supervisor.py
@@ -0,0 +1,175 @@
+"""The download supervisor/watchdog: `_run_supervised` decides when to hard-kill a child, and
+`_stop_child` escalates SIGTERM→SIGKILL. No real subprocess or DB here — a fake proc stands in and
+the DB-liveness lookup + side effects are monkeypatched, so the decision logic is exercised on any
+machine. These are regression tests for the class of bug that motivated the whole change: a wedged
+download that a thread could neither kill nor cancel.
+"""
+import signal
+import subprocess
+
+import pytest
+
+from app import worker
+
+
+class _FakeProc:
+    """Stands in for the child Popen. `waits` scripts each wait() call: "timeout" = still alive,
+    ("exit", rc) = the child finished with that code."""
+
+    def __init__(self, waits, pid=4242):
+        self._waits = list(waits)
+        self.pid = pid
+        self.returncode = None
+        self.signals = []
+
+    def wait(self, timeout=None):
+        step = self._waits.pop(0) if self._waits else "timeout"
+        if step == "timeout":
+            raise subprocess.TimeoutExpired("child", timeout or 0)
+        self.returncode = step[1]
+        return step[1]
+
+    def poll(self):
+        return self.returncode
+
+    def send_signal(self, sig):
+        self.signals.append(sig)
+
+
+@pytest.fixture
+def rec(monkeypatch):
+    """Record supervisor side effects and stub out the real spawn/kill/liveness/staging so a test
+    drives them. `calls` collects (name, kwargs) so assertions read the decisions, not the plumbing."""
+    calls = []
+
+    def record(name):
+        def _f(*args, **kwargs):
+            calls.append((name, args, kwargs))
+        return _f
+
+    monkeypatch.setattr(worker, "_reset_asset_pending", record("reset_asset"))
+    monkeypatch.setattr(worker, "_requeue_if_running", record("requeue"))
+    monkeypatch.setattr(worker, "_set_job", record("set_job"))
+    monkeypatch.setattr(worker, "_cleanup_staging", record("cleanup"))
+    monkeypatch.setattr(worker, "_stop_child", record("stop_child"))
+    monkeypatch.setattr(worker._stop, "is_set", lambda: False)
+    return calls
+
+
+def _names(calls):
+    return [c[0] for c in calls]
+
+
+def _spawn(monkeypatch, proc):
+    monkeypatch.setattr(worker, "_spawn_download_child", lambda job_id: proc)
+
+
+def _liveness(monkeypatch, value):
+    monkeypatch.setattr(worker, "_job_liveness", lambda job_id: value)
+
+
+def test_clean_child_exit_leaves_the_outcome_alone(monkeypatch, rec):
+    # Child finished on its own (exit 0) — it wrote the terminal state; the supervisor must not kill,
+    # reset, or re-fail anything.
+    _spawn(monkeypatch, _FakeProc([("exit", 0)]))
+    worker._run_supervised(1, 99)
+    assert _names(rec) == []
+
+
+def test_no_progress_stall_kills_and_fails_the_job(monkeypatch, rec):
+    _spawn(monkeypatch, _FakeProc(["timeout"]))
+    _liveness(monkeypatch, ("running", 9999.0))  # far past the stall timeout
+    worker._run_supervised(1, 99)
+    names = _names(rec)
+    assert "stop_child" in names and "reset_asset" in names
+    set_job = next(c for c in rec if c[0] == "set_job")
+    assert set_job[2]["status"] == "error" and set_job[2]["error_code"] == "network"
+    assert "stalled" in set_job[2]["error"].lower()
+
+
+def test_route_pause_kills_but_keeps_the_route_status(monkeypatch, rec):
+    # A route flipped the job to 'paused'/'cancelled'. Kill + free the asset, but do NOT overwrite the
+    # status with an error and do NOT requeue.
+    _spawn(monkeypatch, _FakeProc(["timeout"]))
+    _liveness(monkeypatch, ("paused", 1.0))
+    worker._run_supervised(1, 99)
+    names = _names(rec)
+    assert "stop_child" in names and "reset_asset" in names
+    assert "set_job" not in names and "requeue" not in names
+
+
+def test_wall_clock_cap_kills_even_while_progressing(monkeypatch, rec):
+    monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0)
+    _spawn(monkeypatch, _FakeProc(["timeout"]))
+    _liveness(monkeypatch, ("running", 1.0))  # fresh progress, but over the absolute cap
+    worker._run_supervised(1, 99)
+    set_job = next(c for c in rec if c[0] == "set_job")
+    assert set_job[2]["status"] == "error"
+    assert "time limit" in set_job[2]["error"].lower()
+
+
+def test_child_crash_without_finishing_is_failed(monkeypatch, rec):
+    # Non-zero exit and the job is still 'running' → the child died (crash/OOM) before writing state.
+    _spawn(monkeypatch, _FakeProc([("exit", 1)]))
+    _liveness(monkeypatch, ("running", 1.0))
+    worker._run_supervised(1, 99)
+    names = _names(rec)
+    assert "reset_asset" in names
+    set_job = next(c for c in rec if c[0] == "set_job")
+    assert set_job[2]["status"] == "error" and "unexpectedly" in set_job[2]["error"].lower()
+
+
+def test_child_nonzero_exit_that_already_set_state_is_left_alone(monkeypatch, rec):
+    # Child exited non-zero but the job is no longer 'running' (it recorded its own error) → no double.
+    _spawn(monkeypatch, _FakeProc([("exit", 1)]))
+    _liveness(monkeypatch, ("error", None))
+    worker._run_supervised(1, 99)
+    assert "set_job" not in _names(rec)
+
+
+def test_spawn_failure_frees_the_asset_and_requeues(monkeypatch, rec):
+    def boom(job_id):
+        raise OSError("no fork")
+
+    monkeypatch.setattr(worker, "_spawn_download_child", boom)
+    worker._run_supervised(1, 99)
+    names = _names(rec)
+    assert "reset_asset" in names and "requeue" in names
+
+
+def test_shutdown_kills_and_requeues_for_next_boot(monkeypatch, rec):
+    monkeypatch.setattr(worker._stop, "is_set", lambda: True)
+    _spawn(monkeypatch, _FakeProc(["timeout"]))
+    _liveness(monkeypatch, ("running", 1.0))
+    worker._run_supervised(1, 99)
+    names = _names(rec)
+    assert "stop_child" in names and "requeue" in names and "set_job" not in names
+
+
+# --- _stop_child escalation -----------------------------------------------------------------
+
+@pytest.fixture
+def killpg_rec(monkeypatch):
+    sent = []
+    monkeypatch.setattr(worker.os, "getpgid", lambda pid: pid)
+    monkeypatch.setattr(worker.os, "killpg", lambda pgid, sig: sent.append(sig))
+    return sent
+
+
+def test_stop_child_sigterm_is_enough(killpg_rec):
+    proc = _FakeProc([("exit", 0)])  # dies on the SIGTERM wait
+    worker._stop_child(proc, grace_s=0.01)
+    assert killpg_rec == [signal.SIGTERM]
+
+
+def test_stop_child_escalates_to_sigkill(killpg_rec):
+    proc = _FakeProc(["timeout", ("exit", -9)])  # survives SIGTERM, dies on SIGKILL
+    worker._stop_child(proc, grace_s=0.01)
+    assert killpg_rec == [signal.SIGTERM, signal.SIGKILL]
+
+
+def test_stop_child_noop_if_already_exited(killpg_rec):
+    proc = _FakeProc([])
+    proc.returncode = 0  # already gone
+    worker._stop_child(proc, grace_s=0.01)
+    assert killpg_rec == []
diff --git a/docker-compose.localdev.yml b/docker-compose.localdev.yml
index 988afd6f..48560639 100644
--- a/docker-compose.localdev.yml
+++ b/docker-compose.localdev.yml
@@ -76,6 +76,9 @@ services:
         GIT_SHA: ${GIT_SHA:-unknown}
         BUILD_DATE: ${BUILD_DATE:-}
     command: ["python", "-m", "app.worker"]
+    # tini as PID 1: reaps orphaned ffmpeg/yt-dlp grandchildren when the watchdog hard-kills a
+    # download child's process group (the worker reaps its own direct children via subprocess.wait).
+    init: true
     env_file: .env
     environment:
       DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-siftlode}:${POSTGRES_PASSWORD:-siftlode}@db:5432/${POSTGRES_DB:-siftlode}

From 657224c1ad861c7c3988be9c802b79b6160ef8b3 Mon Sep 17 00:00:00 2001
From: npeter83 <npeter83+claude@gmail.com>
Date: Wed, 12 Aug 2026 23:04:48 +0200
Subject: [PATCH 2/5] refactor(downloads): address /code-review on the download
 supervisor
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- worker.py: wrap the watchdog loop so a failure in it (e.g. a transient DB error in
  _job_liveness) still hard-kills the child + fails the job, instead of leaving the child
  running unsupervised — the exact wedge the feature prevents (#1).
- worker.py: _touch_job heartbeat bumps updated_at at each extraction attempt, so a slow /
  multi-client-set fallback extraction (which writes no progress, and re-writes identical
  values so ORM onupdate does not fire) is not false-killed as "stalled" (#2).
- edit.py: extract escalate_kill (terminate->wait->kill escalation); _stop and worker's
  _stop_child (process-group variant) now share it (#3).
- worker.py: document the edit path's second, tighter run_ffmpeg watchdog and that the
  supervisor is POSIX/Linux-container-only (#4, #6).
- tests: a long download stays alive while progress is fresh, and a watchdog-loop error
  still stops the child (#1, #5).
---
 backend/app/downloads/edit.py           | 30 +++++---
 backend/app/worker.py                   | 94 ++++++++++++++++---------
 backend/tests/test_worker_supervisor.py | 26 +++++++
 3 files changed, 108 insertions(+), 42 deletions(-)

diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py
index ed671a47..45337dad 100644
--- a/backend/app/downloads/edit.py
+++ b/backend/app/downloads/edit.py
@@ -288,20 +288,30 @@ def _drain_tail(stream, sink: list[str]) -> None:
         pass
 
 
-def _stop(proc: subprocess.Popen) -> None:
-    """Terminate, then kill — never wait unbounded on a process we've given up on. Safe to call on
-    a process that already exited (the single exit path below may reach it after a clean wait)."""
+def escalate_kill(proc: subprocess.Popen, terminate, kill, grace_s: float, label: str) -> None:
+    """terminate → wait(grace) → kill → wait(grace) escalation for a process we've given up on, so it
+    is never waited on unboundedly. `terminate`/`kill` deliver the stop to whatever scope the caller
+    owns — a single process (`proc.terminate`/`proc.kill`) or its whole group (`os.killpg`). No-op if
+    the process already exited."""
     if proc.poll() is not None:
         return
-    proc.terminate()
+    terminate()
     try:
-        proc.wait(timeout=5)
+        proc.wait(timeout=grace_s)
+        return
     except subprocess.TimeoutExpired:
-        proc.kill()
-        try:
-            proc.wait(timeout=5)
-        except subprocess.TimeoutExpired:
-            log.error("ffmpeg pid %s survived SIGKILL", proc.pid)
+        pass
+    kill()
+    try:
+        proc.wait(timeout=grace_s)
+    except subprocess.TimeoutExpired:
+        log.error("%s pid %s survived SIGKILL", label, proc.pid)
+
+
+def _stop(proc: subprocess.Popen) -> None:
+    """Terminate, then kill — never wait unbounded on a process we've given up on. Safe to call on
+    a process that already exited."""
+    escalate_kill(proc, proc.terminate, proc.kill, grace_s=5, label="ffmpeg")
 
 
 def run_ffmpeg(cmd, total_s, on_progress, should_cancel, stall_timeout_s=_FFMPEG_STALL_TIMEOUT_S) -> None:
diff --git a/backend/app/worker.py b/backend/app/worker.py
index 57dce801..4d3cef68 100644
--- a/backend/app/worker.py
+++ b/backend/app/worker.py
@@ -96,6 +96,20 @@ def _job_status(job_id: int) -> str | None:
         ).scalar()
 
 
+def _touch_job(job_id: int) -> None:
+    """Bump a running job's `updated_at` without changing any value — a stall-watchdog heartbeat for
+    phases that legitimately write no progress. Extraction writes nothing, and a multi-client-set
+    fallback re-writes identical values (progress 0 / phase 'downloading'), so ORM `onupdate` would
+    NOT fire and two slow extraction attempts could otherwise cross the stall timeout and be killed
+    mid-extraction though they'd have succeeded."""
+    with SessionLocal() as db:
+        db.execute(
+            text("UPDATE download_jobs SET updated_at=now() WHERE id=:id AND status='running'"),
+            {"id": job_id},
+        )
+        db.commit()
+
+
 def _requeue_if_running(job_id: int, phase: str) -> None:
     """Put the job back to 'queued' for a later retry, but ONLY while it is still 'running'. The
     route flips status to paused/cancelled to stop a job; a blind `status='queued'` here would
@@ -214,6 +228,12 @@ def _claim_asset(asset_id: int) -> bool:
 # and one such thread starves the rest via the GIL — the incident that motivated this. The parent
 # thread supervises: it kills the child (and its process group, so ffmpeg grandchildren die too) on
 # a route pause/cancel, on a no-progress stall, or past an absolute wall-clock cap.
+#
+# POSIX-only: the kill path uses process groups (start_new_session / os.killpg / os.getpgid), so the
+# worker is Linux-container-only — never run `python -m app.worker` natively on Windows.
+# Edits ALSO carry a second, tighter watchdog inside run_ffmpeg (a per-ffmpeg stall timeout with a
+# cleaner "ffmpeg stalled" error); this process-level supervisor is the uniform backstop for both the
+# download and edit paths, and the last line of defence for anything run_ffmpeg can't see.
 
 _CHILD_OK = 0
 
@@ -257,25 +277,22 @@ def _signal_group(proc: subprocess.Popen, pgid: int | None, sig: int) -> None:
 
 
 def _stop_child(proc: subprocess.Popen, grace_s: float) -> None:
-    """SIGTERM the child's process group, then SIGKILL if it outlives the grace period — the same
-    terminate→wait→kill escalation as run_ffmpeg, so a child that ignores SIGTERM is still reaped."""
+    """SIGTERM the child's process GROUP, then SIGKILL if it outlives the grace period. Shares the
+    terminate→wait→kill escalation with run_ffmpeg (`editmod.escalate_kill`); the only difference is
+    that we signal the whole group so ffmpeg grandchildren die too."""
     if proc.poll() is not None:
         return
     try:
         pgid: int | None = os.getpgid(proc.pid)
     except OSError:
         pgid = None
-    _signal_group(proc, pgid, signal.SIGTERM)
-    try:
-        proc.wait(timeout=grace_s)
-        return
-    except subprocess.TimeoutExpired:
-        pass
-    _signal_group(proc, pgid, signal.SIGKILL)
-    try:
-        proc.wait(timeout=grace_s)
-    except subprocess.TimeoutExpired:
-        log.error("download child pid %s survived SIGKILL", proc.pid)
+    editmod.escalate_kill(
+        proc,
+        terminate=lambda: _signal_group(proc, pgid, signal.SIGTERM),
+        kill=lambda: _signal_group(proc, pgid, signal.SIGKILL),
+        grace_s=grace_s,
+        label="download child",
+    )
 
 
 def _run_supervised(job_id: int, asset_id: int) -> None:
@@ -301,25 +318,36 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
 
     started = time.monotonic()
     reason: str | None = None
-    while True:
-        try:
-            proc.wait(timeout=poll_s)
-            break  # child finished on its own
-        except subprocess.TimeoutExpired:
-            pass
-        status, stale = _job_liveness(job_id)
-        if status is None or status != "running":
-            reason = "cancelled"  # a route paused/cancelled it (or the row vanished)
-            break
-        if stale is not None and stale > stall_s:
-            reason = "stalled"
-            break
-        if time.monotonic() - started > wall_s:
-            reason = "timeout"
-            break
-        if _stop.is_set():
-            reason = "shutdown"
-            break
+    try:
+        while True:
+            try:
+                proc.wait(timeout=poll_s)
+                break  # child finished on its own
+            except subprocess.TimeoutExpired:
+                pass
+            status, stale = _job_liveness(job_id)
+            if status is None or status != "running":
+                reason = "cancelled"  # a route paused/cancelled it (or the row vanished)
+                break
+            if stale is not None and stale > stall_s:
+                reason = "stalled"
+                break
+            if time.monotonic() - started > wall_s:
+                reason = "timeout"
+                break
+            if _stop.is_set():
+                reason = "shutdown"
+                break
+    except Exception:  # noqa: BLE001 — a watchdog error must NEVER leave the child running unsupervised
+        log.exception("supervisor loop failed for job %s; stopping the child", job_id)
+        _stop_child(proc, grace_s)
+        _reset_asset_pending(asset_id)
+        _cleanup_staging(asset_id)
+        _set_job(
+            job_id, status="error", error_code="unknown", phase=None,
+            error="The download was stopped after a worker error.",
+        )
+        return
 
     if reason is None:
         # Child exited. If it died WITHOUT setting a terminal state (non-zero exit / crash / OOM),
@@ -682,6 +710,8 @@ def _extract_with_fallback(job_id: int, asset_id: int, url: str, spec: dict, sta
             settings.download_pot_base_url, clients,
         )
         opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
+        _touch_job(job_id)  # stall-watchdog heartbeat: extraction writes no progress, so a slow /
+        # multi-client-set fallback would otherwise cross the stall timeout with a stale updated_at.
         try:
             with yt_dlp.YoutubeDL(opts) as ydl:
                 return ydl.extract_info(url, download=True)
diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py
index ce8373c6..ca80d82b 100644
--- a/backend/tests/test_worker_supervisor.py
+++ b/backend/tests/test_worker_supervisor.py
@@ -146,6 +146,32 @@ def test_shutdown_kills_and_requeues_for_next_boot(monkeypatch, rec):
     assert "stop_child" in names and "requeue" in names and "set_job" not in names
 
 
+def test_a_long_download_survives_while_progress_stays_fresh(monkeypatch, rec):
+    # A download that outlives the stall timeout must NOT be killed as long as it keeps writing
+    # progress (updated_at stays fresh via the model's onupdate). Model: alive across several polls,
+    # liveness always fresh, then a clean exit — nothing killed, nothing failed.
+    monkeypatch.setattr(worker.settings, "download_stall_timeout_seconds", 10)
+    _spawn(monkeypatch, _FakeProc(["timeout", "timeout", "timeout", ("exit", 0)]))
+    _liveness(monkeypatch, ("running", 2.0))  # always well under the 10s stall
+    worker._run_supervised(1, 99)
+    assert _names(rec) == []
+
+
+def test_a_watchdog_error_still_stops_the_child(monkeypatch, rec):
+    # If the liveness probe raises (e.g. a transient DB error), the child must NOT be left running
+    # unsupervised — the feature's whole point. It is killed and the job is failed.
+    _spawn(monkeypatch, _FakeProc(["timeout"]))
+
+    def boom(job_id):
+        raise RuntimeError("db down")
+
+    monkeypatch.setattr(worker, "_job_liveness", boom)
+    worker._run_supervised(1, 99)
+    names = _names(rec)
+    assert "stop_child" in names and "reset_asset" in names
+    assert next(c for c in rec if c[0] == "set_job")[2]["status"] == "error"
+
+
 # --- _stop_child escalation -----------------------------------------------------------------
 
 @pytest.fixture

From b42b340594f81e10ab4a08b1010dd5667c29fee5 Mon Sep 17 00:00:00 2001
From: npeter83 <npeter83+claude@gmail.com>
Date: Wed, 12 Aug 2026 23:15:41 +0200
Subject: [PATCH 3/5] fix(downloads): don't let the supervisor kill a finishing
 child (review round 2)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- worker.py: the watchdog loop now treats only 'paused'/'cancelled' (or a vanished row) as a
  stop; a terminal status (done/error, or a re-queued contended asset) means the child already
  wrote its outcome and is exiting, so keep polling instead of killing it. Previously a
  proc.wait timeout landing between the child's terminal commit and its exit was misread as a
  route-cancel and SIGKILLed a successful download (dropping its completion notification). The
  stall/wall-clock/shutdown checks are now gated on status=='running' (#1, #5 test).
- worker.py: guard the crash-path _job_liveness probe so a DB blip there can't propagate (the
  child is already dead; boot recovery is the backstop) (#2).
- worker.py: drop _stop_child's redundant poll guard — escalate_kill no-ops on an exited proc
  and getpgid yields None for a reaped pid (#3).
- worker.py: note the extraction heartbeat is per-attempt (a single over-timeout attempt is
  still killed, bounded by socket_timeout) (#4).
---
 backend/app/worker.py                   | 58 +++++++++++++++----------
 backend/tests/test_worker_supervisor.py | 10 +++++
 2 files changed, 44 insertions(+), 24 deletions(-)

diff --git a/backend/app/worker.py b/backend/app/worker.py
index 4d3cef68..3cb9397d 100644
--- a/backend/app/worker.py
+++ b/backend/app/worker.py
@@ -279,9 +279,8 @@ def _signal_group(proc: subprocess.Popen, pgid: int | None, sig: int) -> None:
 def _stop_child(proc: subprocess.Popen, grace_s: float) -> None:
     """SIGTERM the child's process GROUP, then SIGKILL if it outlives the grace period. Shares the
     terminate→wait→kill escalation with run_ffmpeg (`editmod.escalate_kill`); the only difference is
-    that we signal the whole group so ffmpeg grandchildren die too."""
-    if proc.poll() is not None:
-        return
+    that we signal the whole group so ffmpeg grandchildren die too. `escalate_kill` no-ops if the
+    child already exited, and `getpgid` yields None for a reaped pid, so no separate poll guard."""
     try:
         pgid: int | None = os.getpgid(proc.pid)
     except OSError:
@@ -326,18 +325,23 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             except subprocess.TimeoutExpired:
                 pass
             status, stale = _job_liveness(job_id)
-            if status is None or status != "running":
-                reason = "cancelled"  # a route paused/cancelled it (or the row vanished)
-                break
-            if stale is not None and stale > stall_s:
-                reason = "stalled"
-                break
-            if time.monotonic() - started > wall_s:
-                reason = "timeout"
-                break
-            if _stop.is_set():
-                reason = "shutdown"
+            if status == "running":
+                # Only a genuinely-running child is subject to the stall / wall-clock / shutdown kill.
+                if stale is not None and stale > stall_s:
+                    reason = "stalled"
+                    break
+                if time.monotonic() - started > wall_s:
+                    reason = "timeout"
+                    break
+                if _stop.is_set():
+                    reason = "shutdown"
+                    break
+            elif status is None or status in ("paused", "cancelled"):
+                reason = "cancelled"  # a route stopped it (or the row vanished)
                 break
+            # else: a terminal status (done / error, or a re-queued contended asset) means the child
+            # already wrote its outcome and is exiting on its own — keep polling; the next proc.wait
+            # catches the exit. NEVER kill a finishing child (that could drop its completion notice).
     except Exception:  # noqa: BLE001 — a watchdog error must NEVER leave the child running unsupervised
         log.exception("supervisor loop failed for job %s; stopping the child", job_id)
         _stop_child(proc, grace_s)
@@ -351,17 +355,21 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
 
     if reason is None:
         # Child exited. If it died WITHOUT setting a terminal state (non-zero exit / crash / OOM),
-        # the job is still 'running' — fail it here so nothing is left stuck.
+        # the job is still 'running' — fail it here so nothing is left stuck. Guarded: the child is
+        # already dead (no orphan risk), so a DB blip in this probe must not propagate.
         if proc.returncode != _CHILD_OK:
-            status, _ = _job_liveness(job_id)
-            if status == "running":
-                _reset_asset_pending(asset_id)
-                _cleanup_staging(asset_id)
-                _set_job(
-                    job_id, status="error", error_code="unknown", phase=None,
-                    error=f"The download process exited unexpectedly (code {proc.returncode}).",
-                )
-                log.warning("job %s: child exited %s without finishing", job_id, proc.returncode)
+            try:
+                status, _ = _job_liveness(job_id)
+                if status == "running":
+                    _reset_asset_pending(asset_id)
+                    _cleanup_staging(asset_id)
+                    _set_job(
+                        job_id, status="error", error_code="unknown", phase=None,
+                        error=f"The download process exited unexpectedly (code {proc.returncode}).",
+                    )
+                    log.warning("job %s: child exited %s without finishing", job_id, proc.returncode)
+            except Exception:  # noqa: BLE001 — boot recovery is the backstop for a stuck 'running' job
+                log.exception("post-exit cleanup failed for job %s", job_id)
         return
 
     _stop_child(proc, grace_s)
@@ -712,6 +720,8 @@ def _extract_with_fallback(job_id: int, asset_id: int, url: str, spec: dict, sta
         opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
         _touch_job(job_id)  # stall-watchdog heartbeat: extraction writes no progress, so a slow /
         # multi-client-set fallback would otherwise cross the stall timeout with a stale updated_at.
+        # (Per-ATTEMPT only; a single extract_info that itself runs past the stall timeout is still
+        # killed — acceptable, as socket_timeout + extractor_retries bound one attempt well under it.)
         try:
             with yt_dlp.YoutubeDL(opts) as ydl:
                 return ydl.extract_info(url, download=True)
diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py
index ca80d82b..3a17d5a6 100644
--- a/backend/tests/test_worker_supervisor.py
+++ b/backend/tests/test_worker_supervisor.py
@@ -157,6 +157,16 @@ def test_a_long_download_survives_while_progress_stays_fresh(monkeypatch, rec):
     assert _names(rec) == []
 
 
+def test_a_finishing_child_is_not_mistaken_for_a_cancel(monkeypatch, rec):
+    # The child committed its terminal state (done/error) but hasn't exited yet (e.g. slow staging
+    # cleanup outlasting a poll). A non-'running' status must NOT be treated as a route-cancel and
+    # kill the finisher (which could drop its completion notification) — let it exit on its own.
+    _spawn(monkeypatch, _FakeProc(["timeout", ("exit", 0)]))  # alive one poll, then a clean exit
+    _liveness(monkeypatch, ("done", 1.0))
+    worker._run_supervised(1, 99)
+    assert _names(rec) == []  # never killed, asset/job untouched
+
+
 def test_a_watchdog_error_still_stops_the_child(monkeypatch, rec):
     # If the liveness probe raises (e.g. a transient DB error), the child must NOT be left running
     # unsupervised — the feature's whole point. It is killed and the job is failed.

From cce862cde72127ecfae40a9710ff80c1c61b7151 Mon Sep 17 00:00:00 2001
From: npeter83 <npeter83+claude@gmail.com>
Date: Wed, 12 Aug 2026 23:38:37 +0200
Subject: [PATCH 4/5] fix(downloads): supervisor edge-cases from review round 3
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- worker.py: keep an absolute wall-clock backstop for ANY still-alive child. A child that
  reached a terminal state (done/error) but wedges in teardown is now killed to free the worker
  slot ("abandon"), with its DB status left intact; only a still-'running' over-cap job is failed.
  Previously gating the wall-clock on status=='running' let a wedged finisher loop forever (#1).
- worker.py: reconcile the claimed asset on ANY clean child exit (not only non-zero), so a child
  that exits without finalizing it — e.g. an early return on a job DELETEd mid-flight — no longer
  leaves the asset stuck 'downloading' until boot recovery (#2).
- worker.py: extract _stop_and_free + _fail_after_kill; the exception / stall / timeout / cancel /
  shutdown / abandon paths now share one kill+free (+fail) tail instead of three copies (#3).
- tests: a wedged-finished child is killed by the wall-clock without a status change; clean exits
  now assert the reconciliation no-op (#1, #4).
---
 backend/app/worker.py                   | 93 +++++++++++++++----------
 backend/tests/test_worker_supervisor.py | 22 ++++--
 2 files changed, 73 insertions(+), 42 deletions(-)

diff --git a/backend/app/worker.py b/backend/app/worker.py
index 3cb9397d..7bd3f764 100644
--- a/backend/app/worker.py
+++ b/backend/app/worker.py
@@ -294,6 +294,23 @@ def _stop_child(proc: subprocess.Popen, grace_s: float) -> None:
     )
 
 
+def _stop_and_free(job_id: int, asset_id: int, proc: subprocess.Popen, grace_s: float) -> None:
+    """Kill the child, free its claimed asset, and clear its staging — the common tail of every path
+    that stops a child (a route pause/cancel, a shutdown, an over-cap abandon)."""
+    _stop_child(proc, grace_s)
+    _reset_asset_pending(asset_id)
+    _cleanup_staging(asset_id)
+
+
+def _fail_after_kill(
+    job_id: int, asset_id: int, proc: subprocess.Popen, grace_s: float, error: str, error_code: str
+) -> None:
+    """`_stop_and_free` + mark the job errored — for the watchdog's stall/timeout kills and the
+    supervisor-error fallback."""
+    _stop_and_free(job_id, asset_id, proc, grace_s)
+    _set_job(job_id, status="error", error=error, error_code=error_code, phase=None)
+
+
 def _run_supervised(job_id: int, asset_id: int) -> None:
     """Run the heavy download/edit for an already-claimed asset in a child process and watchdog it.
 
@@ -325,70 +342,72 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             except subprocess.TimeoutExpired:
                 pass
             status, stale = _job_liveness(job_id)
+            if status is None or status in ("paused", "cancelled"):
+                reason = "cancelled"  # a route stopped it (or the row vanished)
+                break
             if status == "running":
-                # Only a genuinely-running child is subject to the stall / wall-clock / shutdown kill.
                 if stale is not None and stale > stall_s:
                     reason = "stalled"
                     break
-                if time.monotonic() - started > wall_s:
-                    reason = "timeout"
-                    break
                 if _stop.is_set():
                     reason = "shutdown"
                     break
-            elif status is None or status in ("paused", "cancelled"):
-                reason = "cancelled"  # a route stopped it (or the row vanished)
+            # Absolute wall-clock cap frees the worker slot even for a FINISHED child wedged in
+            # teardown. A still-running job over the cap is failed ("timeout"); a terminal one (already
+            # done/error) just gets its lingering child killed, its DB status left intact ("abandon").
+            if time.monotonic() - started > wall_s:
+                reason = "timeout" if status == "running" else "abandon"
                 break
-            # else: a terminal status (done / error, or a re-queued contended asset) means the child
-            # already wrote its outcome and is exiting on its own — keep polling; the next proc.wait
-            # catches the exit. NEVER kill a finishing child (that could drop its completion notice).
+            # else: a terminal status under the cap (done/error, or a re-queued contended asset) means
+            # the child is exiting on its own — keep polling; NEVER kill a finishing child (that could
+            # drop its completion notice).
     except Exception:  # noqa: BLE001 — a watchdog error must NEVER leave the child running unsupervised
         log.exception("supervisor loop failed for job %s; stopping the child", job_id)
-        _stop_child(proc, grace_s)
-        _reset_asset_pending(asset_id)
-        _cleanup_staging(asset_id)
-        _set_job(
-            job_id, status="error", error_code="unknown", phase=None,
-            error="The download was stopped after a worker error.",
+        _fail_after_kill(
+            job_id, asset_id, proc, grace_s, "The download was stopped after a worker error.", "unknown"
         )
         return
 
     if reason is None:
-        # Child exited. If it died WITHOUT setting a terminal state (non-zero exit / crash / OOM),
-        # the job is still 'running' — fail it here so nothing is left stuck. Guarded: the child is
-        # already dead (no orphan risk), so a DB blip in this probe must not propagate.
-        if proc.returncode != _CHILD_OK:
-            try:
+        # Child exited. Reconcile the asset we claimed: free it if the child left it non-finalized —
+        # a crash/OOM, or an early return on a job deleted mid-flight — a no-op when it finalized
+        # normally. Then, if it died non-zero while the job is still 'running', fail the job so it is
+        # not left stuck. Guarded: the child is already dead, so a DB blip here must not propagate.
+        try:
+            _reset_asset_pending(asset_id)
+            if proc.returncode != _CHILD_OK:
                 status, _ = _job_liveness(job_id)
                 if status == "running":
-                    _reset_asset_pending(asset_id)
                     _cleanup_staging(asset_id)
                     _set_job(
                         job_id, status="error", error_code="unknown", phase=None,
                         error=f"The download process exited unexpectedly (code {proc.returncode}).",
                     )
                     log.warning("job %s: child exited %s without finishing", job_id, proc.returncode)
-            except Exception:  # noqa: BLE001 — boot recovery is the backstop for a stuck 'running' job
-                log.exception("post-exit cleanup failed for job %s", job_id)
+        except Exception:  # noqa: BLE001 — boot recovery is the backstop for a stuck 'running' job
+            log.exception("post-exit reconciliation failed for job %s", job_id)
         return
 
-    _stop_child(proc, grace_s)
-    _reset_asset_pending(asset_id)
-    _cleanup_staging(asset_id)
-    if reason == "cancelled":
-        # The route already set paused/cancelled (or the job is gone); just free the asset.
-        log.info("job %s: child stopped (route pause/cancel)", job_id)
+    if reason == "stalled":
+        _fail_after_kill(
+            job_id, asset_id, proc, grace_s,
+            f"The download stalled (no progress for {int(stall_s)}s) and was stopped.", "network",
+        )
         return
+    if reason == "timeout":
+        _fail_after_kill(
+            job_id, asset_id, proc, grace_s,
+            f"The download exceeded the {int(wall_s)}s time limit and was stopped.", "network",
+        )
+        return
+    # cancelled / abandon / shutdown: stop + free the asset, but leave the job's own DB status.
+    _stop_and_free(job_id, asset_id, proc, grace_s)
     if reason == "shutdown":
         _requeue_if_running(job_id, None)  # resume on the next boot
-        return
-    msg = (
-        f"The download stalled (no progress for {int(stall_s)}s) and was stopped."
-        if reason == "stalled"
-        else f"The download exceeded the {int(wall_s)}s time limit and was stopped."
-    )
-    _set_job(job_id, status="error", error=msg, error_code="network", phase=None)
-    log.warning("job %s: stopped by watchdog (%s)", job_id, reason)
+    elif reason == "abandon":
+        log.warning("job %s: killed a child wedged past the wall-clock; DB status left intact", job_id)
+    else:  # cancelled
+        log.info("job %s: child stopped (route pause/cancel)", job_id)
 
 
 def _cleanup_staging(asset_id: int) -> None:
diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py
index 3a17d5a6..e4b66006 100644
--- a/backend/tests/test_worker_supervisor.py
+++ b/backend/tests/test_worker_supervisor.py
@@ -69,11 +69,11 @@ def _liveness(monkeypatch, value):
 
 
 def test_clean_child_exit_leaves_the_outcome_alone(monkeypatch, rec):
-    # Child finished on its own (exit 0) — it wrote the terminal state; the supervisor must not kill,
-    # reset, or re-fail anything.
+    # Child finished on its own (exit 0) — it wrote the terminal state. The supervisor only reconciles
+    # the asset (a no-op when the child finalized it) and must not kill or re-fail anything.
     _spawn(monkeypatch, _FakeProc([("exit", 0)]))
     worker._run_supervised(1, 99)
-    assert _names(rec) == []
+    assert _names(rec) == ["reset_asset"]
 
 
 def test_no_progress_stall_kills_and_fails_the_job(monkeypatch, rec):
@@ -154,7 +154,7 @@ def test_a_long_download_survives_while_progress_stays_fresh(monkeypatch, rec):
     _spawn(monkeypatch, _FakeProc(["timeout", "timeout", "timeout", ("exit", 0)]))
     _liveness(monkeypatch, ("running", 2.0))  # always well under the 10s stall
     worker._run_supervised(1, 99)
-    assert _names(rec) == []
+    assert _names(rec) == ["reset_asset"]  # only the clean-exit reconciliation (a no-op), no kill
 
 
 def test_a_finishing_child_is_not_mistaken_for_a_cancel(monkeypatch, rec):
@@ -164,7 +164,19 @@ def test_a_finishing_child_is_not_mistaken_for_a_cancel(monkeypatch, rec):
     _spawn(monkeypatch, _FakeProc(["timeout", ("exit", 0)]))  # alive one poll, then a clean exit
     _liveness(monkeypatch, ("done", 1.0))
     worker._run_supervised(1, 99)
-    assert _names(rec) == []  # never killed, asset/job untouched
+    assert _names(rec) == ["reset_asset"]  # not killed; only the clean-exit reconciliation (no-op)
+
+
+def test_a_wedged_finished_child_is_killed_by_the_wall_clock(monkeypatch, rec):
+    # A child that reached a terminal state (done) but never exits (e.g. wedged in teardown on a stuck
+    # mount) must still be killed by the absolute wall-clock cap to free the worker slot — WITHOUT
+    # overwriting its DB status (it already succeeded).
+    monkeypatch.setattr(worker.settings, "download_wall_clock_max_seconds", 0)
+    _spawn(monkeypatch, _FakeProc(["timeout"]))  # never exits within the test
+    _liveness(monkeypatch, ("done", 1.0))  # finished, but the process lingers
+    worker._run_supervised(1, 99)
+    names = _names(rec)
+    assert "stop_child" in names and "set_job" not in names  # killed to free the slot, status kept
 
 
 def test_a_watchdog_error_still_stops_the_child(monkeypatch, rec):

From 099d6d1301b97fbac944a53c624211d1c6f95e84 Mon Sep 17 00:00:00 2001
From: npeter83 <npeter83+claude@gmail.com>
Date: Wed, 12 Aug 2026 23:43:15 +0200
Subject: [PATCH 5/5] chore: bump VERSION 0.63.0 + release notes (supervised
 download-subprocess watchdog)

---
 VERSION                          | 2 +-
 frontend/src/lib/releaseNotes.ts | 7 +++++++
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/VERSION b/VERSION
index 7c6425a5..c149bbff 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-0.62.2
\ No newline at end of file
+0.63.0
\ No newline at end of file
diff --git a/frontend/src/lib/releaseNotes.ts b/frontend/src/lib/releaseNotes.ts
index 4578aacc..3a6673fc 100644
--- a/frontend/src/lib/releaseNotes.ts
+++ b/frontend/src/lib/releaseNotes.ts
@@ -14,6 +14,13 @@ export interface ReleaseEntry {
 }
 
 export const RELEASE_NOTES: ReleaseEntry[] = [
+  {
+    version: "0.63.0",
+    date: "2026-08-12",
+    chores: [
+      "Hardened the download worker: each download or clip now runs in its own supervised process, so a stuck or misbehaving one can be stopped cleanly and can never freeze the rest of the download queue. No change to how downloads look or work.",
+    ],
+  },
   {
     version: "0.62.2",
     date: "2026-08-12",