feat: detect within-source duplicate likes

README.md

@@ -160,6 +160,7 @@ uv run likesurgeon sync --drift-min-confidence 1.0 # only apply 100%-confidence
 | `possible_pointer_drift` | YouTube like → unlike | `videos.rate("like")` then `videos.rate("none")` | — | `confidence >= --drift-min-confidence` (default 0.95) | Re-points the YouTube like at the YT Music track's video_id. Like first, then unlike — a partial failure leaves a duplicate like (cleaned up on the next run) instead of losing the original. |
 | `ytmusic_only` | — | — | — | never | Informational only. Reverse-direction sync (YT Music → YouTube) is out of scope for 0.4. |
 | `metadata_drift` | — | — | — | never | Informational only. Title/artist drift is signal for the user, not a write target. |
+| `duplicate_in_source` | — | — | — | never | Informational. Acts on this finding deferred to 0.5 (dedupe command). |
 
 #### State model
 

docs/ARCHITECTURE.md

@@ -69,7 +69,7 @@ Five tables, all SQLite-backed at `~/.like-surgeon/like-surgeon.sqlite`:
   - `is_music_candidate` / `music_candidate_score` / `music_candidate_reason` — populated for YouTube rows only (ytmusic is always music).
   - `is_available` / `unavailable_reason` — ghost detection result, populated for YouTube rows only. Reasons: `deleted`, `rejected`, `region_blocked` (0.3.1+).
 - **`diagnoses`** — one row per `compare-likes` run. References both source snapshots.
-- **`diagnosis_items`** — findings: `issue_type` ∈ {`possibly_missing_from_ytmusic`, `possible_pointer_drift`, `ytmusic_only`, `unavailable_video`, `metadata_drift`}, plus `confidence`, `reason`, optional `source_track_id` / `related_track_id`.
+- **`diagnosis_items`** — findings: `issue_type` ∈ {`possibly_missing_from_ytmusic`, `possible_pointer_drift`, `ytmusic_only`, `unavailable_video`, `metadata_drift`, `duplicate_in_source`}, plus `confidence`, `reason`, optional `source_track_id` / `related_track_id`.
 
 **Lightweight in-place migration only.** [`_migrate_in_place`](../src/likesurgeon/db.py) (called from `make_engine` on every CLI run) idempotently issues `ALTER TABLE ... ADD COLUMN` for nullable columns added after a table was first created — that's how 0.3's `is_available` / `unavailable_reason` reach pre-0.3 DBs without forcing a re-scan. There is no Alembic-style framework, so any **breaking** change during 0.x (renamed columns, type changes, FK reshuffles) requires dropping `~/.like-surgeon/like-surgeon.sqlite` and re-scanning. The DB only holds derived data; no original-source state is lost.
 

src/likesurgeon/cli.py

@@ -12,7 +12,7 @@
 from sqlalchemy.orm import Session, sessionmaker
 
 from . import __version__
-from .compare import CompareInput, CompareResult, compare_likes
+from .compare import CompareInput, CompareResult, compare_likes, dedupe_by_video_id
 from .config import Config, InvalidRegionError, _validate_region
 from .db import init_db, make_engine, make_session_factory, session_scope
 from .diff import diff_snapshots
@@ -446,7 +446,8 @@ def _render_source(label: str, sh) -> None:
             f"{d.pointer_drift} pointer drift · "
             f"{d.ytmusic_only} YT Music only · "
             f"{d.unavailable_videos} unavailable videos · "
-            f"{d.metadata_drift} metadata drift"
+            f"{d.metadata_drift} metadata drift · "
+            f"{d.duplicate_in_source} duplicate likes"
         )
 
     if report.match_rate_percent is None:
@@ -462,12 +463,23 @@ class _PipelineResult:
     """Compound return value for `_compare_and_persist` so the CLI wrapper
     can render the existing summary table AND the two new finding-type
     rows from a single call. Tests typically only need `.diagnosis_id`.
+
+    Raw counts (``raw_ytmusic_count`` / ``raw_youtube_total_count`` /
+    ``raw_youtube_music_count``) reflect snapshot row counts as scanned —
+    the "what's actually in your account" view, including within-source
+    duplicates. ``compare_result`` counts are post-canonicalization
+    (each ``video_id`` collapsed to one row), matching what the matcher
+    actually saw.
     """
 
     diagnosis_id: int
     compare_result: CompareResult
     unavailable_count: int
     drift_count: int
+    duplicate_count: int
+    raw_ytmusic_count: int
+    raw_youtube_total_count: int
+    raw_youtube_music_count: int
 
 
 def _compare_and_persist(session: Session) -> _PipelineResult:
@@ -484,6 +496,7 @@ def _compare_and_persist(session: Session) -> _PipelineResult:
     """
     from .diagnosis import (
         DiagnosisInput,
+        build_duplicate_in_source_items,
         build_metadata_drift_items,
         build_unavailable_video_items,
         create_diagnosis,
@@ -511,10 +524,17 @@ def _compare_and_persist(session: Session) -> _PipelineResult:
             code=2,
         )
 
-    yt_items = get_snapshot_items(session, yt_snap.id)
-    ytm_items = get_snapshot_items(session, ytm_snap.id)
+    yt_items_raw = get_snapshot_items(session, yt_snap.id)
+    ytm_items_raw = get_snapshot_items(session, ytm_snap.id)
+
+    # Stage A0 — within-source duplicate detection (pre-matching).
+    # We retain the raw rows so the duplicate-finding builder sees every
+    # occurrence; canonicalized streams feed the matcher and drift detector
+    # below so each ``video_id`` is counted at most once per source.
+    yt_items = dedupe_by_video_id(yt_items_raw)
+    ytm_items = dedupe_by_video_id(ytm_items_raw)
 
-    # Stage A — existing cross-source matcher.
+    # Stage A — existing cross-source matcher (over canonicalized items).
     cmp_result = compare_likes(CompareInput(ytmusic=ytm_items, youtube=yt_items))
 
     # Stage B — persist Diagnosis + the existing 0.2 finding buckets.
@@ -527,20 +547,34 @@ def _compare_and_persist(session: Session) -> _PipelineResult:
         ),
     )
 
+    # Stage B1 — append within-source duplicate findings now that we have
+    # ``diagnosis.id``. Builder sees the RAW snapshot items so every
+    # duplicate occurrence is counted.
+    dup_items_yt = build_duplicate_in_source_items(diagnosis.id, yt_items_raw, YOUTUBE_LIKED_VIDEOS)
+    dup_items_ytm = build_duplicate_in_source_items(
+        diagnosis.id, ytm_items_raw, YTMUSIC_LIKED_SONGS
+    )
+    for it in (*dup_items_yt, *dup_items_ytm):
+        session.add(it)
+    duplicate_total = len(dup_items_yt) + len(dup_items_ytm)
+
     # Stage C — append ghost findings.
     ghost_items = build_unavailable_video_items(diagnosis.id, yt_items)
     for it in ghost_items:
         session.add(it)
 
     # Stage D — append drift findings per source against (latest, prev).
+    # Drift re-fetches from DB inside the loop, so canonicalization must
+    # be reapplied here (rebinding ``yt_items`` / ``ytm_items`` above
+    # doesn't affect these fresh fetches).
     drift_total = 0
     for source in (YOUTUBE_LIKED_VIDEOS, YTMUSIC_LIKED_SONGS):
         snaps = latest_snapshots_for_source(session, source, limit=2)
         if len(snaps) < 2:
             continue
         curr_snap, prev_snap = snaps[0], snaps[1]
-        curr_items = get_snapshot_items(session, curr_snap.id)
-        prev_items = get_snapshot_items(session, prev_snap.id)
+        curr_items = dedupe_by_video_id(get_snapshot_items(session, curr_snap.id))
+        prev_items = dedupe_by_video_id(get_snapshot_items(session, prev_snap.id))
         findings = detect_drift(prev_items, curr_items, source=source)
         drift_items = build_metadata_drift_items(diagnosis.id, findings, curr_items)
         drift_total += len(drift_items)
@@ -553,6 +587,10 @@ def _compare_and_persist(session: Session) -> _PipelineResult:
         compare_result=cmp_result,
         unavailable_count=len(ghost_items),
         drift_count=drift_total,
+        duplicate_count=duplicate_total,
+        raw_ytmusic_count=len(ytm_items_raw),
+        raw_youtube_total_count=len(yt_items_raw),
+        raw_youtube_music_count=sum(1 for it in yt_items_raw if it.is_music_candidate),
     )
 
 
@@ -568,9 +606,9 @@ def compare_likes_cmd() -> None:
     table = Table(title="compare-likes summary")
     table.add_column("Bucket")
     table.add_column("Count", justify="right")
-    table.add_row("YouTube Music liked songs", str(result.ytmusic_count))
-    table.add_row("YouTube liked videos (total)", str(result.youtube_total_count))
-    table.add_row("YouTube liked videos (music-like)", str(result.youtube_music_count))
+    table.add_row("YouTube Music liked songs", str(outcome.raw_ytmusic_count))
+    table.add_row("YouTube liked videos (total)", str(outcome.raw_youtube_total_count))
+    table.add_row("YouTube liked videos (music-like)", str(outcome.raw_youtube_music_count))
     table.add_row("Matched (any stage)", str(len(result.matched)))
     table.add_row(
         "Possibly missing from YT Music",
@@ -583,6 +621,7 @@ def compare_likes_cmd() -> None:
     table.add_row("Pointer-drift candidates", str(len(result.pointer_drift_candidates)))
     table.add_row("Unavailable videos (ghost)", str(outcome.unavailable_count))
     table.add_row("Metadata drift candidates", str(outcome.drift_count))
+    table.add_row("Duplicate likes (within-source)", str(outcome.duplicate_count))
     console.print(table)
     console.print("Run [cyan]likesurgeon issues[/cyan] for the full per-item breakdown.")
 
@@ -596,7 +635,8 @@ def issues(
             help=(
                 "Filter findings by issue type. One of: "
                 "possibly_missing_from_ytmusic | possible_pointer_drift | "
-                "ytmusic_only | unavailable_video | metadata_drift."
+                "ytmusic_only | unavailable_video | metadata_drift | "
+                "duplicate_in_source."
             ),
         ),
     ] = None,

src/likesurgeon/compare.py

@@ -24,9 +24,10 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Sequence
 from dataclasses import dataclass, field
 from enum import StrEnum
-from typing import Any, Protocol
+from typing import Any, Protocol, TypeVar
 
 from rapidfuzz import fuzz
 
@@ -43,6 +44,44 @@ class _Itemish(Protocol):
     is_music_candidate: bool | None
 
 
+class _VideoIdPositioned(Protocol):
+    """Minimal duck-typed contract for ``dedupe_by_video_id``.
+
+    Keeps ``compare.py`` decoupled from the SQLAlchemy ORM — both real
+    ``SnapshotItem`` rows and lightweight test stand-ins satisfy this.
+    """
+
+    video_id: str | None
+    position: int
+
+
+_T = TypeVar("_T", bound=_VideoIdPositioned)
+
+
+def dedupe_by_video_id(items: Sequence[_T]) -> list[_T]:
+    """Return one row per ``video_id`` (lowest-position winner), NULL-safe.
+
+    Contract:
+      - Rows with falsy ``video_id`` pass through unchanged (no identity).
+      - For each non-null ``video_id`` group, keep the row with the
+        smallest ``position``; discard the rest.
+      - Output is sorted by ``position`` ascending (position-stable).
+      - Idempotent: running twice returns an equal list.
+    """
+    kept: dict[str, _T] = {}
+    no_id: list[_T] = []
+    for it in items:
+        if not it.video_id:
+            no_id.append(it)
+            continue
+        existing = kept.get(it.video_id)
+        if existing is None or it.position < existing.position:
+            kept[it.video_id] = it
+    combined: list[_T] = list(kept.values()) + no_id
+    combined.sort(key=lambda x: x.position)
+    return combined
+
+
 def _normalize_artists(value: Any) -> list[str]:
     """Return a flat ``list[str]`` regardless of source-shape.
 

src/likesurgeon/diagnosis.py

@@ -16,6 +16,7 @@
 ISSUE_YTMUSIC_ONLY = "ytmusic_only"
 ISSUE_UNAVAILABLE_VIDEO = "unavailable_video"
 ISSUE_METADATA_DRIFT = "metadata_drift"
+ISSUE_DUPLICATE_IN_SOURCE = "duplicate_in_source"
 
 
 @dataclass(frozen=True)
@@ -178,6 +179,49 @@ def build_metadata_drift_items(
     return out
 
 
+def build_duplicate_in_source_items(
+    diagnosis_id: int,
+    snapshot_items: list[SnapshotItem],
+    source: str,
+) -> list[DiagnosisItem]:
+    """Build DiagnosisItem rows for within-source duplicate ``video_id`` groups.
+
+    ``confidence=1.0`` because a duplicate row is a deterministic fact about
+    the snapshot — there's no probability to surface. Filtering with
+    ``--min-confidence`` should never hide a real duplicate.
+
+    Rows with falsy ``video_id`` are skipped (no identity to dedupe). For
+    each duplicated ``video_id`` group, the group is sorted by ``position``
+    ascending and ``source_track_id`` resolves to the position-1 row's
+    ``track_id`` so output is deterministic regardless of caller input order.
+    """
+    groups: dict[str, list[SnapshotItem]] = {}
+    for item in snapshot_items:
+        if not item.video_id:
+            continue
+        groups.setdefault(item.video_id, []).append(item)
+
+    out: list[DiagnosisItem] = []
+    for group in groups.values():
+        if len(group) < 2:
+            continue
+        sorted_group = sorted(group, key=lambda it: it.position)
+        positions = ", ".join(str(it.position) for it in sorted_group)
+        reason = f"appears {len(sorted_group)} times in {source} snapshot (positions: {positions})"
+        out.append(
+            DiagnosisItem(
+                diagnosis_id=diagnosis_id,
+                issue_type=ISSUE_DUPLICATE_IN_SOURCE,
+                confidence=1.0,
+                reason=reason,
+                source_track_id=sorted_group[0].track_id,
+                related_track_id=None,
+                status="open",
+            )
+        )
+    return out
+
+
 def latest_diagnosis(session: Session) -> Diagnosis | None:
     stmt = select(Diagnosis).order_by(Diagnosis.created_at.desc(), Diagnosis.id.desc()).limit(1)
     return session.scalar(stmt)

src/likesurgeon/doctor.py

@@ -10,6 +10,7 @@
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
+from .compare import dedupe_by_video_id
 from .diff import DiffResult, diff_snapshots
 from .models import Snapshot
 from .snapshot import get_snapshot_items, latest_snapshot
@@ -36,6 +37,7 @@ class DiagnosisSummary:
     ytmusic_only: int
     unavailable_videos: int
     metadata_drift: int
+    duplicate_in_source: int
 
 
 @dataclass(frozen=True)
@@ -80,6 +82,7 @@ def _source_health(session: Session, source: str) -> SourceHealth:
 
 def _latest_diagnosis_summary(session: Session) -> tuple[DiagnosisSummary | None, float | None]:
     from .diagnosis import (
+        ISSUE_DUPLICATE_IN_SOURCE,
         ISSUE_METADATA_DRIFT,
         ISSUE_POINTER_DRIFT,
         ISSUE_POSSIBLY_MISSING_FROM_YTMUSIC,
@@ -100,6 +103,7 @@ def _latest_diagnosis_summary(session: Session) -> tuple[DiagnosisSummary | None
         ISSUE_YTMUSIC_ONLY: 0,
         ISSUE_UNAVAILABLE_VIDEO: 0,
         ISSUE_METADATA_DRIFT: 0,
+        ISSUE_DUPLICATE_IN_SOURCE: 0,
     }
     for it in items:
         if it.issue_type in counts:
@@ -111,13 +115,20 @@ def _latest_diagnosis_summary(session: Session) -> tuple[DiagnosisSummary | None
         ytmusic_only=counts[ISSUE_YTMUSIC_ONLY],
         unavailable_videos=counts[ISSUE_UNAVAILABLE_VIDEO],
         metadata_drift=counts[ISSUE_METADATA_DRIFT],
+        duplicate_in_source=counts[ISSUE_DUPLICATE_IN_SOURCE],
     )
 
     # Health: match_rate = (yt_music_candidates - unmatched) / yt_music_candidates.
     # The "unmatched" count is exactly the high-priority bucket from this run.
+    #
+    # Denominator must use the same canonicalized view the diagnosis used —
+    # ``compare`` dedupes by video_id before matching, so the unmatched count
+    # is per unique video_id, not per raw row. Mixing raw denominator with
+    # canonicalized numerator would falsely inflate the score whenever the
+    # YouTube snapshot contains within-source duplicates.
     if diag.youtube_snapshot_id is None:
         return summary, None
-    yt_items = get_snapshot_items(session, diag.youtube_snapshot_id)
+    yt_items = dedupe_by_video_id(get_snapshot_items(session, diag.youtube_snapshot_id))
     yt_music_count = sum(1 for it in yt_items if it.is_music_candidate)
     if yt_music_count == 0:
         return summary, None

src/likesurgeon/models.py

@@ -160,6 +160,9 @@ class DiagnosisItem(Base):
         ``videos.list``.
       - ``metadata_drift`` — the same ``video_id`` appears in two snapshots
         of one source with meaningfully different title or artists.
+      - ``duplicate_in_source`` — the same ``video_id`` appears more than
+        once in a single snapshot (ytmusic accumulates these over time).
+        Informational; actual dedup ships as a separate command in 0.5.
     """
 
     __tablename__ = "diagnosis_items"

src/likesurgeon/sync.py

@@ -27,9 +27,12 @@
 from sqlalchemy.orm import Session
 
 from .diagnosis import (
+    ISSUE_DUPLICATE_IN_SOURCE,
+    ISSUE_METADATA_DRIFT,
     ISSUE_POINTER_DRIFT,
     ISSUE_POSSIBLY_MISSING_FROM_YTMUSIC,
     ISSUE_UNAVAILABLE_VIDEO,
+    ISSUE_YTMUSIC_ONLY,
 )
 from .models import DiagnosisItem, SyncAttempt, Track
 from .youtube_client import YouTubeClient, YouTubeWriteError
@@ -121,8 +124,9 @@ def plan(
     for an explicit manual override ("I never want to act on this finding"
     — e.g. a private/deleted YouTube ghost that ``videos.rate`` can't
     unlike anyway, so retrying would just noise the audit log forever).
-    Findings of type ``ytmusic_only`` or ``metadata_drift`` are also
-    silently dropped (informational, not actionable in 0.4).
+    Findings of type ``ytmusic_only``, ``metadata_drift``, or
+    ``duplicate_in_source`` are also silently dropped (informational, not
+    actionable in 0.4 — within-source dedup is deferred to 0.5).
     """
     actions: list[PlannedAction] = []
     skips: list[SkipRecord] = []
@@ -201,7 +205,16 @@ def plan(
                 )
             )
 
-        # ytmusic_only / metadata_drift: silently ignored — no record at all.
+        elif item.issue_type in {
+            ISSUE_YTMUSIC_ONLY,
+            ISSUE_METADATA_DRIFT,
+            ISSUE_DUPLICATE_IN_SOURCE,
+        }:
+            # Informational findings — no record, no action.
+            continue
+
+        # Unknown future issue types fall through silently (no WARN channel
+        # — sync.plan stays pure).
 
     return actions, skips