release: v0.6.3

- record_frigate_files_batch: copy-before-mutate so a write failure
  doesn't leave cache ahead of disk (same fix as remove_frigate_files_batch)
- executor: replace tracker_ok boolean with try/else
- jobs: collapse duplicate custom_limit is not None checks into one guard

Bump version to 0.6.3.

CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.6.3] - 2026-06-16
+
+### Fixed
+
+- **`record_frigate_files_batch` no longer mutates the tracker cache before write**: the function shared the same cache-corruption-on-write-failure bug that was fixed in `remove_frigate_files_batch` in v0.6.1 — `data.setdefault("by_person", {})` mutated the cached dict in-place, so a disk-full or permission error left the in-memory cache ahead of the on-disk file. Now uses the same copy-before-mutate pattern (shallow copies of the top-level dict and `by_person` sub-dict) so a failed write leaves cache and disk in sync.
+
+- **`tracker_ok` boolean flag replaced with try/else**: the intermediate boolean was a misleading placeholder — the `True` initial value suggested success before the operation ran. The control flow is now expressed directly with a try/except/else block.
+
+- **`LIMIT` env var guard simplified**: the two adjacent `if custom_limit is not None` checks in `_resolve_strategy` are collapsed into a single `if custom_limit is not None:` with nested branches, removing redundant evaluation.
+
 ## [0.6.2] - 2026-06-16
 
 ### Changed

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "winnow"
-version = "0.6.2"
+version = "0.6.3"
 description = "Selects diverse, high-quality photos from Immich as training data for Frigate face recognition."
 license = "AGPL-3.0-or-later"
 requires-python = ">=3.13"

uv.lock

@@ -862,7 +862,7 @@ wheels = [
 
 [[package]]
 name = "winnow"
-version = "0.6.1"
+version = "0.6.2"
 source = { editable = "." }
 dependencies = [
     { name = "croniter" },

winnow/cache.py

@@ -90,7 +90,7 @@ class EmbeddingCache:
             np.save(tmp, embedding)
             os.replace(tmp, final)
         except Exception as e:
-            logger.debug("Cache write failed for %s: %s", asset_id, e)
+            logger.warning("Cache write failed for %s: %s", asset_id, e)
             try:
                 os.remove(tmp)
             except OSError:

winnow/cli.py

@@ -86,6 +86,8 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
             for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
         }
 
+    skip_ids = _smaller_duplicate_ids(duplicates)
+
     if not Config.MERGE_DUPLICATE_PEOPLE:
         rprint("\n[bold yellow]⚠  Duplicate person names detected in Immich:[/bold yellow]")
         for name, ps in sorted(duplicates.items()):
@@ -107,7 +109,7 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
         )
         # Return deduplicated list — keep only the largest per name so that
         # downstream job creation never runs two jobs for the same Frigate folder.
-        return [p for p in people if p["id"] not in _smaller_duplicate_ids(duplicates)]
+        return [p for p in people if p["id"] not in skip_ids]
 
     # Auto-merge: survivor = largest asset count, rest merge into it inside Immich
     merged_any = False
@@ -133,7 +135,6 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
         # IDs still exist in Immich and would produce two jobs for the same folder.
         # IDs from groups that merged successfully are already gone from Immich, so
         # this filter is a no-op for them.
-        skip_ids = _smaller_duplicate_ids(duplicates)
         return [p for p in fresh if p.get("id") not in skip_ids]
 
     # All merges failed — fall back to local deduplication (keep largest per name) so
@@ -142,7 +143,7 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
         "  [yellow]All merges failed — applying local deduplication"
         " to avoid overwriting output.[/yellow]"
     )
-    return [p for p in people if p["id"] not in _smaller_duplicate_ids(duplicates)]
+    return [p for p in people if p["id"] not in skip_ids]
 
 
 _UNSUPPORTED_VARS = [

winnow/executor.py

@@ -27,6 +27,8 @@ from .log_config import console
 from .quality import blur_score_from_image
 from .reconcile import enrich_asset_with_face_data, reconcile_frigate_mappings
 from .upload_tracker import (
+    UPLOAD_TRACKER_FILE,
+    REJECT_TRACKER_FILE,
     get_lowest_quality_mapped_file,
     get_most_redundant_mapped_file,
     get_tracked_frigate_file_count,
@@ -367,6 +369,8 @@ def upload_to_frigate(jobs: list[dict]) -> None:
             person_has_fscores: bool = has_frigate_scores(name)
 
             begin_batch(UPLOAD_TRACKER_FILE)
+            begin_batch(REJECT_TRACKER_FILE)
+            try:
                 for fname in person_files:
                     fpath = os.path.join(person_dir, fname)
 
@@ -481,6 +485,7 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                         )
                         if delete_frigate_person_files(name, [target_frigate_file]):
                             remove_frigate_file(name, target_frigate_file)
+                            person_has_fscores = has_frigate_scores(name)
                             effective_count -= 1
                             min_quality_score_for_slot = None if using_fscore else target_score
                         else:
@@ -521,6 +526,7 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                                             " but asset may be re-selected next run: %s",
                                             fname, tracker_exc,
                                         )
+                                    else:
                                         if pre_fscore is not None:
                                             person_has_fscores = True
                                         actually_uploaded.append((fname, asset_id))
@@ -593,7 +599,15 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                         " was not filled this run — will be available next run"
                     )
 
+            finally:
+                try:
                     flush_batch(UPLOAD_TRACKER_FILE)
+                except Exception as _flush_exc:
+                    logger.warning("flush_batch failed during cleanup — batch will be recovered on next begin_batch: %s", _flush_exc)
+                try:
+                    flush_batch(REJECT_TRACKER_FILE)
+                except Exception as _flush_exc:
+                    logger.warning("flush_batch failed during cleanup — batch will be recovered on next begin_batch: %s", _flush_exc)
 
             # Batch-map Frigate filenames to asset IDs now that all uploads are done.
             if actually_uploaded and not _skip_reconcile:

winnow/jobs.py

@@ -70,7 +70,9 @@ def _resolve_strategy(strategy: str, has_embedding: bool) -> tuple[int | str, st
 
     custom_limit = _getenv_optional_int("LIMIT")
     if custom_limit is not None:
+        if custom_limit > 0:
             return custom_limit, "smart"
+        logger.warning("LIMIT=%s is invalid — ignoring and using auto strategy", custom_limit)
 
     strategy_map = {
         "adaptive": ("auto", "smart"),

winnow/quality.py

@@ -14,6 +14,11 @@ from PIL import Image
 logger = logging.getLogger(__name__)
 
 
+def _laplacian_var(img_np: np.ndarray) -> float:
+    gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) if img_np.ndim == 3 else img_np
+    return float(cv2.Laplacian(gray, cv2.CV_64F).var())
+
+
 @dataclass
 class QualityResult:
     """Result of quality assessment on a face/image crop."""
@@ -32,8 +37,7 @@ def check_blur(img_np: np.ndarray, threshold: float = 100.0) -> tuple[bool, str]
 
     Lower variance = blurrier image. ArcFace needs clear facial features.
     """
-    gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) if img_np.ndim == 3 else img_np
-    variance = cv2.Laplacian(gray, cv2.CV_64F).var()
+    variance = _laplacian_var(img_np)
     if variance < threshold:
         return False, f"Blurry (laplacian={variance:.1f}, threshold={threshold})"
     return True, ""
@@ -115,8 +119,7 @@ def assess_quality(
     reasons = []
 
     # Compute laplacian variance once (used by check_blur and stored as blur_score)
-    gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) if img_np.ndim == 3 else img_np
-    blur_score = float(cv2.Laplacian(gray, cv2.CV_64F).var())
+    blur_score = _laplacian_var(img_np)
 
     checks = [
         (
@@ -154,8 +157,7 @@ def blur_score_from_image(img: Image.Image, max_dim: int = 1440) -> float | None
         if score_img.width > max_dim or score_img.height > max_dim:
             score_img = score_img.copy()
             score_img.thumbnail((max_dim, max_dim), Image.LANCZOS)
-        gray = cv2.cvtColor(np.array(score_img), cv2.COLOR_RGB2GRAY)
-        return float(cv2.Laplacian(gray, cv2.CV_64F).var())
+        return _laplacian_var(np.array(score_img))
     except Exception as exc:
         logger.debug("blur_score_from_image failed: %s", exc)
         return None

winnow/upload_tracker.py

@@ -32,7 +32,7 @@ import logging
 import os
 from pathlib import Path
 
-from .frigate_api import delete_frigate_person_files
+from .frigate_api import _get_frigate_url, delete_frigate_person_files
 
 logger = logging.getLogger(__name__)
 
@@ -44,6 +44,7 @@ REJECT_TRACKER_FILE = "frigate_rejected_ids.json"
 # Keyed by full path so tests with isolated tmp dirs never share entries.
 _cache: dict[str, dict] = {}
 _deferred: set[str] = set()  # paths whose disk writes are batched until flush_batch()
+_dirty: set[str] = set()    # deferred paths that received at least one _save during the batch
 
 
 def _tracker_path(filename: str) -> Path:
@@ -87,6 +88,7 @@ def _save(filename: str, data: dict) -> None:
     key = str(path)
     if key in _deferred:
         _cache[key] = data  # accumulate in cache; disk write deferred until flush_batch()
+        _dirty.add(key)
         return
     _write_to_disk(path, data)
     _cache[key] = data  # update cache only after successful write
@@ -95,17 +97,32 @@ def _save(filename: str, data: dict) -> None:
 def begin_batch(filename: str) -> None:
     """Defer tracker disk writes for filename. All _save calls accumulate in the
     in-memory cache until flush_batch() is called. Use around per-person upload loops
-    to reduce N writes to 1."""
-    _deferred.add(str(_tracker_path(filename)))
+    to reduce N writes to 1.
+
+    If a previous batch for this file was interrupted before flush_batch() was called
+    (e.g. an exception escaped the upload loop), the leftover cache state is flushed
+    to disk here before starting fresh so that partial progress is not silently lost.
+    """
+    path = _tracker_path(filename)
+    key = str(path)
+    if key in _deferred and key in _dirty:
+        try:
+            _write_to_disk(path, _cache[key])
+        except Exception:
+            logger.warning("begin_batch: could not flush leftover deferred state for %s — partial progress may be lost", path)
+        _deferred.discard(key)
+        _dirty.discard(key)
+    _deferred.add(key)
 
 
 def flush_batch(filename: str) -> None:
     """Write the accumulated cache state for filename to disk."""
     path = _tracker_path(filename)
     key = str(path)
-    _deferred.discard(key)
-    if key in _cache:
+    if key in _dirty and key in _cache:
         _write_to_disk(path, _cache[key])
+    _deferred.discard(key)
+    _dirty.discard(key)
 
 
 def _flat_key(filename: str) -> str:
@@ -141,8 +158,10 @@ def _mark(
     crop_dims: tuple[int, int] | None = None,
     frigate_score: float | None = None,
 ) -> None:
+    if not person_name:
+        logger.warning("_mark called with empty person_name for asset %s — asset not recorded", asset_id)
+        return
     data = _load(filename)
-    if person_name:
     by_person = data.setdefault("by_person", {})
     entry = _migrate_entry(by_person.get(person_name, {}))
     ids = set(entry["asset_ids"])
@@ -156,6 +175,7 @@ def _mark(
         entry["frigate_scores"][asset_id] = round(frigate_score, 4)
     by_person[person_name] = entry
     _save(filename, data)
+    logger.debug("Marked %s in %s (%s)", asset_id, filename, person_name)
 
 
 # ── Public API ────────────────────────────────────────────────────────────────
@@ -186,12 +206,10 @@ def mark_uploaded(
     frigate_score: float | None = None,
 ) -> None:
     _mark(UPLOAD_TRACKER_FILE, asset_id, person_name, score=score, crop_dims=crop_dims, frigate_score=frigate_score)
-    logger.debug(f"Marked {asset_id} as uploaded ({person_name})")
 
 
 def mark_rejected(asset_id: str, person_name: str | None = None) -> None:
     _mark(REJECT_TRACKER_FILE, asset_id, person_name)
-    logger.debug(f"Marked {asset_id} as rejected ({person_name})")
 
 
 
@@ -205,11 +223,13 @@ def record_frigate_files_batch(person_name: str, mappings: dict[str, str]) -> No
     """Record multiple Frigate filename → asset_id mappings in a single load/save."""
     if not mappings:
         return
-    data = _load(UPLOAD_TRACKER_FILE)
-    by_person = data.setdefault("by_person", {})
+    src = _load(UPLOAD_TRACKER_FILE)
+    by_person = dict(src.get("by_person", {}))
     entry = _migrate_entry(by_person.get(person_name, {}))
     entry["frigate_files"].update(mappings)
     by_person[person_name] = entry
+    data = dict(src)
+    data["by_person"] = by_person
     _save(UPLOAD_TRACKER_FILE, data)
     logger.debug(f"Batch-mapped {len(mappings)} Frigate file(s) for {person_name}")
 
@@ -225,17 +245,19 @@ def remove_frigate_file(person_name: str, frigate_filename: str) -> None:
 
 def remove_frigate_files_batch(person_name: str, frigate_filenames: list[str]) -> None:
     """Remove multiple Frigate filenames in a single load/save."""
-    data = _load(UPLOAD_TRACKER_FILE)
-    by_person = data.get("by_person", {})
-    raw = by_person.get(person_name)
+    src = _load(UPLOAD_TRACKER_FILE)
+    raw = src.get("by_person", {}).get(person_name)
     if raw is None:
         return
     entry = _migrate_entry(raw)
     for fn in frigate_filenames:
         asset_id = entry["frigate_files"].pop(fn, None)
-        if asset_id:
+        if asset_id is not None and asset_id not in entry["frigate_files"].values():
             entry["frigate_scores"].pop(asset_id, None)
+    by_person = dict(src.get("by_person", {}))  # copy so assignment does not mutate the cache
     by_person[person_name] = entry
+    data = dict(src)
+    data["by_person"] = by_person
     _save(UPLOAD_TRACKER_FILE, data)
     logger.debug(f"Removed {len(frigate_filenames)} Frigate file mapping(s) for {person_name}")
 
@@ -265,9 +287,11 @@ def get_tracked_frigate_filenames(person_name: str) -> set[str]:
 def has_frigate_scores(person_name: str) -> bool:
     """Return True if any mapped file for this person has a stored Frigate recognition score."""
     data = _load(UPLOAD_TRACKER_FILE)
-    entry = _migrate_entry(data.get("by_person", {}).get(person_name, {}))
-    frigate_files = entry.get("frigate_files", {})
-    frigate_scores = entry.get("frigate_scores", {})
+    raw = data.get("by_person", {}).get(person_name)
+    if not raw or isinstance(raw, list):
+        return False
+    frigate_files = raw.get("frigate_files", {})
+    frigate_scores = raw.get("frigate_scores", {})
     return any(asset_id in frigate_scores for asset_id in frigate_files.values())
 
 
@@ -361,14 +385,16 @@ def reset_all_people() -> None:
     approach is O(P²) because each call rebuilds the flat list from all remaining entries.
     """
     upload_data = _load(UPLOAD_TRACKER_FILE)
+    frigate_url = _get_frigate_url()
+    if not frigate_url:
+        logger.info("FRIGATE_URL not set — skipping Frigate file deletion")
     for person_name, raw_entry in upload_data.get("by_person", {}).items():
         entry = _migrate_entry(raw_entry)
         frigate_filenames = list(entry.get("frigate_files", {}).keys())
         if not frigate_filenames:
             continue
-        if not os.environ.get("FRIGATE_URL", "").strip():
-            logger.info(f"FRIGATE_URL not set — skipping Frigate file deletion for {person_name}")
-        elif delete_frigate_person_files(person_name, frigate_filenames):
+        if frigate_url:
+            if delete_frigate_person_files(person_name, frigate_filenames):
                 logger.info(f"Deleted {len(frigate_filenames)} Frigate file(s) for {person_name}")
             else:
                 logger.warning(f"Could not delete Frigate files for {person_name} — tracker reset proceeding anyway")
@@ -389,7 +415,7 @@ def reset_person(person_name: str) -> None:
     entry = _migrate_entry(upload_data.get("by_person", {}).get(person_name, {}))
     frigate_filenames = list(entry.get("frigate_files", {}).keys())
     if frigate_filenames:
-        if not os.environ.get("FRIGATE_URL", "").strip():
+        if not _get_frigate_url():
             logger.info(f"FRIGATE_URL not set — skipping Frigate file deletion for {person_name}")
         elif delete_frigate_person_files(person_name, frigate_filenames):
             logger.info(f"Deleted {len(frigate_filenames)} Frigate file(s) for {person_name}")
@@ -398,15 +424,16 @@ def reset_person(person_name: str) -> None:
 
     changed = False
     for filename in (UPLOAD_TRACKER_FILE, REJECT_TRACKER_FILE):
-        data = upload_data if filename == UPLOAD_TRACKER_FILE else _load(REJECT_TRACKER_FILE)
-        by_person = data.get("by_person", {})
+        src = upload_data if filename == UPLOAD_TRACKER_FILE else _load(REJECT_TRACKER_FILE)
+        by_person = dict(src.get("by_person", {}))  # copy so pop() does not mutate the cache
         tracker_entry = by_person.pop(person_name, None)
         if tracker_entry is not None:
+            data = dict(src)
+            data["by_person"] = by_person
             flat_key = _flat_key(filename)
             person_ids = set(_get_ids(tracker_entry))
             if person_ids and flat_key in data:
                 data[flat_key] = sorted(set(data[flat_key]) - person_ids)
-            data["by_person"] = by_person
             _save(filename, data)
             changed = True
     if changed: