album completeness: fix excluded-sibling missing list + O(N) grouping (follow-up to #936)

#936 added fragmented-row grouping. two follow-ups found in review:

1) BUG: an excluded canonical sibling (a row pinned to the same canonical edition whose tracks
   fail the strict fragment match) was emitted with canonical_items but no owned set, so
   _build_missing_tracks flagged the ENTIRE tracklist as missing — including tracks the row
   already owns (e.g. a 3-track fragment of a 12-track edition reported 12 missing, not 9).
   now compute that sibling's own owned slots from its local tracks (shared
   _owned_reference_for_tracks helper, same logic as the anchor) so it reports only what it's
   actually missing and the count stays internally consistent.

2) PERF: _build_candidate_groups rescanned all albums for every canonical group — O(G*N), which
   degrades badly once many editions are pinned (fine today at G=1, latent at scale). invert to
   a single O(N) pass that assigns each row to its unique group; identical 'exactly one match'
   semantics. also added a Stop/Pause check in _prepare_work_items, which now front-loads the
   canonical lookups + matching.

3 new tests (excluded-sibling reports only its missing tracks; ambiguous candidate stays
   independent; unambiguous candidate joins) — 17 completeness + 123 repair tests green.

core/repair_jobs/album_completeness.py

@@ -435,6 +435,13 @@ class AlbumCompletenessJob(RepairJob):
         work_items = []
 
         for group in groups:
+            # This prep phase front-loads the canonical API lookups + track
+            # matching, so honour Stop/Pause here too — otherwise a stop would
+            # be ignored until every group is processed.
+            check_stop = getattr(context, 'check_stop', None)
+            if callable(check_stop) and check_stop():
+                break
+
             anchor = group['anchor']
             members = group['members']
             canonical_source = anchor.get('canonical_source') or ''
@@ -462,19 +469,13 @@ class AlbumCompletenessJob(RepairJob):
                         [],
                     )
 
-                    anchor_reference, _ = self._match_tracks(
+                    # The anchor's persisted disc/track slots remain
+                    # authoritative even when titles or durations drift.
+                    anchor_reference = self._owned_reference_for_tracks(
                         canonical_items,
                         anchor_tracks,
                         str(canonical_source),
                     )
-                    # The anchor's persisted disc/track slots remain
-                    # authoritative even when titles or durations drift.
-                    anchor_reference.update(
-                        self._reference_slots_for_local_tracks(
-                            canonical_items,
-                            anchor_tracks,
-                        )
-                    )
 
                     included = [anchor]
                     excluded = []
@@ -537,20 +538,43 @@ class AlbumCompletenessJob(RepairJob):
                     })
 
                     for member in excluded:
-                        work_items.append(
+                        # An excluded member that is itself pinned to this
-                            self._independent_work_item(
+                        # canonical edition is still evaluated against it (no
-                                member,
+                        # fallback). It must report only the tracks it *doesn't*
-                                canonical_items=(
+                        # own — compute its own owned slots from its local
-                                    canonical_items
+                        # tracks, exactly like the anchor, instead of leaving
-                                    if self._same_canonical_pair(
+                        # the set empty (which would flag the whole tracklist as
-                                        member,
+                        # missing, including tracks the row already has).
-                                        canonical_source,
+                        if self._same_canonical_pair(
-                                        canonical_album_id,
+                            member,
-                                    )
+                            canonical_source,
-                                    else None
+                            canonical_album_id,
-                                ),
+                        ):
+                            member_tracks = local_by_album.get(
+                                str(member['album_id']),
+                                [],
+                            )
+                            member_owned = (
+                                self._owned_reference_for_tracks(
+                                    canonical_items,
+                                    member_tracks,
+                                    str(canonical_source),
+                                )
+                            )
+                            work_items.append(
+                                self._independent_work_item(
+                                    member,
+                                    canonical_items=canonical_items,
+                                    owned_reference_indexes=member_owned,
+                                    effective_actual_count=len(
+                                        member_owned
+                                    ),
+                                )
+                            )
+                        else:
+                            work_items.append(
+                                self._independent_work_item(member)
                             )
-                        )
                     continue
 
                 # The canonical lookup was attempted and returned no usable
@@ -581,7 +605,13 @@ class AlbumCompletenessJob(RepairJob):
         work_items.sort(key=lambda item: item['_scan_order'])
         return work_items
 
-    def _independent_work_item(self, row, canonical_items=None):
+    def _independent_work_item(
+        self,
+        row,
+        canonical_items=None,
+        owned_reference_indexes=None,
+        effective_actual_count=None,
+    ):
         item = {
             'row': row,
             'related_album_ids': [row['album_id']],
@@ -589,6 +619,10 @@ class AlbumCompletenessJob(RepairJob):
         }
         if canonical_items is not None:
             item['canonical_items'] = canonical_items
+        if owned_reference_indexes is not None:
+            item['owned_reference_indexes'] = owned_reference_indexes
+        if effective_actual_count is not None:
+            item['effective_actual_count'] = effective_actual_count
         return item
 
     def _same_canonical_pair(
@@ -651,9 +685,37 @@ class AlbumCompletenessJob(RepairJob):
                     )
                 ].add(key)
 
+        # Assign each non-canonical row to a group in ONE pass over the albums
+        # (O(N)), instead of rescanning every album for every group (O(G*N),
+        # which degrades badly once many editions are pinned). A row joins a
+        # group only when its stored IDs resolve to exactly one canonical group
+        # — identical to the old `matches == {key}` rule, just computed once.
         assigned_candidate_ids = set()
-        groups = []
+        candidates_by_group = defaultdict(list)
+        for row in albums:
+            row_id = str(row['album_id'])
+            if row_id in canonical_row_ids:
+                continue
 
+            artist_id = str(row.get('artist_id') or '')
+            matches = set()
+            for source, source_id in (
+                self._source_ids_from_row(row).items()
+            ):
+                if source_id:
+                    matches.update(
+                        alias_to_groups.get(
+                            (artist_id, source, str(source_id)),
+                            set(),
+                        )
+                    )
+
+            if len(matches) == 1:
+                (key,) = tuple(matches)
+                candidates_by_group[key].append(row)
+                assigned_candidate_ids.add(row_id)
+
+        groups = []
         for key, anchor_rows in canonical_groups.items():
             anchor = max(
                 anchor_rows,
@@ -663,36 +725,7 @@ class AlbumCompletenessJob(RepairJob):
                     -int(row.get('_scan_order') or 0),
                 ),
             )
-            members = list(anchor_rows)
+            members = list(anchor_rows) + candidates_by_group.get(key, [])
-
-            for row in albums:
-                row_id = str(row['album_id'])
-                if (
-                    row_id in canonical_row_ids
-                    or row_id in assigned_candidate_ids
-                ):
-                    continue
-
-                artist_id = str(row.get('artist_id') or '')
-                matches = set()
-                for source, source_id in (
-                    self._source_ids_from_row(row).items()
-                ):
-                    if source_id:
-                        matches.update(
-                            alias_to_groups.get(
-                                (
-                                    artist_id,
-                                    source,
-                                    str(source_id),
-                                ),
-                                set(),
-                            )
-                        )
-
-                if matches == {key}:
-                    members.append(row)
-                    assigned_candidate_ids.add(row_id)
 
             groups.append({
                 'anchor': anchor,
@@ -890,6 +923,29 @@ class AlbumCompletenessJob(RepairJob):
                 matched.add(indexes[0])
         return matched
 
+    def _owned_reference_for_tracks(
+        self,
+        reference_items,
+        local_tracks,
+        reference_source,
+    ):
+        """Reference indexes a set of local tracks owns: fuzzy one-to-one
+        matches plus the persisted disc/track slots (authoritative even when
+        titles or durations drift). Used for the anchor and for any excluded
+        sibling that is still evaluated against this canonical edition."""
+        owned, _ = self._match_tracks(
+            reference_items,
+            local_tracks,
+            reference_source,
+        )
+        owned.update(
+            self._reference_slots_for_local_tracks(
+                reference_items,
+                local_tracks,
+            )
+        )
+        return owned
+
     def _track_match_score(
         self,
         reference,

tests/test_album_completeness_fragmented_rows.py

@@ -353,6 +353,138 @@ def test_shared_id_without_track_match_stays_independent(
     ]
 
 
+def test_excluded_canonical_sibling_reports_only_its_missing_tracks(
+    monkeypatch,
+):
+    """A second row pinned to the SAME canonical edition whose tracks fail the
+    strict fragment match is still evaluated against that edition — but it must
+    report only the tracks it does NOT own, not the whole tracklist. Regression
+    for the excluded-sibling bug (it previously flagged every canonical track as
+    missing, including the ones the row already had)."""
+    db = _SharedMemoryDB()
+    db.insert_artist("artist-1", "Artist")
+    # Anchor: complete, titles match the canonical edition.
+    db.insert_album(
+        "anchor",
+        "artist-1",
+        "Album",
+        canonical_source="deezer",
+        canonical_album_id="canonical-release",
+    )
+    for number in range(1, 6):
+        db.insert_track("anchor", number, f"Canonical Track {number}")
+
+    # Sibling: same canonical pair, owns tracks 1-3 by NUMBER but with blank
+    # titles → fails the strict fragment match → excluded from the anchor group.
+    db.insert_album(
+        "sibling",
+        "artist-1",
+        "Album",
+        canonical_source="deezer",
+        canonical_album_id="canonical-release",
+    )
+    for number in range(1, 4):
+        db.insert_track("sibling", number, "")
+
+    monkeypatch.setattr(
+        album_completeness_module,
+        "get_album_tracks_for_source",
+        lambda source, album_id: _canonical_tracks(5),
+    )
+    monkeypatch.setattr(
+        album_completeness_module,
+        "get_primary_source",
+        lambda: "deezer",
+    )
+    monkeypatch.setattr(
+        album_completeness_module,
+        "get_source_priority",
+        lambda primary: ["deezer"],
+    )
+
+    findings = []
+    AlbumCompletenessJob().scan(_context(db, findings))
+
+    sibling_finding = next(
+        f for f in findings if f["entity_id"] == "sibling"
+    )
+    details = sibling_finding["details"]
+    # Owns tracks 1-3 (by number) → "3 of 5", and only 2 missing (4 & 5),
+    # NOT the full 5 — and the count stays internally consistent.
+    assert details["actual_tracks"] == 3
+    assert details["expected_tracks"] == 5
+    missing_numbers = sorted(
+        t["track_number"] for t in details["missing_tracks"]
+    )
+    assert missing_numbers == [4, 5]
+
+
+def test_candidate_matching_two_canonical_groups_stays_independent():
+    """A candidate whose shared ID resolves to MORE THAN ONE canonical group is
+    ambiguous and must not be fused into either. Locks the `len(matches) == 1`
+    rule the one-pass grouping relies on. (`_build_candidate_groups` is pure, so
+    drive it directly with album dicts.)"""
+    def _album(album_id, order, **extra):
+        base = {
+            'album_id': album_id, 'artist_id': 'artist-1',
+            'album_title': album_id, 'actual_count': 1, '_scan_order': order,
+            'spotify_album_id': '', 'itunes_album_id': '', 'deezer_album_id': '',
+            'discogs_album_id': '', 'hydrabase_album_id': '',
+            'musicbrainz_album_id': '', 'canonical_source': '',
+            'canonical_album_id': '',
+        }
+        base.update(extra)
+        return base
+
+    # Two distinct canonical editions (same artist) that share spotify-X, plus a
+    # candidate that also carries spotify-X → it resolves to BOTH groups.
+    albums = [
+        _album('anchor-a', 0, spotify_album_id='shared-X',
+               canonical_source='deezer', canonical_album_id='canonical-a'),
+        _album('anchor-b', 1, spotify_album_id='shared-X',
+               canonical_source='deezer', canonical_album_id='canonical-b'),
+        _album('candidate', 2, spotify_album_id='shared-X'),
+    ]
+
+    groups = AlbumCompletenessJob()._build_candidate_groups(albums)
+    candidate_groups = [
+        g for g in groups
+        if any(m['album_id'] == 'candidate' for m in g['members'])
+    ]
+    # Candidate is its OWN singleton group, never fused into A or B.
+    assert len(candidate_groups) == 1
+    assert [m['album_id'] for m in candidate_groups[0]['members']] == [
+        'candidate'
+    ]
+
+
+def test_unambiguous_candidate_joins_its_single_group():
+    """The complement: a candidate that resolves to exactly one canonical group
+    joins it (the one-pass path produces the same membership as before)."""
+    def _album(album_id, order, **extra):
+        base = {
+            'album_id': album_id, 'artist_id': 'artist-1',
+            'album_title': album_id, 'actual_count': 1, '_scan_order': order,
+            'spotify_album_id': '', 'itunes_album_id': '', 'deezer_album_id': '',
+            'discogs_album_id': '', 'hydrabase_album_id': '',
+            'musicbrainz_album_id': '', 'canonical_source': '',
+            'canonical_album_id': '',
+        }
+        base.update(extra)
+        return base
+
+    albums = [
+        _album('anchor', 0, spotify_album_id='shared-X',
+               canonical_source='deezer', canonical_album_id='canonical-a'),
+        _album('candidate', 1, spotify_album_id='shared-X'),
+    ]
+    groups = AlbumCompletenessJob()._build_candidate_groups(albums)
+    assert len(groups) == 1
+    assert sorted(m['album_id'] for m in groups[0]['members']) == [
+        'anchor', 'candidate'
+    ]
+
+
 def test_fragment_grouping_never_crosses_artist_boundary(
     monkeypatch,
 ):