Lyra/docs/superpowers/plans/2026-07-13-cleanup-batch.md

Cleanup Batch (Tier A + B) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Close the deferred backlog's fix-items — PATCH allowlist hardening + a11y/copy polish (Tier A), and the two design tradeoffs the user chose to fix properly: per-seed discovery scoring and a persisted scan worklist (Tier B).

Architecture: Tier A is small web edits to two Next.js route handlers and one client component. Tier B adds two additive tables via one Prisma migration: DiscoverySeedContribution (normalizes discovery scores so a suggestion's score = sum of every seed's latest contribution, independent of chunk/sweep boundaries) and ScanWorkItem (a frozen worklist so scan_chunk pops by indexed cursor instead of re-walking the tree each chunk). All worker DB access is raw psycopg SQL.

Tech Stack: Next.js (App Router, TypeScript, vitest, node test env), Prisma (migration source of truth), Python worker (psycopg, pytest against lyra_test).

Global Constraints

• Tests run against lyra_test only; a guard refuses non-*_test DBs. Never point at live lyra.
• Web vitest env is node, not jsdom — no DOM-render tests. UI-only changes (aria-label, copy) are verified by tsc + build + Playwright, not unit tests.
• Preserve existing aria-labels / button text — web route tests and Playwright key off them.
• Worker SQL uses quoted PascalCase table/column identifiers ("DiscoverySuggestion", "seedMbid").
• Discovery _upsert_artist/_upsert_album MUST keep their WHERE "DiscoverySuggestion".status = 'pending' guard (dismissed/wanted rows are never revived).
• Frequent commits: one per task minimum, following the TDD step order.
• Migration reaches live lyra via the web entrypoint's migrate deploy on rebuild; no down -v, ever.

---

File Structure

Tier A (web):

• web/src/app/api/monitor/config/route.ts — PATCH allowlist → hasOwnProperty.
• web/src/app/api/discover/config/route.ts — PATCH allowlist → hasOwnProperty.
• web/src/app/api/monitor/config/route.test.ts — prototype-chain rejection test.
• web/src/app/api/discover/config/route.test.ts — prototype-chain rejection test.
• web/src/app/api/config/route.test.ts — add constructor case.
• web/src/app/settings/settings-form.tsx — Monitor toggle aria-label + eyebrow copy.

Tier B (schema + worker):

• web/prisma/schema.prisma — add both models.
• web/prisma/migrations/<ts>_add_contributions_and_scan_worklist/migration.sql — CREATE TABLE ×2.
• worker/tests/conftest.py — clean the two new tables between tests.
• worker/lyra_worker/discovery.py — contributions write + recompute.
• worker/tests/test_discovery.py — new cross-chunk / partial-sweep / drop-off tests.
• worker/lyra_worker/scan.py — build_worklist / new scan_chunk / clear_worklist / reimplemented scan_library.
• worker/lyra_worker/main.py — maybe_run_scan drives the worklist by scan.id.
• worker/tests/test_scan_chunk.py — rewritten to the worklist API.

---

Task 1: Tier A1 + A3 — PATCH allowlist hardening + tests

Files:

• Modify: web/src/app/api/monitor/config/route.ts:26-28
• Modify: web/src/app/api/discover/config/route.ts:28-30
• Test: web/src/app/api/monitor/config/route.test.ts
• Test: web/src/app/api/discover/config/route.test.ts
• Test: web/src/app/api/config/route.test.ts (add one case)

Interfaces:

• Consumes: nothing new.
• Produces: nothing consumed by later tasks.

Background: Both PATCH routes filter with ([k]) => k in DEFAULTS, which walks the prototype chain, so a body key of __proto__ or constructor passes the filter and writes a junk Config row (harmless — no delete — but inconsistent with the DELETE handler at web/src/app/api/config/route.ts:51, which already uses Object.prototype.hasOwnProperty.call(...)).

• Step 1: Write the failing test — monitor PATCH rejects prototype-chain keys

Append to web/src/app/api/monitor/config/route.test.ts (mirror the file's existing PATCH test style — read the top of the file for its patchReq helper; if none exists, add one like below):

function patchReq(body: unknown) {
  return new Request("http://localhost/api/monitor/config", {
    method: "PATCH",
    headers: { "content-type": "application/json" },
    body: JSON.stringify(body),
  });
}

it("ignores prototype-chain keys (__proto__, constructor) and writes no Config row", async () => {
  const before = await prisma.config.count();
  const res = await PATCH(patchReq({ __proto__: "x", constructor: "y" }));
  expect((await res.json()).updated).toBe(0);
  expect(await prisma.config.count()).toBe(before);
});

Ensure PATCH and prisma are imported at the top of the test file (import { GET, PATCH } from "./route"; and import { prisma } from "@/lib/db";).

• Step 2: Run it to verify it fails

Run: cd web && npx vitest run src/app/api/monitor/config/route.test.ts Expected: FAIL — updated is 2 (or a Config row was written), because __proto__/constructor pass k in DEFAULTS.

• Step 3: Harden both PATCH routes

In web/src/app/api/monitor/config/route.ts change the filter:

  const entries = Object.entries((body ?? {}) as Record<string, unknown>).filter(
    ([k]) => Object.prototype.hasOwnProperty.call(DEFAULTS, k),
  );

Apply the identical change in web/src/app/api/discover/config/route.ts.

• Step 4: Add the mirror test for discover PATCH

Append the same test (with a patchReq pointing at /api/discover/config and importing this route's PATCH) to web/src/app/api/discover/config/route.test.ts.

• Step 5: Add the constructor case to the DELETE allowlist test

In web/src/app/api/config/route.test.ts, the test at line 85 ("rejects prototype-chain field names") only exercises __proto__. Add constructor alongside it:

    for (const bad of ["__proto__", "constructor"]) {
      const res = await DELETE(delReq(bad));
      expect(res.status).toBe(400);
    }
    // Must not have collapsed to deleteMany({ where: {} }) and wiped the table.
    expect(await prisma.config.count()).toBe(before);
    expect(await prisma.config.findUnique({ where: { key: "qobuz.password" } })).not.toBeNull();

(Replace the existing single __proto__ DELETE block; keep the surrounding before/PUT setup.)

• Step 6: Run all three affected test files to verify they pass

Run: cd web && npx vitest run src/app/api/monitor/config/route.test.ts src/app/api/discover/config/route.test.ts src/app/api/config/route.test.ts Expected: PASS.

• Step 7: Commit

git add web/src/app/api/monitor/config/route.ts web/src/app/api/discover/config/route.ts \
        web/src/app/api/monitor/config/route.test.ts web/src/app/api/discover/config/route.test.ts \
        web/src/app/api/config/route.test.ts
git commit -m "fix(web): harden config PATCH allowlists against prototype-chain keys"

---

Task 2: Tier A2 — Monitor toggle aria-label + Settings eyebrow copy

Files:

• Modify: web/src/app/settings/settings-form.tsx:159 (eyebrow) and :273-277 (toggle)

Interfaces:

• Consumes: nothing.
• Produces: nothing.

No unit test (node vitest env can't render this component; verified by build + Playwright later).

• Step 1: Add an aria-label to the Monitor enable checkbox

At web/src/app/settings/settings-form.tsx:273, add aria-label to the <input type="checkbox"> inside the Monitor <label className="toggle">:

              <input
                type="checkbox"
                aria-label="automatic monitoring"
                checked={monitorCfg["monitor.enabled"] === "true"}
                onChange={(e) => setMon("monitor.enabled", e.target.checked ? "true" : "false")}
              />

• Step 2: Fix the Settings eyebrow copy

At web/src/app/settings/settings-form.tsx:159, the eyebrow lists only three of the five tabs. Update it:

      <PageHead title="Settings" eyebrow="Credentials · library · monitor · discovery" />

• Step 3: Typecheck + build to confirm no breakage

Run: cd web && npx tsc --noEmit && npm run build Expected: clean (no type errors, build succeeds).

• Step 4: Commit

git add web/src/app/settings/settings-form.tsx
git commit -m "fix(web): label Monitor toggle for a11y + correct Settings eyebrow"

---

Task 3: Migration — add DiscoverySeedContribution + ScanWorkItem tables

Files:

• Modify: web/prisma/schema.prisma
• Create: web/prisma/migrations/<timestamp>_add_contributions_and_scan_worklist/migration.sql
• Modify: worker/tests/conftest.py (clean the new tables)

Interfaces:

• Produces (raw SQL tables consumed by Tasks 4 & 5):

  • DiscoverySeedContribution(id, "candidateMbid", "candidateName", "seedMbid", score float8, sources text[], "updatedAt"), UNIQUE("candidateMbid","seedMbid").
  • ScanWorkItem(id, "scanId", artist, album, path, done bool), UNIQUE("scanId", path).

• Step 1: Add both models to web/prisma/schema.prisma

Append after the DiscoverySuggestion model:

model DiscoverySeedContribution {
  id            String   @id @default(cuid())
  candidateMbid String
  candidateName String
  seedMbid      String
  score         Float
  sources       String[]
  updatedAt     DateTime @updatedAt

  @@unique([candidateMbid, seedMbid])
  @@index([candidateMbid])
  @@index([seedMbid])
}

model ScanWorkItem {
  id     String  @id @default(cuid())
  scanId String
  artist String
  album  String
  path   String
  done   Boolean @default(false)

  @@unique([scanId, path])
  @@index([scanId, done, artist, album])
}

• Step 2: Hand-author the migration SQL

Create web/prisma/migrations/20260713120000_add_contributions_and_scan_worklist/migration.sql (use a timestamp lexicographically after 20260711215631):

-- CreateTable
CREATE TABLE "DiscoverySeedContribution" (
    "id" TEXT NOT NULL,
    "candidateMbid" TEXT NOT NULL,
    "candidateName" TEXT NOT NULL,
    "seedMbid" TEXT NOT NULL,
    "score" DOUBLE PRECISION NOT NULL,
    "sources" TEXT[],
    "updatedAt" TIMESTAMP(3) NOT NULL,

    CONSTRAINT "DiscoverySeedContribution_pkey" PRIMARY KEY ("id")
);

-- CreateTable
CREATE TABLE "ScanWorkItem" (
    "id" TEXT NOT NULL,
    "scanId" TEXT NOT NULL,
    "artist" TEXT NOT NULL,
    "album" TEXT NOT NULL,
    "path" TEXT NOT NULL,
    "done" BOOLEAN NOT NULL DEFAULT false,

    CONSTRAINT "ScanWorkItem_pkey" PRIMARY KEY ("id")
);

-- CreateIndex
CREATE UNIQUE INDEX "DiscoverySeedContribution_candidateMbid_seedMbid_key" ON "DiscoverySeedContribution"("candidateMbid", "seedMbid");
CREATE INDEX "DiscoverySeedContribution_candidateMbid_idx" ON "DiscoverySeedContribution"("candidateMbid");
CREATE INDEX "DiscoverySeedContribution_seedMbid_idx" ON "DiscoverySeedContribution"("seedMbid");

-- CreateIndex
CREATE UNIQUE INDEX "ScanWorkItem_scanId_path_key" ON "ScanWorkItem"("scanId", "path");
CREATE INDEX "ScanWorkItem_scanId_done_artist_album_idx" ON "ScanWorkItem"("scanId", "done", "artist", "album");

• Step 3: Regenerate the Prisma client and apply to lyra_test

Run:

cd web && npx prisma generate
DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra_test npx prisma migrate deploy

Expected: prisma generate succeeds; migrate deploy reports the new migration applied (8 migrations). Confirm npx prisma migrate status (against lyra_test) shows "Database schema is up to date".

• Step 4: Clean the new tables in the worker test fixture

In worker/tests/conftest.py, the conn fixture DELETEs a set of tables both before (setup) and after (teardown). Add the two new tables to both DELETE blocks, before the Config delete in each:

        cur.execute('DELETE FROM "DiscoverySeedContribution"')
        cur.execute('DELETE FROM "ScanWorkItem"')

• Step 5: Verify the tables exist and the suite still boots

Run: cd worker && python -m pytest tests/test_discovery_schema.py -q Expected: PASS (existing schema tests unaffected; the new tables are reachable). If it errors on a missing table, re-check Step 3 applied to lyra_test.

• Step 6: Commit

git add web/prisma/schema.prisma web/prisma/migrations worker/tests/conftest.py
git commit -m "feat: add DiscoverySeedContribution + ScanWorkItem tables"

---

Task 4: Tier B1 — per-seed discovery contributions + recompute

Files:

• Modify: worker/lyra_worker/discovery.py (run_discovery and helpers)
• Test: worker/tests/test_discovery.py (add new tests)

Interfaces:

• Consumes: DiscoverySeedContribution table (Task 3); _upsert_artist, _derive_albums, _healthy_sources, _seed_artists, _followed_mbids (existing, unchanged signatures).
• Produces: run_discovery(conn, sources, browser, cfg) -> DiscoveryResult (same signature); score is now sum-of-all-seed-contributions.

Design: Replace the in-memory chunk-local aggregation with contribution rows. Per swept seed, replace its rows; then recompute every affected candidate's suggestion from the summed contributions.

• Step 1: Write the failing test — score sums across separate chunks

Add to worker/tests/test_discovery.py:

def _artist_score(conn, mbid):
    with conn.cursor() as cur:
        cur.execute('SELECT score, "seedCount" FROM "DiscoverySuggestion" WHERE "artistMbid" = %s', (mbid,))
        return cur.fetchone()

def test_score_sums_across_separate_chunks(conn):
    # Two seeds, both eligible, forced into separate one-seed chunks by chunk_size=1.
    insert_watched_artist(conn, mbid="s1", name="Seed One",
                          last_discovered_sql="now() - interval '200 hours'")
    insert_watched_artist(conn, mbid="s2", name="Seed Two",
                          last_discovered_sql="now() - interval '190 hours'")
    src = FakeSimilaritySource(similar={
        "s1": [SimilarArtist("c1", "Shared", 0.6)],
        "s2": [SimilarArtist("c1", "Shared", 0.5)],
    })
    cfg = DiscoveryConfig(chunk_size=1)
    run_discovery(conn, [src], FakeMbBrowser(), cfg)   # processes s1 (older)
    run_discovery(conn, [src], FakeMbBrowser(), cfg)   # processes s2
    # replace-not-accumulate would leave 0.5/seedCount1; contributions give the sum.
    assert _artist_score(conn, "c1") == (1.1, 2)

• Step 2: Run it to verify it fails

Run: cd worker && python -m pytest tests/test_discovery.py::test_score_sums_across_separate_chunks -q Expected: FAIL — score is 0.5, seedCount 1 (last chunk overwrote via score = EXCLUDED.score).

• Step 3: Rewrite the aggregation in discovery.py

Replace the body of run_discovery (lines ~146-194) and add two helpers. Keep _upsert_artist, _upsert_album, _derive_albums, _existing_rg_mbids, _healthy_sources, _seed_artists, _followed_mbids as-is.

def _record_contributions(conn: psycopg.Connection, seed_mbid: str, live, followed: set[str],
                          cfg: DiscoveryConfig) -> set[str]:
    """Replace seed_mbid's contribution rows with its current similar-artist scores.
    Returns the set of candidate mbids whose aggregate may have changed (old contributors
    of this seed ∪ newly inserted), so callers recompute drop-offs too."""
    with conn.cursor() as cur:
        cur.execute('SELECT "candidateMbid" FROM "DiscoverySeedContribution" WHERE "seedMbid" = %s',
                    (seed_mbid,))
        affected = {r[0] for r in cur.fetchall()}
        cur.execute('DELETE FROM "DiscoverySeedContribution" WHERE "seedMbid" = %s', (seed_mbid,))

    per_cand: dict[str, dict] = {}
    for src in live:
        try:
            similar = src.similar_artists(seed_mbid)
        except Exception as e:  # a bad source/seed must not abort the sweep
            print(f"worker: discovery source {src.name} failed for {seed_mbid}: {e}", flush=True)
            continue
        for sa in sorted(similar, key=lambda a: a.score, reverse=True)[: cfg.similar_per_seed]:
            if sa.mbid in followed:
                continue
            c = per_cand.setdefault(sa.mbid, {"name": sa.name, "score": 0.0, "sources": set()})
            c["score"] += sa.score
            c["sources"].add(src.name)
            if sa.name and not c["name"]:
                c["name"] = sa.name

    with conn.cursor() as cur:
        for mbid, c in per_cand.items():
            cur.execute(
                'INSERT INTO "DiscoverySeedContribution" (id, "candidateMbid", "candidateName", '
                '"seedMbid", score, sources, "updatedAt") '
                'VALUES (gen_random_uuid()::text, %s, %s, %s, %s, %s, now())',
                (mbid, c["name"], seed_mbid, c["score"], sorted(c["sources"])),
            )
            affected.add(mbid)
    return affected


def _recompute_suggestions(conn: psycopg.Connection, affected: set[str],
                           cfg: DiscoveryConfig) -> list[dict]:
    """Recompute each affected candidate's DiscoverySuggestion from the sum of its
    contributions. Returns the candidates that produced a live pending suggestion."""
    if not affected:
        return []
    with conn.cursor() as cur:
        cur.execute(
            'SELECT "candidateMbid", "candidateName", score, "seedMbid", sources '
            'FROM "DiscoverySeedContribution" WHERE "candidateMbid" = ANY(%s)',
            (list(affected),))
        rows = cur.fetchall()

    agg: dict[str, dict] = {}
    for mbid, name, score, seed, sources in rows:
        a = agg.setdefault(mbid, {"mbid": mbid, "name": name, "score": 0.0,
                                  "seeds": set(), "sources": set()})
        a["score"] += score
        a["seeds"].add(seed)
        a["sources"].update(sources or [])
        if name and not a["name"]:
            a["name"] = name

    surfaced: list[dict] = []
    for a in sorted(agg.values(), key=lambda a: a["score"], reverse=True):
        if a["score"] < cfg.min_score:
            continue
        cand = {"mbid": a["mbid"], "name": a["name"], "score": a["score"],
                "seed_count": len(a["seeds"]), "sources": a["sources"]}
        if _upsert_artist(conn, cand):  # rowcount 1 => a live pending suggestion
            surfaced.append(cand)
    return surfaced


def run_discovery(conn: psycopg.Connection, sources: list[SimilaritySource],
                  browser: MbBrowser, cfg: DiscoveryConfig) -> DiscoveryResult:
    """Record each seed's similar-artist contributions, then recompute affected
    suggestions from the summed contributions — so a candidate's score is the sum of
    every seed's latest contribution, independent of chunk and sweep boundaries."""
    seeds = _seed_artists(conn, cfg)
    followed = _followed_mbids(conn)
    live = _healthy_sources(sources)

    affected: set[str] = set()
    for seed_mbid, _seed_name in seeds:
        affected |= _record_contributions(conn, seed_mbid, live, followed, cfg)

    surfaced = _recompute_suggestions(conn, affected, cfg)
    albums = _derive_albums(conn, browser, surfaced, cfg)

    with conn.cursor() as cur:
        for seed_mbid, _seed_name in seeds:
            cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() WHERE mbid = %s',
                        (seed_mbid,))
    conn.commit()
    return DiscoveryResult(artists=len(surfaced), albums=albums, seeds=len(seeds))

• Step 4: Run the new test + the full discovery suite to verify green

Run: cd worker && python -m pytest tests/test_discovery.py tests/test_discovery_albums.py tests/test_discovery_trigger.py -q Expected: PASS — including the pre-existing test_aggregates_scores_across_seeds (single call, 2 seeds → c1 1.1/seedCount 2), which the contributions model reproduces.

• Step 5: Write the partial-sweep + drop-off tests

Add to worker/tests/test_discovery.py:

def test_partial_sweep_preserves_other_seeds_contribution(conn):
    # Full sweep: c1 gets s1(0.6)+s2(0.5)=1.1. Then only s1 is due again with a new
    # score; s2's contribution must survive untouched.
    insert_watched_artist(conn, mbid="s1", name="Seed One",
                          last_discovered_sql="now() - interval '200 hours'")
    insert_watched_artist(conn, mbid="s2", name="Seed Two",
                          last_discovered_sql="now() - interval '200 hours'")
    src = FakeSimilaritySource(similar={
        "s1": [SimilarArtist("c1", "Shared", 0.6)],
        "s2": [SimilarArtist("c1", "Shared", 0.5)],
    })
    run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())  # both -> 1.1/2
    assert _artist_score(conn, "c1") == (1.1, 2)

    # s1 due again (older than interval), s2 fresh; s1 now scores c1 at 0.7.
    with conn.cursor() as cur:
        cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() - interval \'200 hours\' WHERE mbid = %s', ("s1",))
        cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() WHERE mbid = %s', ("s2",))
    conn.commit()
    src2 = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Shared", 0.7)]})
    run_discovery(conn, [src2], FakeMbBrowser(), DiscoveryConfig())
    assert _artist_score(conn, "c1") == (1.2, 2)  # 0.7 (new s1) + 0.5 (kept s2)


def test_seed_dropoff_lowers_seedcount(conn):
    insert_watched_artist(conn, mbid="s1", name="Seed One",
                          last_discovered_sql="now() - interval '200 hours'")
    insert_watched_artist(conn, mbid="s2", name="Seed Two",
                          last_discovered_sql="now() - interval '200 hours'")
    src = FakeSimilaritySource(similar={
        "s1": [SimilarArtist("c1", "Shared", 0.6)],
        "s2": [SimilarArtist("c1", "Shared", 0.5)],
    })
    run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())
    assert _artist_score(conn, "c1") == (1.1, 2)

    # s1 re-swept but no longer finds c1 -> its contribution is removed, c1 recomputed.
    with conn.cursor() as cur:
        cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() - interval \'200 hours\' WHERE mbid = %s', ("s1",))
        cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() WHERE mbid = %s', ("s2",))
    conn.commit()
    src2 = FakeSimilaritySource(similar={"s1": [SimilarArtist("c9", "Other", 0.9)]})
    run_discovery(conn, [src2], FakeMbBrowser(), DiscoveryConfig())
    assert _artist_score(conn, "c1") == (0.5, 1)  # only s2 remains

• Step 6: Run to verify the new tests pass

Run: cd worker && python -m pytest tests/test_discovery.py -q Expected: PASS.

• Step 7: Commit

git add worker/lyra_worker/discovery.py worker/tests/test_discovery.py
git commit -m "feat(worker): per-seed discovery contributions so score sums across chunks/sweeps"

---

Task 5: Tier B2 — persisted scan worklist

Files:

• Modify: worker/lyra_worker/scan.py
• Modify: worker/lyra_worker/main.py (maybe_run_scan, imports)
• Test: worker/tests/test_scan_chunk.py (rewrite to the worklist API)

Interfaces:

• Consumes: ScanWorkItem table (Task 3); existing _iter_album_entries, _process_album, ScanResult.
• Produces:
  • build_worklist(conn, scan_id: str, dest_root: str = "/music") -> int
  • scan_chunk(conn, resolver, probe, browser, scan_id: str, limit: int) -> tuple[int, int, bool] — returns (imported, skipped, done).
  • clear_worklist(conn, scan_id: str) -> None
  • scan_library(conn, resolver, probe, browser, dest_root="/music") -> ScanResult — unchanged signature, reimplemented on the worklist.

Note: scan_chunk's signature CHANGES (drops dest_root/cursor, adds scan_id, drops the returned cursor). test_scan.py uses only scan_library and is unaffected; test_scan_trigger.py checks scan.progress/scan.result/scan.inProgress (not scan.cursor) and is unaffected; only test_scan_chunk.py must be rewritten.

• Step 1: Write the failing tests — worklist build/pop/resume

Replace the entire contents of worker/tests/test_scan_chunk.py with:

from lyra_worker.adapters.fakes import FakeMbBrowser
from lyra_worker.browser import ReleaseGroupInfo
from lyra_worker.scan import build_worklist, scan_chunk, clear_worklist
from lyra_worker.types import MBTarget
from tests.test_scan import FakeProbe, FakeResolver, _album, _counts


def _tree(tmp_path):
    """Three albums across two artists: Artist A/Alpha, Artist A/Beta, Artist B/Gamma."""
    _album(tmp_path, "Artist A", "Alpha (2001)")
    _album(tmp_path, "Artist A", "Beta (2002)")
    _album(tmp_path, "Artist B", "Gamma (2003)")
    resolver = FakeResolver({
        ("Artist A", "Alpha"): MBTarget(artist="Artist A", album="Alpha", year=2001, rg_mbid="rg-a1", artist_mbid="ma"),
        ("Artist A", "Beta"): MBTarget(artist="Artist A", album="Beta", year=2002, rg_mbid="rg-a2", artist_mbid="ma"),
        ("Artist B", "Gamma"): MBTarget(artist="Artist B", album="Gamma", year=2003, rg_mbid="rg-b1", artist_mbid="mb"),
    })
    browser = FakeMbBrowser(releases={
        "ma": [ReleaseGroupInfo("rg-a1", "Alpha", "Album", (), "2001"),
               ReleaseGroupInfo("rg-a2", "Beta", "Album", (), "2002")],
        "mb": [ReleaseGroupInfo("rg-b1", "Gamma", "Album", (), "2003")],
    })
    return resolver, browser


def test_build_worklist_inserts_one_row_per_album(conn, tmp_path):
    resolver, _ = _tree(tmp_path)
    n = build_worklist(conn, "scan1", str(tmp_path))
    assert n == 3
    with conn.cursor() as cur:
        cur.execute('SELECT count(*) FROM "ScanWorkItem" WHERE "scanId" = %s AND NOT done', ("scan1",))
        assert cur.fetchone()[0] == 3


def test_build_worklist_is_idempotent(conn, tmp_path):
    resolver, _ = _tree(tmp_path)
    assert build_worklist(conn, "scan1", str(tmp_path)) == 3
    assert build_worklist(conn, "scan1", str(tmp_path)) == 0  # ON CONFLICT DO NOTHING
    with conn.cursor() as cur:
        cur.execute('SELECT count(*) FROM "ScanWorkItem" WHERE "scanId" = %s', ("scan1",))
        assert cur.fetchone()[0] == 3


def test_scan_chunk_respects_limit_and_marks_done(conn, tmp_path):
    resolver, browser = _tree(tmp_path)
    build_worklist(conn, "scan1", str(tmp_path))
    imported, skipped, done = scan_chunk(conn, resolver, FakeProbe(quality_class=2), browser, "scan1", limit=2)
    assert (imported, skipped, done) == (2, 0, False)  # 2 of 3 processed, more remain
    with conn.cursor() as cur:
        cur.execute('SELECT count(*) FROM "ScanWorkItem" WHERE "scanId" = %s AND done', ("scan1",))
        assert cur.fetchone()[0] == 2
        cur.execute('SELECT count(*) FROM "LibraryItem"')
        assert cur.fetchone()[0] == 2


def test_scan_chunk_resumes_and_finishes(conn, tmp_path):
    resolver, browser = _tree(tmp_path)
    build_worklist(conn, "scan1", str(tmp_path))
    scan_chunk(conn, resolver, FakeProbe(quality_class=2), browser, "scan1", limit=2)
    imported, skipped, done = scan_chunk(conn, resolver, FakeProbe(quality_class=2), browser, "scan1", limit=25)
    assert (imported, skipped, done) == (1, 0, True)  # last album, then drained
    with conn.cursor() as cur:
        cur.execute('SELECT album FROM "LibraryItem" ORDER BY album')
        assert cur.fetchall() == [("Alpha",), ("Beta",), ("Gamma",)]


def test_scan_chunk_one_per_chunk_totals_correctly(conn, tmp_path):
    resolver, browser = _tree(tmp_path)
    build_worklist(conn, "scan1", str(tmp_path))
    total_i = total_s = guard = 0
    done = False
    while not done and guard < 10:
        i, s, done = scan_chunk(conn, resolver, FakeProbe(quality_class=2), browser, "scan1", limit=1)
        total_i += i
        total_s += s
        guard += 1
    assert (total_i, total_s) == (3, 0)
    assert guard == 3
    assert _counts(conn) == {"LibraryItem": 3, "MonitoredRelease": 3, "WatchedArtist": 2}


def test_clear_worklist_removes_this_scans_rows(conn, tmp_path):
    resolver, _ = _tree(tmp_path)
    build_worklist(conn, "scan1", str(tmp_path))
    clear_worklist(conn, "scan1")
    with conn.cursor() as cur:
        cur.execute('SELECT count(*) FROM "ScanWorkItem" WHERE "scanId" = %s', ("scan1",))
        assert cur.fetchone()[0] == 0

• Step 2: Run to verify failure

Run: cd worker && python -m pytest tests/test_scan_chunk.py -q Expected: FAIL — ImportError (build_worklist/clear_worklist don't exist; scan_chunk has the old signature).

• Step 3: Rewrite scan.py

Add import uuid at the top. Replace scan_chunk and scan_library (lines ~142-176), keeping everything above (_iter_album_entries, _process_album, _cursor_key/_cursor_tuple may be removed since nothing uses them now — remove _cursor_key and _cursor_tuple):

def build_worklist(conn: psycopg.Connection, scan_id: str, dest_root: str = "/music") -> int:
    """Walk the tree once and persist this scan's album worklist. Idempotent
    (ON CONFLICT DO NOTHING), so a re-run after a crash adds only missing rows.
    Returns the number of rows inserted."""
    inserted = 0
    with conn.cursor() as cur:
        for artist_name, album_folder, album_path in _iter_album_entries(dest_root):
            cur.execute(
                'INSERT INTO "ScanWorkItem" (id, "scanId", artist, album, path, done) '
                "VALUES (gen_random_uuid()::text, %s, %s, %s, %s, false) "
                'ON CONFLICT ("scanId", path) DO NOTHING',
                (scan_id, artist_name, album_folder, album_path),
            )
            inserted += cur.rowcount
    conn.commit()
    return inserted


def scan_chunk(conn: psycopg.Connection, resolver, probe: AudioProbe, browser,
               scan_id: str, limit: int) -> tuple[int, int, bool]:
    """Process up to `limit` not-yet-done worklist items for `scan_id`, in stable
    (artist, album) order, marking each done. Returns (imported, skipped, done) where
    `done` is True when no undone rows remain. Idempotent: all record writes are
    ON CONFLICT, and an item is marked done only after it is processed."""
    with conn.cursor() as cur:
        cur.execute(
            'SELECT id, artist, album, path FROM "ScanWorkItem" '
            'WHERE "scanId" = %s AND NOT done ORDER BY artist, album LIMIT %s',
            (scan_id, limit),
        )
        items = cur.fetchall()

    imported = skipped = 0
    browsed: set[str] = set()  # artists whose discography we've populated this chunk
    for item_id, artist_name, album_folder, album_path in items:
        outcome = _process_album(conn, resolver, probe, browser, artist_name, album_folder, album_path, browsed)
        if outcome == "imported":
            imported += 1
        elif outcome == "skipped":
            skipped += 1
        with conn.cursor() as cur:
            cur.execute('UPDATE "ScanWorkItem" SET done = true WHERE id = %s', (item_id,))
        conn.commit()

    with conn.cursor() as cur:
        cur.execute('SELECT count(*) FROM "ScanWorkItem" WHERE "scanId" = %s AND NOT done', (scan_id,))
        remaining = cur.fetchone()[0]
    return imported, skipped, remaining == 0


def clear_worklist(conn: psycopg.Connection, scan_id: str) -> None:
    with conn.cursor() as cur:
        cur.execute('DELETE FROM "ScanWorkItem" WHERE "scanId" = %s', (scan_id,))
    conn.commit()


def scan_library(conn: psycopg.Connection, resolver, probe: AudioProbe, browser,
                 dest_root: str = "/music") -> ScanResult:
    """Walk `{dest_root}/{Artist}/{Album}/` in one unbounded pass (build the worklist,
    drain it, clean up), recording matched albums as have + monitored + followed plus
    each followed artist's full discography. Test/one-shot convenience; the worker loop
    uses the chunked build_worklist + scan_chunk so a large library doesn't block
    job-claim."""
    scan_id = "oneshot-" + uuid.uuid4().hex
    build_worklist(conn, scan_id, dest_root)
    imported = skipped = 0
    done = False
    while not done:
        imp, skp, done = scan_chunk(conn, resolver, probe, browser, scan_id, limit=1000)
        imported += imp
        skipped += skp
    clear_worklist(conn, scan_id)
    return ScanResult(imported, skipped)

• Step 4: Run scan.py's direct tests to verify green

Run: cd worker && python -m pytest tests/test_scan_chunk.py tests/test_scan.py -q Expected: PASS (test_scan.py uses scan_library, unchanged behavior; test_scan_chunk.py exercises the new API).

• Step 5: Rewire maybe_run_scan in main.py

Update the import at worker/lyra_worker/main.py:17:

from lyra_worker.scan import build_worklist, clear_worklist, scan_chunk

Add import uuid near the top imports. Replace maybe_run_scan (lines ~58-96) with:

def maybe_run_scan(conn, resolver, probe, browser, config, dest_root: str = DEST_ROOT) -> None:
    """Advance a chunked library scan by at most one chunk per call, so the worker loop
    keeps claiming jobs and ticking the monitor between chunks. `scan.requested` starts a
    fresh scan (walk once -> ScanWorkItem worklist); `scan.inProgress`/`scan.id`/`scan.progress`
    persist state until the worklist drains, at which point `scan.result` is written."""
    requested = str(config.get("scan.requested", "")).strip().lower() in _TRUE
    in_progress = str(config.get("scan.inProgress", "")).strip().lower() in _TRUE
    if not requested and not in_progress:
        return
    if requested and not in_progress:  # start a fresh scan
        scan_id = uuid.uuid4().hex
        _set_config(conn, "scan.inProgress", "true")
        _set_config(conn, "scan.requested", "false")
        _set_config(conn, "scan.id", scan_id)
        _set_config(conn, "scan.progress", "0/0")
        try:
            build_worklist(conn, scan_id, dest_root)
        except Exception as e:  # a scan error must never kill the worker
            print(f"worker: library scan worklist build failed: {e}", flush=True)
            conn.rollback()
            _set_config(conn, "scan.inProgress", "false")
            _set_config(conn, "scan.requested", "false")
            return
        imported_so_far, skipped_so_far = 0, 0
    else:  # resume the in-progress scan
        scan_id = config.get("scan.id", "") or ""
        imported_so_far, skipped_so_far = _parse_progress(config.get("scan.progress", ""))
        if not scan_id:  # corrupt/absent state -> abandon; a new request restarts it
            _set_config(conn, "scan.inProgress", "false")
            return
    try:
        imp, skp, done = scan_chunk(conn, resolver, probe, browser, scan_id, _scan_chunk_size(config))
    except Exception as e:  # a scan error must never kill the worker
        print(f"worker: library scan chunk failed: {e}", flush=True)
        conn.rollback()
        _set_config(conn, "scan.inProgress", "false")
        _set_config(conn, "scan.requested", "false")
        return
    imported_total = imported_so_far + imp
    skipped_total = skipped_so_far + skp
    if done:
        _set_config(conn, "scan.result", f"imported {imported_total}, skipped {skipped_total}")
        _set_config(conn, "scan.progress", "")
        _set_config(conn, "scan.inProgress", "false")
        clear_worklist(conn, scan_id)
        _set_config(conn, "scan.id", "")
        print(f"worker: library scan done — {imported_total} imported, {skipped_total} skipped", flush=True)
    else:
        _set_config(conn, "scan.progress", f"{imported_total}/{skipped_total}")

• Step 6: Run the scan trigger tests + the full worker suite

Run: cd worker && python -m pytest tests/test_scan_trigger.py -q && python -m pytest -q Expected: PASS. test_scan_trigger.py is unchanged (it asserts on scan.progress/scan.result/scan.inProgress, all preserved). Full suite green (previous baseline 173 passed / 7 skipped, now higher with the added tests).

• Step 7: Commit

git add worker/lyra_worker/scan.py worker/lyra_worker/main.py worker/tests/test_scan_chunk.py
git commit -m "feat(worker): persisted scan worklist (O(N) scan, robust to mid-scan changes)"

---

Final verification (before merge)

• Worker suite: cd worker && python -m pytest -q → all green.
• Web suite: cd web && npx vitest run → all green.
• Typecheck + build: cd web && npx tsc --noEmit && npm run build → clean.
• Migration status (lyra_test): cd web && DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra_test npx prisma migrate status → up to date, 8 migrations.
• Opus whole-branch review (via subagent-driven-development's final review), then ff-merge cleanup/tier-a-b → main, push, docker compose up -d --build web worker, verify live (web 200, worker healthy, migrate deploy reports the new migration applied).

---

Self-review notes

• Spec coverage: Tier A1 (Task 1), A2 (Task 2), A3 (Task 1 Step 5); Tier B1 (Tasks 3+4); Tier B2 (Tasks 3+5); migration (Task 3); testing + delivery (Final verification). All spec sections mapped.
• Type consistency: scan_chunk new signature (conn, resolver, probe, browser, scan_id, limit) -> (imported, skipped, done) used identically in scan.py, main.py, and test_scan_chunk.py. build_worklist/clear_worklist signatures match across producer and callers. Discovery run_discovery signature unchanged; helpers _record_contributions/_recompute_suggestions are internal.
• Backward-compat: existing test_aggregates_scores_across_seeds, test_scan.py, test_scan_trigger.py pass unchanged by design (verified against their current assertions).