Files

T

Jonathan 915bd96cd3 docs: add monitoring-core implementation plan (slice 2, plan 1)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

2026-07-11 12:01:57 +02:00

50 KiB

Raw Blame History

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)

# 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:

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

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:

enum MonitoredReleaseState {
  wanted
  grabbed
  fulfilled
}

Add to model Request (a new field + relation, alongside job / libraryItem):

  monitoredReleaseId String?
  monitoredRelease   MonitoredRelease? @relation(fields: [monitoredReleaseId], references: [id], onDelete: SetNull)

Add two new models:

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

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:

    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:

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

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)

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

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:

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

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:

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:

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

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

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:

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

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:

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

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

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:

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

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:

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

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

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:

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

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:

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

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

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:

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

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:

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

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:

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:

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:

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:

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

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

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:

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

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:

        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

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

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

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.

50 KiB Raw Blame History

Lyra Monitoring Core (Worker + Data Model) Implementation Plan

Global Constraints

Shared interfaces (defined by tasks below)

Task 1: Data model — WatchedArtist, MonitoredRelease, Request link

Task 2: MbBrowser seam + FakeMbBrowser

Task 3: Discovery phase (monitor.discover) + MonitorConfig

Task 4: Retry/enqueue phase (monitor.enqueue_due)

Task 5: Outcome reconcile (monitor.reconcile)

Task 6: sweep() + real browser + wire into the worker loop

Task 7: Pipeline import upsert (quality upgrades replace)

Notes: monitoring core complete

50 KiB

Raw Blame History

Task 3: Discovery phase (`monitor.discover`) + `MonitorConfig`

Task 4: Retry/enqueue phase (`monitor.enqueue_due`)

Task 5: Outcome reconcile (`monitor.reconcile`)

Task 6: `sweep()` + real browser + wire into the worker loop