From 20c9f9321d44df5c9aea0320e273c5ed780b609e Mon Sep 17 00:00:00 2001 From: Jeff Crouse Date: Sun, 4 Oct 2026 15:58:06 -0400 Subject: [PATCH] feat(analysis): trust a community cache hit, check one in fifty (ADR-0144) A features hit at the current FEATURES_VERSION is stored as it is and the track is not decoded; one hit in fifty, chosen from the fingerprint, is also analysed locally, its disagreements recorded and its result not contributed. A miss or an unanswered lookup is analysed exactly as before. The worker logs which path each track took. Building it found the cache's section analysis was being discarded: fetched into analysis_detail, then reset to None and recomputed by the next block, so a hit with cached detail decoded the track anyway. It is now kept; a hit without it is left to the backfill (point 3). tests/test_features_cache_trust.py runs run_track_features against the test database with the decoder rigged to record calls; five of its eight tests fail on the replaced code. Full backend suite: 2475 passed. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_014hHXQAoTT1g8ATYHgVhGwb --- CHANGELOG.md | 8 + backend/app/services/llm/tools.py | 2 +- .../app/services/tasks/analysis_pipeline.py | 147 ++++++++++----- backend/tests/test_features_cache_trust.py | 178 ++++++++++++++++++ ...he-hit-is-trusted-and-checked-by-sample.md | 24 ++- 5 files changed, 305 insertions(+), 54 deletions(-) create mode 100644 backend/tests/test_features_cache_trust.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 34ec589b..34022ac5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Analysis + +- **A track the community cache already knows is no longer analysed again** (ADR-0144). Its cached + features are stored as they are, without decoding the file — about 28 of a cache hit's 35 + seconds on Familiar Server, and 760 MB of the worker's memory. One hit in fifty is still + analysed locally to check the cache, and any disagreement is recorded. +- **The cache's section analysis is kept.** It was fetched and then discarded and recomputed. + ## [0.2.0-beta17] - 2026-10-04 No migrations; the API contract stays at v1. diff --git a/backend/app/services/llm/tools.py b/backend/app/services/llm/tools.py index eb6a2c84..1294c556 100644 --- a/backend/app/services/llm/tools.py +++ b/backend/app/services/llm/tools.py @@ -1026,7 +1026,7 @@ **Feature Confidence:** - Features have confidence scores (0-1) stored in feature_confidence. Most reliable: energy (0.95), key (varies). Least reliable: valence (0.4), speechiness (0.3 without VAD). -- When external features (ReccoBeats) exist, local analysis is also run for cross-validation. Disagreements are flagged. +- Features may come from the community cache (computed by Familiar on another installation). One cached track in fifty is also analysed locally and any disagreement is recorded as `_disagreement`. **Mood mapping examples:** - "chill" → energy<0.4, valence 0.3-0.6 diff --git a/backend/app/services/tasks/analysis_pipeline.py b/backend/app/services/tasks/analysis_pipeline.py index ba92ef62..6c32d913 100644 --- a/backend/app/services/tasks/analysis_pipeline.py +++ b/backend/app/services/tasks/analysis_pipeline.py @@ -5,6 +5,7 @@ """ import gc +import hashlib import logging from datetime import timedelta from pathlib import Path @@ -56,9 +57,10 @@ def run_track_features(track_id: str) -> dict[str, Any]: This is separated from embedding extraction to reduce peak memory usage. Each phase runs in its own subprocess that exits after completion. - External Features Lookup: - Before running expensive librosa analysis, checks if pre-computed features - are available from external services (e.g., ReccoBeats via Spotify track ID). + Community cache (ADR-0144): + A features hit at the current ``FEATURES_VERSION`` is trusted and the track is not decoded for + features. One hit in ``HIT_SAMPLE_RATE``, chosen from the fingerprint, is also analysed locally + and its disagreements recorded. A miss, or a lookup that did not answer, is analysed locally. """ # Configure logging for subprocess (spawned processes don't inherit parent's config) import asyncio @@ -157,15 +159,15 @@ def run_track_features(track_id: str) -> dict[str, Any]: if fp_result: _, acoustid_fingerprint = fp_result - # Try external feature lookup first (ReccoBeats via Spotify ID) features: dict[str, Any] = {} features_source = "local" app_settings = get_app_settings_service().get() - - # Try community cache for features if no external features found + # The community cache is the only outside source of features (ADR-0144's Context: + # the ReccoBeats lookup this once sat behind is gone). deep_scalars: dict[str, Any] = {} - if not features.get("bpm") and app_settings.community_cache_enabled and acoustid_fingerprint: + cached_analysis_detail: dict[str, Any] | None = None + if app_settings.community_cache_enabled and acoustid_fingerprint: try: cache_service = get_community_cache_service( cache_url=app_settings.community_cache_url @@ -198,7 +200,7 @@ def run_track_features(track_id: str) -> dict[str, Any]: cache_service.lookup_analysis_detail(acoustid_fingerprint) ) if cached_detail: - analysis_detail = cached_detail.detail + cached_analysis_detail = cached_detail.detail logger.info( f"Community cache analysis detail hit for " f"{track.title}" @@ -217,56 +219,73 @@ def run_track_features(track_id: str) -> dict[str, Any]: except Exception as e: logger.warning(f"Community cache features lookup failed: {e}") - # Always run local librosa analysis for features + cross-validation + # ADR-0144: trust a hit, check one in fifty, analyse a miss. + # + # This used to decode and analyse every track, hit or not ("always run local + # librosa analysis for features + cross-validation"), a rule from when the outside + # source was ReccoBeats, a different algorithm. It cost a hit ~28 of its ~35 s and + # raised the worker from 110 MB to 870 MB, for values nothing read. It also threw away + # the cache's section analysis: it was fetched into `analysis_detail`, which this + # block then reset to None and recomputed. + cache_hit = features_source == "community_cache" and bool(features.get("bpm")) + path = features_path(cache_hit, acoustid_fingerprint) computed_locally = False - analysis_detail = None + analysis_detail = cached_analysis_detail feature_confidence: dict[str, Any] = {} local_features_dict: dict[str, Any] | None = None - has_external = bool(features.get("bpm")) - - # Use unified pipeline: shared precomputation → features + cheap sections - y, sr, shared = precompute_shared(file_path) - local_result = derive_features(y, sr, shared, file_path) - local_feats, local_confidence = local_result - - if not has_external: - # No external features — use local as primary - features = local_feats - features_source = "local" - computed_locally = True - feature_confidence = dict(local_confidence) - else: - # External features are primary — store local for cross-validation - local_features_dict = local_feats + + if path != "hit": + # Unified pipeline: shared precomputation → features + cheap sections + y, sr, shared = precompute_shared(file_path) + local_feats, local_confidence = derive_features(y, sr, shared, file_path) feature_confidence = dict(local_confidence) - # Compute disagreements between local and external - disagreements = _compute_disagreements(local_feats, features) - for feat_key, disagreement_val in disagreements.items(): - feature_confidence[f"{feat_key}_cross_validated"] = True - if disagreement_val is not None: - feature_confidence[f"{feat_key}_disagreement"] = disagreement_val - - # Run cheap analysis sections (harmonic, rhythmic, spectral, structural, energy) - # Only if track is long enough for meaningful analysis - if track.duration_seconds and track.duration_seconds >= 30: - try: - analysis_detail, deep_scalars, section_errors = run_cheap_sections( - y, sr, shared, str(file_path), track_id - ) - if section_errors: - logger.warning( - f"Analysis section errors for {track.title}: {section_errors}" + + if path == "local": + features = local_feats + features_source = "local" + computed_locally = True + else: + # A sampled hit: the cache's values stay primary; the local ones are kept to + # compare, and never contributed (point 2 — the cache already holds the track, + # and a second submission from the same algorithm is manufactured agreement). + local_features_dict = local_feats + disagreements = _compute_disagreements(local_feats, features) + for feat_key, disagreement_val in disagreements.items(): + feature_confidence[f"{feat_key}_cross_validated"] = True + if disagreement_val is not None: + feature_confidence[f"{feat_key}_disagreement"] = disagreement_val + + # Cheap analysis sections (harmonic, rhythmic, spectral, structural, energy), + # only for a track long enough for them to mean anything. + if track.duration_seconds and track.duration_seconds >= 30: + try: + local_detail, local_deep, section_errors = run_cheap_sections( + y, sr, shared, str(file_path), track_id ) - # Capture structural boundary confidence - if analysis_detail: - structural = analysis_detail.get("structural", {}) - if structural.get("boundary_confidence") is not None: - feature_confidence["structural"] = structural["boundary_confidence"] - except Exception as e: - logger.warning(f"Cheap sections failed for {track.title}: {e}") + if section_errors: + logger.warning( + f"Analysis section errors for {track.title}: {section_errors}" + ) + if path == "local": + analysis_detail, deep_scalars = local_detail, local_deep + else: + # The cache's stay primary where it had them. + analysis_detail = cached_analysis_detail or local_detail + deep_scalars = {**local_deep, **deep_scalars} + if analysis_detail: + structural = analysis_detail.get("structural", {}) + if structural.get("boundary_confidence") is not None: + feature_confidence["structural"] = structural["boundary_confidence"] + except Exception as e: + logger.warning(f"Cheap sections failed for {track.title}: {e}") - del y, shared - gc.collect() + del y, shared + gc.collect() + # A trusted hit without cached section analysis keeps `analysis_detail` None, and the + # backfill phase (rows with `analysis_detail IS NULL`) computes it later (point 3). + + # Point 5: which path, beside the per-line times, so the saving is measured. + logger.info(f"Features path for {track.title}: {path}") # Contribute features to community cache if computed locally if ( @@ -390,6 +409,7 @@ def run_track_features(track_id: str) -> dict[str, Any]: "artwork_extracted": artwork_hash is not None, "features_extracted": bool(all_features.get("bpm")), "features_source": features_source, + "features_path": path, "bpm": all_features.get("bpm"), "key": all_features.get("key"), } @@ -689,6 +709,29 @@ def run_track_analysis(track_id: str) -> dict[str, Any]: } +HIT_SAMPLE_RATE = 50 +"""One community cache hit in this many is also analysed locally (ADR-0144 point 2).""" + + +def is_sampled_hit(fingerprint: str | None, rate: int = HIT_SAMPLE_RATE) -> bool: + """Whether a hit is one of the few also analysed locally. + + Chosen from the fingerprint rather than at random, so a track is or is not sampled on every + installation and every run, and a disagreement can be reproduced. + """ + if not fingerprint: + return False + digest = hashlib.sha256(fingerprint.encode()).digest() + return int.from_bytes(digest[:8], "big") % rate == 0 + + +def features_path(cache_hit: bool, fingerprint: str | None) -> str: + """``hit`` (trusted, not decoded), ``hit (sampled)`` (also analysed), or ``local``.""" + if not cache_hit: + return "local" + return "hit (sampled)" if is_sampled_hit(fingerprint) else "hit" + + def _compute_disagreements( local: dict[str, Any], external: dict[str, Any] ) -> dict[str, str | None]: diff --git a/backend/tests/test_features_cache_trust.py b/backend/tests/test_features_cache_trust.py new file mode 100644 index 00000000..bd4b5e37 --- /dev/null +++ b/backend/tests/test_features_cache_trust.py @@ -0,0 +1,178 @@ +"""ADR-0144: a community cache hit is trusted, and checked by sample. + +The end-to-end tests run ``run_track_features`` against the test database with every outside +service replaced — the cache, the fingerprint, artwork, AcoustID and MusicBrainz — and with the +decoder rigged to fail, so a trusted hit that decodes the file fails the test rather than only +running slowly. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from sqlalchemy import delete, select + +from app.db.models import Track, TrackAnalysis, TrackStatus +from app.db.session import sync_session_maker +from app.services.tasks import analysis_pipeline as pipeline +from app.services.tasks.analysis_pipeline import HIT_SAMPLE_RATE, features_path, is_sampled_hit + + +def _fingerprint(sampled: bool) -> str: + """A fingerprint that the sampling rule does, or does not, pick.""" + for i in range(10_000): + fp = f"AQAD{i:06d}" + if is_sampled_hit(fp) == sampled: + return fp + raise AssertionError("no fingerprint found") + + +# --- The rule ------------------------------------------------------------------------------------ + + +def test_the_sample_is_the_same_on_every_run(): + fp = _fingerprint(sampled=True) + assert all(is_sampled_hit(fp) for _ in range(5)) + + +def test_about_one_hit_in_fifty_is_sampled(): + picked = sum(is_sampled_hit(f"AQAD-{i}") for i in range(20_000)) + assert 20_000 / HIT_SAMPLE_RATE * 0.8 < picked < 20_000 / HIT_SAMPLE_RATE * 1.2 + + +def test_a_miss_is_always_local_and_no_fingerprint_is_never_sampled(): + assert features_path(False, _fingerprint(sampled=True)) == "local" + assert features_path(True, _fingerprint(sampled=False)) == "hit" + assert features_path(True, _fingerprint(sampled=True)) == "hit (sampled)" + assert is_sampled_hit(None) is False + + +# --- The pipeline -------------------------------------------------------------------------------- + +CACHED = {"bpm": 128.0, "key": "A minor", "energy": 0.8, "danceability": 0.7} +LOCAL = {"bpm": 96.0, "key": "C major", "energy": 0.3, "danceability": 0.2} +CACHED_DETAIL = {"structural": {"boundary_confidence": 0.9}, "from": "cache"} +LOCAL_DETAIL = {"structural": {"boundary_confidence": 0.4}, "from": "local"} + + +@dataclass +class Harness: + decoded: list + contributed: list + + +@pytest.fixture +def harness(monkeypatch, tmp_path): + decoded: list = [] + contributed: list = [] + state = SimpleNamespace(hit=True, detail=True) + + class FakeCache: + async def lookup_features(self, fp): + if not state.hit: + return None + return SimpleNamespace(features=dict(CACHED), contributor_count=2) + + async def lookup_analysis_detail(self, fp): + return SimpleNamespace(detail=dict(CACHED_DETAIL)) if state.detail else None + + async def contribute_features(self, fp, features): + contributed.append(("features", fp)) + + async def contribute_analysis_detail(self, fp, detail): + contributed.append(("detail", fp)) + + def decode(path): + decoded.append(path) + return object(), 22050, {} + + import app.services.analysis as analysis + import app.services.app_settings as app_settings + import app.services.artwork as artwork + import app.services.community_cache as community_cache + import app.services.metadata.musicbrainz as musicbrainz + import app.services.track_analysis as track_analysis + + monkeypatch.setattr(artwork, "extract_and_save_artwork", lambda *a, **k: None) + monkeypatch.setattr(analysis, "precompute_shared", decode) + monkeypatch.setattr(analysis, "derive_features", lambda y, sr, shared, path: (dict(LOCAL), {"bpm": 0.9})) + monkeypatch.setattr(analysis, "identify_track", lambda path: {}) + monkeypatch.setattr(musicbrainz, "enrich_track", lambda **k: None) + monkeypatch.setattr(track_analysis, "run_cheap_sections", + lambda *a, **k: (dict(LOCAL_DETAIL), {"spectral_centroid_mean": 1500.0}, [])) + monkeypatch.setattr(community_cache, "get_community_cache_service", lambda **k: FakeCache()) + settings = SimpleNamespace(community_cache_enabled=True, community_cache_url="http://cache.test", + community_cache_contribute=True) + monkeypatch.setattr(app_settings, "get_app_settings_service", lambda: SimpleNamespace(get=lambda: settings)) + + audio = tmp_path / "track.flac" + audio.write_bytes(b"not really audio") + track_ids: list = [] + + def run(fingerprint: str, *, hit: bool = True, detail: bool = True): + state.hit, state.detail = hit, detail + monkeypatch.setattr(analysis, "generate_fingerprint", lambda path: (200, fingerprint)) + track_id = uuid4() + track_ids.append(track_id) + with sync_session_maker() as db: + db.add(Track(id=track_id, file_path=str(audio), file_hash=uuid4().hex, title="Song", + artist="Artist", album="Album", duration_seconds=200.0, + status=TrackStatus.ACTIVE)) + db.commit() + result = pipeline.run_track_features(str(track_id)) + with sync_session_maker() as db: + row = db.execute(select(TrackAnalysis).where(TrackAnalysis.track_id == track_id)).scalar_one() + db.expunge(row) + return result, row + + yield SimpleNamespace(run=run, h=Harness(decoded, contributed)) + + with sync_session_maker() as db: + db.execute(delete(TrackAnalysis).where(TrackAnalysis.track_id.in_(track_ids))) + db.execute(delete(Track).where(Track.id.in_(track_ids))) + db.commit() + + +def test_a_trusted_hit_is_stored_without_decoding_the_track(harness): + result, row = harness.run(_fingerprint(sampled=False)) + assert result["features_path"] == "hit" + assert harness.h.decoded == [], "point 1: a trusted hit is not decoded" + assert row.features_source == "community_cache" + assert row.bpm == CACHED["bpm"] and row.key == CACHED["key"] + assert row.local_features is None + + +def test_a_hits_cached_section_analysis_is_kept(harness): + # Before ADR-0144 this was fetched and then reset to None and recomputed. + _, row = harness.run(_fingerprint(sampled=False)) + assert row.analysis_detail == CACHED_DETAIL + + +def test_a_hit_without_cached_sections_leaves_them_to_the_backfill(harness): + _, row = harness.run(_fingerprint(sampled=False), detail=False) + assert harness.h.decoded == [] + assert row.analysis_detail is None, "point 3: the backfill selects analysis_detail IS NULL" + + +def test_a_sampled_hit_is_analysed_compared_and_not_contributed(harness): + result, row = harness.run(_fingerprint(sampled=True)) + assert result["features_path"] == "hit (sampled)" + assert len(harness.h.decoded) == 1 + assert row.bpm == CACHED["bpm"], "the cache's values stay primary" + assert row.local_features["bpm"] == LOCAL["bpm"] + assert "bpm_disagreement" in row.feature_confidence + assert row.analysis_detail == CACHED_DETAIL + assert harness.h.contributed == [], "point 2: a sampled track's local result is not contributed" + + +def test_a_miss_is_analysed_locally_and_contributed(harness): + result, row = harness.run(_fingerprint(sampled=False), hit=False) + assert result["features_path"] == "local" + assert len(harness.h.decoded) == 1 + assert row.features_source == "local" + assert row.bpm == LOCAL["bpm"] + assert row.analysis_detail == LOCAL_DETAIL + assert ("features", _fingerprint(sampled=False)) in harness.h.contributed diff --git a/docs/decisions/ADR-0144-a-community-cache-hit-is-trusted-and-checked-by-sample.md b/docs/decisions/ADR-0144-a-community-cache-hit-is-trusted-and-checked-by-sample.md index d53eca8d..fa2cce9d 100644 --- a/docs/decisions/ADR-0144-a-community-cache-hit-is-trusted-and-checked-by-sample.md +++ b/docs/decisions/ADR-0144-a-community-cache-hit-is-trusted-and-checked-by-sample.md @@ -5,7 +5,29 @@ Status: accepted Date: 2026-10-02 Implementation: -- **Accepted 2026-10-02**, as written. Not yet built. +- **Accepted 2026-10-02**, as written. **Built 2026-10-04**, released in the version after + `v0.2.0-beta17`. + - `tasks/analysis_pipeline.py`: `features_path()` chooses `hit`, `hit (sampled)` or `local`; + `is_sampled_hit()` takes the first eight bytes of the fingerprint's SHA-256 modulo + `HIT_SAMPLE_RATE` (50). A trusted hit never calls `precompute_shared`. A sampled hit keeps the + cache's features and deep scalars as primary, stores the local ones in `local_features` and + the disagreements in `feature_confidence`, and is not contributed. A miss is unchanged. The + worker logs `Features path for …: `, and the task result carries `features_path`. + - `tests/test_features_cache_trust.py` runs `run_track_features` against the test database with + the decoder rigged to record every call: a trusted hit decodes nothing. Five of its eight tests + fail on the code this replaced. + - **The cache's section analysis was being thrown away.** The pipeline fetched it into + `analysis_detail`, and the next block reset `analysis_detail = None` and recomputed it, so a hit + *with* cached detail decoded the track anyway, and the cached detail was never stored. Point 3 + now holds for the first time: cached detail is kept, and only a hit without it is left to the + backfill. + - **Point 1 named the wrong `features_source`.** It said a hit is written as + `community_cache:recording` or `community_cache:hash` "as today". That split is ADR-0119's and + exists for embeddings only; the features lookup is by fingerprint alone and writes + `community_cache`, which is what it still writes. Asking the corpus by recording for features + too is not part of this ADR. + - The MCP tool description that named ReccoBeats (`services/llm/tools.py`) now describes the + cache and the sample. Extends [ADR-0119](ADR-0119-the-corpus-is-asked-by-recording-before-it-is-asked-by-hash.md) and [ADR-0138](ADR-0138-a-desktop-server-yields-to-its-owner.md)