feat: Qobuz pacing budget gate (human-like volume + cadence)

Avoid re-triggering a USER_BLOCKED account by capping Qobuz to a human-looking
volume and cadence. A per-album gate (worker/lyra_worker/qobuz_gate.py) allows
Qobuz only within active hours, under today's cap (a warm-up ramp of a steady
cap), and past a randomized spacing since the last Qobuz download; the gap spreads
the day's budget across active hours with jitter. When the gate is closed the
pipeline drops Qobuz candidates so the job falls through to another source (and
monitored albums upgrade to Qobuz later); a Qobuz-only setup is never gated.
Config keys read live each loop; counters reset at the day boundary via the DB
clock. Tunable in Settings -> Qobuz -> Pacing (new /api/qobuz/pacing route).

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

worker/lyra_worker/pipeline.py

@@ -1,4 +1,5 @@
 import os
+import random
 import shutil
 import threading
 import time
@@ -11,6 +12,7 @@ import psycopg
 from lyra_worker.adapters.base import SourceAdapter
 from lyra_worker.confidence import score_confidence
 from lyra_worker.library import album_dir, import_album, list_audio_files, staging_dir
+from lyra_worker.qobuz_gate import gate_open, record_download
 from lyra_worker.quality import quality_class
 from lyra_worker.ranker import rank_candidates
 from lyra_worker.types import Candidate, MBTarget
@@ -365,6 +367,13 @@ def run_pipeline(
             found.append(replace(c, source_tier=adapter.tier, confidence=score_confidence(target, c)))
     _persist_candidates(conn, job_id, found)
 
+    # Qobuz pacing: when the budget gate is closed (off-hours / daily cap / spacing), drop Qobuz
+    # candidates so the job falls through to another source and Qobuz stays under the radar — UNLESS
+    # Qobuz is the only option (nothing to fall through to, so don't strand the album).
+    if any(c.source == "qobuz" for c in found) and any(c.source != "qobuz" for c in found):
+        if not gate_open(conn):
+            found = [c for c in found if c.source != "qobuz"]
+
     # 3. rank
     _set_state(conn, job_id, "matched", "rank")
     ranked = rank_candidates(target, found, min_confidence)
@@ -415,6 +424,12 @@ def run_pipeline(
             _fail(conn, job_id, last_problem or "all downloads failed")
             return
         _set_download_progress(conn, job_id, 1.0)  # download done — UI shows 100% into Finishing
+        if winner.source == "qobuz":
+            # Count this Qobuz download against today's budget and set the next randomized spacing.
+            try:
+                record_download(conn, random.random())
+            except Exception as e:  # pacing bookkeeping must never fail a good download
+                print(f"pipeline: qobuz pacing record failed for job {job_id}: {e}", flush=True)
 
         # 5. promote + tag the verified winner
         _set_state(conn, job_id, "tagging", "tag")

worker/lyra_worker/qobuz_gate.py

@@ -0,0 +1,159 @@
+"""Qobuz pacing ("budget gate").
+
+Keep using Qobuz for hi-res, but at a human-looking volume and cadence so the account isn't
+flagged for bulk downloading (which got it 403 USER_BLOCKED). One gate decides, per album,
+whether Qobuz is an eligible source right now: within active hours, under today's cap (a warm-up
+ramp of a steady cap), and past a randomized spacing since the last Qobuz download. When the gate
+is closed the pipeline drops Qobuz candidates and falls through to the next source; monitored
+albums upgrade to Qobuz later once budget frees up.
+
+Pure decision logic (`today_cap`, `is_allowed`, `next_gap_seconds`) is separated from the two DB
+helpers (`gate_open`, `record_download`) so the policy is unit-tested without a clock or DB.
+"""
+from dataclasses import dataclass
+from datetime import date
+
+_TRUE = {"1", "true", "yes", "on"}
+_JITTER = 0.3   # ± fraction around the spread interval
+_GAP_FLOOR_S = 60.0
+
+
+@dataclass(frozen=True)
+class QobuzPacing:
+    enabled: bool = True
+    active_start: int = 8
+    active_end: int = 23
+    daily_cap: int = 40
+    warmup_start_date: str = ""  # ISO "YYYY-MM-DD", or "" for no warm-up ramp
+
+    @classmethod
+    def from_config(cls, config: dict) -> "QobuzPacing":
+        def _int(key, default):
+            try:
+                return int(config[key])
+            except (KeyError, TypeError, ValueError):
+                return default
+
+        return cls(
+            enabled=str(config.get("qobuz.pacing.enabled", "true")).strip().lower() in _TRUE,
+            active_start=_int("qobuz.pacing.activeStartHour", 8),
+            active_end=_int("qobuz.pacing.activeEndHour", 23),
+            daily_cap=_int("qobuz.pacing.dailyCap", 40),
+            warmup_start_date=str(config.get("qobuz.pacing.warmupStartDate", "") or "").strip(),
+        )
+
+
+def today_cap(pacing: QobuzPacing, today: date) -> int:
+    """Today's Qobuz download cap: the steady cap, ramped up by account age (days since
+    ``warmup_start_date``). No warm-up date -> steady cap. A small steady cap is never exceeded."""
+    steady = max(0, pacing.daily_cap)
+    if not pacing.warmup_start_date:
+        return steady
+    try:
+        start = date.fromisoformat(pacing.warmup_start_date)
+    except ValueError:
+        return steady
+    days = max(0, (today - start).days)
+    if days < 3:
+        return min(5, steady)
+    if days < 7:
+        return min(10, steady)
+    if days < 14:
+        return min(20, steady)
+    return steady
+
+
+def is_allowed(pacing: QobuzPacing, now_hour: int, count_today: int, cap: int,
+               now_epoch: float, next_allowed_epoch: float) -> bool:
+    """Pure gate decision. Qobuz is eligible only within active hours, under today's cap, and past
+    the randomized spacing since the last download. Pacing disabled -> always allowed."""
+    if not pacing.enabled:
+        return True
+    if not (pacing.active_start <= now_hour < pacing.active_end):
+        return False
+    if count_today >= cap:
+        return False
+    if now_epoch < next_allowed_epoch:
+        return False
+    return True
+
+
+def next_gap_seconds(pacing: QobuzPacing, cap: int, rand01: float) -> float:
+    """Seconds until the next Qobuz download is allowed: the day's budget spread across the active
+    window (``activeWindow / cap``), ± _JITTER, floored at _GAP_FLOOR_S. ``rand01`` in [0,1] is the
+    injected randomness (so tests are deterministic; callers pass random.random())."""
+    window_s = max(1, pacing.active_end - pacing.active_start) * 3600
+    base = window_s / max(1, cap)
+    lo, hi = base * (1 - _JITTER), base * (1 + _JITTER)
+    gap = lo + (hi - lo) * rand01
+    return max(_GAP_FLOOR_S, gap)
+
+
+def _now_parts(conn):
+    """(hour:int, today:date, epoch:float) from the DB clock — avoids container-TZ ambiguity."""
+    with conn.cursor() as cur:
+        cur.execute("SELECT extract(hour from now())::int, current_date, extract(epoch from now())")
+        hour, today, epoch = cur.fetchone()
+    return int(hour), today, float(epoch)
+
+
+def _read(conn, keys):
+    with conn.cursor() as cur:
+        cur.execute('SELECT key, value FROM "Config" WHERE key = ANY(%s)', (list(keys),))
+        return dict(cur.fetchall())
+
+
+def gate_open(conn) -> bool:
+    """Whether Qobuz may be used for a download right now (reads pacing config + counter/spacing
+    state from Config and the DB clock). A counter from a previous day is treated as 0."""
+    from lyra_worker.config import get_config
+
+    pacing = QobuzPacing.from_config(get_config(conn))
+    hour, today, epoch = _now_parts(conn)
+    state = _read(conn, ("qobuz.pacing.countDay", "qobuz.pacing.countToday", "qobuz.pacing.nextAllowedAt"))
+    count = 0
+    if state.get("qobuz.pacing.countDay") == today.isoformat():
+        try:
+            count = int(state.get("qobuz.pacing.countToday") or 0)
+        except (TypeError, ValueError):
+            count = 0
+    try:
+        next_allowed = float(state.get("qobuz.pacing.nextAllowedAt") or 0)
+    except (TypeError, ValueError):
+        next_allowed = 0.0
+    return is_allowed(pacing, hour, count, today_cap(pacing, today), epoch, next_allowed)
+
+
+def record_download(conn, rand01: float) -> None:
+    """Record a successful Qobuz download: bump today's counter (resetting it if the stored day is
+    stale) and set the randomized next-allowed time. ``rand01`` is injected randomness."""
+    from lyra_worker.config import get_config
+
+    pacing = QobuzPacing.from_config(get_config(conn))
+    _hour, today, epoch = _now_parts(conn)
+    state = _read(conn, ("qobuz.pacing.countDay", "qobuz.pacing.countToday"))
+    if state.get("qobuz.pacing.countDay") == today.isoformat():
+        try:
+            count = int(state.get("qobuz.pacing.countToday") or 0)
+        except (TypeError, ValueError):
+            count = 0
+    else:
+        count = 0
+    count += 1
+    next_allowed = epoch + next_gap_seconds(pacing, today_cap(pacing, today), rand01)
+    _write(conn, {
+        "qobuz.pacing.countDay": today.isoformat(),
+        "qobuz.pacing.countToday": str(count),
+        "qobuz.pacing.nextAllowedAt": repr(next_allowed),
+    })
+
+
+def _write(conn, kv: dict) -> None:
+    with conn.cursor() as cur:
+        for key, value in kv.items():
+            cur.execute(
+                'INSERT INTO "Config"(key,value,secret,"updatedAt") VALUES (%s,%s,false,now()) '
+                'ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value, "updatedAt"=now()',
+                (key, value),
+            )
+    conn.commit()