diff --git a/backend/app/db.py b/backend/app/db.py
index 8ca8ecf2..51a06301 100644
--- a/backend/app/db.py
+++ b/backend/app/db.py
@@ -20,6 +20,10 @@ engine = create_engine(
     max_overflow=30,
     pool_timeout=30,
     future=True,
+    # A database error renders its bound parameters into its message — a title, an uploader, a
+    # file path, a URL — and that message is logged and stored as a job's error text. The SQL
+    # stays; the values are data, and the ids an operator needs are on the line already.
+    hide_parameters=True,
 )
 SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
 
diff --git a/backend/app/downloads/adapters.py b/backend/app/downloads/adapters.py
index 3f0c4ca5..4749adc5 100644
--- a/backend/app/downloads/adapters.py
+++ b/backend/app/downloads/adapters.py
@@ -18,6 +18,7 @@ import yt_dlp
 from yt_dlp.networking.impersonate import ImpersonateTarget
 
 from app.config import settings
+from app.downloads.stated import StatedFailure
 
 # Facebook's "Cannot parse data": the page yt-dlp got was a wall, not the content. Both the Chrome retry
 # below and the failure catalog (errors.py) decide on it, so the wording exists once.
@@ -178,11 +179,11 @@ class FacebookAdapter(SourceAdapter):
         # Measured (E11.5 review): a refused or expired cookie gets the login page (~450 KB, a login
         # form, no post id at all), which is not a post without a video — the user's cookie is the cause.
         if _quoted(post_id) not in page:
-            raise RuntimeError(POST_NOT_SHOWN_MESSAGE)
+            raise StatedFailure(POST_NOT_SHOWN_MESSAGE)
         video_id = facebook_post_video_id(page, post_id)
         if video_id is None:
             # Never fall back to the post link: that is exactly where the wrong video comes from.
-            raise RuntimeError(POST_WITHOUT_VIDEO_MESSAGE)
+            raise StatedFailure(POST_WITHOUT_VIDEO_MESSAGE)
         return f"https://www.facebook.com/watch/?v={video_id}"
 
     def client_sets(self, authed: bool = False) -> list:
diff --git a/backend/app/downloads/edit.py b/backend/app/downloads/edit.py
index dfa07932..d4967169 100644
--- a/backend/app/downloads/edit.py
+++ b/backend/app/downloads/edit.py
@@ -34,6 +34,7 @@ from pathlib import Path
 
 from app import logs
 from app.config import settings
+from app.downloads.stated import StatedFailure
 
 log = logs.get_logger(__name__)
 
@@ -451,11 +452,11 @@ def run_ffmpeg(
             if should_cancel():
                 raise EditAborted
             if time.monotonic() - last_output > stall_timeout_s:
-                raise RuntimeError(f"ffmpeg stalled: no progress for {int(stall_timeout_s)}s")
+                raise StatedFailure(f"ffmpeg stalled: no progress for {int(stall_timeout_s)}s")
         try:
             ret = proc.wait(timeout=_FFMPEG_EXIT_GRACE_S)
         except subprocess.TimeoutExpired:  # pipes closed but the process lingers
-            raise RuntimeError("ffmpeg did not exit after closing its output")
+            raise StatedFailure("ffmpeg did not exit after closing its output")
     except BaseException:
         # ONE exit path stops the process, so the invariant is unconditional: nothing leaves this
         # function with ffmpeg still running. Not just our own EditAborted/stall — `on_progress`
@@ -470,7 +471,7 @@ def run_ffmpeg(
             if pipe:
                 pipe.close()
     if ret != 0:
-        raise RuntimeError(f"ffmpeg failed ({ret}): {''.join(err_tail)[-300:]}")
+        raise StatedFailure(f"ffmpeg failed ({ret}): {''.join(err_tail)[-300:]}")
 
 
 # --- ffmpeg: universal browser playability --------------------------------------------------
@@ -723,6 +724,15 @@ def _run_ffmpeg_best_effort(cmd: list[str], timeout_s: int) -> Exception | None:
         return exc
 
 
+def _step_failure(err: BaseException) -> str:
+    """What a best-effort image step's failure line says: the kind and, for ffmpeg, its exit
+    status. Never `str(err)`: a CalledProcessError or TimeoutExpired quotes ffmpeg's argv, and an
+    OSError its file name — the source path either way, which is built from the channel and the
+    title."""
+    code = getattr(err, "returncode", None)
+    return f"{type(err).__name__} {code}" if code is not None else type(err).__name__
+
+
 def _is_jpeg(path: Path) -> bool:
     """True if the file starts with the JPEG SOI marker (FF D8 FF)."""
     try:
@@ -750,13 +760,15 @@ def thumbnail_to_jpg(src: Path | None, dest: Path, timeout_s: int = 30) -> Path
             shutil.copyfile(src, dest)
             return dest
         except OSError as exc:
-            log.info("thumbnail copy failed (%s); poster falls back to a cut frame", str(exc)[:120])
+            log.info(
+                "thumbnail copy failed (%s); poster falls back to a cut frame", _step_failure(exc)
+            )
             return None
     err = _run_ffmpeg_best_effort(build_thumbnail_jpg_cmd(src, dest), timeout_s)
     if err is not None:
         log.info(
             "thumbnail->jpg conversion failed (%s); poster falls back to a cut frame",
-            str(err)[:120],
+            _step_failure(err),
         )
         return None
     return dest if dest.exists() else None
@@ -781,7 +793,7 @@ def ensure_poster(asset, download_root: str, timeout_s: int = 30) -> str | None:
     at = min(3.0, (asset.duration_s or 10) * 0.1)
     err = _run_ffmpeg_best_effort(build_poster_cmd(src, dest, at), timeout_s)
     if err is not None:
-        log.info("poster gen skipped for asset %s: %s", asset.id, str(err)[:120])
+        log.info("poster gen skipped for asset %s: %s", asset.id, _step_failure(err))
         return None
     return rel if dest.exists() else None
 
@@ -812,7 +824,7 @@ def ensure_storyboard(asset, download_root: str, timeout_s: int = 90) -> dict |
     cmd = build_storyboard_cmd(src, dest, asset.duration_s, geom)
     err = _run_ffmpeg_best_effort(cmd, timeout_s)
     if err is not None:
-        log.info("storyboard gen skipped for asset %s: %s", asset.id, str(err)[:120])
+        log.info("storyboard gen skipped for asset %s: %s", asset.id, _step_failure(err))
         return None
     if not dest.exists():
         return None
diff --git a/backend/app/downloads/formats.py b/backend/app/downloads/formats.py
index 8ae2fcad..1180cf97 100644
--- a/backend/app/downloads/formats.py
+++ b/backend/app/downloads/formats.py
@@ -26,16 +26,29 @@ from app import logs
 
 log = logs.get_logger(__name__)
 
-# yt-dlp names what it is working on in three shapes, and each one is a row of someone's download
-# history: the source URL, the `[extractor] <id>:` prefix on nearly every warning and error (for
-# the generic extractor the "id" is the URL's own basename), and the file path it writes to, which
-# `storage.rel_path` builds from the channel and the title. The job id is already on every line
-# (`logs.set_job` in the download child), so none of the three adds anything an operator needs.
-# What stays: the host (it says which extractor is failing) and the extractor tag.
+# yt-dlp names what it is working on in these shapes, and each one is a row of someone's download
+# history: the source URL, the `[extractor] <id>:` prefix on nearly every warning and error, the
+# file path it writes to (`storage.rel_path` builds it from the channel and the title), a
+# playlist's title, and an id quoted mid-sentence. The job id is already on every line
+# (`logs.set_job` in the download child), so none of them adds anything an operator needs. What
+# stays: the host (it says which extractor is failing) and the extractor tag. Every pattern below
+# is a shape measured in yt-dlp's own source (2026.07.04), pinned in test_ytdlp_logging.py.
 _URL_RE = re.compile(r"https?://([^/\s]+)\S*")
 # `Destination:` is yt-dlp's own word in `[download] Destination: <path>`, not an id — the path
-# after it is what the third pattern takes out.
+# after it is what `_PATH_RE` takes out.
 _ID_PREFIX_RE = re.compile(r"(\[[\w:.-]+\] )(?!Destination:)[^\s:\[\]]+:")
+# The generic extractor's "id" is the URL's basename, percent-DECODED: `My%20Holiday.mp4` is
+# `My Holiday` — spaces and all, which the pattern above stops at. Only this extractor: every
+# other one names a site id, and letting the prefix run to `: ` everywhere would take the words
+# out of lines like `[VideoConvertor] Converting video from webm to mp4; Destination: …`.
+_GENERIC_ID_RE = re.compile(r"(\[generic\] )(?!Destination:)[^\[\]\n]+?:(?= |$)", re.M)
+# `[youtube:tab] Playlist <title>: Downloading 5 items` — the title is not an id, and has spaces.
+_PLAYLIST_TITLE_RE = re.compile(r"(\] Playlist ).+?(?=: Downloading )")
+# `[download] Finished downloading playlist: <title>`, `[download] Downloading playlist: <title>`,
+# `… Skipping already downloaded playlist: <title>` — the title runs to the end of the line.
+_TITLE_TAIL_RE = re.compile(r"(playlist: |\[download\] Downloading \w+: ).*", re.I)
+# An id quoted mid-sentence, not as a prefix.
+_BARE_ID_RE = re.compile(r"((?:subtitles not available for|Downloading just video) )[^\s,;]+")
 # Two segments or more, and not glued to a word: `www.youtube.com/…` is what the URL pattern
 # leaves behind, and it must survive this one.
 _PATH_RE = re.compile(r"(?<![\w.…])/(?:[^/\s\"']+/)+[^/\s\"']*")
@@ -48,6 +61,10 @@ def log_safe(text: object) -> str:
     it would put back one line later exactly what the adapter took out."""
     text = _URL_RE.sub(lambda m: f"{m.group(1)}/…", str(text))
     text = _ID_PREFIX_RE.sub(r"\1…:", text)
+    text = _GENERIC_ID_RE.sub(r"\1…:", text)
+    text = _PLAYLIST_TITLE_RE.sub(r"\1…", text)
+    text = _TITLE_TAIL_RE.sub(r"\1…", text)
+    text = _BARE_ID_RE.sub(r"\1…", text)
     return _PATH_RE.sub("…", text)
 
 
diff --git a/backend/app/downloads/pipeline.py b/backend/app/downloads/pipeline.py
index ad737af1..39006d64 100644
--- a/backend/app/downloads/pipeline.py
+++ b/backend/app/downloads/pipeline.py
@@ -23,6 +23,7 @@ from app.downloads.jobdb import (
     _staging_dir,
     _touch_job,
 )
+from app.downloads.stated import StatedFailure
 from app.downloads.status import JobStatus
 from app.models import DownloadJob, MediaAsset
 from app.notifications import create_notification
@@ -307,7 +308,7 @@ def _produced_file(info: dict, staging: Path) -> Path:
         and not _YTDLP_TEMP_RE.search(p.name)
     ]
     if not cands:
-        raise RuntimeError("yt-dlp produced no output file")
+        raise StatedFailure("yt-dlp produced no output file")
     return max(cands, key=lambda p: p.stat().st_size)
 
 
@@ -481,7 +482,7 @@ def _download(
             try:
                 urlguard.check_url(url)
             except urlguard.BlockedUrl as e:
-                raise RuntimeError(f"Refusing to download from this address ({e.reason}).") from e
+                raise StatedFailure(f"Refusing to download from this address ({e.reason}).") from e
         # S6: attach the acting user's per-host cookie ONLY when it's host-scoped to this URL AND
         # this asset is that user's private, salted identity (the double gate lives in credentials).
         remaining_bytes: int | None = None
@@ -653,25 +654,30 @@ def _download(
         diagnosis = dlerrors.diagnose(msg, url=url, cookie_attached=cookiefile is not None)
         # SEC-12: redact a transport-level readout for user-supplied URLs (see user_error_text).
         msg = dlerrors.user_error_text(source_kind, msg, diagnosis.code)
-        with SessionLocal() as db:
-            asset = db.get(MediaAsset, asset_id)
-            if asset:
-                asset.status = "error"
-                asset.error = msg
-            # Atomic fail: only while still running, so a route-set cancel/pause is never overwritten
-            # with `error` (error also holds the asset → would re-open the double-release class).
-            service.finalize_if_running(
-                db,
-                job_id,
-                status=JobStatus.ERROR,
-                error=msg,
-                error_code=diagnosis.code,
-                error_detail=diagnosis.detail,
-                phase=None,
-            )
-            db.commit()
-        # silent-ok: not silent — `_log_download_failed` is the report.
-        _log_download_failed(job_id, msg, diagnosis.code, exc)
+        # The line goes out whether or not the write below succeeds: when the database is what
+        # failed, that write fails too, and the failure would otherwise leave no line at all.
+        try:
+            with SessionLocal() as db:
+                asset = db.get(MediaAsset, asset_id)
+                if asset:
+                    asset.status = "error"
+                    asset.error = msg
+                # Atomic fail: only while still running, so a route-set cancel/pause is never
+                # overwritten with `error` (error also holds the asset → would re-open the
+                # double-release class).
+                service.finalize_if_running(
+                    db,
+                    job_id,
+                    status=JobStatus.ERROR,
+                    error=msg,
+                    error_code=diagnosis.code,
+                    error_detail=diagnosis.detail,
+                    phase=None,
+                )
+                db.commit()
+        finally:
+            # silent-ok: not silent — `_log_download_failed` is the report.
+            _log_download_failed(job_id, msg, diagnosis.code, exc)
     finally:
         # Unlink the decrypted cookie file explicitly (the rmtree below removes the whole staging
         # tree too, but never leave a plaintext session cookie on disk a moment longer than needed).
@@ -680,13 +686,20 @@ def _download(
         shutil.rmtree(staging, ignore_errors=True)
 
 
-def _stack_worth_logging(exc: BaseException) -> bool:
-    """Whether a failure's traceback goes with its line: only when it is our own code failing (a
-    TypeError on an odd info dict, a KeyError). yt-dlp's DownloadError, a RuntimeError we raised
-    with the whole story in its message (`ffmpeg failed (1): …`, `yt-dlp produced no output
-    file`) and an OSError naming its file all have the same stack — the `except` that caught them
-    — and the message at its foot is the unscrubbed text: the one copy `log_safe` cannot reach."""
-    return not isinstance(exc, (yt_dlp.utils.DownloadError, RuntimeError, OSError))
+def _failure_stack(exc: BaseException) -> dict[str, str]:
+    """The `extra=` fragment that puts a failure's traceback on its line — or nothing.
+
+    Nothing for yt-dlp's DownloadError and for our own `StatedFailure` (`ffmpeg failed (1): …`):
+    their message is the whole story, and their stack is only the `except` that caught them.
+    Everything else keeps it — a TypeError on an odd info dict, a plain RuntimeError, an OSError,
+    a database error: that is our code or our platform failing, and the stack is the lead.
+
+    Scrubbed, link by link (`logs.scrubbed_stack`): a stack's messages are data — an OSError names
+    its file, and a database error raised inside `except DownloadError` carries yt-dlp's
+    unscrubbed text as its context."""
+    if isinstance(exc, (yt_dlp.utils.DownloadError, StatedFailure)):
+        return {}
+    return logs.scrubbed_stack(exc, formats.log_safe)
 
 
 def _log_download_failed(job_id: int, msg: str, code: str | None, exc: BaseException) -> None:
@@ -694,14 +707,20 @@ def _log_download_failed(job_id: int, msg: str, code: str | None, exc: BaseExcep
     log's copy goes through `log_safe`, or it would put back one line later the URL and id the
     yt-dlp adapter took out of the line before it. The stack only when the classifier could not
     place the failure: a recognised one is fully described by its code."""
+    unplaced = code in (None, "", "unknown")
     log.warning(
         "job %s failed: %s",
         job_id,
         formats.log_safe(msg),
-        extra=logs.event(
-            "download.job.failed", job_id=job_id, error_code=code, error_type=type(exc).__name__
-        ),
-        exc_info=code in (None, "", "unknown") and _stack_worth_logging(exc),
+        extra={
+            **logs.event(
+                "download.job.failed",
+                job_id=job_id,
+                error_code=code,
+                error_type=type(exc).__name__,
+            ),
+            **(_failure_stack(exc) if unplaced else {}),
+        },
     )
 
 
@@ -709,41 +728,41 @@ def _fail_edit(job_id: int, asset_id: int, msg: str, exc: BaseException | None =
     """Fail an edit job, and say so ONCE — every way an edit fails comes through here, so this is
     where its line and its event code live. `exc` is the exception when there was one.
 
-    Its traceback goes with the line only when it is our own code failing
-    (`_stack_worth_logging`)."""
-    with SessionLocal() as db:
-        asset = db.get(MediaAsset, asset_id)
-        if asset:
-            asset.status = "error"
-            asset.error = msg
-        # Atomic fail: only while still running (see the download error path).
-        service.finalize_if_running(
-            db,
+    Its traceback goes with the line unless the message already tells it all (`_failure_stack`).
+    The line goes out whether or not the write succeeds (see the download error path)."""
+    try:
+        with SessionLocal() as db:
+            asset = db.get(MediaAsset, asset_id)
+            if asset:
+                asset.status = "error"
+                asset.error = msg
+            # Atomic fail: only while still running (see the download error path).
+            service.finalize_if_running(
+                db,
+                job_id,
+                status=JobStatus.ERROR,
+                error=msg,
+                error_code=dlerrors.classify(msg),
+                phase=None,
+            )
+            db.commit()
+    finally:
+        # `msg` is the user's copy; ffmpeg's stderr tail in it can name the source file, whose
+        # path is built from the channel and the title.
+        log.warning(
+            "edit job %s failed: %s",
             job_id,
-            status=JobStatus.ERROR,
-            error=msg,
-            error_code=dlerrors.classify(msg),
-            phase=None,
+            formats.log_safe(msg),
+            extra={
+                **logs.event(
+                    "edit.job.failed",
+                    job_id=job_id,
+                    asset_id=asset_id,
+                    **({"error_type": type(exc).__name__} if exc is not None else {}),
+                ),
+                **(_failure_stack(exc) if exc is not None else {}),
+            },
         )
-        db.commit()
-    # `msg` is the user's copy; ffmpeg's stderr tail in it can name the source file, whose path is
-    # built from the channel and the title. The traceback frames name code, not data.
-    log.warning(
-        "edit job %s failed: %s",
-        job_id,
-        formats.log_safe(msg),
-        extra=logs.event(
-            "edit.job.failed",
-            job_id=job_id,
-            asset_id=asset_id,
-            **({"error_type": type(exc).__name__} if exc is not None else {}),
-        ),
-        exc_info=(
-            (type(exc), exc, exc.__traceback__)
-            if exc is not None and _stack_worth_logging(exc)
-            else None
-        ),
-    )
 
 
 def _process_edit(job_id: int, asset_id: int) -> None:
@@ -795,7 +814,7 @@ def _process_edit(job_id: int, asset_id: int) -> None:
             should_cancel=_cancel_poll(job_id),
         )
         if not dest_staging.exists():
-            raise RuntimeError("ffmpeg produced no output file")
+            raise StatedFailure("ffmpeg produced no output file")
 
         rel = storage.edit_rel_path(user_id, asset_id, out_ext)
         storage.place_file(dest_staging, settings.download_root, rel)
diff --git a/backend/app/downloads/stated.py b/backend/app/downloads/stated.py
new file mode 100644
index 00000000..e4a2c991
--- /dev/null
+++ b/backend/app/downloads/stated.py
@@ -0,0 +1,13 @@
+"""The failures the download pipeline raises itself, with the whole story in the message.
+
+`ffmpeg failed (1): …`, `yt-dlp produced no output file`, a post that shows no video: the message
+says what happened, and the stack is only the `except` that caught it. A failure line leaves that
+stack off (`pipeline._failure_stack`) — but ONLY for these. A plain `RuntimeError` is also what
+Python raises for a bug of ours (a dict changed size during iteration, a generator's
+StopIteration, `RecursionError`, `NotImplementedError`), and that one's stack is the only lead.
+"""
+
+
+class StatedFailure(RuntimeError):
+    """A failure we raised on purpose, whose message is its full explanation. A `RuntimeError` so
+    every `except RuntimeError` that caught these before still does."""
diff --git a/backend/app/logs.py b/backend/app/logs.py
index 91dab5a1..013076ed 100644
--- a/backend/app/logs.py
+++ b/backend/app/logs.py
@@ -37,7 +37,9 @@ import re
 import sys
 import threading
 import time
+import traceback
 import uuid
+from collections.abc import Callable
 from datetime import datetime, timezone
 from typing import Any
 from urllib.parse import parse_qsl
@@ -263,11 +265,53 @@ def redact(value: Any, *, key: str = "") -> Any:
 
 
 # --- formatters ----------------------------------------------------------------------------
+_SCRUBBED_STACK = "scrubbed_stack"
+
+
+def scrubbed_stack(exc: BaseException, scrub: Callable[[BaseException], str]) -> dict[str, str]:
+    """`extra=` fragment: `exc`'s traceback, rendered where an `exc_info` stack goes, with the
+    message of every exception in it put through `scrub`.
+
+    For a failure whose message can name someone's data (a download's URL, a file path, a video
+    id). The frames name code and are kept whole; the text at the foot of each link is the data.
+    A plain `exc_info` renders the chain too — the `__cause__`/`__context__` a DB error raised
+    inside `except DownloadError` carries — so scrubbing only the line's own message is not
+    enough. Merge it into the call's `extra`: `{**event(...), **scrubbed_stack(...)}`."""
+    links: list[BaseException] = []
+    link: BaseException | None = exc
+    while link is not None and all(link is not seen for seen in links):
+        links.append(link)
+        link = link.__cause__ or (None if link.__suppress_context__ else link.__context__)
+    parts: list[str] = []
+    for i in range(len(links) - 1, -1, -1):  # the oldest first, as Python prints a chain
+        link = links[i]
+        kind = type(link)
+        name = kind.__qualname__
+        if kind.__module__ not in ("builtins", "__main__"):
+            name = f"{kind.__module__}.{name}"
+        try:
+            message = scrub(link)
+        # silent-ok: the message is what could not be rendered, and the placeholder says so — the
+        # caller is in the middle of reporting one failure and must not get a second from here.
+        except Exception:  # noqa: BLE001 — a `__str__` that raises, of any shape
+            message = "<unprintable>"
+        frames = "".join(traceback.format_tb(link.__traceback__))
+        head = f"Traceback (most recent call last):\n{frames}" if frames else ""
+        parts.append(f"{head}{name}: {message}" if message else f"{head}{name}")
+        if i:
+            parts.append(
+                "\nThe above exception was the direct cause of the following exception:\n"
+                if links[i - 1].__cause__ is link
+                else "\nDuring handling of the above exception, another exception occurred:\n"
+            )
+    return {_SCRUBBED_STACK: "\n".join(parts)}
+
+
 # Attributes every LogRecord carries; anything else a caller passed via `extra=` is a structured
 # attribute and is merged into `attrs` (so `extra={"job_id": 7}` works as well as `event(...)`).
 _RECORD_STANDARD = frozenset(
     vars(logging.LogRecord("", 0, "", 0, "", None, None)).keys()
-    | {"message", "asctime", "taskName", "event", "attrs", PERSIST, "elevated"}
+    | {"message", "asctime", "taskName", "event", "attrs", PERSIST, "elevated", _SCRUBBED_STACK}
 )
 
 
@@ -348,6 +392,8 @@ def record_fields(record: logging.LogRecord) -> dict[str, Any]:
         fields["attrs"] = redact(attrs)
     if record.exc_info:
         fields["exc"] = redact_text(logging.Formatter().formatException(record.exc_info))
+    elif getattr(record, _SCRUBBED_STACK, None):
+        fields["exc"] = redact_text(getattr(record, _SCRUBBED_STACK))
     elif record.exc_text:
         # A QueueHandler-style hand-off formats the traceback into exc_text and drops exc_info.
         fields["exc"] = redact_text(record.exc_text)
@@ -381,6 +427,8 @@ class TextFormatter(logging.Formatter):
             line += f" {json.dumps(redact(attrs), ensure_ascii=False, default=str)}"
         if record.exc_info:
             line += "\n" + redact_text(self.formatException(record.exc_info))
+        elif getattr(record, _SCRUBBED_STACK, None):
+            line += "\n" + redact_text(getattr(record, _SCRUBBED_STACK))
         return line
 
 
diff --git a/backend/app/worker.py b/backend/app/worker.py
index 5db9b49f..77648159 100644
--- a/backend/app/worker.py
+++ b/backend/app/worker.py
@@ -23,6 +23,7 @@ import subprocess
 import sys
 import threading
 import time
+from collections.abc import Callable
 from datetime import datetime, timezone
 
 from sqlalchemy import text
@@ -267,11 +268,11 @@ def _fail_after_kill(
     error: str,
     error_code: str,
     cause: str | None = None,
-) -> None:
+) -> bool:
     """`_stop_and_free` + mark the job errored — for the watchdog's stall/timeout kills and the
-    supervisor-error fallback."""
+    supervisor-error fallback. Returns whether the job failed here (`_fail_running`)."""
     _stop_and_free(job_id, asset_id, proc, grace_s)
-    _fail_running(job_id, error=error, error_code=error_code, cause=cause)
+    return _fail_running(job_id, error=error, error_code=error_code, cause=cause)
 
 
 def _fail_running(
@@ -281,17 +282,19 @@ def _fail_running(
     error_code: str | None,
     error_detail: str | None = None,
     cause: str | None = None,
-) -> None:
+    exit_code: int | None = None,
+) -> bool:
     """Fail a claimed job — only while it is still `running` (`service.finalize_if_running`), so a
     cancel/pause a route landed first is never turned back into `error`, which holds the asset the
     route already released. `error_detail` is written every time: a failure never shows an earlier
     failure's explanation.
 
     `cause` is the worker's half of `download.job.failed`: the failures the child never got to
-    report because it was killed or never ran — a stall, the wall-clock cap. Without it those left
-    no line at all. A caller that already logs its own coded line (a crash, with its stack) passes
-    none. Logged only when the job actually failed here: a route's cancel that landed first is not
-    a failure."""
+    report because it was killed, never ran or died — a stall, the wall-clock cap, a non-zero exit
+    (with its `exit_code`). Without it those left no line at all. A caller that logs its own line
+    (a worker error, with its stack — `_report_worker_error`) passes none and uses the returned
+    verdict. Logged only when the job actually failed here: a route's cancel that landed first is
+    not a failure. Returns whether it failed here."""
     with SessionLocal() as db:
         failed = service.finalize_if_running(
             db,
@@ -309,7 +312,37 @@ def _fail_running(
             job_id,
             cause,
             extra=logs.event(
-                "download.job.failed", job_id=job_id, error_code=error_code, cause=cause
+                "download.job.failed",
+                job_id=job_id,
+                error_code=error_code,
+                cause=cause,
+                **({"exit_code": exit_code} if exit_code is not None else {}),
+            ),
+        )
+    return failed
+
+
+def _report_worker_error(job_id: int, exc: Exception, cause: str, fail: Callable[[], bool]) -> None:
+    """A worker error on a job (the supervisor loop, or processing the job, raised): fail the job,
+    then ONE line with the stack. Its event says what the error did to the job —
+    `download.job.failed` when the job failed with it, `download.worker.error` when a route had
+    already stopped it (a cancel that landed first is not a failure, `_fail_running`).
+
+    The line goes out even when failing the job raises (the database is down): as
+    `download.worker.error`, since the job was not failed, and the raise goes on to the caller."""
+    failed = False
+    try:
+        failed = fail()
+    finally:
+        log.error(
+            "job %s: worker error (%s)",
+            job_id,
+            cause,
+            exc_info=exc,
+            extra=logs.event(
+                "download.job.failed" if failed else "download.worker.error",
+                job_id=job_id,
+                cause=cause,
             ),
         )
 
@@ -366,19 +399,20 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             # 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",
+    except Exception as exc:  # noqa: BLE001 — a watchdog error must NEVER leave the child running unsupervised
+        # silent-ok: not silent — `_report_worker_error` is the report.
+        _report_worker_error(
             job_id,
-            extra=logs.event("download.job.failed", job_id=job_id, cause="supervisor_error"),
-        )
-        _fail_after_kill(
-            job_id,
-            asset_id,
-            proc,
-            grace_s,
-            "The download was stopped after a worker error.",
-            "unknown",
+            exc,
+            "supervisor_error",
+            lambda: _fail_after_kill(
+                job_id,
+                asset_id,
+                proc,
+                grace_s,
+                "The download was stopped after a worker error.",
+                "unknown",
+            ),
         )
         return
 
@@ -397,17 +431,8 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
                         job_id,
                         error=f"The download process exited unexpectedly (code {proc.returncode}).",
                         error_code="unknown",
-                    )
-                    log.warning(
-                        "job %s: child exited %s without finishing",
-                        job_id,
-                        proc.returncode,
-                        extra=logs.event(
-                            "download.job.failed",
-                            job_id=job_id,
-                            cause="child_exit",
-                            exit_code=proc.returncode,
-                        ),
+                        cause="child_exit",
+                        exit_code=proc.returncode,
                     )
         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)
@@ -446,10 +471,13 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             extra=logs.event("download.child.abandoned", job_id=job_id),
         )
     else:  # cancelled
+        # Not `download.job.aborted`: that is the child's line, logged when it notices the pause
+        # itself — and it often has, and is only tearing down when this poll comes round. This one
+        # says the worker had to stop it.
         log.info(
             "job %s: child stopped (route pause/cancel)",
             job_id,
-            extra=logs.event("download.job.aborted", job_id=job_id),
+            extra=logs.event("download.child.stopped", job_id=job_id),
         )
 
 
@@ -559,13 +587,19 @@ def _worker_loop(idx: int) -> None:
         with logs.bind(job_id=job_id):
             try:
                 _process_job(job_id)
-            except Exception:  # noqa: BLE001
-                log.exception(
-                    "processing job %s crashed",
-                    job_id,
-                    extra=logs.event("download.job.failed", job_id=job_id, cause="crash"),
-                )
-                _fail_running(job_id, error="Worker error", error_code=None)
+            except Exception as exc:  # noqa: BLE001
+                try:
+                    _report_worker_error(
+                        job_id,
+                        exc,
+                        "crash",
+                        lambda: _fail_running(job_id, error="Worker error", error_code=None),
+                    )
+                # silent-ok: not silent — `_report_worker_error` logged the crash before this
+                # re-raise. Failing the job hit the database, and letting that out would end this
+                # worker thread for good; boot recovery settles a job left `running`.
+                except Exception:  # noqa: BLE001
+                    log.exception("could not fail job %s after its crash", job_id)
 
 
 def main() -> None:
diff --git a/backend/tests/test_download_failure_catalog.py b/backend/tests/test_download_failure_catalog.py
index af661721..e56f15bc 100644
--- a/backend/tests/test_download_failure_catalog.py
+++ b/backend/tests/test_download_failure_catalog.py
@@ -125,6 +125,25 @@ def test_the_download_child_knows_the_cookie_went_with_it(db, monkeypatch, tmp_p
     assert (job.error_code, job.error_detail) == ("login_required", "facebook.cookie_rejected")
 
 
+def test_the_failure_line_goes_out_even_when_its_write_fails(db, monkeypatch, tmp_path):
+    """The database is often what failed; the write of the failure then fails too, and the line
+    that says so must not go down with it."""
+    u = _user(db, "fb-dbdown@example.com")
+    a, j = _asset_and_job(db, u, sig="dbdown", asset_status="downloading", job_status="running")
+    _fail_the_child(monkeypatch, tmp_path, cookie=False)
+    said = []
+    monkeypatch.setattr(pipeline, "_log_download_failed", lambda job_id, *rest: said.append(job_id))
+
+    def db_down(*a, **k):
+        raise ConnectionError("the database went away")
+
+    monkeypatch.setattr(pipeline.service, "finalize_if_running", db_down)
+
+    with pytest.raises(ConnectionError):
+        pipeline._download(j.id, a.id, "url", _REEL, {}, u.id)
+    assert said == [j.id]
+
+
 def test_a_job_meeting_a_failed_shared_asset_gets_the_same_explanation(db):
     # Before the catalog this path wrote the text and NO code: "Download failed" for the second
     # job, "Facebook sign-in required" for the first.
diff --git a/backend/tests/test_failure_lines.py b/backend/tests/test_failure_lines.py
index 51a8c0cd..c5332f0f 100644
--- a/backend/tests/test_failure_lines.py
+++ b/backend/tests/test_failure_lines.py
@@ -16,6 +16,8 @@ import yt_dlp
 
 from app import logs, sysconfig, worker
 from app.downloads import pipeline
+from app.downloads.stated import StatedFailure
+from app.models import DownloadJob, MediaAsset
 from app.plex import stream as plex_stream
 
 
@@ -102,6 +104,98 @@ def test_a_caller_with_its_own_line_is_not_reported_twice(sink, finalized):
     assert sink.events("download.job.failed") == []
 
 
+def test_a_child_that_died_is_reported_with_its_exit_code(sink, finalized):
+    worker._fail_running(
+        42, error="exited", error_code="unknown", cause="child_exit", exit_code=137
+    )
+
+    (line,) = sink.events("download.job.failed")
+    assert line["attrs"]["exit_code"] == 137
+
+
+@pytest.mark.parametrize(
+    ("fires", "code"), [(True, "download.job.failed"), (False, "download.worker.error")]
+)
+def test_a_worker_error_is_one_line_that_says_what_it_did_to_the_job(sink, fires, code):
+    """A route's cancel that landed first is not a failure — the error is still reported, with
+    its stack, but not as the job failing."""
+    try:
+        raise KeyError("a worker bug")
+    except KeyError as exc:
+        worker._report_worker_error(42, exc, "crash", lambda: fires)
+
+    (line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
+    assert line["event"] == code
+    assert "KeyError" in line["exc"]
+
+
+def test_a_worker_error_is_reported_even_when_failing_the_job_raises(sink):
+    def db_down():
+        raise ConnectionError("the database went away")
+
+    with pytest.raises(ConnectionError):
+        worker._report_worker_error(42, KeyError("a worker bug"), "crash", db_down)
+
+    (line,) = sink.events("download.worker.error")
+    assert line["attrs"]["cause"] == "crash"
+
+
+def test_a_crash_the_database_will_not_record_does_not_end_the_worker_thread(monkeypatch, sink):
+    """The crash handler writes to the database; if that raises too, the thread must go on to
+    the next job — a dead thread is a download slot gone until the worker restarts."""
+    claims = iter([42, 43])
+    seen = []
+
+    def claim():
+        job = next(claims, None)
+        if job is None:
+            worker._stop.set()
+        return job
+
+    def crash(job_id):
+        seen.append(job_id)
+        raise RuntimeError("boom")
+
+    def db_down(*a, **k):
+        raise ConnectionError("the database went away")
+
+    monkeypatch.setattr(worker, "_claim_job", claim)
+    monkeypatch.setattr(worker, "_process_job", crash)
+    monkeypatch.setattr(worker, "_fail_running", db_down)
+    monkeypatch.setattr(worker.settings, "download_poll_seconds", 0)
+    worker._stop.clear()
+    try:
+        worker._worker_loop(0)
+    finally:
+        worker._stop.clear()
+    assert seen == [42, 43]
+
+
+class _StoppedProc:
+    """A child that is still tearing down when the supervisor polls, after a route paused it."""
+
+    pid = 4242
+    returncode = None
+
+    def wait(self, timeout=None):
+        import subprocess
+
+        raise subprocess.TimeoutExpired("child", timeout or 0)
+
+
+def test_the_worker_stopping_a_paused_child_is_not_a_second_abort_line(monkeypatch, sink):
+    """The child logs `download.job.aborted` when it notices the pause itself; the worker's line
+    is its own fact — it had to stop the child — so one pause is never two aborts."""
+    monkeypatch.setattr(worker, "_spawn_download_child", lambda job_id: _StoppedProc())
+    monkeypatch.setattr(worker, "_job_liveness", lambda job_id: ("paused", 1.0))
+    monkeypatch.setattr(worker, "_stop_and_free", lambda *a: None)
+
+    worker._run_supervised(42, 7)
+
+    assert sink.events("download.job.aborted") == []
+    assert len(sink.events("download.child.stopped")) == 1
+
+
 # --- an edit failure, once ---------------------------------------------------------------------
 def test_an_edit_failure_is_one_line_with_its_code_and_our_own_stack(sink, finalized):
     try:
@@ -117,15 +211,65 @@ def test_an_edit_failure_is_one_line_with_its_code_and_our_own_stack(sink, final
 
 def test_an_edit_that_ffmpeg_refused_names_no_file_and_carries_no_stack(sink, finalized):
     """ffmpeg's stderr tail names its input, whose path is the channel and the title. The message
-    is scrubbed; the stack is left off, because its last line is the same text unscrubbed."""
+    is scrubbed; the stack is left off, because the message is the whole story."""
     msg = "ffmpeg failed (1): /downloads/Some_Channel/Season_2024/A_Private_Title_a3.mp4: Invalid data"
-    pipeline._fail_edit(9, 3, msg, RuntimeError(msg))
+    pipeline._fail_edit(9, 3, msg, StatedFailure(msg))
 
     (line,) = sink.events("edit.job.failed")
     assert "A_Private_Title" not in json.dumps(line)
     assert "exc" not in line
 
 
+def test_an_edit_failure_is_reported_even_when_its_write_fails(sink, monkeypatch):
+    """The database is often what failed; then the write fails too, and the line must not go
+    down with it."""
+
+    class _Down:
+        def __enter__(self):
+            raise ConnectionError("the database went away")
+
+        def __exit__(self, *exc):
+            return False
+
+    monkeypatch.setattr(pipeline, "SessionLocal", _Down)
+    with pytest.raises(ConnectionError):
+        pipeline._fail_edit(9, 3, "boom", KeyError("x"))
+
+    assert len(sink.events("edit.job.failed")) == 1
+
+
+class _EditDb(_NoDb):
+    """An edit job whose source is ready, for `_process_edit` to get as far as its encode."""
+
+    def get(self, model, _id):
+        if model is DownloadJob:
+            return SimpleNamespace(source_asset_id=7, edit_spec={}, user_id=1)
+        if model is MediaAsset:
+            return SimpleNamespace(
+                status="ready", rel_path="src.mp4", width=640, height=360, duration_s=10
+            )
+        return None
+
+
+def test_an_edit_that_crashes_is_said_once(sink, finalized, monkeypatch, tmp_path):
+    """The line `_process_edit`'s handler used to log before `_fail_edit` logged it again."""
+    (tmp_path / "src.mp4").write_bytes(b"x")
+    monkeypatch.setattr(pipeline, "SessionLocal", _EditDb)
+    monkeypatch.setattr(pipeline.settings, "download_root", str(tmp_path))
+    monkeypatch.setattr(pipeline, "_edit_staging_dir", lambda asset_id: tmp_path / "staging")
+
+    def odd_spec(*a, **k):
+        raise TypeError("a spec shape nobody expected")
+
+    monkeypatch.setattr(pipeline, "_set_job", odd_spec)
+
+    pipeline._process_edit(9, 3)
+
+    (line,) = [o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]
+    assert line["event"] == "edit.job.failed"
+    assert "TypeError" in line["exc"]
+
+
 def test_an_edit_whose_source_is_gone_is_found_by_the_same_code(sink, finalized):
     pipeline._fail_edit(9, 3, "The source file is missing.")
 
@@ -147,6 +291,48 @@ def test_a_failed_download_line_does_not_put_back_what_the_adapter_took_out(sink
     assert "exc" not in line
 
 
+@pytest.mark.parametrize(
+    "exc",
+    [
+        RuntimeError("dictionary changed size during iteration"),
+        RecursionError("maximum recursion depth exceeded"),
+        NotImplementedError(),
+        FileNotFoundError(2, "No such file", "/downloads/Some_Channel/Season_2024/A_Private_Title"),
+    ],
+    ids=lambda e: type(e).__name__,
+)
+def test_a_bug_that_python_raises_keeps_its_stack_and_names_no_file(sink, exc):
+    """Only the failures WE raise with the whole story in the message go without a stack. The
+    same types are what Python raises for a bug of ours, and then the stack is the only lead."""
+    try:
+        raise exc
+    except (RuntimeError, OSError) as caught:
+        pipeline._log_download_failed(5, str(caught), "unknown", caught)
+
+    (line,) = sink.events("download.job.failed")
+    assert type(exc).__name__ in line["exc"]
+    assert "A_Private_Title" not in json.dumps(line)
+
+
+def test_a_stack_does_not_carry_the_download_error_it_was_raised_over(sink):
+    """A database error raised inside `except DownloadError` has yt-dlp's text as its context,
+    and a plain `exc_info` prints that link unscrubbed."""
+    # A variable, as in the pipeline: a frame prints its SOURCE line, and that is code, not data.
+    text = "ERROR: [youtube] dQw4w9WgXcQ: Sign in to confirm"
+    try:
+        try:
+            raise yt_dlp.utils.DownloadError(text)
+        except yt_dlp.utils.DownloadError:
+            raise ConnectionError("server closed the connection unexpectedly")  # noqa: B904
+    except ConnectionError as exc:
+        pipeline._log_download_failed(5, str(exc), "unknown", exc)
+
+    (line,) = sink.events("download.job.failed")
+    assert "During handling of the above exception" in line["exc"]
+    assert "Sign in to confirm" in line["exc"]
+    assert "dQw4w9WgXcQ" not in json.dumps(line)
+
+
 def test_an_unclassified_failure_of_our_own_keeps_its_stack(sink):
     try:
         raise KeyError("requested_formats")
@@ -218,3 +404,62 @@ def test_a_refusing_session_directory_is_reported_once_per_session(sink):
 
     (line,) = sink.events("plex.stream.segment_probe_failed")
     assert line["attrs"]["reason"] == "PermissionError"
+
+
+def test_a_retried_attempt_names_the_host_and_not_the_video(sink, monkeypatch, tmp_path):
+    """The first attempt's DownloadError is logged when the fallback set takes over — one line
+    before the adapter's scrubbed copy of the same text would have hidden it."""
+    url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
+    attempts = iter(
+        [yt_dlp.utils.DownloadError(f"ERROR: [youtube] dQw4w9WgXcQ: Sign in ({url})"), {"id": "x"}]
+    )
+
+    class _YDL:
+        def __init__(self, opts):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+        def extract_info(self, target, download):
+            outcome = next(attempts)
+            if isinstance(outcome, Exception):
+                raise outcome
+            return outcome
+
+    adapter = SimpleNamespace(
+        client_sets=lambda authed: [["web"], ["tv"]],
+        resolve_url=lambda target, opts: target,
+        ydl_opts_extra=lambda clients: {},
+        retriable=lambda text: True,
+    )
+    monkeypatch.setattr(pipeline.yt_dlp, "YoutubeDL", _YDL)
+    monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
+    monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **fields: None)
+
+    assert pipeline._extract_with_fallback(1, 1, url, {}, tmp_path, adapter) == {"id": "x"}
+
+    (line,) = [o for o in sink.objects() if "retrying with fallback" in o["msg"]]
+    assert "dQw4w9WgXcQ" not in json.dumps(line)
+
+
+def test_a_failed_poster_or_storyboard_names_no_file(sink, tmp_path):
+    """ffmpeg's CalledProcessError quotes its argv, and the source path in it is built from the
+    channel and the title."""
+    from app.downloads import edit as editmod
+
+    rel = "Some_Channel/Season_2024/Some_Channel_-_20240101_-_A_Private_Title_a3.mp4"
+    (tmp_path / rel).parent.mkdir(parents=True)
+    (tmp_path / rel).write_bytes(b"not a video")
+    asset = SimpleNamespace(id=3, rel_path=rel, duration_s=10.0, storyboard_path=None)
+
+    assert editmod.ensure_poster(asset, str(tmp_path)) is None
+    assert editmod.ensure_storyboard(asset, str(tmp_path)) is None
+
+    lines = [o for o in sink.objects() if "gen skipped" in o["msg"]]
+    assert len(lines) == 2
+    assert all("Some_Channel" not in json.dumps(o) for o in lines)
+    assert all("CalledProcessError" in o["msg"] for o in lines)
diff --git a/backend/tests/test_log_pii.py b/backend/tests/test_log_pii.py
index 06c5a6f3..92a52d73 100644
--- a/backend/tests/test_log_pii.py
+++ b/backend/tests/test_log_pii.py
@@ -127,15 +127,31 @@ def test_a_scope_with_no_elevation_is_not_consent():
     assert not logs.consented_trace()  # a scheduler thread: no request at all
 
 
-def test_a_refused_token_refresh_names_the_oauth_reason_and_not_its_prose(sink):
+def test_a_refused_token_refresh_names_the_oauth_reason_and_not_its_prose(sink, monkeypatch):
     """The token endpoint's error is `{"error": "<code>", "error_description": <prose>}` — not the
-    Data API's shape. It goes through the same door, so it gets the same treatment."""
+    Data API's shape. The refresh goes through the same door, so it gets the same treatment."""
+    from types import SimpleNamespace
+
+    from app.youtube import client as yt
+
     resp = httpx.Response(
         400,
         json={"error": "invalid_grant", "error_description": "Token has been expired or revoked."},
         request=httpx.Request("POST", "https://oauth2.googleapis.com/token"),
     )
-    err = _upstream_failure("POST", "oauth2/token", resp)
+    client = yt.YouTubeClient.__new__(yt.YouTubeClient)  # no HTTP client: `_send` is the stub
+    client.db = None
+    client.user = SimpleNamespace(
+        token=SimpleNamespace(access_token=None, expiry=None, refresh_token_enc=b"x")
+    )
+    monkeypatch.setattr(yt, "decrypt_optional", lambda value: None)
+    monkeypatch.setattr(yt, "decrypt", lambda value: "a-refresh-token")
+    monkeypatch.setattr(yt.sysconfig, "get_str", lambda db, key: "")
+    monkeypatch.setattr(client, "_send", lambda *a, **k: resp)
+
+    with pytest.raises(YouTubeError) as caught:
+        client._access_token()
+    err = caught.value
 
     (line,) = sink.at("WARNING")
     assert line["attrs"]["reason"] == "invalid_grant"
@@ -267,16 +283,88 @@ def test_no_log_call_passes_what_a_person_typed():
     )
 
 
+_AT_LEAST_DEBUG = {"DEBUG", "INFO", "WARNING", "WARN", "ERROR", "CRITICAL", "FATAL"}
+
+
+def _may_run_at_trace(call: ast.Call) -> bool:
+    """A `log.log(level, …)` whose level is not provably DEBUG or above: `logs.TRACE`, a bare
+    `TRACE`, a literal 5 — or a variable, which could be any of them."""
+    if call.func.attr != "log" or not call.args:
+        return False
+    level = call.args[0]
+    if isinstance(level, ast.Attribute | ast.Name):
+        return (level.attr if isinstance(level, ast.Attribute) else level.id) not in _AT_LEAST_DEBUG
+    if isinstance(level, ast.Constant) and isinstance(level.value, int):
+        return level.value < logging.DEBUG
+    return True
+
+
+def _is_consent_check(test: ast.expr) -> bool:
+    """Exactly `…consented_trace()`: not `not …`, not `x or …` — each of those runs the body when
+    there is no consent."""
+    return (
+        isinstance(test, ast.Call)
+        and isinstance(test.func, ast.Attribute | ast.Name)
+        and (test.func.attr if isinstance(test.func, ast.Attribute) else test.func.id)
+        == "consented_trace"
+    )
+
+
 def _guarded_by_consent(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> bool:
+    """The node runs only inside the BODY of an `if <consented_trace()>:` — not its `else`."""
     while node in parents:
-        node = parents[node]
-        if isinstance(node, ast.If):
-            for sub in ast.walk(node.test):
-                if isinstance(sub, ast.Attribute) and sub.attr == "consented_trace":
-                    return True
+        parent = parents[node]
+        if isinstance(parent, ast.If) and _is_consent_check(parent.test) and node in parent.body:
+            return True
+        node = parent
     return False
 
 
+def _parents(tree: ast.AST) -> dict[ast.AST, ast.AST]:
+    return {child: node for node in ast.walk(tree) for child in ast.iter_child_nodes(node)}
+
+
+_GATE_SHAPES = {
+    "the body of the check": ("if logs.consented_trace():\n    log.log(logs.TRACE, 'x')", True),
+    "its else": (
+        "if logs.consented_trace():\n    pass\nelse:\n    log.log(logs.TRACE, 'x')",
+        False,
+    ),
+    "a negated check": ("if not logs.consented_trace():\n    log.log(logs.TRACE, 'x')", False),
+    "an or": ("if debug or logs.consented_trace():\n    log.log(logs.TRACE, 'x')", False),
+    "a nested block in the body": (
+        "if logs.consented_trace():\n    for t in ts:\n        log.log(logs.TRACE, 'x')",
+        True,
+    ),
+}
+
+
+@pytest.mark.parametrize("shape", list(_GATE_SHAPES))
+def test_the_consent_gate_is_the_body_of_the_check_and_nothing_else(shape):
+    src, gated = _GATE_SHAPES[shape]
+    tree = ast.parse(src)
+    (call,) = [n for n in ast.walk(tree) if isinstance(n, ast.Call) and _is_log_call(n)]
+    assert _guarded_by_consent(call, _parents(tree)) is gated
+
+
+@pytest.mark.parametrize(
+    ("level", "trace"),
+    [
+        ("logs.TRACE", True),
+        ("TRACE", True),
+        ("5", True),
+        ("level", True),
+        ("levels[scope]", True),
+        ("logging.DEBUG", False),
+        ("logging.WARNING", False),
+        ("20", False),
+    ],
+)
+def test_a_level_that_may_be_trace_is_treated_as_trace(level, trace):
+    (call,) = [n for n in ast.walk(ast.parse(f"log.log({level}, 'x')")) if isinstance(n, ast.Call)]
+    assert _may_run_at_trace(call) is trace
+
+
 def test_every_escaped_trace_line_is_gated_on_the_users_own_consent():
     """`log-pii-ok` on a TRACE line means "this is what the user's Teljes consent is for". The TRACE
     level alone does not hold that promise: an admin can run the whole process at TRACE, and a
@@ -287,15 +375,33 @@ def test_every_escaped_trace_line_is_gated_on_the_users_own_consent():
         src = path.read_text(encoding="utf-8")
         lines = src.splitlines()
         tree = ast.parse(src)
-        parents = {child: node for node in ast.walk(tree) for child in ast.iter_child_nodes(node)}
+        parents = _parents(tree)
         for node in ast.walk(tree):
             if not _is_log_call(node) or not _offending(node):
                 continue
-            first = node.args[0] if node.args else None
-            at_trace = isinstance(first, ast.Attribute) and first.attr == "TRACE"
             span = " ".join(lines[max(0, node.lineno - 2) : (node.end_lineno or node.lineno)])
-            if at_trace and _ESCAPE in span and not _guarded_by_consent(node, parents):
+            if (
+                _may_run_at_trace(node)
+                and _ESCAPE in span
+                and not _guarded_by_consent(node, parents)
+            ):
                 offenders.append(f"{path.relative_to(root)}:{node.lineno}")
     assert offenders == [], (
         f"a TRACE line with user content is not under consented_trace(): {offenders}"
     )
+
+
+def test_a_database_error_does_not_carry_the_values_it_was_given(db):
+    """DB lane. A failed write's message is logged and stored as the job's error text; with its
+    bound parameters in it, that is the title or the file path the row was about."""
+    from sqlalchemy import exc as sa_exc
+    from sqlalchemy import text
+
+    # Postgres's own text for this one names no value ("value too long for type character
+    # varying(3)"), so what is left to leak is SQLAlchemy's `[parameters: …]` suffix.
+    db.execute(text("CREATE TEMP TABLE pii_probe (x varchar(3))"))
+    with pytest.raises(sa_exc.DBAPIError) as caught:
+        db.execute(text("INSERT INTO pii_probe VALUES (:v)"), {"v": "A_Private_Title"})
+    db.rollback()
+    assert "value too long" in str(caught.value)
+    assert "A_Private_Title" not in str(caught.value)
diff --git a/backend/tests/test_logsink.py b/backend/tests/test_logsink.py
index 3dc4323a..7b9eb7cf 100644
--- a/backend/tests/test_logsink.py
+++ b/backend/tests/test_logsink.py
@@ -271,6 +271,23 @@ def test_the_writer_drains_what_was_queued_when_it_is_stopped(db):
     )
 
 
+def test_a_stopped_writer_says_what_it_dropped_before_it_goes(db):
+    """The download child's exit is `stop()` → `_run`'s drain: the notice for the repeats it
+    dropped is written there, or a child that lived under a minute never reports them."""
+    sink = logsink.LogSink(engine)
+    log = logging.getLogger("siftlode.sinktest.dropped")
+    log.addHandler(sink)
+    try:
+        for i in range(logsink.REPEATS_PER_MINUTE + 4):
+            log.warning("RSS poll skipped for channel %s: HTTP 404", f"UC{i}")
+    finally:
+        log.removeHandler(sink)
+    sink._stopping.set()
+    sink._run()
+    last = db.execute(select(LogEvent).order_by(LogEvent.id.desc()).limit(1)).scalar_one()
+    assert (last.event, last.attrs["suppressed"]) == ("log.repeats_suppressed", 4)
+
+
 def test_the_running_sink_keeps_the_base_level_for_everyone_the_elevation_is_not_for(
     db, monkeypatch
 ):
diff --git a/backend/tests/test_worker_supervisor.py b/backend/tests/test_worker_supervisor.py
index bc4adb38..53818780 100644
--- a/backend/tests/test_worker_supervisor.py
+++ b/backend/tests/test_worker_supervisor.py
@@ -87,6 +87,7 @@ def test_no_progress_stall_kills_and_fails_the_job(monkeypatch, rec):
     fail = next(c for c in rec if c[0] == "fail")
     assert fail[2]["error_code"] == "network"
     assert "stalled" in fail[2]["error"].lower()
+    assert fail[2]["cause"] == "stalled"  # the line the killed child never gets to write
 
 
 def test_route_pause_kills_but_keeps_the_route_status(monkeypatch, rec):
@@ -120,6 +121,7 @@ def test_wall_clock_cap_kills_even_while_progressing(monkeypatch, rec):
     worker._run_supervised(1, 99)
     fail = next(c for c in rec if c[0] == "fail")
     assert "time limit" in fail[2]["error"].lower()
+    assert fail[2]["cause"] == "timeout"
 
 
 def test_child_crash_without_finishing_is_failed(monkeypatch, rec):
@@ -131,6 +133,8 @@ def test_child_crash_without_finishing_is_failed(monkeypatch, rec):
     assert "reset_asset" in names
     fail = next(c for c in rec if c[0] == "fail")
     assert "unexpectedly" in fail[2]["error"].lower()
+    # Its line is `_fail_running`'s, so it is logged only if the job really failed here.
+    assert (fail[2]["cause"], fail[2]["exit_code"]) == ("child_exit", 1)
 
 
 def test_child_nonzero_exit_that_already_set_state_is_left_alone(monkeypatch, rec):
@@ -258,7 +262,7 @@ def test_a_crashed_jobs_crash_line_and_cleanup_carry_the_job_id(monkeypatch):
     monkeypatch.setattr(worker, "_claim_job", claim)
     monkeypatch.setattr(worker, "_process_job", crash)
     monkeypatch.setattr(
-        worker.log, "exception", lambda *a, **k: seen.setdefault("crash", logs._job_id.get())
+        worker.log, "error", lambda *a, **k: seen.setdefault("crash", logs._job_id.get())
     )
     monkeypatch.setattr(
         worker, "_fail_running", lambda *a, **k: seen.setdefault("cleanup", logs._job_id.get())
diff --git a/backend/tests/test_ytdlp_logging.py b/backend/tests/test_ytdlp_logging.py
index a17cc84a..afee4519 100644
--- a/backend/tests/test_ytdlp_logging.py
+++ b/backend/tests/test_ytdlp_logging.py
@@ -101,6 +101,44 @@ def test_the_source_url_does_not_travel_with_the_explanation(sink):
             "WARNING: [generic] …: Falling back on generic information",
             "my-private-holiday-clip",
         ),
+        # ...percent-decoded, so it can have spaces: `https://x/My%20Holiday%20in%20Crete.mp4`.
+        (
+            "ERROR: [generic] My Holiday in Crete: Unable to download webpage: HTTP Error 404",
+            "ERROR: [generic] …: Unable to download webpage: HTTP Error 404",
+            "Holiday",
+        ),
+        (
+            "[generic] My Holiday in Crete: Downloading webpage",
+            "[generic] …: Downloading webpage",
+            "Holiday",
+        ),
+        # A playlist URL is extracted too (`playlist_items: "1"`), and its title is printed.
+        (
+            "[youtube:tab] Playlist Our Wedding Videos: Downloading 1 items of 12",
+            "[youtube:tab] Playlist …: Downloading 1 items of 12",
+            "Wedding",
+        ),
+        (
+            "[download] Finished downloading playlist: Our Wedding Videos",
+            "[download] Finished downloading playlist: …",
+            "Wedding",
+        ),
+        (
+            "[download] Downloading playlist: Our Wedding Videos",
+            "[download] Downloading playlist: …",
+            "Wedding",
+        ),
+        # An id quoted mid-sentence.
+        (
+            "hu subtitles not available for dQw4w9WgXcQ",
+            "hu subtitles not available for …",
+            "dQw4w9WgXcQ",
+        ),
+        (
+            "[youtube:tab] Downloading just video dQw4w9WgXcQ because of --no-playlist",
+            "[youtube:tab] Downloading just video … because of --no-playlist",
+            "dQw4w9WgXcQ",
+        ),
         # `Destination` is yt-dlp's word, not an id; the path after it is built from the title.
         (
             "[download] Destination: /downloads/Some_Channel/Season_2024/A_Private_Title_a7.mp4",
@@ -125,6 +163,20 @@ def test_a_download_error_is_scrubbed_the_same_way_wherever_it_is_logged():
     assert formats.log_safe(text) == "ERROR: Unsupported URL: example.org/…"
 
 
+@pytest.mark.parametrize(
+    "line",
+    [
+        "nsig extraction failed: Some formats may be missing",
+        "[VideoConvertor] Converting video from webm to mp4; Destination: x",
+        "[download]  53.2% of ~  12.00MiB at    1.20MiB/s ETA 00:05",
+        "[download] 100% of   12.00MiB in 00:00:03 at 3.90MiB/s",
+    ],
+)
+def test_a_line_that_names_nothing_keeps_its_words(line):
+    """The scrub takes out what names a download, and nothing an operator reads the line for."""
+    assert formats.log_safe(line) == line
+
+
 def test_a_line_with_no_url_is_left_alone(sink):
     formats._YDL_LOG.warning("nsig extraction failed: Some formats may be missing")