0.6.5: HTML Telegram digest + fix silently-empty beet sections + watermark-art resilience

The daily digest is now formatted with Telegram HTML: bold section headers
with <pre>-aligned columns, colored circle emoji as status dots, a branded
header line, and a top blockquote callout (all clear vs N issues). All free
text is &/</> escaped so a stray angle bracket in a log snippet can't break
the parse and swallow the whole message. notify-telegram.sh gains an --html
flag (parse_mode=HTML) used by the digest send; plain-text callers are
unchanged. Truncation is now tag-safe in HTML mode: cut on a line boundary,
re-close an open <pre>, and use an escaped marker -- the old literal
"...<truncated>" tail would itself have 400'd the send.

Two real bugs surfaced while testing:

- pipeline-status.sh pins its own PATH, which lacked /opt/venv/bin where
  beet lives, so every beet probe ("added today", "Library by format", the
  mp3-now count) has been silently empty behind 2>/dev/null since the cron
  migration. PATH now includes the venv; both sections show real numbers.

- strip-watermark-art.py aborted its entire weekly run when metaflac stalled
  on ONE file (seen today: a healthy 190KB cover took >15s under disk
  contention, TimeoutExpired killed the job). Per-file timeout is now 60s
  and a timeout skips that file with a warning instead of failing the run.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

README.md

@@ -73,10 +73,10 @@ Optional, add these later if you want them:
 Pull the prebuilt image onto your Docker host:
 
 ```bash
-docker pull git.kretzer.club/andrew/alembic:0.6.2
+docker pull git.kretzer.club/andrew/alembic:0.6.5
 ```
 
-That is the whole install. You do not need to download the source or build anything. The `0.6.2` is the version; you can pin to it so nothing changes under you, or use `latest` to always get the newest.
+That is the whole install. You do not need to download the source or build anything. The `0.6.5` is the version; you can pin to it so nothing changes under you, or use `latest` to always get the newest.
 
 (If you would rather build it yourself from source, you can, but you do not need to.)
 
@@ -136,7 +136,7 @@ Create a file called `docker-compose.yml` on your server (put it wherever you ke
 ```yaml
 services:
   alembic:
-    image: git.kretzer.club/andrew/alembic:0.6.2
+    image: git.kretzer.club/andrew/alembic:0.6.5
     container_name: alembic
     ports:
       - "8420:8420"
@@ -211,6 +211,8 @@ Steps:
 
 That's it, done once. alembic stores the resulting OAuth refresh token (encrypted, same as your other credentials) and uses it to silently mint a fresh access token whenever it needs one — you never have to repeat this unless you disconnect or revoke access on Spotify's side.
 
+One more limit that comes with a new Spotify app: Spotify only returns a playlist's contents to the account that owns it (or collaborates on it), even after you connect. So sync playlists that belong to the account you connect, and if someone shares a playlist with you, save your own copy of it first (or have them add you as a collaborator).
+
 If you happen to be running a Spotify app created *before* the 2025/2026 change, plain Client ID/Secret still works and this step is optional — but connecting is harmless either way, so there's no reason not to.
 
 > **If you connected Spotify before version 0.6.1:** click **Connect Spotify** again. Versions before 0.6.1 requested a narrower OAuth scope than playlist reads actually need, which could get you a token that refreshes fine but then gets 403 Forbidden loading a playlist (see Troubleshooting). Reconnecting grants the correct scope; your old connection doesn't upgrade itself.
@@ -290,11 +292,13 @@ Your Spotify app can't read the playlist with app-only (client-credentials) acce
 
 The fix: go to **Settings → Credentials** and click **Connect Spotify** to complete the OAuth login (see "Connecting your Spotify account (OAuth)" above). This mints a user token that works regardless of when your app was created. If you haven't already, add the redirect URI `https://<your-host>/connect/spotify/callback` in your Spotify app's dashboard first, or the connect step itself will fail with a redirect_uri mismatch.
 
-**A playlist's job log shows `Spotify returned 403 loading this playlist` even though you've connected Spotify.** (On versions before 0.6.2 the same problem showed up as `sldl crashed (exit 134)` mentioning a 403 Forbidden.)
-Two possible causes, in order of likelihood:
+**A playlist's job log shows `Spotify returned 403 loading this playlist`, `Spotify returned this playlist without its contents`, or `0 visible tracks` even though you've connected Spotify.** (On versions before 0.6.2 these showed up as `sldl crashed (exit 134)` mentioning a 403 Forbidden; on 0.6.2 a playlist you don't own could finish "successfully" while downloading nothing.)
+Possible causes, in order of likelihood:
 
-1. **You connected before version 0.6.1.** Earlier versions requested a narrower OAuth scope than playlist reads actually need, so the token refreshes fine but then gets 403 loading a playlist regardless of ownership. Fix: go to **Settings → Credentials** and click **Connect Spotify** again to grant the correct scope.
-2. **The connected account genuinely can't see this playlist.** Once connected, playlist reads are scoped to what that account can actually see — its own playlists, ones it collaborates on, or ones marked Public — not any arbitrary playlist by ID the way app-only access could. Fix: make sure the playlist is owned by the connected account, or set it to Public on Spotify (Playlist → ⋯ → Make public).
+1. **The playlist isn't owned by the connected account.** For Spotify apps created after the 2025/2026 API change, Spotify only returns a playlist's contents to its owner or a collaborator on it. Being public is NOT enough. Fix: recreate the playlist under the account you connected via **Connect Spotify**, or have the owner make it collaborative and add you.
+2. **You connected before version 0.6.1.** Earlier versions requested a narrower OAuth scope than playlist reads actually need, so the token refreshes fine but then gets 403 loading a playlist regardless of ownership. Fix: go to **Settings → Credentials** and click **Connect Spotify** again to grant the correct scope.
+
+(If your Spotify app is old enough to be grandfathered, both public playlists and your own keep working like before.)
 
 Either way, re-run the playlist after fixing it.
 

app/routers/playlists.py

@@ -14,12 +14,16 @@ templates = Jinja2Templates(directory="app/templates")
 def _friendly_status_error(exc: Exception) -> str:
     """Turn a raw Spotify API error into something a non-technical user can act
     on. 403 here almost always means the Spotify app can't read the playlist:
-    either the credentials are wrong or the playlist isn't public."""
+    wrong credentials, or the connected account can't see it. (The other
+    can't-read case, contents hidden for playlists the account doesn't own,
+    arrives as a RuntimeError from spotify_client with its own message.)"""
     if isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code == 403:
         return (
             "Spotify denied access (403). Check your Spotify credentials under "
-            "Settings then Credentials, and make sure the playlist is set to public "
-            "on Spotify -- alembic can't read private playlists."
+            "Settings then Credentials, and make sure the account connected via "
+            "Connect Spotify can see this playlist. For newly created Spotify "
+            "apps the connected account must own the playlist or be a "
+            "collaborator on it."
         )
     return str(exc)
 

app/services/pipeline_runner.py

@@ -12,6 +12,15 @@ from app.settings import settings
 # all-jobs-share-one-lock behavior exactly rather than over-engineering it.
 _lock = asyncio.Lock()
 
+# How long a SCHEDULED run waits for the lock before giving up (recorded as
+# skipped_lock). The old flock -n instant-skip starves fixed-time schedules:
+# the Sunday 08:30 strip-mb-tags run took 10m19s on 2026-07-12, so the 08:35
+# and 08:40 jobs both hit the held lock and silently lost their only slot of
+# the week. Scheduled jobs have nobody watching, so queueing (bounded) beats
+# skipping; manual runs keep the instant skip because a person clicking "Run
+# now" should get an immediate answer, not a silent 30-minute wait.
+SCHEDULED_LOCK_WAIT_SECONDS = 1800.0
+
 _ENV_PASSTHROUGH_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TZ")
 
 
@@ -94,18 +103,30 @@ async def _execute(
     timeout: float | None,
     use_lock: bool,
     capture: bool,
+    lock_wait: float | None,
 ) -> tuple[JobRun, str]:
     """Shared core for run_job/run_job_capture: acquire the lock (unless
     use_lock=False), record the run, exec the subprocess, and finalize. When
     capture=True, stdout+stderr is captured as text and returned; otherwise it
     streams straight to the log file. Returns (JobRun, output_text) -- the text
-    is "" in the non-capture and skipped-lock cases."""
+    is "" in the non-capture and skipped-lock cases.
+
+    lock_wait: how long to wait for a held lock before recording
+    skipped_lock. None picks the policy default: SCHEDULED_LOCK_WAIT_SECONDS
+    for triggered_by="schedule", instant skip for everything else."""
     started_at = time.time()
+    if lock_wait is None:
+        lock_wait = SCHEDULED_LOCK_WAIT_SECONDS if triggered_by == "schedule" else 0.0
 
     acquired = False
     if use_lock:
         if not await _try_acquire_nowait():
-            return _record_run(job_key, started_at, triggered_by, finished_at=started_at, status="skipped_lock"), ""
+            if lock_wait <= 0:
+                return _record_run(job_key, started_at, triggered_by, finished_at=started_at, status="skipped_lock"), ""
+            try:
+                await asyncio.wait_for(_lock.acquire(), timeout=lock_wait)
+            except asyncio.TimeoutError:
+                return _record_run(job_key, started_at, triggered_by, finished_at=time.time(), status="skipped_lock"), ""
         acquired = True
 
     try:
@@ -170,18 +191,21 @@ async def run_job(
     triggered_by: str = "schedule",
     timeout: float | None = None,
     use_lock: bool = True,
+    lock_wait: float | None = None,
 ) -> JobRun:
     """Run a pipeline command under the shared mutual-exclusion lock,
-    recording one job_runs row start-to-finish. If the lock is already
-    held, records status='skipped_lock' immediately and returns without
-    running anything -- the old flock -n behavior, now visible in the UI
-    instead of silently skipping.
+    recording one job_runs row start-to-finish. If the lock is already held,
+    scheduled runs (triggered_by="schedule") wait up to
+    SCHEDULED_LOCK_WAIT_SECONDS for it before recording skipped_lock -- see
+    that constant for why -- while manual/other runs record skipped_lock
+    immediately (the old flock -n behavior, visible in the UI instead of
+    silently skipping). Pass lock_wait to override either way.
 
     use_lock=False runs the command WITHOUT taking the pipeline lock, for
     read-only jobs that never touch the library (e.g. the status report).
     Those must never be starved by a long-running write job, and running
     them concurrently is safe."""
-    run, _ = await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=False)
+    run, _ = await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=False, lock_wait=lock_wait)
     return run
 
 
@@ -191,6 +215,7 @@ async def run_job_capture(
     triggered_by: str = "manual",
     timeout: float | None = None,
     use_lock: bool = True,
+    lock_wait: float | None = None,
 ) -> tuple[JobRun, str]:
     """Like run_job(), but captures stdout+stderr as text and returns it
     alongside the JobRun, instead of only writing it to the log file --
@@ -198,7 +223,7 @@ async def run_job_capture(
     dedup_review_service's dry-run scan. The full output is still written
     to a log file afterward so job_runs.log_path works the same as any
     other job."""
-    return await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=True)
+    return await _execute(job_key, argv, triggered_by, timeout, use_lock, capture=True, lock_wait=lock_wait)
 
 
 async def run_playlist(playlist_name: str, no_m3u: bool = False, triggered_by: str = "schedule") -> JobRun:

app/services/spotify_client.py

@@ -77,18 +77,34 @@ def get_playlist_tracks(db: Session, playlist_url: str) -> list[dict]:
 
     tracks = []
     # /items, not /tracks: Spotify removed GET /playlists/{id}/tracks in its
-    # February 2026 API changes in favor of /items (same response shape). New
-    # apps are 403'd on the old endpoint; grandfathered apps still tolerate it
-    # for now, but /items is the correct, future-proof one.
+    # February 2026 API changes. The response shape ALSO changed, but only for
+    # apps on the new behavior: each entry's payload moved from "track" to
+    # "item" (tracks.tracks.track -> items.items.item). Extended Quota Mode
+    # (grandfathered) apps keep the old "track" key, so parse both. No
+    # `fields` filter: it selects by key name, so on the renamed shape a
+    # track(...) filter silently returns empty pages -- exactly the failure
+    # we're avoiding.
     url = f"{API_BASE}/playlists/{playlist_id}/items"
-    params = {"limit": 100, "fields": "items(track(name,artists(name),external_ids)),next"}
+    params = {"limit": 100}
 
     while url:
         resp = httpx.get(url, params=params, headers=headers, timeout=15)
         resp.raise_for_status()
         data = resp.json()
-        for item in data.get("items", []):
-            track = item.get("track")
+        if "items" not in data:
+            # New-behavior apps get metadata only (no items field at all) for
+            # playlists the connected account doesn't own or collaborate on --
+            # public is no longer sufficient. Same message style the playlist
+            # page shows for a 403.
+            raise RuntimeError(
+                "Spotify returned this playlist without its contents. For newly "
+                "created Spotify apps, the account connected via Connect Spotify "
+                "must own the playlist (or be a collaborator on it) -- ask the "
+                "owner to share it as collaborative, or recreate it under the "
+                "connected account."
+            )
+        for item in data["items"]:
+            track = item.get("track") or item.get("item")
             if not track:
                 continue  # local files / removed tracks show up as null
             artists = track.get("artists") or []

pipeline/bin/run-playlist.sh

@@ -84,6 +84,18 @@ if ! SPOTIFY_CLIENT_ID="$SPOTIFY_CLIENT_ID" SPOTIFY_CLIENT_SECRET="$SPOTIFY_CLIE
   exit 1
 fi
 
+# Surface the fetched count in the timestamped summary. Zero tracks from a
+# successful fetch is almost never a truly empty playlist -- it usually means
+# Spotify hid the contents from the connected account (see the pointed errors
+# in spotify-playlist-csv.py), so call it out instead of letting sldl no-op
+# with a clean exit 0.
+TRACK_COUNT=$(($(wc -l < "$CSV_FILE") - 1))
+if [[ "$TRACK_COUNT" -le 0 ]]; then
+  log "WARNING: Spotify returned 0 visible tracks for this playlist -- nothing to download. If the playlist isn't actually empty, the connected Spotify account can't see its contents (it must own or collaborate on the playlist for newly created Spotify apps)."
+else
+  log "Fetched $TRACK_COUNT track(s) from Spotify"
+fi
+
 # ==== Run sldl (vendored binary, subprocess of this container) ====
 log "Running sldl for: $PLAYLIST_NAME"
 

pipeline/lib/notify-telegram.sh

@@ -5,14 +5,29 @@
 # Usage:
 #   echo "single line message" | notify-telegram.sh
 #   notify-telegram.sh "single line message"
-#   notify-telegram.sh < ${ALEMBIC_CONFIG_DIR:-/config}/logs/STATUS.log
+#   notify-telegram.sh --html < ${ALEMBIC_CONFIG_DIR:-/config}/logs/STATUS.log
 #
-# Telegram messages are capped at 4096 chars. Anything longer is truncated
-# with a "...<truncated>" tail. Returns exit 0 on send-OK, non-zero otherwise.
+# --html sends with parse_mode=HTML for messages authored as Telegram-HTML
+# (pipeline-status.sh's digest: <b>/<i>/<code>/<pre>/<blockquote>). The
+# CALLER is responsible for escaping &, <, > in any free text; this script
+# only guarantees the truncation below can't cut a tag in half. Without the
+# flag, messages go as plain text exactly as before.
+#
+# Telegram messages are capped at 4096 chars. Anything longer is truncated;
+# in HTML mode the cut lands on a line boundary and re-closes an open <pre>
+# so the truncated message still parses (a mid-tag cut, or a bare "<" in the
+# tail marker, makes the Bot API reject the ENTIRE message with a 400).
+# Returns exit 0 on send-OK, non-zero otherwise.
 
 set -euo pipefail
 PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
 
+PARSE_MODE=""
+if [[ "${1:-}" == "--html" ]]; then
+  PARSE_MODE="HTML"
+  shift
+fi
+
 CONFIG="${ALEMBIC_CONFIG_DIR:-/config}/pipeline/telegram/notify.env"
 if [[ ! -f "$CONFIG" ]]; then
   echo "[notify-telegram] no config at $CONFIG" >&2
@@ -30,24 +45,37 @@ else
   MSG=$(cat)
 fi
 
-# Telegram caps at 4096 chars (UTF-8 codepoints). Trim conservatively at 3900.
-# Use python for proper UTF-8 length handling.
+# Telegram caps at 4096 chars. Trim conservatively at 3900. Use python for
+# proper UTF-8 length handling. Plain mode appends a literal marker; HTML
+# mode cuts at the last newline inside the budget (our tags never span
+# lines except <pre> blocks) and re-closes an unbalanced <pre>.
 MSG=$(python3 -c "
 import sys
 s = sys.argv[1]
+html = sys.argv[2] == 'HTML'
 if len(s) > 3900:
-    s = s[:3900] + '\n...<truncated>'
+    s = s[:3900]
+    if html:
+        cut = s.rfind('\n')
+        if cut > 0:
+            s = s[:cut]
+        if s.count('<pre>') > s.count('</pre>'):
+            s += '</pre>'
+        s += '\n<i>… truncated</i>'
+    else:
+        s += '\n...(truncated)'
 print(s, end='')
-" "$MSG")
+" "$MSG" "${PARSE_MODE:-plain}")
 
-# Send via Bot API. Disable web-page preview and use plain text (no parse_mode)
-# so log content with special chars doesn't get interpreted as Markdown.
+# Send via Bot API. Preview disabled; parse_mode only when requested so log
+# content with special chars can't be misread as markup in plain sends.
 # `|| true`: a curl timeout/network error must fall through to the explicit
 # ok-check below (which reports "send failed" and exits 2), not abort here.
 resp=$(curl -s --max-time 15 -X POST \
   "https://api.telegram.org/bot${TG_BOT_TOKEN}/sendMessage" \
   --data-urlencode "chat_id=${TG_CHAT_ID}" \
   --data-urlencode "text=${MSG}" \
+  ${PARSE_MODE:+--data-urlencode "parse_mode=${PARSE_MODE}"} \
   --data-urlencode "disable_web_page_preview=true" || true)
 
 ok=$(echo "$resp" | python3 -c "import sys,json;print(json.load(sys.stdin).get('ok',False))" 2>/dev/null || echo False)

pipeline/lib/pipeline-status.sh

@@ -4,6 +4,17 @@
 # Writes a one-screen summary to ${ALEMBIC_CONFIG_DIR:-/config}/logs/STATUS.log (overwritten daily)
 # and ships it to Telegram via notify-telegram.sh.
 #
+# Formatting note: the digest now uses Telegram HTML entities (<b>, <i>, <pre>,
+# <blockquote>) — the brand's chrome/dot-badge language translated into what
+# Telegram can actually render (no color, no custom font, so status "dots"
+# become colored circle emoji and section cards become bold header + <pre>
+# block, matching the web UI's h2 + mono table pattern). This REQUIRES
+# notify-telegram.sh to be called with --html (parse_mode=HTML) — sent as
+# plain text the tags would show up literally.
+# All free-text going through mark_ok/mark_warn/mark_skip/mark_info/section is
+# escaped (esc()) for &, <, > so a stray angle bracket in a log line can't
+# break the HTML parse and swallow the whole message.
+#
 # Reports on:
 #   - Sibling service reachability (slskd, navidrome) via HTTP, not docker ps —
 #     alembic has no Docker socket access
@@ -27,7 +38,10 @@
 # the opposite of what a health report should do. It writes no library state,
 # so there is nothing to leave half-applied.
 set -u
-PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+# /opt/venv/bin first: `beet` lives in the app venv. Without it every beet
+# probe here ("added today", "Library by format", the mp3-now count) silently
+# came up empty behind its 2>/dev/null.
+PATH=/opt/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
 
 OUT=${ALEMBIC_CONFIG_DIR:-/config}/logs/STATUS.log
 TODAY=$(date +%Y%m%d)
@@ -43,13 +57,33 @@ ND_PASS="${ND_PASS:-}"
 OK=0
 WARN=0
 SKIP=0
+PRE_OPEN=0
 
-mark_ok()   { OK=$((OK+1));     printf "  ✓ %s\n" "$*"; }
-mark_warn() { WARN=$((WARN+1)); printf "  ⚠ %s\n" "$*"; }
-mark_skip() { SKIP=$((SKIP+1)); printf "  ⊘ %s\n" "$*"; }
-mark_info() {                   printf "    %s\n" "$*"; }
+# Escapes &, <, > so free text (log snippets, exception messages, filenames)
+# can't be mistaken for an HTML entity by Telegram's parser.
+esc() {
+  local s=$1
+  s=${s//&/&amp;}
+  s=${s//</&lt;}
+  s=${s//>/&gt;}
+  printf '%s' "$s"
+}
 
-section() { printf "\n▎ %s\n" "$*"; }
+# Status lines double as the brand's badge-dot language (success/warning/muted)
+# translated into Telegram's only color channel: emoji.
+mark_ok()   { OK=$((OK+1));     printf "  🟢 %s\n" "$(esc "$*")"; }
+mark_warn() { WARN=$((WARN+1)); printf "  🟠 %s\n" "$(esc "$*")"; }
+mark_skip() { SKIP=$((SKIP+1)); printf "  ⚪ %s\n" "$(esc "$*")"; }
+mark_info() {                   printf "     %s\n" "$(esc "$*")"; }
+
+# Bold section header (proportional text, reads like the web app's <h2>),
+# followed by a <pre> block so the printf-padded columns below it actually
+# line up in a fixed-width font. Closes the previous section's <pre> first.
+section() {
+  if [[ "$PRE_OPEN" -eq 1 ]]; then printf "</pre>\n"; fi
+  printf "\n<b>▎ %s</b>\n<pre>" "$(esc "$*")"
+  PRE_OPEN=1
+}
 
 # syslog / `logger` is not present in the container image. Only call it if it
 # exists, so these lines don't spew "logger: command not found" into the job
@@ -70,17 +104,6 @@ human_age() {
   fi
 }
 
-# Newest matching log → "<age_seconds>|<path>", empty if no match.
-newest_log() {
-  local glob="$1"
-  local newest
-  newest=$(ls -1t $glob 2>/dev/null | head -1)
-  [[ -z "$newest" ]] && return
-  local mtime
-  mtime=$(stat -c %Y "$newest" 2>/dev/null) || return
-  echo "$((NOW_TS - mtime))|$newest"
-}
-
 # Per-playlist log status.
 playlist_status() {
   local log="$1"
@@ -219,27 +242,25 @@ dedup_today_status() {
   mark_ok "$(printf '%-13s %d new tracks in beets' 'added today' "$added_total")"
   if [[ "$added_total" -gt 0 ]]; then
     # Per-playlist breakdown. Empty $grouping (bandcamp/manual) shown as "(none)".
+    # Piped through the same &/</> escaping as mark_*, since this prints
+    # straight into the <pre> block without going through esc() otherwise.
     echo "$ADDED_TODAY" \
       | awk '{ if ($0 == "") print "(none)"; else print }' \
       | sort | uniq -c | sort -rn \
-      | awk '{ g=$2; for(i=3;i<=NF;i++) g=g" "$i; printf "                  %3d  %s\n", $1, g }'
+      | awk '{ g=$2; for(i=3;i<=NF;i++) g=g" "$i; printf "                  %3d  %s\n", $1, g }' \
+      | sed 's/&/\&amp;/g; s/</\&lt;/g; s/>/\&gt;/g'
   fi
 
   # 3. Maintenance freshness
-  # Each weekly/monthly task: find its newest log and check age.
+  # Read from the app's job_runs table (the scheduler's own record) instead of
+  # globbing for log files. Jobs run through pipeline_runner write their logs
+  # to logs/<job_key>/<timestamp>.log, so the old cron-era filename globs
+  # matched nothing for jobs that don't also write their own log, and healthy
+  # jobs were reported as "never run yet". The DB also distinguishes lock
+  # skips and failures, which a missing log file can't.
   # Limits: weekly tasks should be <9 days old; daily <2; monthly <35.
   section "Maintenance (last run)"
 
-  declare -A MAINT_GLOB=(
-    [strip-mb-tags]="${ALEMBIC_CONFIG_DIR:-/config}/logs/mb-strip-*.log"
-    [strip-watermark-art]="${ALEMBIC_CONFIG_DIR:-/config}/logs/strip-watermark-*.log"
-    [scrub-watermark-text]="${ALEMBIC_CONFIG_DIR:-/config}/logs/scrub-text-*.log"
-    [clean-sldl-index]="${ALEMBIC_CONFIG_DIR:-/config}/logs/clean-index-*.log"
-    [clear-bad-genres]="${ALEMBIC_CONFIG_DIR:-/config}/logs/clear-bad-genres-*.log"
-    [spotify-genre]="${ALEMBIC_CONFIG_DIR:-/config}/logs/spotify-genre-*.log"
-    [dedup-library]="${ALEMBIC_CONFIG_DIR:-/config}/logs/dedup-*.log"
-    [upgrade-mp3-to-flac]="${ALEMBIC_CONFIG_DIR:-/config}/logs/upgrade-mp3-*.log"
-  )
   declare -A MAINT_LIMIT=(
     [strip-mb-tags]=$((9*86400))
     [strip-watermark-art]=$((9*86400))
@@ -250,42 +271,106 @@ dedup_today_status() {
     [dedup-library]=$((2*86400))
     [upgrade-mp3-to-flac]=$((35*86400))
   )
+  # Emits one line per task: task|latest_status|latest_age_s|success_age_s|success_log
+  # (ages are -1 when there is no such run). Read-only DB open; prints nothing
+  # if the DB is missing so every task falls through to "never run yet".
+  maint_rows=$(python3 - "${ALEMBIC_CONFIG_DIR:-/config}/alembic.db" <<'PYEOF'
+import sqlite3
+import sys
+import time
+
+# Digest label -> job_runs.job_key. The scheduled genre refresh records under
+# genre:run (via genre_review_service) and dedup under dedup:scan (via
+# dedup_review_service); everything else is its MAINTENANCE_JOBS key.
+KEYS = [
+    ("strip-mb-tags", "maintenance:strip_mb_tags"),
+    ("strip-watermark-art", "maintenance:strip_watermark_art"),
+    ("scrub-watermark-text", "maintenance:scrub_watermark_text"),
+    ("clean-sldl-index", "maintenance:clean_sldl_index"),
+    ("clear-bad-genres", "maintenance:clear_bad_genres"),
+    ("spotify-genre", "genre:run"),
+    ("dedup-library", "dedup:scan"),
+    ("upgrade-mp3-to-flac", "maintenance:upgrade_mp3_to_flac"),
+]
+now = time.time()
+try:
+    conn = sqlite3.connect(f"file:{sys.argv[1]}?mode=ro", uri=True)
+    conn.execute("SELECT 1 FROM job_runs LIMIT 1")
+except sqlite3.Error:
+    sys.exit(0)
+for task, key in KEYS:
+    latest = conn.execute(
+        "SELECT status, started_at FROM job_runs WHERE job_key = ? ORDER BY started_at DESC LIMIT 1",
+        (key,),
+    ).fetchone()
+    if latest is None:
+        print(f"{task}|none|-1|-1|")
+        continue
+    success = conn.execute(
+        "SELECT started_at, log_path FROM job_runs WHERE job_key = ? AND status = 'success' "
+        "ORDER BY started_at DESC LIMIT 1",
+        (key,),
+    ).fetchone()
+    s_age = int(now - success[0]) if success else -1
+    s_log = (success[1] or "") if success else ""
+    print(f"{task}|{latest[0]}|{int(now - latest[1])}|{s_age}|{s_log}")
+PYEOF
+)
+
   for task in strip-mb-tags strip-watermark-art scrub-watermark-text clean-sldl-index clear-bad-genres spotify-genre dedup-library upgrade-mp3-to-flac; do
-    res=$(newest_log "${MAINT_GLOB[$task]}")
-    if [[ -z "$res" ]]; then
+    row=$(grep "^${task}|" <<< "$maint_rows" || true)
+    IFS='|' read -r _ latest_status latest_age age_s log <<< "$row"
+    if [[ -z "$row" || "$latest_status" == "none" ]]; then
       mark_skip "$(printf '%-22s never run yet' "$task")"
       continue
     fi
-    age_s=${res%%|*}
-    log=${res##*|}
+    if [[ "$age_s" -lt 0 ]]; then
+      # attempted, but never once succeeded -- say what the last attempt did
+      case "$latest_status" in
+        skipped_lock) note="skipped, lock busy" ;;
+        running)      note="running now" ;;
+        *)            note="$latest_status" ;;
+      esac
+      mark_warn "$(printf '%-22s never succeeded (last attempt %s: %s)' "$task" "$(human_age "$latest_age")" "$note")"
+      continue
+    fi
     age_h=$(human_age "$age_s")
-    case "$task" in
-      spotify-genre)
-        snip=$(grep -oE "written: [0-9]+, unchanged: [0-9]+" "$log" | tail -1) ;;
-      clear-bad-genres)
-        snip=$(grep -oE "(would-clear|cleared): [0-9]+ tracks" "$log" | tail -1) ;;
-      clean-sldl-index)
-        snip=$(grep -oE "[0-9]+ m3us, total [0-9]+ kept, [0-9]+ dropped" "$log" | tail -1) ;;
-      strip-watermark-art)
-        snip=$(grep -oE "stripped from [0-9]+ tracks|no images shared" "$log" | tail -1) ;;
-      scrub-watermark-text)
-        snip=$(grep -oE "stripped [0-9]+ frame\(s\) across [0-9]+ file" "$log" | tail -1) ;;
-      strip-mb-tags)
-        snip=$(grep -oE "Done\. Log:" "$log" | head -1 | sed 's/Done\. Log:/done/') ;;
-      dedup-library)
-        snip=$(grep "=== Summary:" "$log" | tail -1 \
-               | sed -E 's/.*=== Summary: //; s/ ===.*//; s/ \([^)]*\)//') ;;
-      upgrade-mp3-to-flac)
-        # The log's "to attempt" count is frozen at the last monthly run, so on
-        # its own it looks stale against the live "Library by format" section.
-        # Pair it with the current format:mp3 count so the drop (upgrades that
-        # have landed since) is visible instead of looking like a mismatch.
-        attempted=$(grep -oE "[0-9]+ MP3 tracks to attempt" "$log" | grep -oE "^[0-9]+" | tail -1)
-        mp3_now=$(beet ls 'format:mp3' 2>/dev/null | grep -c '^')
-        snip="${attempted:-?} attempted · ${mp3_now} MP3 now" ;;
+    snip=""
+    if [[ -n "$log" && -f "$log" ]]; then
+      case "$task" in
+        spotify-genre)
+          snip=$(grep -oE "written: [0-9]+, unchanged: [0-9]+" "$log" | tail -1) ;;
+        clear-bad-genres)
+          snip=$(grep -oE "rewrote: [0-9]+ tracks, blanked: [0-9]+ tracks|(would-clear|cleared): [0-9]+ tracks" "$log" | tail -1) ;;
+        clean-sldl-index)
+          snip=$(grep -oE "[0-9]+ m3us, total [0-9]+ kept, [0-9]+ dropped" "$log" | tail -1) ;;
+        strip-watermark-art)
+          snip=$(grep -oE "stripped from [0-9]+ tracks|no images shared" "$log" | tail -1) ;;
+        scrub-watermark-text)
+          snip=$(grep -oE "stripped [0-9]+ frame\(s\) across [0-9]+ file" "$log" | tail -1) ;;
+        strip-mb-tags)
+          snip=$(grep -oE "Done\. Log:" "$log" | head -1 | sed 's/Done\. Log:/done/') ;;
+        dedup-library)
+          snip=$(grep "=== Summary:" "$log" | tail -1 \
+                 | sed -E 's/.*=== Summary: //; s/ ===.*//; s/ \([^)]*\)//') ;;
+        upgrade-mp3-to-flac)
+          # The log's "to attempt" count is frozen at the last monthly run, so on
+          # its own it looks stale against the live "Library by format" section.
+          # Pair it with the current format:mp3 count so the drop (upgrades that
+          # have landed since) is visible instead of looking like a mismatch.
+          attempted=$(grep -oE "[0-9]+ MP3 tracks to attempt" "$log" | grep -oE "^[0-9]+" | tail -1)
+          mp3_now=$(beet ls 'format:mp3' 2>/dev/null | grep -c '^')
+          snip="${attempted:-?} attempted · ${mp3_now} MP3 now" ;;
+      esac
+    fi
+    # Age/snippet describe the last SUCCESS; if a newer attempt failed or was
+    # lock-skipped, say so rather than hiding it behind the healthy line.
+    case "$latest_status" in
+      failed)       snip="${snip:+$snip, }latest attempt failed" ;;
+      skipped_lock) snip="${snip:+$snip, }latest attempt skipped, lock busy" ;;
     esac
     line=$(printf '%-22s %-9s %s' "$task" "$age_h" "${snip:-}")
-    if (( age_s > MAINT_LIMIT[$task] )); then
+    if (( age_s > MAINT_LIMIT[$task] )) || [[ "$latest_status" == "failed" ]]; then
       mark_warn "$line"
     else
       mark_ok "$line"
@@ -384,16 +469,34 @@ except Exception as e:
   beet ls -f '$format' 2>/dev/null | sort | uniq -c | sort -rn \
     | awk '{printf "    %-6s %d tracks\n", $2, $1}'
 
+  [[ "$PRE_OPEN" -eq 1 ]] && printf "</pre>\n"
+
 } > "$OUT"
 
 # ---- Build header banner and prepend ----
+# Bold brand mark (⚗️ is literally the "ALEMBIC" unicode glyph — no generic
+# music-note stand-in needed), a hostname chip in <code>, and an italic tally
+# that reuses the same 🟢/🟠/⚪ dots as the body. When something needs
+# attention, a <blockquote> callout mirrors the web app's .notice.warning
+# panel; an all-clear run gets the calm .notice.success equivalent instead.
 HOSTNAME_SHORT=$(hostname -s)
 DATE_HUMAN=$(TZ="${TZ:-UTC}" date '+%a %Y-%m-%d %H:%M %Z')
-BANNER_TITLE="🎵 Music pipeline · $HOSTNAME_SHORT · $DATE_HUMAN"
-BANNER_TALLY="   $OK OK · $WARN warn · $SKIP skip"
-sed -i "1c\\
-${BANNER_TITLE}\\
-${BANNER_TALLY}" "$OUT"
+HEADER="⚗️ <b>Alembic</b> · music pipeline · <code>${HOSTNAME_SHORT}</code> · ${DATE_HUMAN}
+<i>🟢 ${OK} OK   🟠 ${WARN} warn   ⚪ ${SKIP} skip</i>"
+if (( WARN > 0 )); then
+  HEADER="${HEADER}
+<blockquote>🟠 <b>${WARN} issue(s)</b> need attention — see below</blockquote>"
+else
+  HEADER="${HEADER}
+<blockquote>🟢 All clear — nothing needs attention.</blockquote>"
+fi
+
+# Swap the __HEADER_PLACEHOLDER__ line for the (2-3 line) banner above. A
+# straight prepend, not sed 1c — the banner's line count varies with the
+# blockquote, which sed's `c` command can't take as a variable easily.
+tail -n +2 "$OUT" > "${OUT}.body"
+{ printf '%s\n' "$HEADER"; cat "${OUT}.body"; } > "$OUT"
+rm -f "${OUT}.body"
 
 # Always mirror the one-line summary to syslog so journalctl shows last-run
 # state even if Telegram is broken. Warnings get a separate WARN tag.
@@ -402,10 +505,12 @@ if (( WARN > 0 )); then
   slog "WARN: pipeline-status flagged $WARN issue(s) — see $OUT"
 fi
 
-# Send to Telegram. If it fails, log the failure to syslog so the absence of
-# a message in the chat has a corresponding journal entry to grep for.
+# Send to Telegram with --html: the digest is Telegram-HTML (see header
+# comment), so notify-telegram.sh must send it with parse_mode=HTML. If the
+# send fails, log to syslog so the absence of a message in the chat has a
+# corresponding journal entry to grep for.
 if [[ -x ${PIPELINE_DIR:-/app/pipeline}/lib/notify-telegram.sh ]]; then
-  if ! ${PIPELINE_DIR:-/app/pipeline}/lib/notify-telegram.sh < "$OUT" 2>/tmp/tg-err; then
+  if ! ${PIPELINE_DIR:-/app/pipeline}/lib/notify-telegram.sh --html < "$OUT" 2>/tmp/tg-err; then
     slog "WARN: Telegram send failed: $(cat /tmp/tg-err 2>/dev/null | head -c 200)"
     rm -f /tmp/tg-err
   fi

pipeline/lib/spotify-playlist-csv.py

@@ -45,10 +45,12 @@ def fetch_playlist(url: str, token: str) -> list[dict]:
         sys.exit(f"Could not parse playlist ID from {url!r}")
     pid = m.group(1)
     tracks = []
-    next_url = (
-        f"{API}/playlists/{pid}/items"
-        "?limit=100&fields=items(track(name,artists(name),album(name),duration_ms,is_local)),next"
-    )
+    # No `fields` filter: Spotify's February 2026 changes renamed each entry's
+    # payload from "track" to "item" for apps on the new behavior (extended
+    # quota / grandfathered apps keep "track"), and a fields filter selects by
+    # key name -- filtering on track(...) silently returns EMPTY pages on the
+    # renamed shape. Fetch unfiltered and parse both key names below.
+    next_url = f"{API}/playlists/{pid}/items?limit=100"
     while next_url:
         req = urllib.request.Request(next_url, headers={"Authorization": f"Bearer {token}"})
         try:
@@ -56,18 +58,26 @@ def fetch_playlist(url: str, token: str) -> list[dict]:
                 data = json.loads(r.read())
         except urllib.error.HTTPError as e:
             if e.code in (403, 404):
-                # The one non-bug cause of this: OAuth playlist reads are
-                # scoped to what the CONNECTED account can see (its own
-                # playlists, collaborations, or ones marked Public).
                 sys.exit(
                     f"Spotify returned {e.code} loading this playlist. The connected "
-                    "Spotify account can't see it: make sure it's owned by, or "
-                    "shared/followed by, whichever account is connected via "
-                    "/connect/spotify, or set it to Public on Spotify."
+                    "Spotify account can't see it: it must own the playlist, be a "
+                    "collaborator on it, or (for grandfathered apps) the playlist "
+                    "must be Public. Check which account is connected via "
+                    "/connect/spotify."
                 )
             raise
-        for item in data.get("items", []):
-            t = item.get("track")
+        if "items" not in data:
+            # New-behavior apps get metadata only (no items field) for
+            # playlists the connected account doesn't own or collaborate on.
+            sys.exit(
+                "Spotify returned this playlist without its contents. For newly "
+                "created Spotify apps, the connected account must OWN the playlist "
+                "or be a collaborator on it (public is no longer enough). Ask the "
+                "owner to make it collaborative and add you, or recreate the "
+                "playlist under the connected account."
+            )
+        for item in data["items"]:
+            t = item.get("track") or item.get("item")
             if not t or t.get("is_local"):
                 continue
             # All artists, comma-joined -- matches what sldl's own Spotify

pipeline/lib/spotify-retag.py

@@ -56,8 +56,11 @@ def fetch_playlist(url: str, token: str) -> list[dict]:
         )
         with urllib.request.urlopen(req, timeout=15) as r:
             data = json.loads(r.read())
-        for item in data["items"]:
-            t = item.get("track")
+        # Entries carry "track" (grandfathered apps) or "item" (apps on the
+        # February 2026 renamed shape); items is absent entirely when the
+        # connected account can't read the playlist's contents.
+        for item in data.get("items", []):
+            t = item.get("track") or item.get("item")
             if not t or t.get("is_local"):
                 continue
             tracks.append(

pipeline/lib/strip-watermark-art.py

@@ -51,10 +51,17 @@ def extract_flac_picture(flac_path):
     with tempfile.NamedTemporaryFile(delete=False, suffix=".pic") as tf:
         tmp = tf.name
     try:
-        r = subprocess.run(
-            ["metaflac", f"--export-picture-to={tmp}", flac_path],
-            capture_output=True, timeout=15
-        )
+        try:
+            r = subprocess.run(
+                ["metaflac", f"--export-picture-to={tmp}", flac_path],
+                capture_output=True, timeout=60
+            )
+        except subprocess.TimeoutExpired:
+            # A transient IO stall on one file must not abort the whole weekly
+            # run (seen 2026-07-15: a healthy 190KB cover took >15s under disk
+            # contention and the raised TimeoutExpired failed the entire job).
+            print(f"[strip-art] WARN: metaflac timed out on {flac_path}, skipping file")
+            return None
         if r.returncode != 0 or not os.path.exists(tmp):
             return None
         sz = os.path.getsize(tmp)

tests/test_pipeline_runner.py

@@ -25,11 +25,41 @@ def test_run_job_failure():
     assert run.exit_code == 1
 
 
-def test_lock_skip_when_held():
+def test_manual_run_skips_immediately_when_lock_held():
     async def scenario():
         await pr._lock.acquire()
         try:
-            return await pr.run_job("test:locked", ["true"], use_lock=True)
+            return await pr.run_job("test:locked", ["true"], use_lock=True, triggered_by="manual")
+        finally:
+            pr._lock.release()
+
+    run = _run(scenario())
+    assert run.status == "skipped_lock"
+
+
+def test_scheduled_run_waits_for_lock_by_default(monkeypatch):
+    # scheduled runs must QUEUE behind a held lock (bounded), not skip --
+    # instant-skip starved the Sunday maintenance block (see
+    # SCHEDULED_LOCK_WAIT_SECONDS)
+    monkeypatch.setattr(pr, "SCHEDULED_LOCK_WAIT_SECONDS", 5.0)
+
+    async def scenario():
+        await pr._lock.acquire()
+        task = asyncio.ensure_future(pr.run_job("test:queued", ["true"], triggered_by="schedule"))
+        await asyncio.sleep(0.2)  # let the job start waiting on the lock
+        pr._lock.release()
+        return await task
+
+    run = _run(scenario())
+    assert run.status == "success"
+    assert not pr._lock.locked()
+
+
+def test_scheduled_run_skips_after_lock_wait_timeout():
+    async def scenario():
+        await pr._lock.acquire()
+        try:
+            return await pr.run_job("test:waited_out", ["true"], triggered_by="schedule", lock_wait=0.2)
         finally:
             pr._lock.release()
 

tests/test_spotify_client.py

@@ -0,0 +1,71 @@
+"""get_playlist_tracks must parse BOTH /items response shapes: entries keyed
+"track" (grandfathered / extended-quota apps) and "item" (apps on Spotify's
+February 2026 renamed shape), and must fail loudly, not return [], when
+Spotify withholds a playlist's contents (no items field at all)."""
+import pytest
+
+from app.services import spotify_client
+
+URL = "https://open.spotify.com/playlist/XYZ123"
+
+
+class _Resp:
+    def __init__(self, data):
+        self._data = data
+
+    def raise_for_status(self):
+        pass
+
+    def json(self):
+        return self._data
+
+
+def _fake_api(monkeypatch, pages):
+    it = iter(pages)
+    monkeypatch.setattr(spotify_client, "_get_token", lambda db: "tok")
+    monkeypatch.setattr(
+        spotify_client.httpx,
+        "get",
+        lambda url, params=None, headers=None, timeout=None: _Resp(next(it)),
+    )
+
+
+def _track(name, artist, isrc=None):
+    return {
+        "name": name,
+        "artists": [{"name": artist}],
+        "external_ids": {"isrc": isrc} if isrc else {},
+    }
+
+
+def test_parses_legacy_track_shape(monkeypatch):
+    _fake_api(monkeypatch, [{"items": [{"track": _track("Sandstorm", "Darude", "FIDAR9900011")}], "next": None}])
+    tracks = spotify_client.get_playlist_tracks(None, URL)
+    assert tracks == [{"artist": "Darude", "title": "Sandstorm", "isrc": "FIDAR9900011"}]
+
+
+def test_parses_renamed_item_shape(monkeypatch):
+    # February 2026 shape: payload under "item", not "track"
+    _fake_api(monkeypatch, [{"items": [{"item": _track("Sandstorm", "Darude")}], "next": None}])
+    tracks = spotify_client.get_playlist_tracks(None, URL)
+    assert tracks == [{"artist": "Darude", "title": "Sandstorm", "isrc": None}]
+
+
+def test_missing_items_field_raises_not_empty(monkeypatch):
+    # metadata-only response (playlist not owned by the connected account on a
+    # new app) must be an error the UI can show, never a silent empty list
+    _fake_api(monkeypatch, [{"name": "DJ-USB", "public": True}])
+    with pytest.raises(RuntimeError, match="without its contents"):
+        spotify_client.get_playlist_tracks(None, URL)
+
+
+def test_skips_null_entries_and_paginates(monkeypatch):
+    _fake_api(
+        monkeypatch,
+        [
+            {"items": [{"track": None}, {"track": _track("A", "X")}], "next": "page2"},
+            {"items": [{"item": _track("B", "Y")}], "next": None},
+        ],
+    )
+    tracks = spotify_client.get_playlist_tracks(None, URL)
+    assert [t["title"] for t in tracks] == ["A", "B"]