feat: 0.5 ytm_dedupe — auto-clean ytmusic duplicate likes

CLAUDE.md

@@ -40,7 +40,8 @@ Feature work goes through a PR (squash merge to main, see PR #7 for the template
 ## Sync state model (when touching sync.py / models)
 
 - `DiagnosisItem.reason` is **never overwritten** — it's diagnosis-time evidence. Sync-side detail (errors, threshold notes, missing video_ids) goes on `SyncAttempt.reason`.
-- `DiagnosisItem.status` is terminal in two cases: `'applied'` (code-set when every API call for the action succeeded) and `'skipped'` (manual override for findings the user never wants to retry). Anything else (`'open'`) gets re-evaluated next run.
+- `DiagnosisItem.status` is terminal in two cases: `'applied'` (code-set when every API call for the action succeeded, with one carve-out: `ytm_dedupe` is non-idempotent and flips to `'applied'` after any attempt regardless of outcome to prevent auto-retry over-removal) and `'skipped'` (manual override for findings the user never wants to retry). Anything else (`'open'`) gets re-evaluated next run.
 - Drift fixes are like-then-unlike (fail-safe order). Both halves must succeed before flipping to `'applied'`.
 - Continue-on-error is the dispatcher contract — `rate_video` / `like_song` wrap all exceptions as typed write errors so the loop never aborts mid-run.
 - Before a scaled `sync` run (>~50 actions), re-scan with `scan youtube-likes` + `compare-likes`. Region restrictions in particular flip between days — acting on a stale diagnosis can unlike videos that have since become available again (real example: 94 region-blocked ghosts un-blocked themselves overnight in testing).
+- Duplicate dedupe (`ytm_dedupe`) is **one `rate_song(INDIFFERENT)` per finding, ytmusic-source + N=2 only**. `rate_song(LIKE)` is non-idempotent (every call appends to LM), so our 0.4 drift sync was the dup producer in the first place. Propagation is minute-scale; the hazard is running `scan ytmusic` → `compare-likes` → `sync` cycle before propagation settles (stale snapshot recreates the finding → over-remove). **Always re-scan ytmusic + compare-likes immediately before a dedupe sync** — a stale finding can also remove the only remaining LM entry if the dup was fixed manually or by propagation since the previous diagnosis.

README.md

@@ -160,12 +160,13 @@ 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). |
+| `duplicate_in_source` | YT Music unlike (-1 entry) | — | `rate_song("INDIFFERENT")` | ytmusic source, **N=2 only** | Removes one LM entry per finding. **Non-idempotent** (validated for N=2; N≥3 unvalidated → skip). Terminal after attempt — failures are NOT auto-retried within the same Diagnosis (avoids over-removing if the server processed but the client errored). Real failures self-correct via the next `compare-likes`. Propagation is minute-scale — wait a few minutes between a dedupe `sync` and the next `scan ytmusic` / `compare-likes`. |
 
 #### State model
 
 - Each HTTP call (or skip decision) writes one `SyncAttempt` row with `kind`, `status` (`applied`/`failed`/`skipped`), and a `reason`. The originating `DiagnosisItem.reason` (the diagnosis-time evidence) is **never overwritten** — sync detail lives on `SyncAttempt.reason` instead.
-- `DiagnosisItem.status` only flips to `'applied'` when every API call for the action succeeded. For drift that means BOTH halves. Anything else (failure, low-confidence skip) leaves it at `'open'` so the next `sync` re-evaluates it.
+- `DiagnosisItem.status` only flips to `'applied'` when every API call for the action succeeded — *except* for `ytm_dedupe`, which is non-idempotent and flips to `'applied'` after any attempt (success OR failure) to prevent auto-retry over-removal. For drift that means BOTH halves. For other failures and low-confidence skips, the status stays at `'open'` so the next `sync` re-evaluates.
+- **Re-scan ytmusic + compare-likes immediately before a dedupe sync.** A stale `duplicate_in_source` finding can remove the only remaining LM entry if the dup was already fixed manually or by propagation between diagnose and sync. Eventual consistency also runs the other way: avoid re-scanning for a few minutes *after* a dedupe sync, since a stale snapshot can recreate the same finding and the next sync over-removes.
 - Re-running `sync` is idempotent: applied items are skipped; failures and previously-skipped findings are re-tried (so lowering `--drift-min-confidence` will pick up borderline drifts on the next run).
 - **Permanent manual skip.** If you want to tell sync to *never* attempt a particular finding again — e.g. a private/deleted YouTube ghost that `videos.rate` rejects with 403/404 — set its `DiagnosisItem.status` to `'skipped'` directly (currently via SQL on `~/.like-surgeon/like-surgeon.sqlite`). The planner treats `'applied'` and `'skipped'` identically as terminal at item level.
 - Continue-on-error: a failure (quota exhausted, transport error, revoked token) records the per-item `failed` row and moves on. The run exits non-zero if any action failed.

src/likesurgeon/cli.py

@@ -800,10 +800,9 @@ def sync(
         if not yes and not typer.confirm("Proceed?", default=False):
             raise typer.Abort()
 
-        # Build the YouTube client only if the plan needs it. ytm-only
-        # runs (no yt_unlike / yt_relike actions) skip the scope check
-        # entirely so a user without the YouTube write scope can still
-        # apply the YT Music half.
+        # Always construct (cheap); only the write-scope check and write method calls are
+        # gated on needs_youtube.
+        # ytm_dedupe is ytmusic-only; not in needs_youtube
         needs_youtube = any(a.kind in {"yt_unlike", "yt_relike"} for a in actions)
         from .youtube_client import YouTubeClient
 

src/likesurgeon/models.py

@@ -162,7 +162,10 @@ class DiagnosisItem(Base):
         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.
+        Actionable via ``ytm_dedupe`` (ytmusic source, N=2 only) as of 0.5.
+        Non-idempotent — ``status='applied'`` is set after attempt regardless
+        of outcome (carve-out from the standard 'applied = full success'
+        invariant; see ``sync.execute()`` docstring).
     """
 
     __tablename__ = "diagnosis_items"
@@ -192,7 +195,7 @@ class SyncAttempt(Base):
     """One row per ``sync`` API call (or skip decision).
 
     ``kind`` is one of ``yt_unlike``, ``ytm_like``, ``yt_relike_like``,
-    ``yt_relike_unlike`` — the two halves of a drift fix get separate rows
+    ``yt_relike_unlike``, ``ytm_dedupe`` — the two halves of a drift fix get separate rows
     so the audit trail stays atomic per HTTP call. ``status`` is one of
     ``applied``, ``failed``, ``skipped``. ``reason`` carries sync-side
     detail (error message, threshold note, missing video_id, etc.) — the

src/likesurgeon/sync.py

@@ -12,14 +12,17 @@
   * ``DiagnosisItem.status`` flips to ``"applied"`` only when every API
     call for the action succeeded (drift = both halves). Anything else
     (failure, skip) leaves it as ``"open"`` so the next ``sync`` run
-    re-evaluates it.
+    re-evaluates it. Exception: ``ytm_dedupe`` is non-idempotent and
+    flips to ``"applied"`` after any attempt (success or failure) to
+    prevent auto-retry over-removal.
   * Per-action commit cadence: a crash mid-run preserves prior actions'
     SyncAttempt rows AND any status updates already committed. Drift's
     two HTTP calls count as one action (one commit).
 """
 
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass
 from typing import Literal
 
@@ -35,10 +38,11 @@
     ISSUE_YTMUSIC_ONLY,
 )
 from .models import DiagnosisItem, SyncAttempt, Track
+from .snapshot import YTMUSIC_LIKED_SONGS
 from .youtube_client import YouTubeClient, YouTubeWriteError
 from .ytmusic_client import YTMusicClient, YTMusicWriteError
 
-ActionKind = Literal["yt_unlike", "ytm_like", "yt_relike"]
+ActionKind = Literal["yt_unlike", "ytm_like", "yt_relike", "ytm_dedupe"]
 
 _TRACK_LOOKUP_BATCH_SIZE = 500
 
@@ -109,6 +113,40 @@ class ExecResult:
     skipped: int
 
 
+@dataclass(frozen=True)
+class _DuplicateReason:
+    count: int
+    source: str
+    positions: tuple[int, ...]
+
+
+_DUPLICATE_REASON_RE = re.compile(
+    r"^appears (?P<count>\d+) times in (?P<source>\w+) snapshot "
+    r"\(positions: (?P<positions>[\d, ]+)\)$"
+)
+
+
+def _parse_duplicate_in_source_reason(reason: str) -> _DuplicateReason | None:
+    """Parse the diagnosis-time reason emitted by ``build_duplicate_in_source_items``.
+
+    Returns ``None`` for any deviation: malformed input, multi-source
+    contamination, count-vs-positions mismatch (e.g.
+    ``"appears 2 times ... (positions: 0, 1, 5)"``). Callers MUST treat
+    ``None`` as "do not auto-act."
+    """
+    m = _DUPLICATE_REASON_RE.fullmatch(reason)
+    if m is None:
+        return None
+    count = int(m.group("count"))
+    try:
+        positions = tuple(int(p.strip()) for p in m.group("positions").split(","))
+    except ValueError:
+        return None
+    if len(positions) != count:
+        return None
+    return _DuplicateReason(count=count, source=m.group("source"), positions=positions)
+
+
 def plan(
     items: list[DiagnosisItem],
     video_ids: dict[int, str],
@@ -124,9 +162,12 @@ 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``, ``metadata_drift``, or
-    ``duplicate_in_source`` are also silently dropped (informational, not
-    actionable in 0.4 — within-source dedup is deferred to 0.5).
+    Findings of type ``ytmusic_only`` or ``metadata_drift`` are silently
+    dropped (informational, not actionable in 0.5).
+    ``duplicate_in_source`` maps to ``ytm_dedupe`` only when the parsed
+    reason validates as ytmusic-source + N=2 + matching position count.
+    All other shapes produce a ``SkipRecord`` — we never default to a
+    destructive YT Music write.
     """
     actions: list[PlannedAction] = []
     skips: list[SkipRecord] = []
@@ -205,14 +246,61 @@ def plan(
                 )
             )
 
-        elif item.issue_type in {
-            ISSUE_YTMUSIC_ONLY,
-            ISSUE_METADATA_DRIFT,
-            ISSUE_DUPLICATE_IN_SOURCE,
-        }:
+        elif item.issue_type in {ISSUE_YTMUSIC_ONLY, ISSUE_METADATA_DRIFT}:
             # Informational findings — no record, no action.
             continue
 
+        elif item.issue_type == ISSUE_DUPLICATE_IN_SOURCE:
+            parsed = _parse_duplicate_in_source_reason(item.reason)
+            if parsed is None:
+                skips.append(
+                    SkipRecord(
+                        item_id=item.id,
+                        kind="ytm_dedupe",
+                        reason="unparseable duplicate_in_source reason",
+                    )
+                )
+                continue
+            if parsed.source != YTMUSIC_LIKED_SONGS:
+                skips.append(
+                    SkipRecord(
+                        item_id=item.id,
+                        kind="ytm_dedupe",
+                        reason=f"duplicate in {parsed.source} source; not handled in 0.5",
+                    )
+                )
+                continue
+            if parsed.count != 2:
+                skips.append(
+                    SkipRecord(
+                        item_id=item.id,
+                        kind="ytm_dedupe",
+                        reason=(
+                            f"count={parsed.count}; only N=2 is auto-handled in 0.5. "
+                            "File an issue for higher counts."
+                        ),
+                    )
+                )
+                continue
+            primary = _video_id_for(video_ids, item.source_track_id)
+            if not primary:
+                skips.append(
+                    SkipRecord(
+                        item_id=item.id,
+                        kind="ytm_dedupe",
+                        reason="no video_id available for source track",
+                    )
+                )
+                continue
+            actions.append(
+                PlannedAction(
+                    item_id=item.id,
+                    kind="ytm_dedupe",
+                    primary_video_id=primary,
+                    secondary_video_id=None,
+                )
+            )
+
         # Unknown future issue types fall through silently (no WARN channel
         # — sync.plan stays pure).
 
@@ -225,12 +313,15 @@ def _video_id_for(video_ids: dict[int, str], track_id: int | None) -> str | None
     return video_ids.get(track_id)
 
 
+_PLAN_ACTION_KINDS = ("yt_unlike", "ytm_like", "yt_relike", "ytm_dedupe")
+
 # Quota cost per action kind (YouTube `videos.rate` = 50 units; ytm is free).
 # Drift's worst case = 100 (like 50 + unlike 50 if the like succeeds).
 _QUOTA_COST: dict[ActionKind, int] = {
     "yt_unlike": 50,
     "ytm_like": 0,
     "yt_relike": 100,
+    "ytm_dedupe": 0,
 }
 
 
@@ -246,11 +337,11 @@ def summarize(actions: list[PlannedAction], skips: list[SkipRecord]) -> str:
     quota = sum(_QUOTA_COST[a.kind] for a in actions)
 
     lines = ["Sync plan:"]
-    for kind in ("yt_unlike", "ytm_like", "yt_relike"):
+    for kind in _PLAN_ACTION_KINDS:
         lines.append(f"  {kind}: {by_action.get(kind, 0)}")
     lines.append(f"  skipped: {len(skips)}")
     if by_skip:
-        for kind in ("yt_unlike", "ytm_like", "yt_relike"):
+        for kind in _PLAN_ACTION_KINDS:
             count = by_skip.get(kind, 0)
             if count:
                 lines.append(f"    {kind}: {count}")
@@ -276,6 +367,12 @@ def execute(
     Commit cadence: per-action. A crash mid-run preserves all prior
     commits — the next ``sync`` re-evaluates anything not at
     ``status='applied'``.
+
+    Carve-out: ``ytm_dedupe`` flips ``item.status`` to ``"applied"``
+    regardless of success. The call is non-idempotent — auto-retrying
+    a failed unlike risks over-removal because client failure can't
+    distinguish "server processed, client errored" from "server didn't
+    process". Genuine failures self-correct via the next compare-likes.
     """
     applied = 0
     failed = 0
@@ -303,10 +400,16 @@ def execute(
 
         ok = _dispatch(action, ytm=ytm, yt=yt, session=session)
         if ok:
-            item.status = "applied"
             applied += 1
         else:
             failed += 1
+        # ytm_dedupe is non-idempotent — terminal-on-attempt regardless of success.
+        # Client failure doesn't distinguish "server processed, client errored"
+        # from "server didn't process", so auto-retry would risk over-removal.
+        # Genuine failures self-correct via the next compare-likes (lingering dup
+        # → new finding → new attempt).
+        if ok or action.kind == "ytm_dedupe":
+            item.status = "applied"
         session.commit()
 
     return ExecResult(applied=applied, failed=failed, skipped=skipped)
@@ -323,6 +426,10 @@ def _dispatch(
 
     ``True`` iff every call for the action succeeded — i.e. the dispatcher
     is allowed to flip ``DiagnosisItem.status`` to ``'applied'``.
+
+    Note: for ``ytm_dedupe``, the bool return is used only for
+    ``ExecResult`` counting. Terminality (``status='applied'`` regardless
+    of success) is handled by ``execute()``, not here.
     """
     if action.kind == "yt_unlike":
         return _try_yt_rate(
@@ -369,6 +476,14 @@ def _dispatch(
             rating="none",
         )
 
+    if action.kind == "ytm_dedupe":
+        return _try_ytm_unlike(
+            session,
+            action.item_id,
+            ytm=ytm,
+            video_id=action.primary_video_id,
+        )
+
     # Unreachable for the closed ActionKind set, but keeps the function
     # total for static analyzers.
     return False
@@ -434,3 +549,33 @@ def _try_ytm_like(
         )
     )
     return True
+
+
+def _try_ytm_unlike(
+    session: Session,
+    item_id: int,
+    *,
+    ytm: YTMusicClient,
+    video_id: str,
+) -> bool:
+    try:
+        ytm.unlike_song(video_id)
+    except YTMusicWriteError as exc:
+        session.add(
+            SyncAttempt(
+                diagnosis_item_id=item_id,
+                kind="ytm_dedupe",
+                status="failed",
+                reason=str(exc),
+            )
+        )
+        return False
+    session.add(
+        SyncAttempt(
+            diagnosis_item_id=item_id,
+            kind="ytm_dedupe",
+            status="applied",
+            reason="ok",
+        )
+    )
+    return True

src/likesurgeon/ytmusic_client.py

@@ -219,6 +219,20 @@ def like_song(self, video_id: str) -> None:
         except Exception as exc:  # noqa: BLE001 — system-boundary catch
             raise YTMusicWriteError(video_id, str(exc)) from exc
 
+    def unlike_song(self, video_id: str) -> None:
+        """Remove ONE LM-playlist entry for ``video_id`` via ``rate_song(..., "INDIFFERENT")``.
+
+        This is the only viable dedupe path: ``LIKE`` is non-idempotent (every call
+        appends to LM), and ``setVideoId`` isn't returned by ``get_liked_songs`` so
+        ``remove_playlist_items`` can't target a specific occurrence. Propagation is
+        eventually consistent on the order of minutes — see callers' cooldown notes.
+        """
+        client = self._build()
+        try:
+            client.rate_song(video_id, "INDIFFERENT")
+        except Exception as exc:  # noqa: BLE001 — system-boundary catch
+            raise YTMusicWriteError(video_id, str(exc)) from exc
+
     def fetch_liked_songs(self, limit: int = 5000) -> list[dict[str, Any]]:
         """Fetch up to ``limit`` liked songs. Returns the raw track dicts.