Import a shared playlist link and build it in Navidrome

Paste a Spotify, Apple Music or Deezer playlist URL; muse resolves it, matches
every track through the same Deezer path a pasted line takes, and writes an
.m3u the library scanner picks up.

Written per import rather than at the end, so the playlist appears in Navidrome
with the first track that lands and grows as the rest arrive -- twenty minutes
of nothing, then everything, is the worse shape. A track rejected in review
drops out on the next rewrite, and one that never resolves is reported instead
of silently vanishing.

Resolve is split from queue on purpose: a shared link is opaque, you cannot
tell twelve tracks from four hundred by looking at it, and four hundred is a
quota decision.

Two of the three sources are scrapes and the code says so plainly. Deezer has a
real API. Spotify and Apple Music have no free way to read a public playlist --
Apple's MusicKit needs a paid developer account -- so those read the public page
and parse the JSON it already embeds for its own player. No key, no OAuth, no
account, and no guarantee: they break whenever either company reshuffles their
frontend. So the resolvers fail loudly and name themselves, because an empty
playlist and a changed page must not produce the same message.

The .m3u uses MUSE_PLAYLIST_PREFIX (/music), not muse's own library path. Both
containers mount the same directory at different points and a playlist written
with muse's view resolves to nothing on Navidrome's side.

Also reorders the hooks in search-bar so the URL-seeding effect is declared
after the state it sets, rather than closing over a binding declared below it.

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

README.md

@@ -152,6 +152,37 @@ Failed jobs get a **Retry** button, but far fewer jobs reach that state now:
 a request that loses its peer falls through to the next candidate on its own.
 See "When a peer says no" below.
 
+## Shared playlist links
+
+Paste a Spotify, Apple Music or Deezer playlist link and muse resolves it,
+matches every track the same way a pasted line is matched, and writes an `.m3u`
+into the library that Navidrome picks up on its next scan — within seconds, in
+practice. The playlist is written **per import**, not at the end, so it appears
+with the first track that lands and grows as the rest arrive. Tracks that never
+resolve are reported rather than silently dropped, and a track rejected in
+review disappears from the playlist on the next rewrite.
+
+Resolving is deliberately separate from queueing. A shared link is opaque — you
+cannot tell twelve tracks from four hundred by looking at it — so the track list
+and count are shown before anything is committed against a quota.
+
+**Two of the three sources are scrapes, and this is worth understanding before
+you rely on it.** Deezer has a real public API. Spotify and Apple Music have no
+free way to read a public playlist — Apple's MusicKit requires a paid developer
+account — so muse reads the public web page and parses the JSON blob the page
+already embeds for its own player. No key, no OAuth, no account. It also means
+those two will break the day either company reshuffles their frontend. The
+resolvers therefore fail loudly and specifically ("Spotify's embed page no
+longer carries `__NEXT_DATA__` — the resolver in app.py needs updating") rather
+than quietly returning nothing, because an empty playlist and a changed page
+must not look the same to whoever has to fix it.
+
+One detail that is easy to get wrong: the `.m3u` has to use **Navidrome's**
+mount path, not muse's. Both containers see the same directory, muse at
+`/data/media/music` and Navidrome at `/music`, and a playlist written with the
+former resolves to nothing on the other side. `MUSE_PLAYLIST_PREFIX` is that
+path.
+
 ### Everything goes through a bounded queue
 
 `MUSE_WORKERS` (default 3) caps concurrent Soulseek searches. This matters:
@@ -218,9 +249,24 @@ and finally the artist itself. Since most requests are answered by the first
 query, the extra searches cost nothing in the common case — `MUSE_MAX_QUERIES`
 caps how far it will go.
 
-Dropping the artist from the *query* does not mean ignoring it. The artist is
-then matched against the whole **folder path**, which is where a well-organised
-share puts it anyway:
+Dropping the artist from the *query* does not mean ignoring it — it makes the
+artist matter more. A bare-title search stops returning "this song from various
+peers" and starts returning *every song anyone ever called Chicago*: The Fall,
+Groove Armada, Sufjan Stevens. The artist is then the only thing separating the
+right recording from a different song entirely, so it is a **gate**: candidates
+whose path names the artist are preferred outright, and ones that do not are
+used only when nothing else is left.
+
+That has to be a gate rather than a weighting, and the reason is worth stating.
+A perfect title with no artist scores 0.75, comfortably above `MUSE_MIN_MATCH`,
+so as a mere contribution it loses to availability — and an idle peer holding
+the wrong song beats a busy one holding the right song. That is not
+hypothetical either: it filed The Fall's "Chicago, Now!" as Michael Jackson.
+The same principle the confidence score is built on applies here: a fast peer
+must not be able to promote a wrong match.
+
+The artist is matched against the whole **folder path**, which is where a
+well-organised share puts it anyway:
 
 ```
 anjya music/Michael Jackson/1982 - Thriller [flac]/06 - Billie Jean.flac
@@ -258,6 +304,23 @@ A download that sits with no bytes for `MUSE_STALL_MINUTES` is abandoned the
 same way, as is one slskd has forgotten about. Both used to leave the job in
 `downloading` forever, with a cheerful pulsing dot and no log line.
 
+**Unreachable peers are the norm, not the exception.** slskd answers 500 when it
+cannot open a connection to a peer within its five-second window, and on a live
+instance that is common: one request walked four such peers before the fifth
+served it. This is why the ladder is worth having and why `MUSE_MAX_ATTEMPTS`
+defaults to the full `MUSE_CANDIDATES` rather than a smaller number — a cap
+below the ladder size throws away candidates that would have worked.
+
+Attempts are spaced by `MUSE_ATTEMPT_BACKOFF`. That is politeness rather than a
+fix for anything measured: each enqueue holds a peer connection slot for up to
+five seconds, and there is no reason to ask for eight of them in immediate
+succession. Separately, `MUSE_SYSTEMIC_LIMIT` consecutive *connection* errors —
+meaning slskd itself is unreachable, not a peer — stop the walk and **keep the
+untried candidates on the row**, so retrying resumes the ladder rather than
+paying for a search whose answer has not changed. A 500 is deliberately not
+counted as systemic: it describes one peer, and treating it as systemic would
+have abandoned the request that succeeded on its fifth attempt.
+
 ## Review: when muse isn't sure
 
 Picking a file and *trusting* it are different questions, so they use different
@@ -276,6 +339,16 @@ filename alone:
 
 Then:
 
+**Length is checked against the audio, not the name.** Every textual signal can
+agree and the file still be the wrong song: right title, right folder, right
+album — and six minutes of a track that runs four. So once the download
+finishes, the file's real duration is compared against what Deezer said, and a
+drift past `MUSE_LENGTH_TOLERANCE` multiplies confidence down and says so in
+words: *"runs 5:59 but the track is 4:05 (47% out), so this is probably a
+different recording"*. This is applied on import rather than inside
+`confidence()` for the obvious reason that the file does not exist yet when the
+candidate is picked.
+
 - **≥ 0.95** — filed automatically. Nothing to do.
 - **0.70 – 0.95** — held in **Review** with the reasons shown.
 - **< 0.70** — also held, but it will almost certainly be rejected.
@@ -547,13 +620,18 @@ stat -c "%h" "/data/media/music/<Artist>/<Album> (<Year>)/<NN> - <Title>.flac"
 | `MUSE_MIN_MATCH` | `0.45` | filename-to-track similarity floor |
 | `MUSE_TITLE_FLOOR` | `0.5` | the title alone must clear this before the artist counts |
 | `MUSE_CANDIDATES` | `8` | ranked candidates kept per request, for fallback |
-| `MUSE_MAX_ATTEMPTS` | `5` | peers tried before a request is given up on |
+| `MUSE_MAX_ATTEMPTS` | `MUSE_CANDIDATES` | peers tried before a request is given up on |
+| `MUSE_ATTEMPT_BACKOFF` | `4` | seconds between candidate attempts |
+| `MUSE_SYSTEMIC_LIMIT` | `2` | consecutive connection failures before stopping |
 | `MUSE_STALL_MINUTES` | `20` | a transfer moving no bytes for this long is abandoned |
 | `MUSE_MAX_QUERIES` | `4` | rungs of the search ladder to try |
 | `MUSE_MIN_CANDIDATES` | `3` | below this, escalate to the next query |
 | `MUSE_SEARCH_RETRIES` | `3` | attempts at issuing a search before failing |
 | `MUSE_AUTO_CONFIDENCE` | `0.95` | at or above this, import without asking |
 | `MUSE_REVIEW_CONFIDENCE` | `0.70` | below this, held but probably wrong |
+| `MUSE_LENGTH_TOLERANCE` | `0.2` | duration drift before a file is called a different recording |
+| `MUSE_MAX_PLAYLIST_TRACKS` | `500` | cap on one playlist link |
+| `MUSE_PLAYLIST_PREFIX` | `/music` | the library path *Navidrome* sees, for `.m3u` lines |
 | `MUSE_WORKERS` | `3` | concurrent Soulseek searches |
 | `MUSE_SMTP_HOST` | — | set to enable email; see **Email** above |
 | `MUSE_SMTP_PORT` | `587` | |

app.py

@@ -43,6 +43,7 @@ from pathlib import Path
 from urllib.parse import urlencode
 
 import httpx
+import mutagen
 from fastapi import FastAPI, HTTPException, Query, Request
 from fastapi.responses import (FileResponse, HTMLResponse, JSONResponse,
                                RedirectResponse, Response)
@@ -80,7 +81,14 @@ TITLE_FLOOR = float(os.getenv("MUSE_TITLE_FLOOR", "0.5"))
 # one is tried instead of failing the request -- "File not shared." is a
 # per-file condition, so candidate two usually just works.
 CANDIDATES = int(os.getenv("MUSE_CANDIDATES", "8"))
-MAX_ATTEMPTS = int(os.getenv("MUSE_MAX_ATTEMPTS", "5"))
+
+# Defaults to the size of the ladder, because keeping eight candidates and then
+# refusing to try three of them is a cap with no reasoning behind it. Measured
+# on a live instance: slskd answers 500 whenever it cannot open a connection to
+# a peer, which is common enough that requests routinely need four or five
+# attempts. One succeeded on exactly the fifth; another gave up at the cap with
+# three candidates still in hand.
+MAX_ATTEMPTS = int(os.getenv("MUSE_MAX_ATTEMPTS", str(CANDIDATES)))
 
 # A download that has moved no bytes for this long is not going to. The peer
 # queued it and vanished, or slskd lost the transfer; either way the job used
@@ -100,6 +108,17 @@ SEARCH_RETRIES = int(os.getenv("MUSE_SEARCH_RETRIES", "3"))
 # Soulseek results trickle in: what you have at t=5s is much worse than t=15s.
 SEARCH_POLLS = int(os.getenv("MUSE_SEARCH_POLLS", "20"))
 
+# Pause between candidate attempts. Without one, a ladder of 8 walks the whole
+# list inside a single bad slskd window -- each enqueue blocks ~5s on a peer
+# connection, so eight attempts is 40 seconds of hammering one subsystem, and
+# it takes other people's searches down with it.
+ATTEMPT_BACKOFF = float(os.getenv("MUSE_ATTEMPT_BACKOFF", "4"))
+
+# Consecutive systemic failures before we stop walking the ladder. A file that
+# is not shared says something about that file; a timeout says only that slskd
+# is having a moment, and burning eight candidates to learn it once is waste.
+SYSTEMIC_LIMIT = int(os.getenv("MUSE_SYSTEMIC_LIMIT", "2"))
+
 # How many Soulseek searches may run at once.
 #
 # This is a hard cap, not a suggestion. Each search takes 10-40s of polling
@@ -463,6 +482,30 @@ def similarity(want: str, got: str) -> float:
     return max(seq, len(a & b) / len(a) if a else 0.0)
 
 
+def artist_confirmed(remote_path: str, artist: str) -> bool:
+    """Is the requested artist actually named somewhere in this path?
+
+    This is a gate rather than a score because of what the query ladder does.
+    Once a filtered artist name forces the search down to the bare title, the
+    result set stops being "this song from various peers" and becomes "every
+    song anyone ever called Chicago" -- The Fall, Groove Armada, Sufjan
+    Stevens. The artist is then the only thing distinguishing the right
+    recording from a completely different song, and a weighted contribution is
+    not enough: 0.75 for a perfect title with no artist still clears MIN_MATCH,
+    and an available stranger will out-score a queued match every time.
+    """
+    na = norm(artist)
+    if not na:
+        return False
+    path = norm(remote_path.replace("\\", "/"))
+    if na in path:
+        return True
+    # Every distinctive word, in any order: catches "Jackson, Michael" and
+    # "MJ - Chicago (Michael Jackson)" without matching on "the" or "and".
+    words = [w for w in na.split() if len(w) > 3]
+    return bool(words) and all(w in path for w in words)
+
+
 def match_ratio(remote_path: str, artist: str, title: str) -> float:
     """How well a remote path matches the track asked for, 0..1.
 
@@ -553,9 +596,29 @@ def init_db():
                           ("candidates", "TEXT"),
                           ("tried", "TEXT"),
                           ("attempts", "INTEGER NOT NULL DEFAULT 0"),
-                          ("started_at", "TEXT")):
+                          ("started_at", "TEXT"),
+                          # Which shared playlist this track came from, and
+                          # where it sits in it -- an m3u in shuffled order is
+                          # not the playlist anyone sent you.
+                          ("playlist_id", "TEXT"),
+                          ("playlist_seq", "INTEGER")):
             if col not in cols:
                 c.execute(f"ALTER TABLE jobs ADD COLUMN {col} {decl}")
+        c.execute("""
+            CREATE TABLE IF NOT EXISTS playlists (
+                id TEXT PRIMARY KEY,
+                name TEXT NOT NULL,
+                source TEXT NOT NULL,          -- deezer|spotify|apple
+                source_url TEXT,
+                requested_by TEXT,
+                tracks_total INTEGER NOT NULL DEFAULT 0,
+                tracks_written INTEGER NOT NULL DEFAULT 0,
+                m3u_path TEXT,
+                created_at TEXT DEFAULT (datetime('now'))
+            )
+        """)
+        c.execute("CREATE INDEX IF NOT EXISTS jobs_by_playlist "
+                  "ON jobs (playlist_id)")
         ucols = {r["name"] for r in c.execute("PRAGMA table_info(users)").fetchall()}
         for col, decl in (("storage_quota", "INTEGER NOT NULL DEFAULT 0"),
                           ("invited_by", "TEXT"),
@@ -895,13 +958,32 @@ def no_candidate_detail(seen: int, stats: "collections.Counter",
 
 
 def exhausted_detail(tried: list[dict], attempts: int) -> str:
-    """Summarise the whole ladder instead of overwriting it with the last error."""
+    """Summarise the whole ladder instead of overwriting it with the last error.
+
+    Counts `tried`, not `attempts`. They agree for anything this version
+    enqueued, but a row carried over from before the ladder existed has
+    attempts=0 with a real failure recorded against it, which rendered as the
+    nonsense "tried 0 peers: 1x no bytes in 20m". The list is the evidence.
+    """
     if not tried:
         return "no candidate could be enqueued"
+    n_peers = len({t.get("username") for t in tried if t.get("username")}) or len(tried)
     reasons = collections.Counter(t["reason"].split(":")[0] for t in tried)
-    plural = "" if attempts == 1 else "s"
+    plural = "" if n_peers == 1 else "s"
     joined = ", ".join(f"{n}x {r}" for r, n in reasons.most_common(3))
-    return f"tried {attempts} peer{plural}: {joined}"
+    return f"tried {n_peers} peer{plural}: {joined}"
+
+
+# Substrings in a failure that describe slskd's state rather than the file's.
+# "File not shared" is about one file and the next candidate is unaffected;
+# a timeout means the client could not talk to anyone, so the next candidate
+# would fail for the same reason and only costs us the candidate.
+SYSTEMIC = ("timed out", "timeout", "connection", "refused", "unreachable")
+
+
+def is_systemic(reason: str) -> bool:
+    r = reason.lower()
+    return any(w in r for w in SYSTEMIC)
 
 
 async def try_next_candidate(job_id: str, artist: str, title: str,
@@ -923,7 +1005,15 @@ async def try_next_candidate(job_id: str, artist: str, title: str,
     tried = json.loads(row["tried"] or "[]")
     attempts = row["attempts"] or 0
 
+    systemic_run = 0
+    first = True
     while cands and attempts < MAX_ATTEMPTS:
+        # Space the attempts out. The first one is immediate; the rest wait,
+        # because back-to-back enqueues are what saturated slskd.
+        if not first:
+            await asyncio.sleep(ATTEMPT_BACKOFF)
+        first = False
+
         cand = cands.pop(0)
         attempts += 1
         try:
@@ -934,11 +1024,33 @@ async def try_next_candidate(job_id: str, artist: str, title: str,
                            "size": cand["size"]}])
                 er.raise_for_status()
         except Exception as e:
+            reason = f"could not reach peer: {errstr(e)}"
             tried.append({"username": cand["username"],
                           "filename": cand["filename"],
-                          "reason": f"could not reach peer: {errstr(e)}"})
+                          "reason": reason})
+            if is_systemic(reason):
+                systemic_run += 1
+                if systemic_run >= SYSTEMIC_LIMIT:
+                    # Stop, and leave the untried candidates on the row so a
+                    # retry resumes the ladder instead of re-running a search
+                    # that already told us the right answer.
+                    with db() as c:
+                        c.execute("UPDATE jobs SET candidates=?, tried=?,"
+                                  " attempts=? WHERE id=?",
+                                  (json.dumps(cands), json.dumps(tried),
+                                   attempts, job_id))
+                    set_state(job_id, "failed",
+                              f"slskd could not reach any peer after "
+                              f"{attempts} attempts -- {len(cands)} candidate"
+                              f"{'' if len(cands) == 1 else 's'} left to try, "
+                              f"so a retry resumes where this stopped")
+                    return False
+            else:
+                systemic_run = 0
             continue
 
+        systemic_run = 0
+
         conf, why = confidence(cand["filename"], artist, title, album)
         with db() as c:
             c.execute(
@@ -1011,6 +1123,20 @@ async def pick_and_enqueue(job_id: str, artist: str, title: str,
         set_state(job_id, "failed", no_candidate_detail(seen, stats, queries))
         return
 
+    # Prefer candidates whose path actually names the artist, and only settle
+    # for ones that do not when there is nothing else. Availability is a
+    # tiebreak among plausible files, never a reason to take the wrong song.
+    named = {k: v for k, v in pool.items()
+             if artist_confirmed(v[2]["filename"], artist)}
+    if named:
+        if len(named) < len(pool):
+            log.info("job %s keeping %d/%d candidates that name %r",
+                     job_id[:8], len(named), len(pool), artist)
+        pool = named
+    else:
+        log.info("job %s no candidate names %r; falling back to title-only "
+                 "matches, which review will have to judge", job_id[:8], artist)
+
     ranked = sorted(pool.values(), key=lambda r: -r[0])
     cands = spread_by_peer(ranked)[:CANDIDATES]
     with db() as c:
@@ -1199,6 +1325,43 @@ def tag_and_file(job: sqlite3.Row, src: Path) -> Path:
     return dest
 
 
+# How far a file's length may drift from what Deezer said before it stops being
+# plausibly the same recording. Generous, because edits, fades, hidden tracks
+# and a stray second of silence are all normal; a different song is not.
+LENGTH_TOLERANCE = float(os.getenv("MUSE_LENGTH_TOLERANCE", "0.2"))
+
+
+def length_check(src: Path, want_ms) -> tuple[float, str] | None:
+    """Compare the downloaded audio's real length against the request.
+
+    The one signal filename matching cannot fake. A file named exactly right,
+    in a folder named exactly right, is still the wrong song if it runs six
+    minutes when the track is four -- and that is not hypothetical: The Fall's
+    "Chicago, Now!" (359s) was filed as Michael Jackson's "Chicago" (245s)
+    because every *textual* signal agreed.
+
+    Returns a multiplier and a reason, or None when there is nothing to say.
+    """
+    want = (want_ms or 0) / 1000.0
+    if want <= 0:
+        return None
+    try:
+        got = mutagen.File(str(src)).info.length
+    except Exception:
+        return None                      # unreadable is review's problem
+    if got <= 0:
+        return None
+    drift = abs(got - want) / want
+    if drift <= LENGTH_TOLERANCE:
+        return None
+    return (0.35, f"runs {mmss(got)} but the track is {mmss(want)} "
+                  f"({drift:.0%} out), so this is probably a different recording")
+
+
+def mmss(seconds: float) -> str:
+    return f"{int(seconds) // 60}:{int(seconds) % 60:02d}"
+
+
 def downloaded_index() -> dict[str, Path]:
     """Every completed file, by basename.
 
@@ -1232,6 +1395,18 @@ async def search_worker(n: int):
                 job = c.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()
             if job and job["state"] in ("queued", "searching"):
                 set_state(job_id, "searching", "")
+                # A ladder left over from a run that stopped on a slskd stall
+                # is still the right answer -- the search that produced it has
+                # not become wrong. Resume it rather than paying for another.
+                keys = job.keys()
+                left = json.loads(
+                    (job["candidates"] if "candidates" in keys else None) or "[]")
+                if left:
+                    log.info("job %s resuming ladder, %d candidate(s) left",
+                             job_id[:8], len(left))
+                    await try_next_candidate(job_id, job["artist"],
+                                             job["title"], job["album"])
+                else:
                     await pick_and_enqueue(job_id, job["artist"], job["title"],
                                            job["album"])
         except Exception as e:
@@ -1261,8 +1436,16 @@ def transfer_index(payload: list) -> dict[str, dict]:
 
 
 def stalled_for(job: sqlite3.Row) -> float:
-    """Minutes since this transfer last had anything to show for itself."""
-    started = job["started_at"] if "started_at" in job.keys() else None
+    """Minutes since this transfer last had anything to show for itself.
+
+    Falls back to created_at, which is what makes this work on rows written
+    before started_at existed. Those are `downloading` with no clock at all,
+    so returning zero for them would exempt precisely the jobs that have been
+    stuck the longest -- the ones this check exists to catch.
+    """
+    keys = job.keys()
+    started = ((job["started_at"] if "started_at" in keys else None)
+               or (job["created_at"] if "created_at" in keys else None))
     if not started:
         return 0.0
     try:
@@ -1345,18 +1528,36 @@ async def import_worker():
                         actual = job["size"]
 
                     conf = job["confidence"]
+                    why = job["match_why"] or ""
+                    # Applied here rather than in confidence(), because the
+                    # file does not exist yet when the candidate is chosen.
+                    penalty = length_check(src, job["duration"])
+                    if penalty is not None and conf is not None:
+                        mult, reason = penalty
+                        conf = round(conf * mult, 3)
+                        why = f"{why}; {reason}" if why else reason
+                        with db() as c:
+                            c.execute("UPDATE jobs SET confidence=?,"
+                                      " match_why=? WHERE id=?",
+                                      (conf, why, job["id"]))
+                        log.info("job %s length mismatch: %s",
+                                 job["id"][:8], reason)
                     if conf is not None and conf < AUTO_CONFIDENCE:
                         band = ("almost certainly wrong"
                                 if conf < REVIEW_CONFIDENCE else "worth a look")
                         set_state(job["id"], "review",
-                                  f"{conf:.0%} confident, {band} -- "
-                                  f"{job['match_why'] or ''}", size=actual)
+                                  f"{conf:.0%} confident, {band} -- {why}",
+                                  size=actual)
                         await notify_review(job)
                         continue
 
                     dest = tag_and_file(job, src)
                     set_state(job["id"], "imported", str(dest),
                               imported_path=str(dest), size=actual)
+                    # Rewritten per track, so the playlist appears in Navidrome
+                    # with the first arrival rather than the last.
+                    if playlist_of(job):
+                        write_playlist(job["playlist_id"])
                     await notify_imported(job, dest)
 
                 elif "Completed" in st and "Succeeded" not in st:
@@ -2123,7 +2324,8 @@ async def artist_detail(artist_id: int):
 
 async def enqueue_job(artist: str, title: str, album=None, year=None,
                       track=None, deezer_id=None, requested_by=None,
-                      cover=None, genre=None, duration=None) -> str:
+                      cover=None, genre=None, duration=None,
+                      playlist_id=None, playlist_seq=None) -> str:
     """Create a job and hand it to the search queue."""
     job_id = str(uuid.uuid4())
     # The single choke point for all three request paths, so normalise here.
@@ -2136,10 +2338,12 @@ async def enqueue_job(artist: str, title: str, album=None, year=None,
     with db() as c:
         c.execute(
             "INSERT INTO jobs (id,state,detail,artist,title,album,year,track,"
-            "recording_mbid,release_mbid,requested_by,cover,genre,duration) "
-            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+            "recording_mbid,release_mbid,requested_by,cover,genre,duration,"
+            "playlist_id,playlist_seq) "
+            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
             (job_id, "queued", "", artist, title, album, year, track,
-             str(deezer_id or ""), None, requested_by, cover, genre, duration))
+             str(deezer_id or ""), None, requested_by, cover, genre, duration,
+             playlist_id, playlist_seq))
     await work_q.put(job_id)
     return job_id
 
@@ -2221,6 +2425,218 @@ MAX_IMPORT_LINES = 500
 MAX_IMPORT_TRACKS = 1000
 
 
+# ─────────────────────────── playlists ───────────────────────────
+#
+# Paste a playlist link from wherever it was shared and get the tracks.
+#
+# Deezer is a real API. Spotify and Apple Music are NOT: neither offers a free
+# way to read a public playlist -- Apple's MusicKit needs a paid developer
+# account -- so those two read the public web page and pull the JSON blob the
+# page already embeds for its own client. That works, and it needs no key, no
+# OAuth and no account. It is also a scrape, and it will break the day either
+# company reshuffles their frontend.
+#
+# So the failure is made loud and specific rather than silent. A resolver that
+# finds the page but not the tracks says the format changed, because "0 tracks"
+# and "this playlist is empty" must not look the same to whoever has to fix it.
+
+SPOTIFY_URL = re.compile(
+    r"open\.spotify\.com/(?:intl-[a-z-]+/)?(playlist|album)/([A-Za-z0-9]+)")
+APPLE_URL = re.compile(
+    r"music\.apple\.com/[a-z]{2}/(playlist|album)/[^/]+/(pl\.[A-Za-z0-9.-]+|\d+)")
+
+# Playlists are capped harder than a paste: one link can be thousands of tracks
+# and the quota check cannot run until the whole thing has been resolved.
+MAX_PLAYLIST_TRACKS = int(os.getenv("MUSE_MAX_PLAYLIST_TRACKS", "500"))
+
+BROWSERISH = {
+    # These are public pages; a default httpx agent gets a different response
+    # from a real one on both hosts.
+    "User-Agent": ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+                   "AppleWebKit/537.36 (KHTML, like Gecko) "
+                   "Chrome/122.0 Safari/537.36"),
+    "Accept-Language": "en-US,en;q=0.9",
+}
+
+
+class PlaylistUnavailable(Exception):
+    """The link could not be turned into a track list."""
+
+
+def _walk(node, want: tuple[str, ...], out: list):
+    """Collect every dict carrying all of `want`, at any depth."""
+    if isinstance(node, dict):
+        if all(k in node for k in want):
+            out.append(node)
+        for v in node.values():
+            _walk(v, want, out)
+    elif isinstance(node, list):
+        for v in node:
+            _walk(v, want, out)
+
+
+async def _fetch(url: str) -> str:
+    async with httpx.AsyncClient(timeout=25, follow_redirects=True,
+                                 headers=BROWSERISH) as c:
+        r = await c.get(url)
+        if r.status_code == 404:
+            raise PlaylistUnavailable("that playlist does not exist, or is private")
+        r.raise_for_status()
+        return r.text
+
+
+async def _deezer_playlist(kind: str, ident: str) -> tuple[str, list[tuple[str, str]]]:
+    async with httpx.AsyncClient(timeout=25) as c:
+        r = await c.get(f"{DEEZER}/{kind}/{ident}")
+        if r.status_code != 200:
+            raise PlaylistUnavailable(f"Deezer returned {r.status_code}")
+        d = r.json()
+    if d.get("error"):
+        raise PlaylistUnavailable(str(d["error"])[:120])
+    tracks = [((t.get("artist") or {}).get("name") or "", t.get("title") or "")
+              for t in (d.get("tracks") or {}).get("data", [])]
+    return d.get("title") or f"Deezer {kind}", [t for t in tracks if t[1]]
+
+
+async def _spotify_playlist(kind: str, ident: str) -> tuple[str, list[tuple[str, str]]]:
+    # The embed player is server-rendered with the whole track list inline,
+    # which is why this needs no credentials.
+    html = await _fetch(f"https://open.spotify.com/embed/{kind}/{ident}")
+    m = re.search(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', html, re.S)
+    if not m:
+        raise PlaylistUnavailable(
+            "Spotify's embed page no longer carries __NEXT_DATA__ — the "
+            "resolver in app.py needs updating")
+    try:
+        entity = (json.loads(m.group(1))["props"]["pageProps"]["state"]
+                  ["data"]["entity"])
+    except (KeyError, ValueError) as e:
+        raise PlaylistUnavailable(
+            f"Spotify's embed payload changed shape ({errstr(e)}) — the "
+            "resolver in app.py needs updating") from e
+    items = entity.get("trackList") or []
+    if not items:
+        raise PlaylistUnavailable("Spotify returned no tracks for that link")
+    # title is the song, subtitle is the artist.
+    return (entity.get("name") or f"Spotify {kind}",
+            [(t.get("subtitle") or "", t.get("title") or "") for t in items
+             if t.get("title")])
+
+
+async def _apple_playlist(kind: str, ident: str, url: str) -> tuple[str, list[tuple[str, str]]]:
+    html = await _fetch(url)
+    m = re.search(r'id="serialized-server-data"[^>]*>(.*?)</script>', html, re.S)
+    if not m:
+        raise PlaylistUnavailable(
+            "Apple Music's page no longer carries serialized-server-data — "
+            "the resolver in app.py needs updating")
+    try:
+        data = json.loads(m.group(1))
+    except ValueError as e:
+        raise PlaylistUnavailable(
+            f"Apple Music's payload did not parse ({errstr(e)})") from e
+    rows: list = []
+    _walk(data, ("title", "artistName"), rows)
+    if not rows:
+        raise PlaylistUnavailable(
+            "Apple Music's payload changed shape — the resolver in app.py "
+            "needs updating")
+    name = ""
+    heads: list = []
+    _walk(data, ("headerTitle",), heads)
+    if heads:
+        name = heads[0].get("headerTitle") or ""
+    seen, tracks = set(), []
+    for t in rows:
+        key = (t.get("artistName"), t.get("title"))
+        if key in seen:
+            continue
+        seen.add(key)
+        tracks.append((t.get("artistName") or "", t.get("title") or ""))
+    return name or f"Apple Music {kind}", tracks
+
+
+async def resolve_playlist(url: str) -> tuple[str, str, list[tuple[str, str]]]:
+    """Turn a shared playlist link into (source, name, [(artist, title), ...])."""
+    url = url.strip()
+    if m := DEEZER_URL.search(url):
+        name, tracks = await _deezer_playlist(*m.groups())
+        source = "deezer"
+    elif m := SPOTIFY_URL.search(url):
+        name, tracks = await _spotify_playlist(*m.groups())
+        source = "spotify"
+    elif m := APPLE_URL.search(url):
+        name, tracks = await _apple_playlist(*m.groups(), url)
+        source = "apple"
+    else:
+        raise PlaylistUnavailable(
+            "that link is not a Deezer, Spotify or Apple Music playlist")
+    if not tracks:
+        raise PlaylistUnavailable("that playlist appears to be empty")
+    if len(tracks) > MAX_PLAYLIST_TRACKS:
+        raise PlaylistUnavailable(
+            f"that playlist has {len(tracks)} tracks; the limit is "
+            f"{MAX_PLAYLIST_TRACKS}")
+    return source, name, tracks
+
+
+# Navidrome mounts the library at a different path than muse does -- /music
+# against /data/media/music -- so an m3u written with muse's own paths resolves
+# to nothing on the other side of the wall.
+PLAYLIST_PREFIX = os.getenv("MUSE_PLAYLIST_PREFIX", "/music")
+
+
+def write_playlist(playlist_id: str) -> Path | None:
+    """Write the .m3u for a playlist, or remove it when nothing has landed yet.
+
+    Rewritten from scratch every time one of its tracks imports, which is both
+    simpler than patching a file in place and the reason the playlist shows up
+    in Navidrome as soon as the first track arrives instead of after the last.
+    """
+    with db() as c:
+        pl = c.execute("SELECT * FROM playlists WHERE id=?",
+                       (playlist_id,)).fetchone()
+        if not pl:
+            return None
+        rows = c.execute(
+            "SELECT artist,title,imported_path FROM jobs WHERE playlist_id=? "
+            "AND state='imported' AND imported_path IS NOT NULL "
+            "ORDER BY playlist_seq", (playlist_id,)).fetchall()
+
+    dest = LIBRARY_DIR / f"{safe(pl['name'])}.m3u"
+    root = LIBRARY_DIR.resolve()
+    if not Path(os.path.normpath(dest)).resolve().is_relative_to(root):
+        log.warning("playlist %s: refusing to write outside the library", playlist_id)
+        return None
+
+    lines = ["#EXTM3U", f"#PLAYLIST:{pl['name']}"]
+    for r in rows:
+        try:
+            rel = Path(r["imported_path"]).resolve().relative_to(root)
+        except ValueError:
+            continue        # imported somewhere else entirely; not ours to list
+        lines.append(f"#EXTINF:-1,{r['artist']} - {r['title']}")
+        lines.append(f"{PLAYLIST_PREFIX}/{rel.as_posix()}")
+
+    if len(lines) == 2:
+        # Nothing imported yet. Leaving a two-line file behind would make
+        # Navidrome show an empty playlist, which reads as a failure.
+        if dest.exists():
+            dest.unlink()
+        return None
+    dest.write_text("\n".join(lines) + "\n", encoding="utf-8")
+    with db() as c:
+        c.execute("UPDATE playlists SET m3u_path=?, tracks_written=? WHERE id=?",
+                  (str(dest), len(rows), playlist_id))
+    log.info("playlist %s -> %s (%d tracks)", playlist_id[:8], dest, len(rows))
+    return dest
+
+
+def playlist_of(job) -> str | None:
+    keys = job.keys() if hasattr(job, "keys") else ()
+    return job["playlist_id"] if "playlist_id" in keys else None
+
+
 async def expand_url(kind: str, ident: str) -> list[str]:
     """Turn a Deezer album/playlist URL into 'Artist - Title' lines."""
     out = []
@@ -2298,6 +2714,101 @@ async def import_list(body: ImportBody, request: Request):
     return {"queued": queued, "skipped": skipped, "pending": work_q.qsize()}
 
 
+class PlaylistBody(BaseModel):
+    url: str
+    # Resolve and show what is in it without queueing anything. A shared link
+    # is opaque -- you cannot tell 12 tracks from 400 by looking at it -- and
+    # queueing 400 against a quota you cannot see first is a bad trade.
+    preview: bool = False
+    name: str | None = None
+
+
+@app.post("/api/playlist")
+async def playlist_import(body: PlaylistBody, request: Request):
+    """Turn a shared playlist link into queued requests and an m3u.
+
+    Every track goes through the same Deezer resolution a pasted line does, so
+    the matching, ranking and review behaviour is identical; the only new thing
+    is that the results stay grouped and ordered.
+    """
+    try:
+        source, name, tracks = await resolve_playlist(body.url)
+    except PlaylistUnavailable as e:
+        raise HTTPException(422, str(e))
+    except Exception as e:
+        raise HTTPException(502, f"could not read that playlist: {errstr(e)}")
+
+    name = (body.name or name).strip()[:MAX_COMPONENT] or "Playlist"
+
+    if body.preview:
+        return {"source": source, "name": name, "count": len(tracks),
+                "tracks": [{"artist": a, "title": t} for a, t in tracks[:100]]}
+
+    check_quota(request, adding=len(tracks))
+
+    playlist_id = str(uuid.uuid4())
+    with db() as c:
+        c.execute("INSERT INTO playlists (id,name,source,source_url,"
+                  "requested_by,tracks_total) VALUES (?,?,?,?,?,?)",
+                  (playlist_id, name, source, body.url[:500],
+                   session_user(request), len(tracks)))
+
+    queued, skipped = [], []
+    for seq, (artist, title) in enumerate(tracks):
+        line = f"{artist} {title}".strip()
+        try:
+            hits = await deezer_search(SPLIT.sub(" ", line), limit=10)
+        except Exception as e:
+            skipped.append({"line": line, "why": errstr(e)})
+            continue
+        if not hits:
+            skipped.append({"line": line, "why": "no match on Deezer"})
+            continue
+        h = hits[0]
+        e = await enrich(h["deezer_id"])
+        jid = await enqueue_job(
+            e.get("artist") or h["artist"], h["title"],
+            e.get("album") or h["album"], e.get("year"), e.get("track"),
+            h["deezer_id"], requested_by=session_user(request),
+            cover=h.get("cover"), duration=h.get("length"),
+            playlist_id=playlist_id, playlist_seq=seq)
+        queued.append({"id": jid, "line": line, "cover": h.get("cover"),
+                       "matched": f"{h['artist']} — {h['title']}"})
+
+    log.info("playlist %s (%s) %r: %d queued, %d skipped",
+             playlist_id[:8], source, name, len(queued), len(skipped))
+    return {"playlist_id": playlist_id, "name": name, "source": source,
+            "queued": queued, "skipped": skipped, "pending": work_q.qsize()}
+
+
+@app.get("/api/playlists")
+async def playlists(request: Request):
+    """Playlists and how far along each one is.
+
+    Shared on purpose: an m3u on disk is visible to every Navidrome user
+    regardless, so pretending otherwise here would only be a fiction.
+    """
+    with db() as c:
+        rows = c.execute(
+            "SELECT * FROM playlists ORDER BY created_at DESC LIMIT 100"
+        ).fetchall()
+        counts = {}
+        for r in c.execute(
+                "SELECT playlist_id, state, COUNT(*) n FROM jobs "
+                "WHERE playlist_id IS NOT NULL GROUP BY playlist_id, state"):
+            counts.setdefault(r["playlist_id"], {})[r["state"]] = r["n"]
+    out = []
+    for r in rows:
+        c_ = counts.get(r["id"], {})
+        done = c_.get("imported", 0)
+        out.append({**dict(r), "counts": c_, "done": done,
+                    "pending": sum(c_.get(s, 0) for s in
+                                   ("queued", "searching", "downloading")),
+                    "failed": c_.get("failed", 0),
+                    "review": c_.get("review", 0)})
+    return {"playlists": out}
+
+
 class ReviewBody(BaseModel):
     action: str          # approve | reject
 
@@ -2355,6 +2866,9 @@ async def review_decide(job_id: str, body: ReviewBody, request: Request):
             except OSError as e:
                 log.warning("review: could not delete %s: %s", src, e)
         set_state(job_id, "failed", "rejected in review", reviewed_by=me)
+        # It may already be listed from an earlier pass; the rewrite drops it.
+        if playlist_of(job):
+            write_playlist(job["playlist_id"])
         return {"ok": True, "state": "failed"}
 
     if not src:
@@ -2362,6 +2876,11 @@ async def review_decide(job_id: str, body: ReviewBody, request: Request):
     dest = tag_and_file(job, src)
     set_state(job_id, "imported", str(dest), imported_path=str(dest),
               reviewed_by=me)
+    # After set_state, not before: write_playlist reads imported_path back out
+    # of the row, so running it first would write a playlist missing the very
+    # track that was just approved.
+    if playlist_of(job):
+        write_playlist(job["playlist_id"])
     log.info("review: %s approved %s", me, job_id[:8])
     return {"ok": True, "state": "imported", "path": str(dest)}
 
@@ -2518,13 +3037,18 @@ async def retry(job_id: str, request: Request):
     if row["state"] != "failed":
         raise HTTPException(409, f"only failed jobs can be retried "
                                  f"(this one is {row['state']})")
-    # Clear the previous attempt, or the ladder resumes mid-way through a
-    # candidate list gathered for a search that has long since expired.
+    # Clear the last transfer, but keep any candidates the previous run did
+    # not get to: those are left behind deliberately when slskd could not
+    # reach anyone, and re-running the search would only rediscover them.
     with db() as c:
+        row = c.execute("SELECT candidates FROM jobs WHERE id=?",
+                        (job_id,)).fetchone()
+        keep = bool(json.loads((row["candidates"] if row else None) or "[]"))
         c.execute("UPDATE jobs SET username=NULL, remote_file=NULL, size=NULL,"
-                  " confidence=NULL, match_why=NULL, candidates=NULL,"
-                  " tried=NULL, attempts=0, started_at=NULL WHERE id=?",
-                  (job_id,))
+                  " confidence=NULL, match_why=NULL, attempts=0,"
+                  " started_at=NULL"
+                  + ("" if keep else ", candidates=NULL, tried=NULL")
+                  + " WHERE id=?", (job_id,))
     set_state(job_id, "queued", "")
     await work_q.put(job_id)
     return {"ok": True}

tests/test_acquisition.py

@@ -8,6 +8,7 @@ the server filters -- can be exercised deterministically.
 Run standalone: python tests/test_acquisition.py
 """
 import asyncio, json, os, sys
+from pathlib import Path
 
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 os.environ.setdefault("MUSE_AUTH", "off")
@@ -161,6 +162,35 @@ def main() -> int:
     check("reordered filename still matches",
           app.match_ratio(r"x\Creep - Radiohead.mp3", "Radiohead", "Creep") == 1.0)
 
+    print("\nthe artist gate")
+    check("artist named in the folder is confirmed",
+          app.artist_confirmed(r"m\Michael Jackson\Xscape\02 Chicago.flac",
+                               "Michael Jackson"))
+    check("a different band's song with the same title is not",
+          not app.artist_confirmed(
+              r"@@jwbvv\Music\The Fall\1990-Extricate\08-Chicago, Now!.flac",
+              "Michael Jackson"))
+    check("out-of-order naming still counts",
+          app.artist_confirmed(r"x\Jackson, Michael - Chicago.flac",
+                               "Michael Jackson"))
+
+    # The real regression: bare-title search returns every song called
+    # "Chicago", and an available stranger used to outrank a queued match.
+    gate = FakeSlskd(per_query={"Chicago": [
+        peer("stranger", r"@@x\Music\The Fall\1990-Extricate\08-Chicago, Now!.flac",
+             slot=True, queue=0),
+        peer("correct", r"m\Michael Jackson\XSCAPE\02 - Chicago.flac",
+             slot=False, queue=300),
+    ]})
+    app.slskd_client = gate
+    make_job("j6", "Michael Jackson", "Chicago")
+    asyncio.run(app.pick_and_enqueue("j6", "Michael Jackson", "Chicago"))
+    r = job_row("j6")
+    check("the right artist wins even from a peer with 300 queued",
+          r["username"] == "correct", f"picked {r['username']}")
+    check("and the wrong-artist file is dropped, not merely ranked lower",
+          len(json.loads(r["candidates"] or "[]")) == 0, r["candidates"])
+
     print("\ncandidate ladder")
     fake = FakeSlskd(per_query={
         "Billie Jean": [peer("alice", r"m\Michael Jackson\Billie Jean.flac"),
@@ -195,6 +225,13 @@ def main() -> int:
     check("only fails once the ladder is exhausted", r["state"] == "failed", r["detail"])
     check("failure names every attempt, not just the last",
           "tried 3 peers" in (r["detail"] or ""), r["detail"])
+    # A row carried over from before the ladder has attempts=0 but a real
+    # failure recorded, which must not render as "tried 0 peers: 1x ...".
+    check("a migrated row counts its evidence, not its counter",
+          app.exhausted_detail(
+              [{"username": "u1", "reason": "no bytes in 20m"}], 0)
+          == "tried 1 peer: 1x no bytes in 20m",
+          app.exhausted_detail([{"username": "u1", "reason": "no bytes in 20m"}], 0))
 
     print("\nattempt cap")
     fake2 = FakeSlskd(per_query={"Creep": [
@@ -202,15 +239,46 @@ def main() -> int:
     # every peer refuses to accept the transfer
     fake2.enqueue_status = {f"u{i}": 500 for i in range(8)}
     app.slskd_client = fake2
+    # The cap defaults to the ladder size, so lower it here or this only ever
+    # proves the list ran out -- which is a different branch.
+    cap, app.MAX_ATTEMPTS = app.MAX_ATTEMPTS, 3
     make_job("j2", "Radiohead", "Creep")
     asyncio.run(app.pick_and_enqueue("j2", "Radiohead", "Creep"))
     r = job_row("j2")
     check("gives up at MAX_ATTEMPTS, not at the end of the list",
-          r["state"] == "failed" and r["attempts"] == app.MAX_ATTEMPTS,
+          r["state"] == "failed" and r["attempts"] == 3,
           f"state={r['state']} attempts={r['attempts']}")
+    check("and leaves the untried remainder alone",
+          len(json.loads(r["candidates"] or "[]")) > 0, r["candidates"])
+    app.MAX_ATTEMPTS = cap
     check("unreachable peers are reported as such",
           "could not reach peer" in (r["tried"] or ""), r["detail"])
 
+    print("\nsystemic failures do not burn the ladder")
+    # slskd timing out says nothing about any particular file, so walking the
+    # whole list to learn it once is waste -- and it is what took a real
+    # request down: five enqueues in twenty-five seconds during one slskd stall.
+    fake4 = FakeSlskd(per_query={"Chicago": [
+        peer(f"p{i}", r"x\Michael Jackson\Xscape\Chicago.flac") for i in range(8)]})
+
+    class Timeouts(FakeSlskd):
+        async def post(self, url, json=None):
+            if url != "/api/v0/searches":
+                raise RuntimeError("The wait timed out after 5000 milliseconds")
+            return await FakeSlskd.post(self, url, json=json)
+
+    t = Timeouts(per_query=fake4.per_query)
+    app.slskd_client = t
+    make_job("j5", "Michael Jackson", "Chicago")
+    asyncio.run(app.pick_and_enqueue("j5", "Michael Jackson", "Chicago"))
+    r = job_row("j5")
+    check("stops after SYSTEMIC_LIMIT instead of walking the whole list",
+          r["attempts"] == app.SYSTEMIC_LIMIT, r["attempts"])
+    left = json.loads(r["candidates"] or "[]")
+    check("untried candidates are preserved for a retry", len(left) > 0, len(left))
+    check("the message says a retry will resume",
+          "retry resumes" in (r["detail"] or ""), r["detail"])
+
     print("\nempty results")
     fake3 = FakeSlskd(per_query={})
     app.slskd_client = fake3
@@ -244,6 +312,36 @@ def main() -> int:
     print("\nsearch hygiene")
     check("finished searches are cleaned up", len(flaky.deleted) >= 1)
 
+    print("\nlength cross-check")
+    # The signal a filename cannot fake. Every textual cue agreed on the file
+    # that got filed as Michael Jackson; only the running time disagreed.
+    class FakeInfo:
+        def __init__(self, n): self.length = n
+
+    class FakeAudio:
+        def __init__(self, n): self.info = FakeInfo(n)
+
+    real_file = mutagen_file = app.mutagen.File
+    try:
+        app.mutagen.File = lambda p: FakeAudio(359.0)      # The Fall
+        got = app.length_check(Path("x.flac"), 245 * 1000)  # Michael Jackson
+        check("a 47% length disagreement is caught", got is not None, got)
+        check("and it drops the file below the review floor",
+              got and round(0.75 * got[0], 3) < app.REVIEW_CONFIDENCE,
+              got and round(0.75 * got[0], 3))
+        check("the reason names both lengths",
+              got and "5:59" in got[1] and "4:05" in got[1], got and got[1])
+
+        app.mutagen.File = lambda p: FakeAudio(247.0)
+        check("a two-second drift is not flagged",
+              app.length_check(Path("x.flac"), 245 * 1000) is None)
+
+        app.mutagen.File = lambda p: FakeAudio(359.0)
+        check("nothing to compare against is not a failure",
+              app.length_check(Path("x.flac"), None) is None)
+    finally:
+        app.mutagen.File = mutagen_file
+
     print("\nstall clock")
     from datetime import datetime, timedelta, timezone
     with app.db() as c:
@@ -253,10 +351,15 @@ def main() -> int:
     check("a 90-minute-old transfer reads as stalled",
           app.stalled_for(job_row("j4")) > app.STALL_MINUTES,
           app.stalled_for(job_row("j4")))
+    # A row written before started_at existed still has to age out, or the
+    # jobs that have been stuck longest are the only ones exempt from the check.
     with app.db() as c:
-        c.execute("UPDATE jobs SET started_at=NULL WHERE id=?", ("j4",))
-    check("a job with no clock is not spuriously stalled",
-          app.stalled_for(job_row("j4")) == 0.0)
+        c.execute("UPDATE jobs SET started_at=NULL, created_at=? WHERE id=?",
+                  ((datetime.now(timezone.utc) - timedelta(days=2))
+                   .isoformat(sep=" ")[:19], "j4"))
+    check("a pre-upgrade row falls back to created_at",
+          app.stalled_for(job_row("j4")) > app.STALL_MINUTES,
+          app.stalled_for(job_row("j4")))
 
     print(f"\n{ok} passed, {fail} failed")
     return 1 if fail else 0

tests/test_playlist.py

@@ -0,0 +1,176 @@
+"""Playlist import: link resolution, ordering, and the m3u Navidrome reads.
+
+The resolvers are exercised against synthetic payloads rather than the live
+sites. Two reasons: a test that fails when Spotify has a bad afternoon is not
+testing this code, and the failure that actually matters here is the *shape*
+changing -- which is precisely what a fixture can assert and a live fetch
+cannot.
+
+Run standalone: python tests/test_playlist.py
+"""
+import asyncio, json, os, sys
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+os.environ.setdefault("MUSE_AUTH", "off")
+os.environ.setdefault("MUSE_DB", "dev/playlist.db")
+os.environ.setdefault("MUSE_LIBRARY_DIR", "dev/pl-library")
+
+for f in ("dev/playlist.db", "dev/playlist.db-wal", "dev/playlist.db-shm"):
+    if os.path.exists(f):
+        os.remove(f)
+
+import app  # noqa: E402
+
+app.init_db()
+app.LIBRARY_DIR.mkdir(parents=True, exist_ok=True)
+
+ok = fail = 0
+
+
+def check(name, cond, detail=""):
+    global ok, fail
+    if cond:
+        ok += 1
+        print(f"  PASS  {name}")
+    else:
+        fail += 1
+        print(f"  FAIL  {name}  {detail}")
+
+
+SPOTIFY_PAGE = (
+    '<html><script id="__NEXT_DATA__" type="application/json">'
+    + json.dumps({"props": {"pageProps": {"state": {"data": {"entity": {
+        "name": "Late Night",
+        "trackList": [
+            {"title": "The Color Violet", "subtitle": "Tory Lanez"},
+            {"title": "No Pole", "subtitle": "Don Toliver"},
+        ]}}}}}})
+    + "</script></html>")
+
+APPLE_PAGE = (
+    '<html><script id="serialized-server-data" type="application/json">'
+    + json.dumps([{"data": {"sections": [
+        {"headerTitle": "Heavy Rotation"},
+        {"items": [
+            {"title": "YUKON", "artistName": "Justin Bieber"},
+            {"title": "LET THE WORLD BURN", "artistName": "Chris Grey"},
+            {"title": "YUKON", "artistName": "Justin Bieber"},   # dupe
+        ]}]}}])
+    + "</script></html>")
+
+
+def fake_fetch(body):
+    async def _f(url):
+        return body
+    return _f
+
+
+def make_playlist(pid, name, rows):
+    """rows: (seq, artist, title, state, imported_path)"""
+    with app.db() as c:
+        c.execute("INSERT OR REPLACE INTO playlists (id,name,source,"
+                  "source_url,tracks_total) VALUES (?,?,?,?,?)",
+                  (pid, name, "spotify", "https://x", len(rows)))
+        for seq, artist, title, state, path in rows:
+            c.execute(
+                "INSERT INTO jobs (id,state,detail,artist,title,"
+                "imported_path,playlist_id,playlist_seq) "
+                "VALUES (?,?,?,?,?,?,?,?)",
+                (f"{pid}-{seq}", state, "", artist, title, path, pid, seq))
+
+
+def main() -> int:
+    print("\nlink routing")
+    for url, want in [
+        ("https://open.spotify.com/playlist/abc123", "spotify"),
+        ("https://open.spotify.com/intl-pt/playlist/abc123", "spotify"),
+        ("https://music.apple.com/us/playlist/hits/pl.abc-123", "apple"),
+        ("https://www.deezer.com/en/playlist/1313621735", "deezer"),
+    ]:
+        matched = ("spotify" if app.SPOTIFY_URL.search(url)
+                   else "apple" if app.APPLE_URL.search(url)
+                   else "deezer" if app.DEEZER_URL.search(url) else "none")
+        check(f"{want} link routes to {want}", matched == want, f"got {matched}")
+
+    print("\nresolvers, against a fixed payload shape")
+    real = app._fetch
+    try:
+        app._fetch = fake_fetch(SPOTIFY_PAGE)
+        name, tracks = asyncio.run(app._spotify_playlist("playlist", "x"))
+        check("spotify name", name == "Late Night", name)
+        check("spotify pulls artist from subtitle",
+              tracks[0] == ("Tory Lanez", "The Color Violet"), tracks[:1])
+
+        app._fetch = fake_fetch(APPLE_PAGE)
+        name, tracks = asyncio.run(app._apple_playlist("playlist", "pl.x", "u"))
+        check("apple name", name == "Heavy Rotation", name)
+        check("apple pulls artist from artistName",
+              ("Justin Bieber", "YUKON") in tracks, tracks)
+        check("apple de-duplicates repeated entries", len(tracks) == 2, tracks)
+
+        # The failure that actually matters: they changed the page.
+        app._fetch = fake_fetch("<html>nothing useful here</html>")
+        for fn, label in ((app._spotify_playlist, "spotify"),
+                          (app._apple_playlist, "apple")):
+            try:
+                asyncio.run(fn("playlist", "x", "u") if label == "apple"
+                            else fn("playlist", "x"))
+                check(f"{label} shape change is reported", False, "no raise")
+            except app.PlaylistUnavailable as e:
+                check(f"{label} shape change names the resolver",
+                      "needs updating" in str(e), str(e))
+    finally:
+        app._fetch = real
+
+    print("\nthe m3u Navidrome reads")
+    lib = app.LIBRARY_DIR
+    (lib / "Tory Lanez").mkdir(parents=True, exist_ok=True)
+    (lib / "Don Toliver").mkdir(parents=True, exist_ok=True)
+    a = lib / "Tory Lanez" / "The Color Violet.flac"; a.write_text("x")
+    b = lib / "Don Toliver" / "No Pole.flac"; b.write_text("x")
+
+    make_playlist("p1", "Late Night", [
+        (1, "Don Toliver", "No Pole", "imported", str(b)),
+        (0, "Tory Lanez", "The Color Violet", "imported", str(a)),
+        (2, "Nobody", "Missing", "failed", None),
+    ])
+    dest = app.write_playlist("p1")
+    text = dest.read_text()
+    check("m3u is written", dest and dest.exists(), dest)
+    check("declares the playlist name for Navidrome",
+          "#PLAYLIST:Late Night" in text, text)
+    check("uses Navidrome's mount, not muse's",
+          "/music/Tory Lanez/The Color Violet.flac" in text
+          and str(lib) not in text, text)
+    check("keeps the playlist's own order, not insertion order",
+          text.index("The Color Violet") < text.index("No Pole"), text)
+    check("omits tracks that never imported", "Missing" not in text, text)
+
+    print("\nedge cases")
+    make_playlist("p2", "Nothing Landed Yet", [
+        (0, "A", "B", "searching", None)])
+    check("no m3u while the playlist is empty",
+          app.write_playlist("p2") is None)
+    check("and no stray file left behind",
+          not (lib / "Nothing Landed Yet.m3u").exists())
+
+    make_playlist("p3", "../../escape", [
+        (0, "Tory Lanez", "The Color Violet", "imported", str(a))])
+    out = app.write_playlist("p3")
+    check("a traversing playlist name cannot escape the library",
+          out is None or out.resolve().is_relative_to(lib.resolve()), out)
+
+    # A track approved and then rejected has to leave the file.
+    with app.db() as c:
+        c.execute("UPDATE jobs SET state='failed' WHERE id=?", ("p1-0",))
+    app.write_playlist("p1")
+    check("rejecting a track drops it from the m3u",
+          "The Color Violet" not in dest.read_text(), dest.read_text())
+
+    print(f"\n{ok} passed, {fail} failed")
+    return 1 if fail else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

web/app/page.tsx

@@ -14,6 +14,7 @@ import { TrackRow } from "@/components/track-row";
 import { AlbumView } from "@/components/album-view";
 import { ArtistView } from "@/components/artist-view";
 import { BulkImport } from "@/components/bulk-import";
+import { PlaylistImport } from "@/components/playlist-import";
 import { Requests, type JobsSnapshot } from "@/components/requests";
 import { Review } from "@/components/review";
 import { DiscoverPanel } from "@/components/discover";
@@ -203,7 +204,12 @@ function Muse() {
             {nav.tab === "discover" && <DiscoverPanel onArtist={doSearch} />}
             {nav.tab === "jobs" && <Requests snap={snap} refresh={refresh} />}
             {nav.tab === "review" && <Review onDecided={refresh} />}
-            {nav.tab === "bulk" && <BulkImport onQueued={refresh} />}
+            {nav.tab === "bulk" && (
+              <div className="flex flex-col gap-4">
+                <PlaylistImport onQueued={refresh} />
+                <BulkImport onQueued={refresh} />
+              </div>
+            )}
             {nav.tab === "settings" && me && <Settings me={me} refresh={refresh} />}
             {nav.tab === "admin" && isAdmin && <AdminPanel me={me?.user ?? null} />}
           </motion.div>

web/components/playlist-import.tsx

@@ -0,0 +1,204 @@
+"use client";
+
+import * as React from "react";
+import { AnimatePresence, motion } from "framer-motion";
+import { Link2, ListMusic, Loader2, CheckCircle2 } from "lucide-react";
+import { api, type PlaylistPreview, type PlaylistRow } from "@/lib/api";
+import { useToast } from "@/components/toast";
+import { Button } from "@/components/ui/button";
+import { Card, CardTitle, CardNote } from "@/components/ui/card";
+import { Input } from "@/components/ui/field";
+import { Badge } from "@/components/ui/badge";
+import { Progress } from "@/components/ui/progress";
+import { useMotionOK, item } from "@/lib/motion";
+
+const SOURCE: Record<string, string> = {
+  spotify: "Spotify",
+  apple: "Apple Music",
+  deezer: "Deezer",
+};
+
+/**
+ * Paste a shared link, get the playlist.
+ *
+ * Resolve is deliberately split from queue. A shared link is opaque — you
+ * cannot tell twelve tracks from four hundred by looking at it — and four
+ * hundred is a quota decision, so the tracks are shown before anything is
+ * committed.
+ */
+export function PlaylistImport({ onQueued }: { onQueued?: () => void }) {
+  const toast = useToast();
+  const ok = useMotionOK();
+  const [url, setUrl] = React.useState("");
+  const [name, setName] = React.useState("");
+  const [preview, setPreview] = React.useState<PlaylistPreview | null>(null);
+  const [busy, setBusy] = React.useState<"idle" | "resolving" | "queueing">("idle");
+  const [rows, setRows] = React.useState<PlaylistRow[] | null>(null);
+
+  // Not async, and resolving through .then: setState then happens in a promise
+  // callback rather than synchronously inside the effect, which is both what
+  // React wants and what the rest of the polling components here already do.
+  const refresh = React.useCallback(() => {
+    api
+      .playlists()
+      .then((d) => setRows(d.playlists))
+      .catch(() => {
+        /* a 401 here just means the poll outlived the session */
+      });
+  }, []);
+
+  React.useEffect(() => {
+    refresh();
+    const t = setInterval(refresh, 5000);
+    return () => clearInterval(t);
+  }, [refresh]);
+
+  const resolve = async () => {
+    if (!url.trim()) return toast("Paste a playlist link first", true);
+    setBusy("resolving");
+    setPreview(null);
+    try {
+      const d = await api.playlistPreview(url.trim());
+      setPreview(d);
+      setName(d.name);
+    } catch (e) {
+      toast((e as Error).message, true);
+    }
+    setBusy("idle");
+  };
+
+  const queue = async () => {
+    setBusy("queueing");
+    try {
+      const d = await api.playlistImport(url.trim(), name.trim() || undefined);
+      toast(
+        `Queued ${d.queued.length} of ${d.queued.length + d.skipped.length} from ${d.name}`,
+      );
+      setUrl("");
+      setPreview(null);
+      onQueued?.();
+      refresh();
+    } catch (e) {
+      toast((e as Error).message, true);
+    }
+    setBusy("idle");
+  };
+
+  return (
+    <div className="flex flex-col gap-4">
+      <Card pad="lg">
+        <CardTitle>Import a shared playlist</CardTitle>
+        <CardNote className="mt-1 mb-3.5 max-w-[64ch]">
+          Paste a link from Spotify, Apple Music or Deezer. Every track is matched on
+          Deezer and queued like any other request, and the playlist itself appears in
+          Navidrome — growing as tracks arrive rather than all at once.
+        </CardNote>
+
+        <div className="flex flex-col gap-2 sm:flex-row">
+          <Input
+            value={url}
+            onChange={(e) => setUrl(e.target.value)}
+            onKeyDown={(e) => e.key === "Enter" && busy === "idle" && resolve()}
+            placeholder="https://open.spotify.com/playlist/…"
+            aria-label="Playlist link"
+            inputMode="url"
+          />
+          <Button onClick={resolve} disabled={busy !== "idle"} className="shrink-0">
+            {busy === "resolving" ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : <Link2 className="h-3.5 w-3.5" />}
+            Resolve
+          </Button>
+        </div>
+
+        <AnimatePresence>
+          {preview && (
+            <motion.div
+              initial={ok ? { opacity: 0, height: 0 } : false}
+              animate={{ opacity: 1, height: "auto" }}
+              exit={ok ? { opacity: 0, height: 0 } : { opacity: 0 }}
+              className="overflow-hidden"
+            >
+              <div className="mt-4 rounded-card border border-edge bg-[var(--surface-2)] p-3.5">
+                <div className="mb-2.5 flex flex-wrap items-center gap-2">
+                  <Badge tone="accent">{SOURCE[preview.source] ?? preview.source}</Badge>
+                  <span className="text-callout text-dim nums">
+                    {preview.count} track{preview.count === 1 ? "" : "s"}
+                  </span>
+                </div>
+                <label className="mb-1 block text-caption font-medium text-dim">
+                  Name in Navidrome
+                </label>
+                <Input value={name} onChange={(e) => setName(e.target.value)} className="mb-3" />
+
+                <ol className="mb-3 max-h-52 overflow-y-auto text-callout">
+                  {preview.tracks.map((t, i) => (
+                    <li key={i} className="flex gap-2 border-b border-edge/60 py-1 last:border-0">
+                      <span className="w-6 shrink-0 text-right text-mute nums">{i + 1}</span>
+                      <span className="clamp1">
+                        {t.title} <span className="text-dim">· {t.artist}</span>
+                      </span>
+                    </li>
+                  ))}
+                  {preview.count > preview.tracks.length && (
+                    <li className="py-1 text-caption text-mute">
+                      …and {preview.count - preview.tracks.length} more
+                    </li>
+                  )}
+                </ol>
+
+                <Button onClick={queue} disabled={busy !== "idle"}>
+                  {busy === "queueing" ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : null}
+                  Queue all {preview.count}
+                </Button>
+              </div>
+            </motion.div>
+          )}
+        </AnimatePresence>
+      </Card>
+
+      {rows && rows.length > 0 && (
+        <Card pad="lg">
+          <CardTitle className="mb-3">Playlists</CardTitle>
+          <div className="flex flex-col gap-2.5">
+            {rows.map((p, i) => {
+              const total = p.tracks_total || 1;
+              const settled = p.done + p.failed;
+              return (
+                <motion.div
+                  key={p.id}
+                  {...item(i, ok)}
+                  className="rounded-card border border-edge bg-[var(--surface-2)] p-3"
+                >
+                  <div className="mb-1.5 flex items-center gap-2">
+                    <ListMusic className="h-3.5 w-3.5 shrink-0 text-mute" />
+                    <span className="min-w-0 flex-1 text-body font-semibold clamp1">{p.name}</span>
+                    <Badge tone="neutral">{SOURCE[p.source] ?? p.source}</Badge>
+                  </div>
+                  <Progress
+                    value={(p.done / total) * 100}
+                    tone={p.failed && settled >= total ? "warn" : "accent"}
+                    size="sm"
+                    label={`${p.name} progress`}
+                  />
+                  <div className="mt-1.5 flex flex-wrap gap-x-3 text-caption text-dim nums">
+                    <span className="text-ok">{p.done} in library</span>
+                    {p.pending > 0 && <span>{p.pending} in flight</span>}
+                    {p.review > 0 && <span className="text-warn">{p.review} to review</span>}
+                    {p.failed > 0 && <span className="text-err">{p.failed} failed</span>}
+                    <span className="text-mute">of {p.tracks_total}</span>
+                  </div>
+                  {p.m3u_path && (
+                    <div className="mt-1 flex items-center gap-1 text-caption text-mute">
+                      <CheckCircle2 className="h-3 w-3 text-ok" />
+                      in Navidrome with {p.tracks_written} track
+                      {p.tracks_written === 1 ? "" : "s"}
+                    </div>
+                  )}
+                </motion.div>
+              );
+            })}
+          </div>
+        </Card>
+      )}
+    </div>
+  );
+}

web/components/search-bar.tsx

@@ -23,24 +23,26 @@ export function SearchBar({
   onArtist: (id: number) => void;
 }) {
   const [q, setQ] = React.useState(initial ?? "");
-  // Follows the URL on Back/Forward, but never fights what is being typed:
-  // only a genuinely different incoming query replaces the field.
+  const [items, setItems] = React.useState<Item[]>([]);
+  const [open, setOpen] = React.useState(false);
+  const [sel, setSel] = React.useState(-1);
+  const inputRef = React.useRef<HTMLInputElement>(null);
+  // A value seeded from the URL must not pop the suggestion list open over the
+  // results that link was for. Only typing counts as asking for suggestions.
+  const typed = React.useRef(false);
   const seeded = React.useRef(initial);
+
+  // Follows the URL on Back/Forward, but never fights what is being typed:
+  // only a genuinely different incoming query replaces the field. Declared
+  // after the state it sets, or the effect closes over a binding that is not
+  // initialised yet.
   React.useEffect(() => {
-    if (initial !== undefined && initial !== seeded.current) {
+    if (initial === undefined || initial === seeded.current) return;
     seeded.current = initial;
     typed.current = false;
     setQ(initial);
     setOpen(false);
-    }
   }, [initial]);
-  const [items, setItems] = React.useState<Item[]>([]);
-  const [open, setOpen] = React.useState(false);
-  // A value seeded from the URL must not pop the suggestion list open over the
-  // results that link was for. Only typing counts as asking for suggestions.
-  const typed = React.useRef(false);
-  const [sel, setSel] = React.useState(-1);
-  const inputRef = React.useRef<HTMLInputElement>(null);
 
   // Debounced so a fast typist makes one upstream call, not eight. The
   // AbortController matters more than the delay: without it a slow response

web/lib/api.ts

@@ -266,7 +266,40 @@ export type Health = {
   downloads_readable?: boolean;
 };
 
+export type PlaylistPreview = {
+  source: "deezer" | "spotify" | "apple";
+  name: string;
+  count: number;
+  tracks: { artist: string; title: string }[];
+};
+
+export type PlaylistRow = {
+  id: string;
+  name: string;
+  source: string;
+  source_url: string | null;
+  tracks_total: number;
+  tracks_written: number;
+  m3u_path: string | null;
+  created_at: string;
+  done: number;
+  pending: number;
+  failed: number;
+  review: number;
+};
+
 export const api = {
+  playlistPreview: (url: string) =>
+    post<PlaylistPreview>("/api/playlist", { url, preview: true }),
+  playlistImport: (url: string, name?: string) =>
+    post<{
+      playlist_id: string;
+      name: string;
+      source: string;
+      queued: { id: string; line: string; matched: string; cover: string | null }[];
+      skipped: { line: string; why: string }[];
+    }>("/api/playlist", { url, name }),
+  playlists: () => get<{ playlists: PlaylistRow[] }>("/api/playlists"),
   me: () => get<Me>("/api/me"),
   health: () => get<Health>("/api/health"),
   search: (q: string) =>