fix(dedup): tighten cross-source candidate prefilter — kill 1800s hang (GOON-V)

_candidate used OR logic (studio OR date±7d OR dur±30s) → 938,950 pairs;
Etap-2 scoring at ~110/s never finished in 1800s → bulk_dedup_performers HUNG
every run, orphan thread leaked until restart. Require AND: same studio plus
(date±2d OR dur±30s). 939k→16k pairs, full run 213s. Real cross-source dup of
one master shares studio + near date/duration; rare studio_id-mismatch pairs
skipped on purpose — a job that COMPLETES beats one that times out merging nothing.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/scheduler/bulk_dedup.py

@@ -240,23 +240,28 @@ def _pairs_sharing_performer(
             scene_meta[sid] = (studio_id, rel_date, dur)
 
         def _candidate(a_id: uuid.UUID, b_id: uuid.UUID) -> bool:
+            # Prefiltr cross-source par PRZED scoringiem (drogi). Wymagamy KONIUNKCJI
+            # sygnałów: identyczne studio ORAZ zbieżna data/długość. Wcześniej alternatywa
+            # (studio LUB data±7d LUB dur±30s) przepuszczała 939k par — płodny performer w
+            # jednym studio generuje tpdb×stashdb kartezjan, a sam wspólny studio go nie tnie.
+            # Etap 2 (~110 par/s) nie kończył w 1800s → job HUNG co run (GOON-V), wątek leak.
+            # studio match AND (data±2d OR dur±30s) ⇒ 939k→~16k par, ~240s całość. Prawdziwy
+            # cross-source dup tej samej sceny ma to samo studio + ~tę samą datę/długość (jeden
+            # master). Pary o rozjechanym studio_id (rzadkie po studio-dedup) świadomie pomijamy
+            # — częściowe pokrycie które KOŃCZY > pełne które timeoutuje i nie merge'uje nic.
             a = scene_meta.get(a_id)
             b = scene_meta.get(b_id)
             if not a or not b:
                 return False
             a_studio, a_date, a_dur = a
             b_studio, b_date, b_dur = b
-            # studio match (oba znają studio i to samo) — bardzo silny sygnał
+            # Kotwica: oba znają studio i to samo.
-            if a_studio is not None and a_studio == b_studio:
+            if a_studio is None or a_studio != b_studio:
-                return True
-            # date ±7d (oba mają daty)
-            if a_date and b_date and abs((a_date - b_date).days) <= 7:
-                return True
-            # duration ±30s (oba znają długość; 30s zostawia margines na intro/outro
-            # różniący się między TPDB a StashDB metadata)
-            if a_dur and b_dur and abs(a_dur - b_dur) <= 30:
-                return True
                 return False
+            # Plus zgodność czasowa LUB długości (znana-i-bliska, nie tylko brak).
+            date_ok = bool(a_date and b_date and abs((a_date - b_date).days) <= 2)
+            dur_ok = bool(a_dur and b_dur and abs(a_dur - b_dur) <= 30)
+            return date_ok or dur_ok
 
         seen_pairs: set[tuple[uuid.UUID, uuid.UUID]] = set()
         for scene_ids in perf_to_scenes.values():