# 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/<timestamp>_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.