From 915bd96cd36a9a899351c31791ed6ffb72c1719d Mon Sep 17 00:00:00 2001 From: Jonathan Date: Sat, 11 Jul 2026 12:01:57 +0200 Subject: [PATCH] docs: add monitoring-core implementation plan (slice 2, plan 1) Co-Authored-By: Claude Opus 4.8 (1M context) --- .../plans/2026-07-11-lyra-monitoring-core.md | 1270 +++++++++++++++++ 1 file changed, 1270 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-11-lyra-monitoring-core.md diff --git a/docs/superpowers/plans/2026-07-11-lyra-monitoring-core.md b/docs/superpowers/plans/2026-07-11-lyra-monitoring-core.md new file mode 100644 index 0000000..8fb1e35 --- /dev/null +++ b/docs/superpowers/plans/2026-07-11-lyra-monitoring-core.md @@ -0,0 +1,1270 @@ +# Lyra Monitoring Core (Worker + Data Model) 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:** Give the worker a background monitor that discovers each watched artist's releases from MusicBrainz, enqueues wanted/upgrade releases into the existing slice 1 pipeline, and feeds job outcomes back into monitoring state — plus the data model the whole of slice 2 builds on. + +**Architecture:** Two new Prisma tables (`WatchedArtist`, `MonitoredRelease`) and a nullable `Request.monitoredReleaseId` link. A new `worker/lyra_worker/monitor.py` runs two phases (discovery + retry/enqueue) plus a `reconcile()` that reads a finished job's outcome; both are driven off timestamps in the tables and composed by `sweep()`, called on a ~60s tick inside the existing `run_forever()` loop. An `MbBrowser` seam (real via `musicbrainzngs`, faked in tests) supplies artist discography. The pipeline's import stage is upgraded to replace a lower-quality library item. + +**Tech Stack:** Python 3.12 worker (psycopg raw SQL, matching the existing codebase); Prisma/Postgres for the schema; `musicbrainzngs` (already a worker dep) for the real browser. Tests: pytest against a seeded Postgres with a `FakeMbBrowser`; one opt-in live test gated by `LYRA_LIVE_TESTS`. + +## Global Constraints + +- Python 3.12 in the worker; psycopg raw SQL against Prisma-created tables (quoted PascalCase table names, camelCase quoted columns, `gen_random_uuid()::text` ids), matching `claim.py`/`pipeline.py`. +- **No new worker dependency.** `musicbrainzngs` is already in `worker/requirements.txt` (used by the resolver). Import it lazily inside the real browser's methods so importing/constructing the module stays offline. +- Time math uses SQL `now()` + `make_interval(...)` (not injected clocks), consistent with `claim.py`. Tests control timing by seeding timestamps with `now() - interval '...'`. +- **Backward compatibility is mandatory:** with `monitor.enabled` unset/false the worker behaves exactly as slice 1 — no sweep, no reconcile side effects. `run_pipeline` is NOT modified. Every existing worker test keeps passing unchanged. +- Enum and array literals follow the existing pattern: plain string literals for enum columns (e.g. `'wanted'`), Python lists for `text[]` columns (psycopg adapts them). +- `qualityClass` values (slice 1 `quality.py`): 3 = hi-res lossless, 2 = CD lossless, 1 = lossy. Default cutoff is 2. +- Every task ends with a commit. +- Run all worker commands from `/home/jonathan/Projects/lyra/worker` with `export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra` and the venv python `worker/.venv/bin/python`. + +## Shared interfaces (defined by tasks below) + +```python +# worker/lyra_worker/browser.py (Task 2) +@dataclass(frozen=True) +class ArtistHit: # mbid, name, disambiguation + mbid: str; name: str; disambiguation: str = "" + +@dataclass(frozen=True) +class ReleaseGroupInfo: # one MB release-group + rg_mbid: str; title: str; primary_type: str = "" + secondary_types: tuple[str, ...] = (); first_release_date: str = "" + +class MbBrowser(Protocol): + def search_artist(self, name: str) -> list[ArtistHit]: ... + def browse_release_groups(self, artist_mbid: str) -> list[ReleaseGroupInfo]: ... + +# worker/lyra_worker/monitor.py (Tasks 3-6) +@dataclass(frozen=True) +class MonitorConfig: # parsed from the Config dict + enabled: bool = False; poll_interval_hours: int = 24 + retry_interval_hours: int = 6; quality_cutoff: int = 2; upgrade_window_days: int = 14 + @classmethod + def from_config(cls, config: dict) -> "MonitorConfig": ... + +def discover(conn, browser: MbBrowser, cfg: MonitorConfig) -> int: ... # Task 3, returns #rows inserted +def enqueue_due(conn, cfg: MonitorConfig) -> int: ... # Task 4, returns #jobs enqueued +def reconcile(conn, job_id: str, cfg: MonitorConfig) -> None: ... # Task 5 +def sweep(conn, browser: MbBrowser, cfg: MonitorConfig) -> None: ... # Task 6 + +# worker/lyra_worker/registry.py (Task 6): build_browser() -> MbBrowser +``` + +--- + +### Task 1: Data model — WatchedArtist, MonitoredRelease, Request link + +**Files:** +- Modify: `web/prisma/schema.prisma` +- Create: (generated) `web/prisma/migrations/_add_monitoring/migration.sql` +- Modify: `worker/tests/conftest.py` (clean + helpers for the new tables) +- Test: `worker/tests/test_monitoring_schema.py` + +**Interfaces:** +- Consumes: existing `Request` model. +- Produces: `WatchedArtist`, `MonitoredRelease` tables + `MonitoredReleaseState` enum; `Request.monitoredReleaseId`; conftest helpers `insert_watched_artist(...)` and `insert_monitored_release(...)`. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_monitoring_schema.py`: +```python +from tests.conftest import insert_monitored_release, insert_watched_artist + + +def test_watched_artist_and_release_round_trip(conn): + artist_id = insert_watched_artist(conn, mbid="mbid-1", name="John Mayer", auto_monitor_future=True) + rel_id = insert_monitored_release( + conn, watched_artist_id=artist_id, artist_mbid="mbid-1", artist_name="John Mayer", + rg_mbid="rg-1", album="Continuum", primary_type="Album", + secondary_types=["Live"], first_release_date="2006-09-12", monitored=True, + ) + with conn.cursor() as cur: + cur.execute( + 'SELECT "watchedArtistId", album, "primaryType", "secondaryTypes", ' + 'monitored, state FROM "MonitoredRelease" WHERE id = %s', + (rel_id,), + ) + row = cur.fetchone() + assert row == (artist_id, "Continuum", "Album", ["Live"], True, "wanted") + + +def test_request_has_nullable_monitored_release_link(conn): + rel_id = insert_monitored_release( + conn, watched_artist_id=None, artist_mbid="mbid-2", artist_name="X", + rg_mbid="rg-2", album="Y", monitored=True, + ) + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "monitoredReleaseId", "createdAt") ' + "VALUES (gen_random_uuid()::text, 'X', 'Y', 'pending', %s, now()) RETURNING id", + (rel_id,), + ) + req_id = cur.fetchone()[0] + conn.commit() + with conn.cursor() as cur: + cur.execute('SELECT "monitoredReleaseId" FROM "Request" WHERE id = %s', (req_id,)) + assert cur.fetchone()[0] == rel_id +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitoring_schema.py -v +``` +Expected: FAIL — `ImportError` (helpers don't exist) / `UndefinedTable`. + +- [ ] **Step 3: Add the Prisma models** + +In `web/prisma/schema.prisma`, add the enum and models, and the `Request` field. + +Add after the existing `JobState` enum: +```prisma +enum MonitoredReleaseState { + wanted + grabbed + fulfilled +} +``` +Add to `model Request` (a new field + relation, alongside `job` / `libraryItem`): +```prisma + monitoredReleaseId String? + monitoredRelease MonitoredRelease? @relation(fields: [monitoredReleaseId], references: [id], onDelete: SetNull) +``` +Add two new models: +```prisma +model WatchedArtist { + id String @id @default(cuid()) + mbid String @unique + name String + autoMonitorFuture Boolean @default(false) + monitorFrom DateTime @default(now()) + lastPolledAt DateTime? + createdAt DateTime @default(now()) + releases MonitoredRelease[] +} + +model MonitoredRelease { + id String @id @default(cuid()) + watchedArtist WatchedArtist? @relation(fields: [watchedArtistId], references: [id], onDelete: Cascade) + watchedArtistId String? + artistMbid String + artistName String + rgMbid String @unique + album String + primaryType String? + secondaryTypes String[] + firstReleaseDate String? + monitored Boolean @default(false) + state MonitoredReleaseState @default(wanted) + currentQualityClass Int? + firstGrabbedAt DateTime? + lastSearchedAt DateTime? + createdAt DateTime @default(now()) + requests Request[] +} +``` + +- [ ] **Step 4: Generate and apply the migration** + +```bash +cd /home/jonathan/Projects/lyra/web +DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra npx prisma migrate dev --name add_monitoring +``` +Expected: a new migration folder under `web/prisma/migrations/` is created and applied; `prisma generate` runs. Verify: `psql` not needed — the worker test in Step 6 proves the tables exist. + +- [ ] **Step 5: Add conftest cleanup + helpers** + +In `worker/tests/conftest.py`, update the `conn` fixture cleanup (both the pre-test and post-test blocks) to also clear the new tables, in FK-safe order. Replace each of the two cleanup blocks' bodies: +```python + with connection.cursor() as cur: + cur.execute('DELETE FROM "Job"') + cur.execute('DELETE FROM "Request"') + cur.execute('DELETE FROM "MonitoredRelease"') + cur.execute('DELETE FROM "WatchedArtist"') + cur.execute('DELETE FROM "Config"') + connection.commit() +``` +Add these helpers at the end of the file: +```python +def insert_watched_artist(conn, mbid="mbid", name="Artist", auto_monitor_future=False, + monitor_from_sql="now()", last_polled_sql="NULL"): + """Insert a WatchedArtist; monitor_from_sql / last_polled_sql are raw SQL time expressions.""" + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "WatchedArtist" (id, mbid, name, "autoMonitorFuture", ' + f'"monitorFrom", "lastPolledAt", "createdAt") ' + f"VALUES (gen_random_uuid()::text, %s, %s, %s, {monitor_from_sql}, {last_polled_sql}, now()) " + "RETURNING id", + (mbid, name, auto_monitor_future), + ) + artist_id = cur.fetchone()[0] + conn.commit() + return artist_id + + +def insert_monitored_release(conn, watched_artist_id=None, artist_mbid="mbid", artist_name="Artist", + rg_mbid="rg", album="Album", primary_type=None, secondary_types=None, + first_release_date=None, monitored=False, state="wanted", + current_quality_class=None, first_grabbed_sql="NULL", + last_searched_sql="NULL"): + """Insert a MonitoredRelease; *_sql args are raw SQL time expressions.""" + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "MonitoredRelease" (id, "watchedArtistId", "artistMbid", "artistName", ' + '"rgMbid", album, "primaryType", "secondaryTypes", "firstReleaseDate", monitored, ' + f'state, "currentQualityClass", "firstGrabbedAt", "lastSearchedAt", "createdAt") ' + f"VALUES (gen_random_uuid()::text, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, " + f"{first_grabbed_sql}, {last_searched_sql}, now()) RETURNING id", + (watched_artist_id, artist_mbid, artist_name, rg_mbid, album, primary_type, + secondary_types if secondary_types is not None else [], first_release_date, + monitored, state, current_quality_class), + ) + rel_id = cur.fetchone()[0] + conn.commit() + return rel_id +``` + +- [ ] **Step 6: Run the test to verify it passes** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitoring_schema.py -v +``` +Expected: PASS — both tests green. + +- [ ] **Step 7: Run the full worker suite (no regressions)** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest -q +``` +Expected: PASS — all prior tests still green (the conftest cleanup additions are harmless), output pristine. + +- [ ] **Step 8: Commit** + +```bash +cd /home/jonathan/Projects/lyra +git add web/prisma/schema.prisma web/prisma/migrations worker/tests/conftest.py worker/tests/test_monitoring_schema.py +git commit -m "feat: monitoring data model (WatchedArtist, MonitoredRelease, Request link)" +``` + +--- + +### Task 2: MbBrowser seam + FakeMbBrowser + +**Files:** +- Create: `worker/lyra_worker/browser.py` +- Modify: `worker/lyra_worker/adapters/fakes.py` (add `FakeMbBrowser`) +- Test: `worker/tests/test_browser.py` + +**Interfaces:** +- Consumes: nothing. +- Produces: `ArtistHit`, `ReleaseGroupInfo`, `MbBrowser` (Protocol); `FakeMbBrowser(artists=..., releases=...)` test double whose `browse_release_groups` returns canned lists keyed by artist mbid. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_browser.py`: +```python +from lyra_worker.adapters.fakes import FakeMbBrowser +from lyra_worker.browser import ArtistHit, ReleaseGroupInfo + + +def test_fake_browser_returns_canned_artists_and_releases(): + rg = ReleaseGroupInfo(rg_mbid="rg-1", title="Continuum", primary_type="Album", + secondary_types=(), first_release_date="2006-09-12") + browser = FakeMbBrowser( + artists=[ArtistHit(mbid="a1", name="John Mayer", disambiguation="")], + releases={"a1": [rg]}, + ) + assert browser.search_artist("john")[0].name == "John Mayer" + assert browser.browse_release_groups("a1") == [rg] + assert browser.browse_release_groups("unknown") == [] +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_browser.py -v +``` +Expected: FAIL — `ModuleNotFoundError: No module named 'lyra_worker.browser'`. + +- [ ] **Step 3: Implement the seam** + +`worker/lyra_worker/browser.py`: +```python +from dataclasses import dataclass, field +from typing import Protocol + + +@dataclass(frozen=True) +class ArtistHit: + mbid: str + name: str + disambiguation: str = "" + + +@dataclass(frozen=True) +class ReleaseGroupInfo: + rg_mbid: str + title: str + primary_type: str = "" + secondary_types: tuple[str, ...] = () + first_release_date: str = "" + + +class MbBrowser(Protocol): + def search_artist(self, name: str) -> list[ArtistHit]: + """Artist name search → ranked hits (best first).""" + ... + + def browse_release_groups(self, artist_mbid: str) -> list[ReleaseGroupInfo]: + """All official release-groups for an artist MBID.""" + ... +``` + +- [ ] **Step 4: Add the fake** + +Append to `worker/lyra_worker/adapters/fakes.py`: +```python +from lyra_worker.browser import ArtistHit, ReleaseGroupInfo + + +class FakeMbBrowser: + """In-memory MbBrowser for tests. `releases` maps artist mbid -> [ReleaseGroupInfo].""" + + def __init__(self, artists=None, releases=None): + self._artists = list(artists or []) + self._releases = dict(releases or {}) + + def search_artist(self, name: str) -> list[ArtistHit]: + return list(self._artists) + + def browse_release_groups(self, artist_mbid: str) -> list[ReleaseGroupInfo]: + return list(self._releases.get(artist_mbid, [])) +``` + +- [ ] **Step 5: Run the test to verify it passes** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_browser.py -v +``` +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +cd /home/jonathan/Projects/lyra +git add worker/lyra_worker/browser.py worker/lyra_worker/adapters/fakes.py worker/tests/test_browser.py +git commit -m "feat: MbBrowser seam and FakeMbBrowser test double" +``` + +--- + +### Task 3: Discovery phase (`monitor.discover`) + `MonitorConfig` + +**Files:** +- Create: `worker/lyra_worker/monitor.py` +- Test: `worker/tests/test_monitor_discover.py` + +**Interfaces:** +- Consumes: `MbBrowser`, `FakeMbBrowser`, `insert_watched_artist`, the `MonitoredRelease` table. +- Produces: `MonitorConfig` (+ `from_config`), `discover(conn, browser, cfg) -> int`. Discovery inserts each not-yet-seen release-group (`ON CONFLICT (rgMbid) DO NOTHING`) with `monitored = autoMonitorFuture AND first_release_date > monitorFrom`, then sets the artist's `lastPolledAt`. Only artists whose `lastPolledAt` is null or older than `poll_interval_hours` are polled. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_monitor_discover.py`: +```python +from lyra_worker.adapters.fakes import FakeMbBrowser +from lyra_worker.browser import ReleaseGroupInfo +from lyra_worker.monitor import MonitorConfig, discover +from tests.conftest import insert_watched_artist + + +def _releases(): + return [ + ReleaseGroupInfo("rg-old", "Old Album", "Album", (), "2001-01-01"), + ReleaseGroupInfo("rg-new", "New Album", "Album", (), "2999-01-01"), + ] + + +def _monitored_map(conn): + with conn.cursor() as cur: + cur.execute('SELECT "rgMbid", monitored FROM "MonitoredRelease"') + return dict(cur.fetchall()) + + +def test_discover_auto_monitor_future_only_flags_new(conn): + aid = insert_watched_artist(conn, mbid="a1", name="John Mayer", auto_monitor_future=True) + browser = FakeMbBrowser(releases={"a1": _releases()}) + inserted = discover(conn, browser, MonitorConfig(enabled=True)) + assert inserted == 2 + assert _monitored_map(conn) == {"rg-old": False, "rg-new": True} # only the future one is monitored + + +def test_discover_default_off_monitors_nothing(conn): + insert_watched_artist(conn, mbid="a1", name="X", auto_monitor_future=False) + browser = FakeMbBrowser(releases={"a1": _releases()}) + discover(conn, browser, MonitorConfig(enabled=True)) + assert _monitored_map(conn) == {"rg-old": False, "rg-new": False} + + +def test_discover_is_idempotent_and_sets_last_polled(conn): + insert_watched_artist(conn, mbid="a1", name="X", auto_monitor_future=True) + browser = FakeMbBrowser(releases={"a1": _releases()}) + assert discover(conn, browser, MonitorConfig(enabled=True)) == 2 + assert discover(conn, browser, MonitorConfig(enabled=True)) == 0 # lastPolledAt now recent; not re-polled + with conn.cursor() as cur: + cur.execute('SELECT "lastPolledAt" IS NOT NULL FROM "WatchedArtist" WHERE mbid = %s', ("a1",)) + assert cur.fetchone()[0] is True + + +def test_config_from_config_parses_strings(): + cfg = MonitorConfig.from_config({"monitor.enabled": "true", "monitor.qualityCutoff": "3"}) + assert cfg.enabled is True and cfg.quality_cutoff == 3 + assert MonitorConfig.from_config({}).enabled is False +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitor_discover.py -v +``` +Expected: FAIL — `ModuleNotFoundError: No module named 'lyra_worker.monitor'`. + +- [ ] **Step 3: Implement `MonitorConfig` + `discover`** + +`worker/lyra_worker/monitor.py`: +```python +from dataclasses import dataclass +from datetime import date, datetime + +import psycopg + +from lyra_worker.browser import MbBrowser + +_TRUE = {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class MonitorConfig: + enabled: bool = False + poll_interval_hours: int = 24 + retry_interval_hours: int = 6 + quality_cutoff: int = 2 + upgrade_window_days: int = 14 + + @classmethod + def from_config(cls, config: dict) -> "MonitorConfig": + def _int(key: str, default: int) -> int: + try: + return int(config[key]) + except (KeyError, TypeError, ValueError): + return default + + return cls( + enabled=str(config.get("monitor.enabled", "")).strip().lower() in _TRUE, + poll_interval_hours=_int("monitor.pollIntervalHours", 24), + retry_interval_hours=_int("monitor.retryIntervalHours", 6), + quality_cutoff=_int("monitor.qualityCutoff", 2), + upgrade_window_days=_int("monitor.upgradeWindowDays", 14), + ) + + +def _is_new(first_release_date: str, monitor_from: datetime) -> bool: + """A release is 'new going forward' if its first-release date is after monitorFrom.""" + if not first_release_date: + return False + try: + frd = date.fromisoformat(first_release_date[:10]) + except ValueError: + return False + return frd > monitor_from.date() + + +def discover(conn: psycopg.Connection, browser: MbBrowser, cfg: MonitorConfig) -> int: + """Poll each due watched artist; insert newly-seen release-groups. Returns #rows inserted.""" + with conn.cursor() as cur: + cur.execute( + 'SELECT id, mbid, name, "autoMonitorFuture", "monitorFrom" FROM "WatchedArtist" ' + 'WHERE "lastPolledAt" IS NULL ' + ' OR "lastPolledAt" < now() - make_interval(hours => %s)', + (cfg.poll_interval_hours,), + ) + artists = cur.fetchall() + + inserted = 0 + for artist_id, mbid, name, auto_future, monitor_from in artists: + for rg in browser.browse_release_groups(mbid): + monitored = bool(auto_future) and _is_new(rg.first_release_date, monitor_from) + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "MonitoredRelease" (id, "watchedArtistId", "artistMbid", ' + '"artistName", "rgMbid", album, "primaryType", "secondaryTypes", ' + '"firstReleaseDate", monitored, state, "createdAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'wanted', now()) " + 'ON CONFLICT ("rgMbid") DO NOTHING', + (artist_id, mbid, name, rg.rg_mbid, rg.title, rg.primary_type or None, + list(rg.secondary_types), rg.first_release_date or None, monitored), + ) + inserted += cur.rowcount + with conn.cursor() as cur: + cur.execute('UPDATE "WatchedArtist" SET "lastPolledAt" = now() WHERE id = %s', (artist_id,)) + conn.commit() + return inserted +``` + +- [ ] **Step 4: Run the test to verify it passes** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitor_discover.py -v +``` +Expected: PASS — all four tests green. + +- [ ] **Step 5: Commit** + +```bash +cd /home/jonathan/Projects/lyra +git add worker/lyra_worker/monitor.py worker/tests/test_monitor_discover.py +git commit -m "feat: monitor discovery phase + MonitorConfig" +``` + +--- + +### Task 4: Retry/enqueue phase (`monitor.enqueue_due`) + +**Files:** +- Modify: `worker/lyra_worker/monitor.py` +- Test: `worker/tests/test_monitor_enqueue.py` + +**Interfaces:** +- Consumes: `MonitorConfig`, the `MonitoredRelease`/`Request`/`Job`/`LibraryItem` tables, `insert_monitored_release`. +- Produces: `enqueue_due(conn, cfg) -> int`. For each `monitored`, non-`fulfilled` release with no active job and due by `retry_interval_hours`: satisfy-or-skip via library dedupe and the upgrade window (setting `fulfilled` where appropriate), else create a `Request` (with `monitoredReleaseId`) + `Job(state='requested')` and set `lastSearchedAt`. Returns #jobs enqueued. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_monitor_enqueue.py`: +```python +from lyra_worker.monitor import MonitorConfig, enqueue_due +from tests.conftest import insert_monitored_release + +CFG = MonitorConfig(enabled=True, retry_interval_hours=6, quality_cutoff=2, upgrade_window_days=14) + + +def _job_count_for(conn, rel_id): + with conn.cursor() as cur: + cur.execute( + 'SELECT count(*) FROM "Request" r JOIN "Job" j ON j."requestId" = r.id ' + 'WHERE r."monitoredReleaseId" = %s', + (rel_id,), + ) + return cur.fetchone()[0] + + +def _state(conn, rel_id): + with conn.cursor() as cur: + cur.execute('SELECT state FROM "MonitoredRelease" WHERE id = %s', (rel_id,)) + return cur.fetchone()[0] + + +def test_enqueues_a_wanted_release(conn): + rel_id = insert_monitored_release(conn, artist_name="John Mayer", album="Continuum", + rg_mbid="rg-1", monitored=True, state="wanted") + assert enqueue_due(conn, CFG) == 1 + assert _job_count_for(conn, rel_id) == 1 + with conn.cursor() as cur: + cur.execute( + 'SELECT artist, album FROM "Request" WHERE "monitoredReleaseId" = %s', (rel_id,) + ) + assert cur.fetchone() == ("John Mayer", "Continuum") + + +def test_skips_unmonitored_and_fulfilled(conn): + insert_monitored_release(conn, rg_mbid="rg-a", monitored=False, state="wanted") + insert_monitored_release(conn, rg_mbid="rg-b", monitored=True, state="fulfilled") + assert enqueue_due(conn, CFG) == 0 + + +def test_skips_when_recently_searched(conn): + insert_monitored_release(conn, rg_mbid="rg-1", monitored=True, state="wanted", + last_searched_sql="now() - interval '1 hour'") # < 6h retry + assert enqueue_due(conn, CFG) == 0 + + +def test_skips_release_with_active_job(conn): + rel_id = insert_monitored_release(conn, rg_mbid="rg-1", monitored=True, state="wanted") + assert enqueue_due(conn, CFG) == 1 # first sweep enqueues + assert enqueue_due(conn, CFG) == 0 # active (requested) job blocks a second + + +def test_grabbed_within_window_below_cutoff_re_searches(conn): + rel_id = insert_monitored_release(conn, rg_mbid="rg-1", monitored=True, state="grabbed", + current_quality_class=1, first_grabbed_sql="now() - interval '1 day'") + assert enqueue_due(conn, CFG) == 1 + + +def test_grabbed_past_window_is_fulfilled_not_enqueued(conn): + rel_id = insert_monitored_release(conn, rg_mbid="rg-1", monitored=True, state="grabbed", + current_quality_class=1, first_grabbed_sql="now() - interval '30 days'") + assert enqueue_due(conn, CFG) == 0 + assert _state(conn, rel_id) == "fulfilled" + + +def test_already_in_library_at_cutoff_is_fulfilled(conn): + rel_id = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "createdAt") ' + "VALUES (gen_random_uuid()::text, 'A', 'B', 'completed', now()) RETURNING id" + ) + req = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "LibraryItem" (id, "requestId", artist, album, path, source, format, ' + '"qualityClass", "importedAt") ' + "VALUES (gen_random_uuid()::text, %s, 'A', 'B', '/m/A/B', 'qobuz', 'FLAC', 2, now())", + (req,), + ) + conn.commit() + assert enqueue_due(conn, CFG) == 0 + assert _state(conn, rel_id) == "fulfilled" +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitor_enqueue.py -v +``` +Expected: FAIL — `ImportError: cannot import name 'enqueue_due'`. + +- [ ] **Step 3: Implement `enqueue_due`** + +Append to `worker/lyra_worker/monitor.py`: +```python +def _set_state(conn: psycopg.Connection, rel_id: str, state: str) -> None: + with conn.cursor() as cur: + cur.execute('UPDATE "MonitoredRelease" SET state = %s WHERE id = %s', (state, rel_id)) + + +def _in_library_at_cutoff(conn: psycopg.Connection, artist: str, album: str, cutoff: int) -> bool: + with conn.cursor() as cur: + cur.execute( + 'SELECT 1 FROM "LibraryItem" WHERE artist = %s AND album = %s ' + 'AND "qualityClass" >= %s LIMIT 1', + (artist, album, cutoff), + ) + return cur.fetchone() is not None + + +def _enqueue_one(conn: psycopg.Connection, rel_id: str, artist: str, album: str) -> None: + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "monitoredReleaseId", "createdAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, 'pending', %s, now()) RETURNING id", + (artist, album, rel_id), + ) + request_id = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "Job" (id, "requestId", state, "currentStage", attempts, "createdAt", "updatedAt") ' + "VALUES (gen_random_uuid()::text, %s, 'requested', 'intake', 0, now(), now())", + (request_id,), + ) + cur.execute('UPDATE "MonitoredRelease" SET "lastSearchedAt" = now() WHERE id = %s', (rel_id,)) + + +def enqueue_due(conn: psycopg.Connection, cfg: MonitorConfig) -> int: + """Enqueue due wanted/upgrade releases as pipeline jobs. Returns #jobs enqueued.""" + with conn.cursor() as cur: + cur.execute( + 'SELECT mr.id, mr."artistName", mr.album, mr.state, mr."currentQualityClass", ' + ' (mr."firstGrabbedAt" IS NOT NULL ' + ' AND mr."firstGrabbedAt" < now() - make_interval(days => %s)) AS window_expired ' + 'FROM "MonitoredRelease" mr ' + 'WHERE mr.monitored = true ' + " AND mr.state <> 'fulfilled' " + ' AND (mr."lastSearchedAt" IS NULL ' + ' OR mr."lastSearchedAt" < now() - make_interval(hours => %s)) ' + ' AND NOT EXISTS (' + ' SELECT 1 FROM "Request" r JOIN "Job" j ON j."requestId" = r.id ' + ' WHERE r."monitoredReleaseId" = mr.id ' + " AND j.state NOT IN ('imported', 'needs_attention'))", + (cfg.upgrade_window_days, cfg.retry_interval_hours), + ) + rows = cur.fetchall() + + enqueued = 0 + for rel_id, artist, album, state, quality, window_expired in rows: + if _in_library_at_cutoff(conn, artist, album, cfg.quality_cutoff): + _set_state(conn, rel_id, "fulfilled") + continue + if state == "grabbed" and (window_expired or (quality is not None and quality >= cfg.quality_cutoff)): + _set_state(conn, rel_id, "fulfilled") + continue + _enqueue_one(conn, rel_id, artist, album) + enqueued += 1 + conn.commit() + return enqueued +``` + +- [ ] **Step 4: Run the test to verify it passes** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitor_enqueue.py -v +``` +Expected: PASS — all seven tests green. + +- [ ] **Step 5: Commit** + +```bash +cd /home/jonathan/Projects/lyra +git add worker/lyra_worker/monitor.py worker/tests/test_monitor_enqueue.py +git commit -m "feat: monitor retry/enqueue phase" +``` + +--- + +### Task 5: Outcome reconcile (`monitor.reconcile`) + +**Files:** +- Modify: `worker/lyra_worker/monitor.py` +- Test: `worker/tests/test_monitor_reconcile.py` + +**Interfaces:** +- Consumes: `MonitorConfig`, a finished `Job` linked to a `MonitoredRelease` via its `Request`. +- Produces: `reconcile(conn, job_id, cfg) -> None`. If the job's request has no `monitoredReleaseId`, does nothing. On `state='imported'` it reads the current `LibraryItem.qualityClass` for the release's `(artist, album)`, sets `currentQualityClass`, `firstGrabbedAt` (if null), and `state = 'fulfilled'` when quality ≥ cutoff else `'grabbed'`. On `needs_attention` it leaves the release `wanted`. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_monitor_reconcile.py`: +```python +from lyra_worker.monitor import MonitorConfig, reconcile +from tests.conftest import insert_monitored_release + +CFG = MonitorConfig(enabled=True, quality_cutoff=2) + + +def _make_job(conn, rel_id, artist, album, state): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "monitoredReleaseId", "createdAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, 'pending', %s, now()) RETURNING id", + (artist, album, rel_id), + ) + req = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "Job" (id, "requestId", state, "currentStage", attempts, "createdAt", "updatedAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, 'import', 1, now(), now()) RETURNING id", + (req, state), + ) + job = cur.fetchone()[0] + conn.commit() + return req, job + + +def _add_library(conn, req, artist, album, quality): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "LibraryItem" (id, "requestId", artist, album, path, source, format, ' + '"qualityClass", "importedAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, %s, '/m', 'qobuz', 'FLAC', %s, now())", + (req, artist, album, quality), + ) + conn.commit() + + +def _release(conn, rel_id): + with conn.cursor() as cur: + cur.execute( + 'SELECT state, "currentQualityClass", "firstGrabbedAt" IS NOT NULL ' + 'FROM "MonitoredRelease" WHERE id = %s', + (rel_id,), + ) + return cur.fetchone() + + +def test_imported_below_cutoff_becomes_grabbed(conn): + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + req, job = _make_job(conn, rel, "A", "B", "imported") + _add_library(conn, req, "A", "B", 1) + reconcile(conn, job, CFG) + assert _release(conn, rel) == ("grabbed", 1, True) + + +def test_imported_at_cutoff_becomes_fulfilled(conn): + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + req, job = _make_job(conn, rel, "A", "B", "imported") + _add_library(conn, req, "A", "B", 3) + reconcile(conn, job, CFG) + assert _release(conn, rel) == ("fulfilled", 3, True) + + +def test_needs_attention_leaves_wanted(conn): + rel = insert_monitored_release(conn, artist_name="A", album="B", rg_mbid="rg-1", + monitored=True, state="wanted") + _req, job = _make_job(conn, rel, "A", "B", "needs_attention") + reconcile(conn, job, CFG) + assert _release(conn, rel) == ("wanted", None, False) + + +def test_unlinked_job_is_ignored(conn): + with conn.cursor() as cur: + cur.execute( + 'INSERT INTO "Request" (id, artist, album, status, "createdAt") ' + "VALUES (gen_random_uuid()::text, 'A', 'B', 'completed', now()) RETURNING id" + ) + req = cur.fetchone()[0] + cur.execute( + 'INSERT INTO "Job" (id, "requestId", state, "currentStage", attempts, "createdAt", "updatedAt") ' + "VALUES (gen_random_uuid()::text, %s, 'imported', 'import', 1, now(), now()) RETURNING id", + (req,), + ) + job = cur.fetchone()[0] + conn.commit() + reconcile(conn, job, CFG) # no monitoredReleaseId → no-op, must not raise +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitor_reconcile.py -v +``` +Expected: FAIL — `ImportError: cannot import name 'reconcile'`. + +- [ ] **Step 3: Implement `reconcile`** + +Append to `worker/lyra_worker/monitor.py`: +```python +def reconcile(conn: psycopg.Connection, job_id: str, cfg: MonitorConfig) -> None: + """Feed a finished job's outcome back into its MonitoredRelease (if any).""" + with conn.cursor() as cur: + cur.execute( + 'SELECT j.state, r."monitoredReleaseId", r.artist, r.album ' + 'FROM "Job" j JOIN "Request" r ON r.id = j."requestId" WHERE j.id = %s', + (job_id,), + ) + row = cur.fetchone() + if row is None: + return + state, rel_id, artist, album = row + if rel_id is None or state != "imported": + return # needs_attention (or in-flight) leaves the release wanted + + with conn.cursor() as cur: + cur.execute( + 'SELECT "qualityClass" FROM "LibraryItem" WHERE artist = %s AND album = %s', + (artist, album), + ) + lib = cur.fetchone() + quality = lib[0] if lib else None + new_state = "fulfilled" if (quality is not None and quality >= cfg.quality_cutoff) else "grabbed" + with conn.cursor() as cur: + cur.execute( + 'UPDATE "MonitoredRelease" ' + 'SET "currentQualityClass" = %s, state = %s, ' + ' "firstGrabbedAt" = COALESCE("firstGrabbedAt", now()) WHERE id = %s', + (quality, new_state, rel_id), + ) + conn.commit() +``` + +- [ ] **Step 4: Run the test to verify it passes** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitor_reconcile.py -v +``` +Expected: PASS — all four tests green. + +- [ ] **Step 5: Commit** + +```bash +cd /home/jonathan/Projects/lyra +git add worker/lyra_worker/monitor.py worker/tests/test_monitor_reconcile.py +git commit -m "feat: monitor outcome reconcile" +``` + +--- + +### Task 6: `sweep()` + real browser + wire into the worker loop + +**Files:** +- Modify: `worker/lyra_worker/monitor.py` (add `sweep`) +- Create: `worker/lyra_worker/_mbbrowser.py` (real `MusicBrainzBrowser`) +- Modify: `worker/lyra_worker/registry.py` (add `build_browser`) +- Modify: `worker/lyra_worker/main.py` (tick sweep + reconcile after each job) +- Test: `worker/tests/test_monitor_sweep.py` +- Test: `worker/tests/test_mbbrowser_live.py` (opt-in) + +**Interfaces:** +- Consumes: `discover`, `enqueue_due`, `FakeMbBrowser`, `insert_watched_artist`. +- Produces: `sweep(conn, browser, cfg) -> None` (calls `discover` then `enqueue_due`); `registry.build_browser() -> MbBrowser` returning a `MusicBrainzBrowser`; `main.run_forever()` runs `sweep` on a ~60s tick when `monitor.enabled` and calls `reconcile` after each `run_pipeline`. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_monitor_sweep.py`: +```python +from lyra_worker.adapters.fakes import FakeMbBrowser +from lyra_worker.browser import ReleaseGroupInfo +from lyra_worker.monitor import MonitorConfig, sweep +from tests.conftest import insert_watched_artist + +CFG = MonitorConfig(enabled=True) + + +def test_sweep_discovers_then_enqueues(conn): + insert_watched_artist(conn, mbid="a1", name="John Mayer", auto_monitor_future=True) + browser = FakeMbBrowser( + releases={"a1": [ReleaseGroupInfo("rg-new", "New", "Album", (), "2999-01-01")]} + ) + sweep(conn, browser, CFG) + with conn.cursor() as cur: + cur.execute( + 'SELECT count(*) FROM "Request" r ' + 'JOIN "MonitoredRelease" mr ON mr.id = r."monitoredReleaseId" ' + 'WHERE mr."rgMbid" = %s', + ("rg-new",), + ) + assert cur.fetchone()[0] == 1 # the newly-discovered monitored release got a job +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitor_sweep.py -v +``` +Expected: FAIL — `ImportError: cannot import name 'sweep'`. + +- [ ] **Step 3: Implement `sweep`** + +Append to `worker/lyra_worker/monitor.py`: +```python +def sweep(conn: psycopg.Connection, browser: MbBrowser, cfg: MonitorConfig) -> None: + """One monitor pass: discover new releases, then enqueue everything due.""" + discover(conn, browser, cfg) + enqueue_due(conn, cfg) +``` + +- [ ] **Step 4: Run the sweep test to verify it passes** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_monitor_sweep.py -v +``` +Expected: PASS. + +- [ ] **Step 5: Implement the real browser** + +`worker/lyra_worker/_mbbrowser.py`: +```python +from lyra_worker.browser import ArtistHit, ReleaseGroupInfo + + +class MusicBrainzBrowser: + """Real MbBrowser via musicbrainzngs (imported lazily; not unit-tested offline).""" + + def __init__(self, app_name: str = "Lyra", version: str = "0.1", contact: str = "lyra@localhost"): + self._app, self._version, self._contact = app_name, version, contact + + def _ua(self): + import musicbrainzngs + + musicbrainzngs.set_useragent(self._app, self._version, self._contact) + return musicbrainzngs + + def search_artist(self, name: str) -> list[ArtistHit]: + mb = self._ua() + res = mb.search_artists(query=name, limit=8) + return [ + ArtistHit(mbid=a["id"], name=a.get("name", ""), disambiguation=a.get("disambiguation", "")) + for a in res.get("artist-list", []) + ] + + def browse_release_groups(self, artist_mbid: str) -> list[ReleaseGroupInfo]: + mb = self._ua() + out: list[ReleaseGroupInfo] = [] + offset = 0 + while True: + res = mb.browse_release_groups(artist=artist_mbid, limit=100, offset=offset) + groups = res.get("release-group-list", []) + for g in groups: + out.append( + ReleaseGroupInfo( + rg_mbid=g["id"], + title=g.get("title", ""), + primary_type=g.get("primary-type", "") or "", + secondary_types=tuple(g.get("secondary-type-list", []) or ()), + first_release_date=g.get("first-release-date", "") or "", + ) + ) + offset += len(groups) + if len(groups) < 100 or offset >= int(res.get("release-group-count", offset)): + break + return out +``` + +- [ ] **Step 6: Add `build_browser` and wire the loop** + +In `worker/lyra_worker/registry.py`, add: +```python +from lyra_worker._mbbrowser import MusicBrainzBrowser +from lyra_worker.browser import MbBrowser + + +def build_browser() -> MbBrowser: + """The MusicBrainz browser used by the background monitor.""" + return MusicBrainzBrowser() +``` + +Replace `worker/lyra_worker/main.py` with: +```python +import time + +from lyra_worker.claim import claim_next +from lyra_worker.config import get_config +from lyra_worker.db import wait_for_db +from lyra_worker.monitor import MonitorConfig, reconcile, sweep +from lyra_worker.pipeline import run_pipeline +from lyra_worker.registry import build_adapters, build_browser, build_resolver, build_tagger + +IDLE_SLEEP = 2.0 +MONITOR_TICK_SECONDS = 60.0 + + +def run_forever() -> None: + conn = wait_for_db() + adapters = build_adapters(get_config(conn)) + resolver = build_resolver() + tagger = build_tagger() + browser = build_browser() + print(f"worker: {len(adapters)} adapter(s) enabled: {[a.name for a in adapters]}", flush=True) + print("worker: waiting for jobs", flush=True) + last_tick = 0.0 + try: + while True: + mcfg = MonitorConfig.from_config(get_config(conn)) + now = time.monotonic() + if mcfg.enabled and now - last_tick >= MONITOR_TICK_SECONDS: + try: + sweep(conn, browser, mcfg) + except Exception as e: # a monitor error must never kill the worker + print(f"worker: monitor sweep failed: {e}", flush=True) + conn.rollback() + last_tick = now + + job_id = claim_next(conn) + if job_id is None: + time.sleep(IDLE_SLEEP) + continue + print(f"worker: claimed job {job_id}", flush=True) + run_pipeline(conn, job_id, adapters, resolver=resolver, tagger=tagger) + if mcfg.enabled: + try: + reconcile(conn, job_id, mcfg) + except Exception as e: + print(f"worker: reconcile failed for job {job_id}: {e}", flush=True) + conn.rollback() + print(f"worker: finished job {job_id}", flush=True) + finally: + conn.close() + + +if __name__ == "__main__": + run_forever() +``` + +- [ ] **Step 7: Add the opt-in live browser test** + +`worker/tests/test_mbbrowser_live.py`: +```python +import os + +import pytest + +pytestmark = pytest.mark.skipif( + not os.environ.get("LYRA_LIVE_TESTS"), + reason="hits the real MusicBrainz API; set LYRA_LIVE_TESTS=1 to run", +) + + +def test_search_and_browse_real_artist(): + from lyra_worker._mbbrowser import MusicBrainzBrowser + + browser = MusicBrainzBrowser() + hits = browser.search_artist("John Mayer") + assert hits and any("John Mayer" in h.name for h in hits) + releases = browser.browse_release_groups(hits[0].mbid) + assert any(r.title for r in releases) +``` + +- [ ] **Step 8: Verify imports stay offline + run the full suite** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -c "import sys; from lyra_worker._mbbrowser import MusicBrainzBrowser; from lyra_worker import main; print('import ok; mb loaded:', 'musicbrainzngs' in sys.modules)" +worker/.venv/bin/python -m pytest -q +``` +Expected: prints `import ok; mb loaded: False` (musicbrainzngs stays lazy); full suite passes, `test_mbbrowser_live.py` skipped, output pristine. + +- [ ] **Step 9: Commit** + +```bash +cd /home/jonathan/Projects/lyra +git add worker/lyra_worker/monitor.py worker/lyra_worker/_mbbrowser.py worker/lyra_worker/registry.py worker/lyra_worker/main.py worker/tests/test_monitor_sweep.py worker/tests/test_mbbrowser_live.py +git commit -m "feat: monitor sweep, real MB browser, and worker-loop wiring" +``` + +--- + +### Task 7: Pipeline import upsert (quality upgrades replace) + +**Files:** +- Modify: `worker/lyra_worker/pipeline.py` (`_import` ON CONFLICT) +- Test: `worker/tests/test_import_upgrade.py` + +**Interfaces:** +- Consumes: existing `_import`, `LibraryItem` (`@@unique([artist, album])`). +- Produces: `_import` now upserts on `(artist, album)` — a higher `qualityClass` overwrites `path`/`source`/`format`/`qualityClass`/`requestId`/`importedAt`; an equal-or-lower quality leaves the existing row unchanged. + +- [ ] **Step 1: Write the failing test** + +`worker/tests/test_import_upgrade.py`: +```python +from lyra_worker.pipeline import _import +from lyra_worker.types import Candidate, Quality +from tests.conftest import insert_request + + +def _winner(fmt, lossless, bit_depth=None, sample_rate=None): + q = Quality(fmt=fmt, lossless=lossless, bit_depth=bit_depth, sample_rate=sample_rate) + return Candidate(source="qobuz", source_ref="ref", matched_artist="A", matched_album="B", + quality=q, track_count=10, source_tier=0) + + +def _library_row(conn, artist, album): + with conn.cursor() as cur: + cur.execute( + 'SELECT path, "qualityClass", source FROM "LibraryItem" WHERE artist = %s AND album = %s', + (artist, album), + ) + return cur.fetchone() + + +def test_higher_quality_replaces(conn): + from lyra_worker.types import MBTarget + + target = MBTarget(artist="A", album="B") + job1 = insert_request(conn, artist="A", album="B") + _import(conn, job1, target, _winner("MP3", False), "/m/lossy") # qualityClass 1 + job2 = insert_request(conn, artist="A", album="B") + _import(conn, job2, target, _winner("FLAC", True, 24, 96000), "/m/hires") # qualityClass 3 + assert _library_row(conn, "A", "B") == ("/m/hires", 3, "qobuz") + + +def test_lower_quality_does_not_replace(conn): + from lyra_worker.types import MBTarget + + target = MBTarget(artist="A", album="B") + job1 = insert_request(conn, artist="A", album="B") + _import(conn, job1, target, _winner("FLAC", True, 24, 96000), "/m/hires") # qualityClass 3 + job2 = insert_request(conn, artist="A", album="B") + _import(conn, job2, target, _winner("MP3", False), "/m/lossy") # qualityClass 1 + assert _library_row(conn, "A", "B") == ("/m/hires", 3, "qobuz") +``` + +- [ ] **Step 2: Run the test to verify it fails** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_import_upgrade.py -v +``` +Expected: FAIL — `test_higher_quality_replaces` fails (current `DO NOTHING` keeps `/m/lossy`). + +- [ ] **Step 3: Change `_import` to upsert on higher quality** + +In `worker/lyra_worker/pipeline.py`, replace the `INSERT ... ON CONFLICT (artist, album) DO NOTHING` statement inside `_import` with: +```python + cur.execute( + 'INSERT INTO "LibraryItem" (id, "requestId", artist, album, path, source, ' + 'format, "qualityClass", "importedAt") ' + "VALUES (gen_random_uuid()::text, %s, %s, %s, %s, %s, %s, %s, now()) " + 'ON CONFLICT (artist, album) DO UPDATE SET ' + ' "requestId" = EXCLUDED."requestId", path = EXCLUDED.path, source = EXCLUDED.source, ' + ' format = EXCLUDED.format, "qualityClass" = EXCLUDED."qualityClass", ' + ' "importedAt" = now() ' + 'WHERE EXCLUDED."qualityClass" > "LibraryItem"."qualityClass"', + (request_id, target.artist, target.album, path, winner.source, + winner.quality.fmt, quality_class(winner.quality)), + ) +``` +(The rest of `_import` — the `Request` status update — is unchanged.) + +- [ ] **Step 4: Run the test to verify it passes** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest tests/test_import_upgrade.py -v +``` +Expected: PASS — both tests green. + +- [ ] **Step 5: Run the full worker suite** + +```bash +cd /home/jonathan/Projects/lyra/worker +export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra +worker/.venv/bin/python -m pytest -q +``` +Expected: PASS — every suite green, live tests skipped, output pristine. + +- [ ] **Step 6: Commit** + +```bash +cd /home/jonathan/Projects/lyra +git add worker/lyra_worker/pipeline.py worker/tests/test_import_upgrade.py +git commit -m "feat: import stage upserts to replace lower-quality library items" +``` + +--- + +## Notes: monitoring core complete + +With this plan the worker is a standing monitor: it discovers each watched artist's +releases from MusicBrainz, enqueues wanted and in-window upgrade releases through +the unchanged slice 1 pipeline, and reconciles each outcome back into +`MonitoredRelease` state — all inert when `monitor.enabled` is false. The data +model (`WatchedArtist`, `MonitoredRelease`, `Request.monitoredReleaseId`) is the +contract the **monitoring-web** plan builds on: live MusicBrainz artist +search/browse for the follow picker, artists/wanted/releases CRUD API, and the +`/artists`, `/artists/[id]`, and `/wanted` pages. Config is seeded via the +existing settings surface: set `monitor.enabled=true` (plus optional +`monitor.pollIntervalHours`, `monitor.retryIntervalHours`, `monitor.qualityCutoff`, +`monitor.upgradeWindowDays`) in the `Config` table. Live end-to-end validation +(a real watched artist producing a real grab) is a user step once the web plan +lands.