diff --git a/backend/app/downloads/backfill.py b/backend/app/downloads/backfill.py
index ef17ed03..47b523c7 100644
--- a/backend/app/downloads/backfill.py
+++ b/backend/app/downloads/backfill.py
@@ -45,13 +45,14 @@ def _extract_meta(url: str, adapter: adapters.SourceAdapter, asset_id: int) -> d
             # history, and `redact_text` only takes credential params out of it — the rest
             # of it goes in whole. The id names the same asset to anyone holding the
             # database and nothing at all to anyone who is not; the host is kept because it
-            # is what decides which client set was tried.
+            # is what decides which client set was tried. yt-dlp's message goes through
+            # `log_safe` for the same reason: it opens with the URL or the video id.
             log.info(
                 "meta extract failed for asset %s on %s (%s): %s",
                 asset_id,
                 urlsplit(url).hostname or "?",  # log-pii-ok: the host, never the path or query
                 clients,
-                str(exc)[:100],
+                formats.log_safe(exc)[:100],
             )
     return None
 
@@ -77,7 +78,9 @@ def run() -> dict:
                 if uu:
                     a.uploader_url = uu
                     filled_url += 1
-                    log.info("asset %s uploader_url <- %s", a.id, uu)
+                    log.info(
+                        "asset %s uploader_url filled", a.id
+                    )  # the URL names what they downloaded
             if a.poster_path is None:
                 # Record a self-hosted poster for every download (the reliable og:image source):
                 # ensure_poster returns the existing <base>.jpg thumbnail sidecar, or cuts a frame.
@@ -85,7 +88,7 @@ def run() -> dict:
                 if rel:
                     a.poster_path = rel
                     filled_poster += 1
-                    log.info("asset %s poster <- %s", a.id, rel)
+                    log.info("asset %s poster filled", a.id)  # the path carries the title
             db.commit()
     result = {
         "assets": len(assets),
diff --git a/backend/app/downloads/formats.py b/backend/app/downloads/formats.py
index ea87f722..8ae2fcad 100644
--- a/backend/app/downloads/formats.py
+++ b/backend/app/downloads/formats.py
@@ -26,10 +26,29 @@ from app import logs
 
 log = logs.get_logger(__name__)
 
-# yt-dlp puts the full source URL in most of its lines, and a source URL is a row of someone's
-# download history. The job id is already on every line (`logs.set_job` in the download child), so
-# the URL adds nothing an operator needs and plenty they should not keep.
+# 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.
 _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.
+_ID_PREFIX_RE = re.compile(r"(\[[\w:.-]+\] )(?!Destination:)[^\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\"']*")
+
+
+def log_safe(text: object) -> str:
+    """yt-dlp's text (its own lines, or a DownloadError's message) with the parts that name a
+    person's download taken out. Every log line that carries yt-dlp's words goes through here,
+    not just the adapter's: yt-dlp raises the same text it has just logged, so a line that skipped
+    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)
+    return _PATH_RE.sub("…", text)
 
 
 class _YdlLog:
@@ -41,19 +60,22 @@ class _YdlLog:
     2026.07.04: with `logger` set it calls `debug` for extraction and progress lines and
     `warning`/`error` for the rest, and `quiet`/`no_warnings` do NOT gate those calls.
 
-    `debug` goes to TRACE — it is several lines per second during a download, and TRACE is the
-    level a user turns on for their own session. Everything else keeps its level."""
+    `debug` goes to TRACE: it is several lines per second during a download, so it shows only
+    when the whole process runs at TRACE. (Not under a user's own Teljes consent — that lives in
+    the API process's request scope, and the download child that runs yt-dlp has none.)
 
-    @staticmethod
-    def _clean(msg: str) -> str:
-        return _URL_RE.sub(lambda m: f"{m.group(1)}/…", str(msg))
+    `error` goes to WARNING, not ERROR. yt-dlp calls it from `trouble()` just before it raises,
+    on EVERY failed attempt — including the ones the pipeline then recovers from with its fallback
+    client set. Whether the job failed is the pipeline's verdict (`download.job.failed`), and that
+    is the one line whose level should say so."""
 
     def _emit(self, level: int, msg: str) -> None:
         # The line goes as an ARGUMENT, not as the format string. yt-dlp writes progress as
         # `[download]  53.2% of 12.00MiB`, and while stdlib logging only applies `%` formatting to
         # a record that has args, this way the question never arises — and it is the shape ruff's
-        # `G` rules ask for.
-        log.log(level, "%s", self._clean(msg), extra=logs.event("download.ytdlp"))
+        # `G` rules ask for. (The sink's flood limiter groups a bare `%s` by the text itself, so
+        # distinct yt-dlp warnings do not share one bucket — `logsink._template`.)
+        log.log(level, "%s", log_safe(msg), extra=logs.event("download.ytdlp"))
 
     def debug(self, msg):
         self._emit(logs.TRACE, msg)
@@ -65,7 +87,7 @@ class _YdlLog:
         self._emit(logging.WARNING, msg)
 
     def error(self, msg):
-        self._emit(logging.ERROR, msg)
+        self._emit(logging.WARNING, msg)
 
 
 _YDL_LOG = _YdlLog()
diff --git a/backend/app/downloads/pipeline.py b/backend/app/downloads/pipeline.py
index 8353d5f0..ad737af1 100644
--- a/backend/app/downloads/pipeline.py
+++ b/backend/app/downloads/pipeline.py
@@ -383,7 +383,7 @@ def _extract_with_fallback(
                     "job %s: clients %s failed (%s); retrying with fallback",
                     job_id,
                     clients,
-                    str(exc)[:70],
+                    formats.log_safe(exc)[:70],
                 )
                 for p in list(staging.iterdir()):  # drop partials before the retry
                     try:
@@ -603,11 +603,12 @@ def _download(
             db.commit()
             if finalized:
                 _notify_download_done(db, owner_id, meta.title, job_id, kind="download")
+                # The asset id, not `rel`: the path is built from the channel and the title.
                 log.info(
-                    "job %s done: %s",
+                    "job %s done: asset %s",
                     job_id,
-                    rel,
-                    extra=logs.event("download.job.done", job_id=job_id),
+                    asset_id,
+                    extra=logs.event("download.job.done", job_id=job_id, asset_id=asset_id),
                 )
 
     except (_Aborted, editmod.EditAborted):
@@ -642,7 +643,7 @@ def _download(
         log.info(
             "job %s refused for its owner: %s",
             job_id,
-            msg,
+            formats.log_safe(msg),
             extra=logs.event("download.job.refused", job_id=job_id),
         )
     except Exception as exc:  # noqa: BLE001 — surface any failure to the user, keep worker alive
@@ -669,16 +670,8 @@ def _download(
                 phase=None,
             )
             db.commit()
-        # The traceback only when the classifier could not place it: a recognised yt-dlp
-        # failure is fully described by its code, and the stack under it is always this same
-        # `except`. An UNRECOGNISED one leaves nothing else to work from.
-        log.warning(
-            "job %s failed: %s",
-            job_id,
-            msg,
-            extra=logs.event("download.job.failed", job_id=job_id, error_code=diagnosis.code),
-            exc_info=diagnosis.code in (None, "", "unknown"),
-        )
+        # 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).
@@ -687,7 +680,37 @@ def _download(
         shutil.rmtree(staging, ignore_errors=True)
 
 
-def _fail_edit(job_id: int, asset_id: int, msg: str) -> None:
+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 _log_download_failed(job_id: int, msg: str, code: str | None, exc: BaseException) -> None:
+    """The child's `download.job.failed` line. `msg` is the user's copy and keeps their URL; the
+    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."""
+    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),
+    )
+
+
+def _fail_edit(job_id: int, asset_id: int, msg: str, exc: BaseException | None = None) -> 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:
@@ -703,7 +726,24 @@ def _fail_edit(job_id: int, asset_id: int, msg: str) -> None:
             phase=None,
         )
         db.commit()
-    log.warning("edit job %s failed: %s", job_id, msg)
+    # `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:
@@ -791,11 +831,12 @@ def _process_edit(job_id: int, asset_id: int) -> None:
             db.commit()
             if finalized:
                 _notify_download_done(db, owner_id, dl_title, job_id, kind="edit")
+                # The asset id, not `rel`: the path is built from the channel and the title.
                 log.info(
-                    "edit job %s done: %s",
+                    "edit job %s done: asset %s",
                     job_id,
-                    rel,
-                    extra=logs.event("edit.job.done", job_id=job_id),
+                    asset_id,
+                    extra=logs.event("edit.job.done", job_id=job_id, asset_id=asset_id),
                 )
 
     except editmod.EditAborted:
@@ -807,12 +848,7 @@ def _process_edit(job_id: int, asset_id: int) -> None:
             extra=logs.event("edit.job.aborted", job_id=job_id),
         )
     except Exception as exc:  # noqa: BLE001 — surface the failure, keep the worker alive
-        # `_fail_edit` stores `str(exc)` for the user, which is the wrong half for whoever has to
-        # fix it: an edit reaching here is by definition unclassified, so the stack is all there is.
-        log.exception(
-            "the edit job failed",
-            extra=logs.event("edit.job.failed", job_id=job_id, asset_id=asset_id),
-        )
-        _fail_edit(job_id, asset_id, str(exc)[:500])
+        # silent-ok: not silent — `_fail_edit` is the report, and it takes `exc` for its stack.
+        _fail_edit(job_id, asset_id, str(exc)[:500], exc)
     finally:
         shutil.rmtree(staging, ignore_errors=True)
diff --git a/backend/app/downloads/runner.py b/backend/app/downloads/runner.py
index 84d04c28..37c08f73 100644
--- a/backend/app/downloads/runner.py
+++ b/backend/app/downloads/runner.py
@@ -56,8 +56,9 @@ def main() -> None:
     logs.configure_logging("runner")
     _deprioritize()
     if len(sys.argv) != 2:
-        # noqa T201: this runs BEFORE configure_logging, and a usage error has to reach the
-        # operator who typed it whether or not logging was ever set up.
+        # noqa T201: a usage line is for the person who typed the command. Logging is already
+        # configured here, and in a deployment it writes JSON records — the one reader of this
+        # line wants the bare sentence, not a record with a request id they do not have.
         print(  # noqa: T201
             "usage: python -m app.downloads.runner <job_id>", file=sys.stderr
         )
diff --git a/backend/app/logs.py b/backend/app/logs.py
index 121f43a7..91dab5a1 100644
--- a/backend/app/logs.py
+++ b/backend/app/logs.py
@@ -415,7 +415,15 @@ def get_logger(name: str) -> logging.Logger:
     all called themselves `siftlode.plex`, so neither a level pinned per logger nor E22.6's console
     filter could tell a sync line from a stream line. Derived from `__name__` there is exactly one
     name per module, it cannot drift from the file it sits in, and `test_logger_names.py` fails if a
-    module hand-rolls one again."""
+    module hand-rolls one again.
+
+    `__main__` is resolved to the module it really is. The worker runs as `python -m app.worker`
+    and spawns `python -m app.downloads.runner`, so in the two long-running processes `__name__`
+    is `__main__` — taken at its word, both would log as the bare root. `-m` records the real name
+    on the module's spec; a plain script has none and stays the root."""
+    if name == "__main__":
+        spec = getattr(sys.modules.get("__main__"), "__spec__", None)
+        name = getattr(spec, "name", None) or name
     stem = name.removeprefix("app.") if name.startswith("app.") else ""
     # logger-name-ok: this call IS the derivation
     return logging.getLogger(f"siftlode.{stem}" if stem else "siftlode")
@@ -669,6 +677,16 @@ async def log_requests(request, call_next):
     return response
 
 
+def consented_trace() -> bool:
+    """Whether the current request's OWN user turned Teljes (TRACE) on — the gate for a line that
+    carries what they typed or what was quoted back to them. The level alone is not enough: an
+    admin running the whole process at TRACE would pass it for every user, and a scheduler thread,
+    which has no request, for none of them in particular. Guard such a line with this, then log it
+    at TRACE."""
+    scope = _scope.get()
+    return scope is not None and _traced(scope)
+
+
 def _traced(scope: _RequestScope) -> bool:
     """Whether this request's query string is logged: only under its OWN user's Teljes consent. An
     admin setting the whole process to TRACE gets more detail, not every user's search terms — the
diff --git a/backend/app/logsink.py b/backend/app/logsink.py
index e74252a5..924475ee 100644
--- a/backend/app/logsink.py
+++ b/backend/app/logsink.py
@@ -29,6 +29,7 @@ import atexit
 import json
 import logging
 import queue
+import re
 import sys
 import threading
 import time
@@ -137,6 +138,21 @@ def _notice(code: str, msg: str, **attrs: Any) -> dict[str, Any]:
     )
 
 
+_DIGITS = re.compile(r"\d+")
+
+
+def _template(record: logging.LogRecord) -> str:
+    """The flood limiter's grouping key: the message template. Except when the template is a bare
+    `%s` — a pass-through for someone else's text (yt-dlp's lines, `formats._YdlLog`), where the
+    template says nothing and every distinct line would share one bucket. Then the text itself
+    groups, with its numbers folded, so `Retrying fragment 5 (1/10)` and `(2/10)` are still one
+    repeat and a different warning is not crowded out by them."""
+    msg = str(record.msg)
+    if msg != "%s" or not record.args:
+        return msg
+    return _DIGITS.sub("#", record.getMessage())[:200]
+
+
 class _Repeats:
     """Per-template admission for one minute at a time. Used by the writer thread only (unlocked)."""
 
@@ -155,9 +171,11 @@ class _Repeats:
         self.seen[key] = (count or 0) + 1
         return self.seen[key] <= self.per_window
 
-    def roll(self, now: float) -> list[dict[str, Any]]:
-        """Close the window if it is over: one notice per template that went past the limit."""
-        if now - self.started < self.window:
+    def roll(self, now: float, *, final: bool = False) -> list[dict[str, Any]]:
+        """Close the window if it is over — or, `final`, because the process is: one notice per
+        template that went past the limit. Without `final` a process that lives under a minute
+        (every download child) would drop its repeats and never say so."""
+        if not final and now - self.started < self.window:
             return []
         notices = [
             _notice(
@@ -211,7 +229,7 @@ class LogSink(logging.Handler):
         key = (
             None
             if row["event"] == "http.request"
-            else (record.name, str(record.msg), record.levelno)
+            else (record.name, _template(record), record.levelno)
         )
         try:
             self._queue.put_nowait((key, row))
@@ -222,9 +240,10 @@ class LogSink(logging.Handler):
                 self._dropped += 1
 
     # --- writer side ---------------------------------------------------------------------------
-    def write_pending(self, wait: float = 0.0) -> int:
+    def write_pending(self, wait: float = 0.0, *, final: bool = False) -> int:
         """Write one batch: whatever is queued (up to BATCH_MAX), waiting up to `wait` seconds for the
-        first item. Returns the number of rows written, notices included."""
+        first item. Returns the number of rows written, notices included. `final` closes the
+        repeat window early — the last batch before the process exits."""
         rows: list[dict[str, Any]] = []
         try:
             item = self._queue.get(timeout=wait) if wait > 0 else self._queue.get_nowait()
@@ -242,7 +261,7 @@ class LogSink(logging.Handler):
             # silent-ok: the batch ends here because the queue ran dry, which is the normal end.
             except queue.Empty:
                 item = None
-        rows.extend(self._repeats.roll(time.monotonic()))
+        rows.extend(self._repeats.roll(time.monotonic(), final=final))
         with self._dropped_lock:
             dropped, self._dropped = self._dropped, 0
         if dropped:
@@ -276,6 +295,7 @@ class LogSink(logging.Handler):
             self.write_pending(wait=FLUSH_SECONDS)
         while self.write_pending():  # drain what arrived before the stop
             pass
+        self.write_pending(final=True)  # and the repeat notices of a window cut short
 
     def start(self) -> None:
         self._writer = threading.Thread(target=self._run, name="log-sink", daemon=True)
diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py
index e4b5934e..9975e1be 100644
--- a/backend/app/plex/stream.py
+++ b/backend/app/plex/stream.py
@@ -133,6 +133,8 @@ class HlsSession:
         # single index. Derived, not passed, so it cannot disagree with the argv.
         self.entry = _MASTER_PL if self.variants else _INDEX_PL
         self.last_access = time.time()
+        # `artifact_may_arrive` reports a refusing session directory once, not per segment.
+        self.probe_failure_reported = False
 
 
 def _seek_opts(start_s: float) -> list[str]:
@@ -929,17 +931,20 @@ def artifact_may_arrive(s: "HlsSession", filename: str) -> bool:
     except OSError as exc:
         # `exists()` answers False for a missing file without raising, so this is the session
         # directory itself refusing — a full or unmounted disk. Every segment then 404s, which the
-        # player shows as a stall. Per-segment, but the sink groups by message template, so a
-        # broken session is one grouped line rather than one per segment.
-        log.warning(
-            "a session segment could not be checked",
-            extra=logs.event(
-                "plex.stream.segment_probe_failed",
-                key=s.key,
-                seq=seq,
-                reason=type(exc).__name__,
-            ),
-        )
+        # player shows as a stall. Once per session: this runs for every segment request and
+        # every hls.js retry of one, and the sink's limiter still stores 30 a minute of a
+        # template — stdout would get them all.
+        if not s.probe_failure_reported:
+            s.probe_failure_reported = True
+            log.warning(
+                "a session segment could not be checked",
+                extra=logs.event(
+                    "plex.stream.segment_probe_failed",
+                    key=s.key,
+                    seq=seq,
+                    reason=type(exc).__name__,
+                ),
+            )
         return False
 
 
diff --git a/backend/app/plex/subtitles.py b/backend/app/plex/subtitles.py
index cfd04051..c09e97ec 100644
--- a/backend/app/plex/subtitles.py
+++ b/backend/app/plex/subtitles.py
@@ -359,7 +359,7 @@ def _canonical_cue(block: str) -> tuple[float, str, str] | None:
             st = _vtt_ts_to_s(left.strip())
             en = _vtt_ts_to_s(rparts[0])
         # silent-ok: None here means "not a cue", and the caller tells a header apart from a
-        # broken timing line by whether the block had a `-->` at all — see `_merge_vtt`.
+        # broken timing line by whether the block had a `-->` at all — see `merge_vtt`.
         except (ValueError, IndexError):
             return None
         lines[i] = f"{_vtt_s_to_ts(st)} --> {_vtt_s_to_ts(en)}{settings}"
diff --git a/backend/app/routes/channels.py b/backend/app/routes/channels.py
index f97a98eb..604bfaf2 100644
--- a/backend/app/routes/channels.py
+++ b/backend/app/routes/channels.py
@@ -165,9 +165,10 @@ def discover_channels(
                 user.yt_channel_id = own
                 db.commit()
         except YouTubeError as exc:
-            # No traceback anywhere in this file's YouTube handlers: `_upstream_failure` already
-            # logged the structured `youtube.api.error`, and the message carries method, path,
-            # status and reason. The stack under it is this `except` and the client call.
+            # No traceback anywhere in this file's YouTube handlers: for an upstream answer
+            # `_upstream_failure` already logged the structured `youtube.api.error`, and the
+            # message names the call, its status and its reason (or, with no answer at all, the
+            # transport failure). The stack under it is this `except` and the client call.
             log.warning(
                 "discovery: own-channel resolve failed: %s",
                 exc,
@@ -575,16 +576,11 @@ def subscribe(
             # subscription resync fills in the resource id. Any other YouTube error is a real
             # problem — surface it with a clear message (422, not a 502 that the client would
             # mistake for a transient "connection lost").
-            # On the reason code, not on the prose: `_upstream_failure` builds the message from
-            # the code now, and a message YouTube rewords would silently turn "already following"
-            # back into a 422 the user cannot act on. The substring check stays beside it for a
-            # body that carried no reason at all (`<none>` / `<unparseable>`).
-            msg = str(exc).lower()
-            if (
-                exc.reason == "subscriptionDuplicate"
-                or "already exists" in msg
-                or "duplicate" in msg
-            ):
+            # On the reason code, never on the prose: the error no longer carries the body (it is
+            # the request quoted back), and a message YouTube rewords would silently turn "already
+            # following" back into a 422 the user cannot act on. `reason` is comma-joined when the
+            # body listed several.
+            if "subscriptionDuplicate" in exc.reason.split(","):
                 yt_sub_id = ""
             else:
                 raise HTTPException(
diff --git a/backend/app/routes/plex/detail.py b/backend/app/routes/plex/detail.py
index 09386461..8a52df0e 100644
--- a/backend/app/routes/plex/detail.py
+++ b/backend/app/routes/plex/detail.py
@@ -405,13 +405,31 @@ def item_detail(
                         "text": codec not in subtitles.BITMAP_SUBS,
                     }
                 )
-    except Exception:  # noqa: BLE001 — playback works without the extras; never 500 here
-        # Best-effort, but not free: what this drops is the subtitle stream list and the intro
-        # markers, so the player offers no subtitles and no skip button — and looks like a file
-        # that has neither.
+    # Best-effort, but not free: what either branch drops is the subtitle stream list and the intro
+    # markers, so the player offers no subtitles and no skip button — and looks like a file that has
+    # neither.
+    except PlexNotConfigured:
+        # silent-ok: not a failure but a state — no Plex server, and nothing to play from anyway.
+        pass
+    except PlexError as exc:
         log.warning(
             "the live extras for an item could not be fetched",
-            extra=logs.event("plex.detail.extras_failed", rating_key=it.rating_key, kind=it.kind),
+            extra=logs.event(
+                "plex.detail.extras_failed",
+                rating_key=it.rating_key,
+                kind=it.kind,
+                reason=type(exc).__name__,
+            ),
+        )
+    except Exception:  # noqa: BLE001 — playback works without the extras; never 500 here
+        # Plex answered and WE could not read it — a shape our parsing does not expect. The same
+        # split `show_detail` makes: the stack is the only thing that says where.
+        log.warning(
+            "the live extras for an item could not be read",
+            extra=logs.event(
+                "plex.detail.extras_unreadable", rating_key=it.rating_key, kind=it.kind
+            ),
+            exc_info=True,
         )
 
     # Lazy enrichment (movies only): the live fetch above carries the FULL cast + rich detail (the sync
diff --git a/backend/app/routes/search.py b/backend/app/routes/search.py
index 7cf66d12..f06adedf 100644
--- a/backend/app/routes/search.py
+++ b/backend/app/routes/search.py
@@ -317,8 +317,8 @@ def search_youtube(
                 # themselves, and `redact_text` does not cover it — it scrubs addresses and
                 # credential params, not what someone typed into the search box. What the
                 # ERROR line carries instead is the shape of the failure, which is what tells
-                # an ingest bug from an upstream one; the term follows at TRACE, where the
-                # consent that allows it is the only thing that emits the record at all.
+                # an ingest bug from an upstream one; the term follows at TRACE, and only
+                # under this user's own consent — not whenever the process runs at TRACE.
                 log.exception(
                     "live search ingest failed",
                     extra=logs.event(
@@ -328,12 +328,13 @@ def search_youtube(
                         collected=len(collected),
                     ),
                 )
-                log.log(
-                    logs.TRACE,
-                    "the term the ingest failed on",
-                    # log-pii-ok: TRACE is the consented level, and this is what it is for
-                    extra=logs.event("search.live.term", term=term),
-                )
+                if logs.consented_trace():
+                    log.log(
+                        logs.TRACE,
+                        "the term the ingest failed on",
+                        # log-pii-ok: only under this user's own Teljes consent
+                        extra=logs.event("search.live.term", term=term),
+                    )
                 if collected:
                     break  # keep what earlier pages already gathered
                 raise HTTPException(
diff --git a/backend/app/sync/playlists.py b/backend/app/sync/playlists.py
index ece2c3f2..056c7a9d 100644
--- a/backend/app/sync/playlists.py
+++ b/backend/app/sync/playlists.py
@@ -418,8 +418,11 @@ class _Breaker:
     write site can't accidentally implement two thirds of the protocol (forgetting to record a
     failure would silently disable the breaker for that operation)."""
 
-    def __init__(self, limit: int = _MAX_CONSECUTIVE_FAILURES) -> None:
+    def __init__(
+        self, limit: int = _MAX_CONSECUTIVE_FAILURES, playlist_id: int | None = None
+    ) -> None:
         self.limit = limit
+        self.playlist_id = playlist_id  # for the failure line only
         self.streak = 0
         self.tripped = False
 
@@ -440,7 +443,9 @@ class _Breaker:
                 self.limit,
             )
 
-    def run(self, op: Callable[[], T], on_fail: Callable[[], None]) -> T | _Failed:
+    def run(
+        self, op: Callable[[], T], on_fail: Callable[[], None], action: str = "write"
+    ) -> T | _Failed:
         """Attempt one YouTube write. Returns the op's result, or `FAILED` when the breaker is
         already tripped (op not called) or the op raised YouTubeError. `on_fail` records the item
         in the caller's failure list either way."""
@@ -450,11 +455,17 @@ class _Breaker:
         try:
             out = op()
         except YouTubeError as exc:
-            # The client logged the call, but not WHICH item it was for — and the caller's failure
-            # list only ever reaches the user as a count.
+            # The client logged the call, but not which playlist it was for or what kind of write
+            # — and the caller's failure list only ever reaches the user as a count. Not the video:
+            # what is in someone's playlist is a row of their history, the same as a download.
             log.debug(
                 "a playlist write to YouTube failed",
-                extra=logs.event("sync.playlist.item_push_failed", reason=exc.reason),
+                extra=logs.event(
+                    "sync.playlist.item_push_failed",
+                    playlist_id=self.playlist_id,
+                    action=action,
+                    reason=exc.reason,
+                ),
             )
             on_fail()
             self.hit(False)
@@ -474,7 +485,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
     local = _local_video_ids(db, pl)
     inserted = deleted = reordered = created = pulled = removed_local = 0
     failures: list[str] = []
-    breaker = _Breaker()
+    breaker = _Breaker(playlist_id=pl.id)
     with YouTubeClient(db, user) as yt:
         if not pl.yt_playlist_id:
             pl.yt_playlist_id = yt.create_playlist(pl.name)
@@ -485,6 +496,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
                 added = breaker.run(
                     lambda v=vid: yt.add_playlist_item(pl.yt_playlist_id, v),
                     lambda v=vid: failures.append(v),
+                    "add",
                 )
                 if not isinstance(added, _Failed):
                     inserted += 1
@@ -567,6 +579,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
             dropped = breaker.run(
                 lambda i=item_id: yt.delete_playlist_item(i),
                 lambda v=vid: failures.append(v),
+                "delete",
             )
             if not isinstance(dropped, _Failed):
                 deleted += 1
@@ -576,6 +589,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
             item_id = breaker.run(
                 lambda v=vid: yt.add_playlist_item(pl.yt_playlist_id, v),
                 lambda v=vid: failures.append(v),
+                "add",
             )
             if not isinstance(item_id, _Failed):
                 item_id_by_vid[vid] = item_id
@@ -595,6 +609,7 @@ def merge_playlist(db: Session, user: User, pl: Playlist, *, live=None) -> dict:
                     item_id_by_vid[w], pl.yt_playlist_id, w, pos
                 ),
                 lambda w=want: failures.append(w),
+                "move",
             )
             if isinstance(moved, _Failed):
                 continue
diff --git a/backend/app/sysconfig.py b/backend/app/sysconfig.py
index 007cd38c..8b2f49b3 100644
--- a/backend/app/sysconfig.py
+++ b/backend/app/sysconfig.py
@@ -20,6 +20,10 @@ log = logs.get_logger(__name__)
 # Config keys whose stored ciphertext failed to decrypt — tracked so the warning is logged once per
 # key, not once per request (bounded by the number of secret config keys).
 _decrypt_warned: set[str] = set()
+# The same for a stored value that no longer coerces (a choice list that shrank in a release): it
+# fails on EVERY read of a hot knob until someone saves it again. Keyed with the value, so saving a
+# different bad value is reported afresh (bounded by the keys times the values ever saved).
+_coerce_warned: set[tuple[str, str]] = set()
 
 
 @dataclass(frozen=True)
@@ -270,12 +274,14 @@ def get(db: Session, key: str):
     except (TypeError, ValueError) as exc:
         # The admin set this in the UI and it is quietly not in force: the app runs on the env
         # default while the settings page goes on showing the value that was saved.
-        log.warning(
-            "a stored setting could not be coerced; the default is used instead",
-            extra=logs.event(
-                "sysconfig.override_uncoercible", key=s.key, reason=type(exc).__name__
-            ),
-        )
+        if (s.key, raw) not in _coerce_warned:
+            _coerce_warned.add((s.key, raw))
+            log.warning(
+                "a stored setting could not be coerced; the default is used instead",
+                extra=logs.event(
+                    "sysconfig.override_uncoercible", key=s.key, reason=type(exc).__name__
+                ),
+            )
         return _default(s)
 
 
diff --git a/backend/app/worker.py b/backend/app/worker.py
index b1ef63ac..5db9b49f 100644
--- a/backend/app/worker.py
+++ b/backend/app/worker.py
@@ -260,23 +260,40 @@ def _stop_and_free(job_id: int, asset_id: int, proc: subprocess.Popen, grace_s:
 
 
 def _fail_after_kill(
-    job_id: int, asset_id: int, proc: subprocess.Popen, grace_s: float, error: str, error_code: str
+    job_id: int,
+    asset_id: int,
+    proc: subprocess.Popen,
+    grace_s: float,
+    error: str,
+    error_code: str,
+    cause: str | None = None,
 ) -> 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)
-    _fail_running(job_id, error=error, error_code=error_code)
+    _fail_running(job_id, error=error, error_code=error_code, cause=cause)
 
 
 def _fail_running(
-    job_id: int, *, error: str, error_code: str | None, error_detail: str | None = None
+    job_id: int,
+    *,
+    error: str,
+    error_code: str | None,
+    error_detail: str | None = None,
+    cause: str | None = None,
 ) -> None:
     """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."""
+    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."""
     with SessionLocal() as db:
-        service.finalize_if_running(
+        failed = service.finalize_if_running(
             db,
             job_id,
             status=JobStatus.ERROR,
@@ -286,6 +303,15 @@ def _fail_running(
             phase=None,
         )
         db.commit()
+    if cause is not None and failed:
+        log.warning(
+            "job %s failed in the worker: %s",
+            job_id,
+            cause,
+            extra=logs.event(
+                "download.job.failed", job_id=job_id, error_code=error_code, cause=cause
+            ),
+        )
 
 
 def _run_supervised(job_id: int, asset_id: int) -> None:
@@ -341,7 +367,11 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             # 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)
+        log.exception(
+            "supervisor loop failed for job %s; stopping the child",
+            job_id,
+            extra=logs.event("download.job.failed", job_id=job_id, cause="supervisor_error"),
+        )
         _fail_after_kill(
             job_id,
             asset_id,
@@ -369,7 +399,15 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
                         error_code="unknown",
                     )
                     log.warning(
-                        "job %s: child exited %s without finishing", job_id, proc.returncode
+                        "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,
+                        ),
                     )
         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)
@@ -383,6 +421,7 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             grace_s,
             f"The download stalled (no progress for {int(stall_s)}s) and was stopped.",
             "network",
+            cause="stalled",
         )
         return
     if reason == "timeout":
@@ -393,6 +432,7 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
             grace_s,
             f"The download exceeded the {int(wall_s)}s time limit and was stopped.",
             "network",
+            cause="timeout",
         )
         return
     # cancelled / abandon / shutdown: stop + free the asset, but leave the job's own DB status.
@@ -401,10 +441,16 @@ def _run_supervised(job_id: int, asset_id: int) -> None:
         _requeue_if_running(job_id, None)  # resume on the next boot
     elif reason == "abandon":
         log.warning(
-            "job %s: killed a child wedged past the wall-clock; DB status left intact", job_id
+            "job %s: killed a child wedged past the wall-clock; DB status left intact",
+            job_id,
+            extra=logs.event("download.child.abandoned", job_id=job_id),
         )
     else:  # cancelled
-        log.info("job %s: child stopped (route pause/cancel)", job_id)
+        log.info(
+            "job %s: child stopped (route pause/cancel)",
+            job_id,
+            extra=logs.event("download.job.aborted", job_id=job_id),
+        )
 
 
 def _cleanup_staging(asset_id: int) -> None:
@@ -444,7 +490,7 @@ def _process_job(job_id: int) -> None:
             del cookie  # only whether there is one matters here
 
     if asset is None:
-        _fail_running(job_id, error="No media asset", error_code=None)
+        _fail_running(job_id, error="No media asset", error_code=None, cause="no_asset")
         return
 
     if asset_status == "ready":
@@ -456,6 +502,7 @@ def _process_job(job_id: int) -> None:
             error=asset_error or "Download failed",
             error_code=diagnosis.code,
             error_detail=diagnosis.detail,
+            cause="asset_failed",
         )
         return
     if asset_status == "downloading":
@@ -513,7 +560,11 @@ def _worker_loop(idx: int) -> None:
             try:
                 _process_job(job_id)
             except Exception:  # noqa: BLE001
-                log.exception("processing job %s crashed", job_id)
+                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)
 
 
diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py
index 04ee7482..260aff31 100644
--- a/backend/app/youtube/client.py
+++ b/backend/app/youtube/client.py
@@ -37,7 +37,11 @@ def _error_reason(resp: httpx.Response) -> str:
     body; the `message` beside it is prose, and upstream prose can quote the request back — a
     search term, a playlist title."""
     try:
-        errors = resp.json().get("error", {}).get("errors") or []
+        error = resp.json().get("error")
+        # The OAuth token endpoint's shape: `{"error": "invalid_grant", "error_description": …}`.
+        if isinstance(error, str):
+            return error or "<none>"
+        errors = (error or {}).get("errors") or []
         reasons = ",".join(e.get("reason", "") for e in errors if e.get("reason"))
     # silent-ok: the sentinel it returns goes straight into the caller's warning line, which
     # is the report. (Any non-JSON body — an HTML error page from a proxy.)
@@ -47,12 +51,13 @@ def _error_reason(resp: httpx.Response) -> str:
 
 
 def _upstream_failure(method: str, path: str, resp: httpx.Response) -> YouTubeError:
-    """Report one upstream failure and build the error to raise. Every failing YouTube call goes
-    through here, so what may appear in a log line cannot drift apart between the read, the write
-    and the delete path — and the error a caller raises carries the reason code rather than the
-    body, because callers put it in a log line (`routes/channels.py`) and in a 422 answered to the
-    user. The body itself is TRACE-only: it is the request quoted back, which the privacy policy
-    keeps out of the log unless the user turned Teljes on for themselves."""
+    """Report one upstream failure and build the error to raise. Every YouTube response that is
+    not a success goes through here — the Data API's reads, writes and deletes and the OAuth
+    token refresh — so what may appear in a log line cannot drift apart between them, and the
+    error a caller raises carries the reason code rather than the body, because callers put it in
+    a log line (`routes/channels.py`) and in a 422 answered to the user. (A request that got no
+    response at all is `_send`'s, with the reason `transport`.) The body is logged only under
+    the requesting user's own Teljes consent: it is the request quoted back."""
     reason = _error_reason(resp)
     log.warning(
         "YouTube %s %s -> %s (%s)",
@@ -68,12 +73,13 @@ def _upstream_failure(method: str, path: str, resp: httpx.Response) -> YouTubeEr
             reason=reason,
         ),
     )
-    log.log(
-        logs.TRACE,
-        "the upstream body behind it",
-        # log-pii-ok: TRACE is the consented level, and this is what it is for
-        extra=logs.event("youtube.api.body", body=resp.text[:500]),
-    )
+    if logs.consented_trace():
+        log.log(
+            logs.TRACE,
+            "the upstream body behind it",
+            # log-pii-ok: only under this user's own Teljes consent, which is what it is for
+            extra=logs.event("youtube.api.body", body=resp.text[:500]),
+        )
     return YouTubeError(f"{method} {path} -> {resp.status_code}: {reason}", reason=reason)
 
 
@@ -119,7 +125,7 @@ class YouTubeClient:
             return self._http.request(method, url, **kwargs)
         except httpx.HTTPError as exc:
             raise YouTubeError(
-                f"YouTube API unreachable ({exc.__class__.__name__}: {exc})"
+                f"YouTube API unreachable ({exc.__class__.__name__}: {exc})", reason="transport"
             ) from exc
 
     # --- auth ---
@@ -150,7 +156,7 @@ class YouTubeClient:
             },
         )
         if resp.status_code != 200:
-            raise YouTubeError(f"Token refresh failed: {resp.status_code} {resp.text[:200]}")
+            raise _upstream_failure("POST", "oauth2/token", resp)
         data = resp.json()
         access = data["access_token"]
         tok.access_token = encrypt(access)
diff --git a/backend/ruff.toml b/backend/ruff.toml
index e13006cb..9bb4cdf0 100644
--- a/backend/ruff.toml
+++ b/backend/ruff.toml
@@ -14,7 +14,7 @@ line-length = 100
 #   G       — the call's shape: %-style args, never an f-string the level filter cannot skip, and
 #             never `str(exc)` where `exc_info` would carry the traceback
 #   T20     — `print` instead of a log line. Two deliberate ones remain, each with its `# noqa`
-#             and a why: `downloads/runner.py` prints before logging is configured, and
+#             and a why: `downloads/runner.py`'s usage line is for whoever typed it, and
 #             `logsink.py` reports the SINK's own failure, which cannot be logged into the sink.
 #   TRY400  — `.error()` in an `except` where `.exception()` would keep the traceback
 #   TRY401  — the exception spelled into the message AND passed as `exc_info` (it is in both)
diff --git a/backend/tests/test_channel_subscribe_upstream.py b/backend/tests/test_channel_subscribe_upstream.py
new file mode 100644
index 00000000..a7f6331d
--- /dev/null
+++ b/backend/tests/test_channel_subscribe_upstream.py
@@ -0,0 +1,69 @@
+"""Subscribing when YouTube says no — and when "no" means "you already are".
+
+The error no longer carries YouTube's body (it is the request quoted back, and it went into a log
+line and a 422), so "already following" is recognised by the reason code and nothing else.
+DB lane.
+"""
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import select
+
+from app.models import Channel, Subscription
+from app.routes import channels
+from app.youtube.client import YouTubeError
+
+
+def _youtube_refusing(reason: str):
+    class _Client:
+        def __init__(self, db, user):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+        def insert_subscription(self, channel_id):
+            raise YouTubeError(f"POST subscriptions -> 400: {reason}", reason=reason)
+
+        def get_channels(self, ids):
+            return []
+
+    return _Client
+
+
+@pytest.fixture
+def stub_channel(db):
+    db.add(Channel(id="UCdup", title="Already followed"))
+    db.commit()
+    return "UCdup"
+
+
+@pytest.mark.parametrize("reason", ["subscriptionDuplicate", "quotaExceeded,subscriptionDuplicate"])
+def test_already_following_on_youtube_is_recorded_not_refused(
+    db, make_user, monkeypatch, stub_channel, reason
+):
+    user = make_user()
+    monkeypatch.setattr(channels, "YouTubeClient", _youtube_refusing(reason))
+
+    channels.subscribe(stub_channel, user=user, db=db)
+
+    sub = db.execute(
+        select(Subscription).where(
+            Subscription.user_id == user.id, Subscription.channel_id == stub_channel
+        )
+    ).scalar_one()
+    assert sub is not None
+
+
+@pytest.mark.parametrize("reason", ["quotaExceeded", "<none>", "<unparseable>"])
+def test_any_other_refusal_is_answered_with_a_422(db, make_user, monkeypatch, stub_channel, reason):
+    user = make_user()
+    monkeypatch.setattr(channels, "YouTubeClient", _youtube_refusing(reason))
+
+    with pytest.raises(HTTPException) as caught:
+        channels.subscribe(stub_channel, user=user, db=db)
+
+    assert caught.value.status_code == 422
diff --git a/backend/tests/test_failure_lines.py b/backend/tests/test_failure_lines.py
new file mode 100644
index 00000000..51a8c0cd
--- /dev/null
+++ b/backend/tests/test_failure_lines.py
@@ -0,0 +1,220 @@
+"""E22.5 review round 1: a failure is reported once, with its code, and a repeating one is not a
+flood.
+
+Each case here is a place where the first sweep got the line wrong in one of two directions: a
+failure that left no line at all (a download the watchdog killed), or one that left too many (the
+same edit failure twice, a warning per segment request, per settings read).
+"""
+
+import json
+import logging
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import yt_dlp
+
+from app import logs, sysconfig, worker
+from app.downloads import pipeline
+from app.plex import stream as plex_stream
+
+
+class _Sink(logging.Handler):
+    def __init__(self):
+        super().__init__(level=logging.NOTSET)
+        self.lines: list[str] = []
+        self.setFormatter(logs.JsonFormatter())
+        self.addFilter(logs.ContextFilter())
+
+    def emit(self, record):
+        self.lines.append(self.format(record))
+
+    def objects(self) -> list[dict]:
+        return [json.loads(line) for line in self.lines]
+
+    def events(self, code: str) -> list[dict]:
+        return [o for o in self.objects() if o.get("event") == code]
+
+
+@pytest.fixture
+def sink():
+    logger = logging.getLogger("siftlode")
+    handler = _Sink()
+    previous = logger.level
+    logger.addHandler(handler)
+    logger.setLevel(logs.TRACE)
+    try:
+        yield handler
+    finally:
+        logger.removeHandler(handler)
+        logger.setLevel(previous)
+
+
+class _NoDb:
+    """A session that finds nothing and commits nothing — these cases are about the line."""
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+    def get(self, *a, **k):
+        return None
+
+    def commit(self):
+        pass
+
+
+@pytest.fixture
+def finalized(monkeypatch):
+    """Whether `finalize_if_running` fires: True unless a case says a route got there first."""
+    state = {"fires": True}
+    monkeypatch.setattr(worker, "SessionLocal", _NoDb)
+    monkeypatch.setattr(pipeline, "SessionLocal", _NoDb)
+    fire = lambda db, job_id, **fields: state["fires"]  # noqa: E731
+    monkeypatch.setattr(worker.service, "finalize_if_running", fire)
+    return state
+
+
+# --- the worker's own failures ---------------------------------------------------------------
+@pytest.mark.parametrize("cause", ["stalled", "timeout"])
+def test_a_download_the_watchdog_kills_leaves_a_coded_line(sink, finalized, cause):
+    """The child is killed, so it never reports; before this the worker did not either, and a
+    stalled download failed with no line anywhere."""
+    worker._fail_running(42, error="stopped", error_code="network", cause=cause)
+
+    (line,) = sink.events("download.job.failed")
+    assert line["level"] == "WARNING"
+    assert line["attrs"] == {"job_id": 42, "error_code": "network", "cause": cause}
+
+
+def test_a_job_a_route_already_stopped_is_not_reported_as_failed(sink, finalized):
+    finalized["fires"] = False
+    worker._fail_running(42, error="stopped", error_code="network", cause="stalled")
+
+    assert sink.events("download.job.failed") == []
+
+
+def test_a_caller_with_its_own_line_is_not_reported_twice(sink, finalized):
+    worker._fail_running(42, error="Worker error", error_code=None)
+
+    assert sink.events("download.job.failed") == []
+
+
+# --- an edit failure, once ---------------------------------------------------------------------
+def test_an_edit_failure_is_one_line_with_its_code_and_our_own_stack(sink, finalized):
+    try:
+        raise TypeError("a spec shape nobody expected")
+    except TypeError as exc:
+        pipeline._fail_edit(9, 3, str(exc), exc)
+
+    (line,) = sink.events("edit.job.failed")
+    assert line["attrs"] == {"job_id": 9, "asset_id": 3, "error_type": "TypeError"}
+    assert "TypeError" in line["exc"]
+    assert len([o for o in sink.objects() if o["level"] in ("WARNING", "ERROR")]) == 1
+
+
+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."""
+    msg = "ffmpeg failed (1): /downloads/Some_Channel/Season_2024/A_Private_Title_a3.mp4: Invalid data"
+    pipeline._fail_edit(9, 3, msg, RuntimeError(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_whose_source_is_gone_is_found_by_the_same_code(sink, finalized):
+    pipeline._fail_edit(9, 3, "The source file is missing.")
+
+    (line,) = sink.events("edit.job.failed")
+    assert line["attrs"] == {"job_id": 9, "asset_id": 3}
+
+
+# --- a download failure, scrubbed ---------------------------------------------------------------
+def test_a_failed_download_line_does_not_put_back_what_the_adapter_took_out(sink):
+    """yt-dlp logs its error through the adapter (scrubbed) and then raises the same text, which
+    the pipeline logs again. Unscrubbed, that second line is the URL the first one hid."""
+    msg = "ERROR: Unsupported URL: https://example.org/private/path?id=123"
+    pipeline._log_download_failed(5, msg, "unknown", yt_dlp.utils.DownloadError(msg))
+
+    (line,) = sink.events("download.job.failed")
+    assert "private/path" not in json.dumps(line)
+    assert "example.org/…" in line["msg"]
+    # Unclassified, but yt-dlp's own error: its stack is the `except`, its foot the raw text.
+    assert "exc" not in line
+
+
+def test_an_unclassified_failure_of_our_own_keeps_its_stack(sink):
+    try:
+        raise KeyError("requested_formats")
+    except KeyError as exc:
+        pipeline._log_download_failed(5, str(exc), "unknown", exc)
+
+    (line,) = sink.events("download.job.failed")
+    assert "KeyError" in line["exc"]
+    assert line["attrs"]["error_type"] == "KeyError"
+
+
+def test_the_metadata_backfill_names_the_asset_and_not_the_download(sink, monkeypatch):
+    from app.downloads import adapters, backfill
+
+    url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
+
+    class _Refusing:
+        def __init__(self, opts):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+        def extract_info(self, target, download):
+            raise yt_dlp.utils.DownloadError(
+                f"ERROR: [youtube] dQw4w9WgXcQ: Video unavailable ({target})"
+            )
+
+    monkeypatch.setattr(backfill.yt_dlp, "YoutubeDL", _Refusing)
+    assert backfill._extract_meta(url, adapters.YouTubeAdapter(), 11) is None
+
+    lines = [o for o in sink.objects() if "meta extract failed" in o["msg"]]
+    assert lines and all("dQw4w9WgXcQ" not in json.dumps(o) for o in lines)
+
+
+# --- a repeating failure, once ------------------------------------------------------------------
+def test_a_stale_setting_is_reported_once_not_on_every_read(sink, monkeypatch):
+    spec = next(s for s in sysconfig.SPECS if s.type == "int")
+    monkeypatch.setattr(sysconfig, "_raw_override", lambda db, key: "not-a-number")
+    monkeypatch.setattr(sysconfig, "_coerce_warned", set())
+
+    for _ in range(3):
+        sysconfig.get(None, spec.key)
+
+    assert len(sink.events("sysconfig.override_uncoercible")) == 1
+
+
+class _RefusingDir:
+    """A session directory on a disk that has gone away: every probe raises."""
+
+    def __truediv__(self, name):
+        return SimpleNamespace(exists=self._refuse)
+
+    @staticmethod
+    def _refuse():
+        raise PermissionError("the disk went read-only")
+
+
+def test_a_refusing_session_directory_is_reported_once_per_session(sink):
+    """Every segment request probes, and hls.js retries each 404 — a line per probe is a flood."""
+    session = plex_stream.HlsSession("k", Path("."), None, 0.0, ())
+    session.dir = _RefusingDir()
+
+    for seq in (3, 4, 5):
+        assert not plex_stream.artifact_may_arrive(session, plex_stream.segment_name(seq, None))
+
+    (line,) = sink.events("plex.stream.segment_probe_failed")
+    assert line["attrs"]["reason"] == "PermissionError"
diff --git a/backend/tests/test_log_pii.py b/backend/tests/test_log_pii.py
index aa7a10e2..06c5a6f3 100644
--- a/backend/tests/test_log_pii.py
+++ b/backend/tests/test_log_pii.py
@@ -87,15 +87,62 @@ def test_an_upstream_failure_logs_the_reason_code_and_not_the_body(sink):
     assert err.reason == "quotaExceeded"
 
 
-def test_the_body_is_kept_for_the_one_user_who_consented_to_it(sink):
+@pytest.fixture
+def consented():
+    """The request scope of a user who turned Teljes (TRACE) on for themselves."""
+    scope = logs._RequestScope()
+    scope.user_id, scope.level = 7, logs.TRACE
+    token = logs._scope.set(scope)
+    try:
+        yield
+    finally:
+        logs._scope.reset(token)
+
+
+def test_the_body_is_kept_for_the_one_user_who_consented_to_it(sink, consented):
     _upstream_failure("GET", "search", _youtube_403(TERM))
 
-    # TRACE is emitted only while the user has turned Teljes on for their own session, which is
-    # exactly the consent the privacy policy describes. The diagnostic is not lost, it is gated.
+    # The diagnostic is not lost, it is gated — on the consent the privacy policy describes.
     (trace,) = sink.at("TRACE")
     assert TERM in trace["attrs"]["body"]
 
 
+def test_a_process_running_at_trace_is_not_anyones_consent(sink):
+    """`sink` runs the whole tree at TRACE — what an admin setting `log_level` to TRACE does. That
+    buys more detail, not every user's search terms quoted back (`logs._traced`)."""
+    _upstream_failure("GET", "search", _youtube_403(TERM))
+
+    assert sink.at("TRACE") == []
+    assert TERM not in json.dumps(sink.objects())
+
+
+def test_a_scope_with_no_elevation_is_not_consent():
+    scope = logs._RequestScope()
+    scope.user_id = 7
+    token = logs._scope.set(scope)
+    try:
+        assert not logs.consented_trace()
+    finally:
+        logs._scope.reset(token)
+    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):
+    """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."""
+    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)
+
+    (line,) = sink.at("WARNING")
+    assert line["attrs"]["reason"] == "invalid_grant"
+    assert err.reason == "invalid_grant"
+    assert "revoked" not in str(err) and "revoked" not in json.dumps(line)
+
+
 def test_an_unparseable_body_still_names_the_status(sink):
     """A proxy answering HTML has no `reason` to take — the line must still be worth reading, and
     must still not fall back to printing the body."""
@@ -153,9 +200,16 @@ _BANNED_NAMES = frozenset(
         "source_url",
         "source_webpage_url",
         "password",
+        # A download's path is built from its channel and its title (`storage.rel_path`), so it
+        # names what someone downloaded exactly as the URL does. Log the asset id.
+        "rel",
+        "rel_path",
+        "title",
     }
 )
-_BANNED_ATTRS = frozenset({"text", "content", "email", "url", "query"})
+_BANNED_ATTRS = frozenset(
+    {"text", "content", "email", "url", "query", "rel_path", "title", "display_name"}
+)
 _ESCAPE = "log-pii-ok"
 _LOG_METHODS = frozenset(
     {"debug", "info", "warning", "warn", "error", "exception", "critical", "log"}
@@ -211,3 +265,37 @@ def test_no_log_call_passes_what_a_person_typed():
         f"the argument (or the line directly above the call) `# {_ESCAPE}: <why it is "
         f"safe>` — {offenders}"
     )
+
+
+def _guarded_by_consent(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> bool:
+    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
+    return False
+
+
+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
+    scheduler thread has no user at all. So the line must sit under `logs.consented_trace()`."""
+    root = pathlib.Path(__file__).resolve().parents[1] / "app"
+    offenders = []
+    for path in sorted(root.rglob("*.py")):
+        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)}
+        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):
+                offenders.append(f"{path.relative_to(root)}:{node.lineno}")
+    assert offenders == [], (
+        f"a TRACE line with user content is not under consented_trace(): {offenders}"
+    )
diff --git a/backend/tests/test_logger_names.py b/backend/tests/test_logger_names.py
index 364cf326..033e2c1d 100644
--- a/backend/tests/test_logger_names.py
+++ b/backend/tests/test_logger_names.py
@@ -13,6 +13,8 @@ the rule, because the rule is only worth having if a hand-rolled name fails the
 import ast
 import logging
 import pathlib
+import sys
+import types
 
 import pytest
 
@@ -29,8 +31,8 @@ _ESCAPE = "logger-name-ok"
         ("app.plex.stream", "siftlode.plex.stream"),
         ("app.routes.channels", "siftlode.routes.channels"),
         ("app.worker", "siftlode.worker"),
-        # Not a module of ours (a script, a doctest): it lands on the root of our tree rather than
-        # inventing `siftlode.__main__`, which nothing would think to filter on.
+        # A script that is not a module of ours (pytest itself, here): it lands on the root of our
+        # tree rather than inventing `siftlode.__main__`, which nothing would think to filter on.
         ("__main__", "siftlode"),
     ],
 )
@@ -38,6 +40,32 @@ def test_the_name_follows_the_module(module, expected):
     assert logs.get_logger(module).name == expected
 
 
+@pytest.mark.parametrize(
+    ("spec_name", "expected"),
+    [
+        # How the worker and the download child actually run: `python -m app.worker`, and the
+        # worker's own `[sys.executable, "-m", "app.downloads.runner", ...]`. `__name__` is
+        # `__main__` in both, and taken at its word both processes logged as the bare root.
+        ("app.worker", "siftlode.worker"),
+        ("app.downloads.runner", "siftlode.downloads.runner"),
+    ],
+)
+def test_a_module_run_with_dash_m_keeps_its_own_name(monkeypatch, spec_name, expected):
+    main = types.ModuleType("__main__")
+    main.__spec__ = types.SimpleNamespace(name=spec_name)
+    monkeypatch.setitem(sys.modules, "__main__", main)
+
+    assert logs.get_logger("__main__").name == expected
+
+
+def test_a_plain_script_has_no_spec_and_stays_the_root(monkeypatch):
+    main = types.ModuleType("__main__")
+    main.__spec__ = None
+    monkeypatch.setitem(sys.modules, "__main__", main)
+
+    assert logs.get_logger("__main__").name == "siftlode"
+
+
 def test_every_derived_logger_stays_under_the_one_tree():
     """`configure_logging` hangs the level and the handler off the root, and the tests attach their
     sink to `siftlode` — a name outside that tree would be invisible to both."""
@@ -62,11 +90,25 @@ def test_no_module_hand_rolls_a_logger_name():
             if not isinstance(node, ast.Call):
                 continue
             func = node.func
-            if not (isinstance(func, ast.Attribute) and func.attr == "getLogger"):
-                continue
-            # A no-argument `logging.getLogger()` is the ROOT logger — level plumbing, not a
-            # module's own logger, and nothing to name.
-            if not node.args:
+            called = (
+                func.attr
+                if isinstance(func, ast.Attribute)
+                else func.id
+                if isinstance(func, ast.Name)
+                else ""
+            )
+            if called == "get_logger":
+                # The derivation itself is only as good as its argument: a literal here does not
+                # name a module, it collapses to the root and pools modules together again.
+                arg = node.args[0] if len(node.args) == 1 else None
+                if isinstance(arg, ast.Name) and arg.id == "__name__" and not node.keywords:
+                    continue
+            elif called == "getLogger":
+                # A no-argument `logging.getLogger()` is the ROOT logger — level plumbing, not a
+                # module's own logger, and nothing to name. A `name=` keyword is still a name.
+                if not node.args and not node.keywords:
+                    continue
+            else:
                 continue
             start = max(0, node.lineno - 2)  # the call's lines, plus the one above it
             span = " ".join(lines[start : (node.end_lineno or node.lineno)])
diff --git a/backend/tests/test_logsink.py b/backend/tests/test_logsink.py
index b107c3dd..3dc4323a 100644
--- a/backend/tests/test_logsink.py
+++ b/backend/tests/test_logsink.py
@@ -90,6 +90,40 @@ def test_a_flood_of_one_message_is_stored_up_to_the_limit_and_then_summarised(st
     assert notice.attrs["template"] == "RSS poll skipped for channel %s: HTTP 404"
 
 
+def test_a_pass_through_template_groups_by_its_text_not_by_the_bare_placeholder(stored):
+    """`"%s"` is how the yt-dlp adapter hands over yt-dlp's own line, and as a key it says nothing:
+    every distinct warning would share one bucket and the thirty-first — whatever it said — would be
+    dropped as a "repeat" of the first. The text groups instead, numbers folded, so a retry storm is
+    still one repeat and the warning that explains the failure still gets through."""
+    sink, log, rows = stored
+    limit = logsink.REPEATS_PER_MINUTE
+    for i in range(limit + 10):
+        log.warning("%s", f"[download] Got error: timed out. Retrying fragment {i} (1/10)...")
+    log.warning("%s", "[youtube] …: Requested format is not available")
+    stored_rows = rows()
+    assert len(stored_rows) == limit + 1
+    assert stored_rows[-1].msg == "[youtube] …: Requested format is not available"
+    sink._repeats.started -= logsink.REPEAT_WINDOW_SECONDS
+    notice = rows()[-1]
+    assert notice.attrs["suppressed"] == 10
+    assert notice.attrs["template"].startswith(
+        "[download] Got error: timed out. Retrying fragment #"
+    )
+
+
+def test_a_process_that_ends_inside_the_minute_still_says_what_it_dropped(stored):
+    """Every download child lives well under the limiter's minute. Its last batch closes the window
+    early, or the repeats it dropped would go unreported."""
+    sink, log, rows = stored
+    for i in range(logsink.REPEATS_PER_MINUTE + 3):
+        log.warning("RSS poll skipped for channel %s: HTTP 404", f"UC{i}")
+    rows()
+    sink.write_pending(final=True)
+    notice = rows()[-1]
+    assert notice.event == "log.repeats_suppressed"
+    assert notice.attrs["suppressed"] == 3
+
+
 def test_request_lines_are_never_treated_as_a_flood(stored):
     _sink, log, rows = stored
     for _ in range(logsink.REPEATS_PER_MINUTE + 5):
diff --git a/backend/tests/test_ytdlp_logging.py b/backend/tests/test_ytdlp_logging.py
index b6ac4421..a17cc84a 100644
--- a/backend/tests/test_ytdlp_logging.py
+++ b/backend/tests/test_ytdlp_logging.py
@@ -59,7 +59,10 @@ def sink():
         ("debug", "TRACE"),
         ("info", "DEBUG"),
         ("warning", "WARNING"),
-        ("error", "ERROR"),
+        # yt-dlp calls `error` from `trouble()` just before it raises, on every failed ATTEMPT —
+        # the pipeline's fallback client set then often succeeds. The job's verdict is the
+        # pipeline's `download.job.failed`, not this.
+        ("error", "WARNING"),
     ],
 )
 def test_each_ytdlp_level_lands_where_it_belongs(sink, method, level):
@@ -83,6 +86,45 @@ def test_the_source_url_does_not_travel_with_the_explanation(sink):
     assert "Requested format is not available" in line["msg"]
 
 
+@pytest.mark.parametrize(
+    ("line", "kept", "gone"),
+    [
+        # The shape nearly every yt-dlp warning and error has: no URL, the id as a prefix.
+        (
+            "ERROR: [youtube] dQw4w9WgXcQ: Video unavailable",
+            "ERROR: [youtube] …: Video unavailable",
+            "dQw4w9WgXcQ",
+        ),
+        # The generic extractor's "id" is the URL's own basename.
+        (
+            "WARNING: [generic] my-private-holiday-clip: Falling back on generic information",
+            "WARNING: [generic] …: Falling back on generic information",
+            "my-private-holiday-clip",
+        ),
+        # `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",
+            "[download] Destination: …",
+            "A_Private_Title",
+        ),
+    ],
+)
+def test_what_names_the_download_comes_out_of_real_ytdlp_shapes(sink, line, kept, gone):
+    formats._YDL_LOG.warning(line)
+
+    (obj,) = sink.objects()
+    assert obj["msg"] == kept
+    assert gone not in json.dumps(obj)
+
+
+def test_a_download_error_is_scrubbed_the_same_way_wherever_it_is_logged():
+    """yt-dlp raises the text it has just logged, and the pipeline logs the exception again one
+    line later (`download.job.failed`). The same scrub, or that line puts the URL back."""
+    text = "ERROR: Unsupported URL: https://example.org/private/path?id=123"
+
+    assert formats.log_safe(text) == "ERROR: Unsupported URL: example.org/…"
+
+
 def test_a_line_with_no_url_is_left_alone(sink):
     formats._YDL_LOG.warning("nsig extraction failed: Some formats may be missing")