Merge pull request #47 from sudolulo/chore/simplify-lockfile-ci

chore: replace lockfile auto-update with uv lock --check

.githooks/pre-commit

@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if git diff --cached --name-only | grep -q "^pyproject\.toml$"; then
+    uv lock
+    git add uv.lock
+fi

.github/workflows/test.yml

@@ -21,6 +21,9 @@ jobs:
       - name: Set up Python
         run: uv python install 3.13
 
+      - name: Check lockfile is up to date
+        run: uv lock --check
+
       - name: Install dependencies
         run: uv sync --extra cpu
 

.github/workflows/update-lockfile.yml

@@ -1,46 +0,0 @@
-# .github/workflows/update-lockfile.yml
-name: Update lockfile
-
-on:
-  push:
-    branches:
-      - '**'
-    paths:
-      - 'pyproject.toml'
-  workflow_dispatch:
-
-jobs:
-  update-lockfile:
-    runs-on: ubuntu-latest
-    permissions:
-      contents: write
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
-
-      - name: Install uv
-        uses: astral-sh/setup-uv@fac544c07dec837d0ccb6301d7b5580bf5edae39 # v8.2.0
-
-      - name: Set up Python
-        run: uv python install 3.13
-
-      - name: Regenerate lockfile
-        run: uv lock
-
-      - name: Check for changes
-        id: diff
-        run: |
-          if git diff --quiet uv.lock; then
-            echo "changed=false" >> "$GITHUB_OUTPUT"
-          else
-            echo "changed=true" >> "$GITHUB_OUTPUT"
-          fi
-
-      - name: Commit and push updated lockfile
-        if: steps.diff.outputs.changed == 'true'
-        run: |
-          git config user.name "github-actions[bot]"
-          git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
-          git add uv.lock
-          git commit -m "chore: update lockfile"
-          git push

CHANGELOG.md

@@ -7,6 +7,142 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.6.6] - 2026-06-18
+
+### Changed
+
+- **`MAX_AUTO_IMAGES` default lowered from 20 to 5** — existing users who have not set this variable and already have more than 5 winnow-managed images in Frigate will find themselves at cap on the next run. With `QUALITY_REPLACEMENT=true` (the default), winnow will attempt to swap weaker images rather than uploading new ones. Set `MAX_AUTO_IMAGES=20` to restore the previous behaviour.
+
+## [0.6.5] - 2026-06-17
+
+### Added
+
+- **Version displayed in startup banner**: winnow now prints its installed version at launch.
+
+### Fixed
+
+- **GPU image: `CUDAExecutionProvider` missing due to parallel install race**: `insightface` declares `onnxruntime` (CPU) as a dependency, causing `uv sync` to install both `onnxruntime` and `onnxruntime-gpu` in parallel — both packages claim the same `pybind11_state.so` binary. On GitHub Actions the CPU binary consistently won the race, leaving the GPU build without CUDA support at runtime despite all CUDA libraries being present. Fixed by reinstalling `onnxruntime-gpu` sequentially after `uv sync` to guarantee its GPU binary is on disk.
+
+- **GPU extra was missing three required nvidia pip packages**: `onnxruntime-gpu` 1.26.0 gates CUDA EP loading on the Python-importability of `nvidia-cuda-runtime-cu12`, `nvidia-cufft-cu12`, and `nvidia-curand-cu12`. These packages were not declared in the `gpu` extra and were absent on fresh installs, silently disabling GPU inference.
+
+- **`_handle_duplicate_people` raises `KeyError` on id-less person records**: bare `p["id"]` subscripts in the auto-merge loop and `_smaller_duplicate_ids` raised `KeyError` when Immich returned a person dict without an `id` field (e.g. unconfirmed face clusters). Fixed by using `p.get("id")` and filtering `None` from `skip_ids`.
+
+- **`_smaller_duplicate_ids` could include `None` in the skip set**: `p.get("id")` without a `None` guard populated `skip_ids` with `None`, causing `p.get("id") not in skip_ids` to pass for every id-less person, so unnamed face clusters were silently re-included in all return paths.
+
+- **`_handle_duplicate_people` dead code removed**: guards `if not survivor_id` and `if not merge_ids` became unreachable after the id-gate fix; their presence suggested they still ran.
+
+- **`_valid_people` in `jobs.py` used wrong name filter**: whitespace-only names (e.g. `"  "`) passed the `p.get("name")` truthiness check and were included in the person list. Fixed using `(p.get("name") or "").strip()` consistent with the cli.py gate.
+
+- **`interactive_configure` queued-marker check was O(N²)**: `[j for j in jobs if j["person"]["id"] == p.get("id")]` ran a full scan over jobs for every person in the display loop. Replaced with a `queued_ids` set hoisted before the loop.
+
+- **`executor.py` slot restore did not clear `min_quality_score_for_slot`**: when a replacement upload failed all retries after a deletion, `effective_count` was restored but the stale quality-score floor from the deleted file remained, blocking the next candidate from filling the slot.
+
+- **`get_immich_version` swallowed `KeyError` on unexpected schema**: bare `data["major"]` / `data["minor"]` / `data["patch"]` subscripts were silently caught by the surrounding `except Exception`, returning `None` without logging. Replaced with `.get()` calls that log a debug warning on unexpected schemas.
+
+- **Face embedding selects nearest face to crop centre, not largest by area**: a 25 % margin on the crop window can pull a larger neighbouring face into the bounding box; selecting the biggest face by area then embeds the wrong person. Centre-proximity is now used instead.
+
+- **Zero-norm face embeddings skipped before diversity selection**: InsightFace occasionally returns a zero vector for low-quality detections; zero embeddings pass deduplication with similarity 0 and score distance 1.0, causing them to be selected first as maximally diverse.
+
+- **`executor.py` slot restore did not clear `min_quality_score_for_slot`**: stale quality floor from the deleted file blocked the next candidate from filling the restored slot in quality-replacement mode.
+
+## [0.6.4] - 2026-06-17
+
+### Fixed
+
+- **Face bbox scaled to thumbnail space before quality filtering**: `assess_quality` now receives coordinates in thumbnail-pixel space rather than detection-image space. Previously, a face detected on a full-resolution image (e.g. 4000 px wide) was compared against `MIN_FACE_WIDTH` using its original pixel dimensions, causing faces that appear small on the thumbnail to pass the quality filter — and faces that appear large to be incorrectly rejected.
+
+- **`conf_array` default restored to 1.0 for faces with missing confidence**: the default was incorrectly set to 0.5, causing images with no `score` field in the Immich faces API response to receive a 1.7× FPS diversity boost and be selected ahead of genuinely high-confidence detections. The default is now 1.0 (no boost), treating missing confidence as neutral.
+
+- **`hard_weight` computed once outside FPS loop**: `conf_array` is constant after initialisation; moving the `np.where` call outside the `while` loop eliminates one O(n) numpy pass per selected image.
+
+- **`has_frigate_model` snapshot prevents mid-batch `recognize_face` calls on first run**: `effective_count` is incremented inside the upload loop, so using it as the `recognize_face` gate would incorrectly trigger scoring after the first upload on a first run. A boolean snapshot is now taken before the loop.
+
+- **`person_has_fscores` only set when tracker write succeeds**: the flag was moved outside the `try/except else` block, causing at-cap replacement to switch into Frigate-score mode even when the score was never written to the tracker — `get_most_redundant_mapped_file` then returned `None` and all replacement candidates were silently skipped. The flag is now set only in the `else` branch.
+
+- **`STRATEGY=skip` honoured before embedding and limit checks**: the strategy was silently converted to `auto` when InsightFace was available, because two early-returns in `_resolve_strategy` ran before the `strategy_map` lookup.
+
+- **`limit="auto"` preserved on first run**: switching to `limit = capacity` unconditionally caused the FPS adaptive early-stop to never fire on a person's first upload run. `limit="auto"` is now kept when `already_uploaded == 0`.
+
+- **`EmbeddingCache.get` falls back gracefully on all load errors**: a `MemoryError` during `np.load` of a cached embedding was re-raised, crashing the entire diversity-selection batch for that person. Cache-read failures of any kind now return `None` so the embedding is recomputed fresh.
+
+- **`get_people` returns `[]` when Immich sends `{"people": null}`**: `.get("people", [])` only uses the default when the key is absent, not when its value is `null`. Changed to `data.get("people") or []` so null-valued responses are handled the same as missing keys.
+
+- **`get_people` and `fetch_all_assets` guard against non-dict responses**: a proxy or CDN returning a JSON array (or other non-dict body) previously caused an `AttributeError` from `.get()`. Both functions now check `isinstance(data, dict)` and return an empty result with an error log.
+
+- **`filter_recent_assets` counts and logs assets with missing or unparseable timestamps** instead of silently dropping them.
+
+- **`_suppress_output` fd cleanup restructured**: the context manager now initialises `devnull_fd`, `saved_out`, and `saved_err` to `None` before the `try` block, so the `finally` can close only the descriptors that were successfully opened. Each `os.close` is wrapped in its own `try/except OSError` so a failed close cannot prevent subsequent descriptors from being released. `OSError` from `os.dup2` restore is logged at DEBUG rather than silently swallowed.
+
+- **`blur_score_from_image` copies the image before thumbnail resize**: `Image.thumbnail` modifies the image in-place. When the caller's image was already in RGB mode (no convert copy), the resize would have mutated the caller's object. A copy is now made when `score_img is img`.
+
+- **`imageWidth`/`imageHeight` zero-value treated as missing** in `image_processing.py`: the old `or img_w` fallback silently set `scale = 1.0` for a zero-valued dimension (correct) but also for `None` (also correct) with no distinction. The explicit `scale = img_w / meta_w if meta_w else 1.0` form matches the pattern used in the new `_scale_bbox_to_thumbnail` helper and makes the fallback intent clear.
+
+- **`_mark` and `update_frigate_count` copy before mutate**: both functions now create a shallow copy of the top-level tracker dict before assigning into `by_person`, so a failed `_save` cannot leave the in-memory cache ahead of the on-disk file.
+
+- **`reset_person` flat-list guard only warns when cleanup would have run**: the `isinstance(data[flat_key], list)` check previously emitted a warning even when `person_ids` was empty (a no-op call). The warning is now gated behind `person_ids and`, matching the guard on the cleanup branch.
+
+- **`_handle_duplicate_people` uses `p.get("id")` consistently**: all four return-path filter comprehensions and the `_smaller_duplicate_ids` set comprehension now use `.get("id")` instead of bare `p["id"]`, preventing a `KeyError` if the Immich API returns a person record without an `id` field.
+
+- **`K-Medoids` non-medoid membership test is O(1)**: `non_medoids` now filters against `set(medoids)` instead of the list, eliminating an O(k) scan per candidate on each outer iteration.
+
+## [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
+
+- **Flat `uploaded_asset_ids` / `rejected_asset_ids` lists dropped as primary storage**: asset IDs are now derived on read from `by_person` entries, which are the single source of truth. The legacy flat lists in existing tracker files are still read (union) so no assets become re-eligible after upgrading. New writes no longer maintain the flat lists. This removes the dual-representation sync hazard and paves the way for multi-instance support (per-instance `by_person` keying in a future release).
+
+- **Tracker writes batched per person**: `mark_uploaded` calls inside the per-person upload loop are now accumulated in memory (`begin_batch`) and flushed in a single `os.replace` write at the end of each person's loop (`flush_batch`), reducing N tracker writes per person to 1. Benefits users on slow storage (NAS, SD card, spinning disks).
+
+- **`RESET_PERSON=*` is now O(1) disk writes**: replaced the per-person `reset_person` loop with `reset_all_people()`, which makes one Frigate API call per person for file deletion and then clears both tracker files in two writes. Previously it was O(P²) iterations and 2P writes.
+
+- **`blur_score_from_image` inlines Laplacian computation**: replaced the `assess_quality()` call (which ran grayscale, exposure, and confidence checks whose results were discarded) with a direct `cv2.Laplacian` computation. The function is now self-contained and does not silently inherit future costs added to the full quality pipeline.
+
+## [0.6.1] - 2026-06-16
+
+### Fixed
+
+- **Corrupt or truncated full-res thumbnails now marked rejected**: `OSError` (truncated file) is caught alongside `PIL.UnidentifiedImageError` in the thumbnail path so persistently bad assets are tombstoned instead of retried forever. Full-res download failures (`USE_FULL_RESOLUTION=true`) remain transient — not marked rejected — so a Immich blip doesn't permanently blacklist valid assets.
+
+- **Quality replacement mode no longer flips mid-loop**: `person_has_fscores` was re-evaluated after each file deletion, which could switch the remaining replacements from Frigate-score mode to blur-score mode if the deleted file was the last scored one. The mode is now fixed for the duration of the upload loop.
+
+- **`reset_person` no longer removes shared asset IDs**: the flat `uploaded_asset_ids` list is now rebuilt from all remaining `by_person` entries rather than subtracting the reset person's IDs. Previously, resetting Alice could remove an asset ID that also appeared under Bob, making it re-eligible for upload.
+
+- **`_save` cache updated only after successful write**: the in-memory tracker cache is now updated after `os.replace` succeeds rather than before. A disk-full or permission error no longer leaves the cache permanently ahead of the on-disk file.
+
+- **Stale Frigate file cleanup batched**: the per-file `remove_frigate_file` loop is replaced with a single `remove_frigate_files_batch` call, reducing N tracker writes to 1 when stale mappings are cleaned up.
+
+- **`_migrate_entry` no longer mutates the cache through nested dict aliases**: all five nested dicts (`asset_ids`, `scores`, `frigate_scores`, `frigate_files`, `crop_dims`) are now individually copied so `.pop()` calls in write paths cannot reach the in-memory cache.
+
+- **`find_by_crop_dimension` and `_pick_mapped_file` now agree on duplicate asset→file handling**: both use first-seen-wins when the same `asset_id` maps to multiple Frigate filenames, preventing inconsistent replacement decisions.
+
+- **Non-atomic JSON write**: tracker files are written to a `.tmp` sibling then renamed with `os.replace` so a crash mid-write never leaves a truncated file.
+
+- **`get_person_summary` uses `_migrate_entry`**: replaced three ad-hoc `isinstance` guards with a single `_migrate_entry` call, making old-format (list) entries consistent with every other read path.
+
+- **Quality replacement floor check**: a candidate with a `None` blur score (PIL error during scoring) no longer blocks a freed slot — the `<=` floor comparison is only applied when a score is actually available.
+
+- **`executor.py` syntax error**: the `if img is None:` block in the full-res download path was comment-only and would have raised `IndentationError` on import. Added `pass`.
+
+- **Duplicate `if stale:` guard**: two consecutive identical guards around stale-cleanup and its log print were merged into one.
+
+- **`_flat_key` uses constant equality** instead of substring match, removing a latent routing bug for any filename that happens to contain "uploaded".
+
+- **`remove_frigate_file` no longer creates ghost entries**: returns early when the person is absent rather than writing an empty stub.
+
+- **`skip_ids` extracted to helper**: the identical set comprehension in `_handle_duplicate_people` that appeared in three branches is now a single `_smaller_duplicate_ids()` inner function.
+
+- **`blur_score_from_image` returns `None` on error** instead of `0.0`, so callers can distinguish a failed measurement from a legitimately near-zero Laplacian variance score.
+
 ## [0.6.0] - 2026-06-15
 
 ### Changed

CONTRIBUTING.md

@@ -17,8 +17,11 @@ git clone https://github.com/sudolulo/winnow.git
 cd winnow
 git checkout dev
 uv sync
+git config core.hooksPath .githooks
 ```
 
+The last line activates the project's git hooks. The pre-commit hook automatically runs `uv lock` and stages the result whenever `pyproject.toml` is part of a commit, keeping the lockfile in sync without any extra steps.
+
 ## Running Tests and Lint
 
 ```bash

Dockerfile

@@ -50,7 +50,9 @@ RUN if [ "$VARIANT" = "cpu" ]; then \
     elif [ "$VARIANT" = "intel" ]; then \
         uv sync --frozen --no-dev --extra intel; \
     elif [ "$VARIANT" = "gpu" ]; then \
-        uv sync --frozen --no-dev --extra gpu; \
+        uv sync --frozen --no-dev --extra gpu && \
+        ORT_GPU_VER=$(.venv/bin/python -c "import importlib.metadata; print(importlib.metadata.version('onnxruntime-gpu'))") && \
+        uv pip install --python .venv/bin/python --no-deps --reinstall "onnxruntime-gpu==$ORT_GPU_VER"; \
     else \
         echo "Unknown VARIANT: '$VARIANT'. Must be one of: cpu, rocm, intel, gpu" >&2; \
         exit 1; \

README.md

@@ -187,7 +187,7 @@ In scheduled mode the process (and loaded models) stays resident between runs. T
 
 | Variable | Default | Description |
 | :--- | :--- | :--- |
-| `MAX_AUTO_IMAGES` | `20` | Maximum training images per person in Frigate |
+| `MAX_AUTO_IMAGES` | `5` | Maximum training images per person in Frigate |
 | `QUALITY_REPLACEMENT` | `true` | When at cap, swap a weaker tracked image for a better candidate. With Frigate scoring active, targets the most redundant image (highest pre-upload recognize score); otherwise uses blur score. Never touches manually added Frigate files. Set `false` to skip people at cap |
 
 #### Advanced Tuning *(calibrated — do not adjust)*

compose.yml

@@ -31,7 +31,7 @@ services:
       # - USE_FULL_RESOLUTION=true    # Use full-res images vs thumbnails (default: true)
       # - MIN_CONFIDENCE=0.7          # Minimum face detection confidence (default: 0.7)
       # - BLUR_THRESHOLD=100.0        # Laplacian blur threshold; lower = accept more blur (default: 100.0)
-      # - MAX_AUTO_IMAGES=80          # Hard cap on auto-diversity selection (default: 20)
+      # - MAX_AUTO_IMAGES=80          # Hard cap on auto-diversity selection (default: 5)
 
       # ── Caching & Models ──────────────────────────────────────────────────
       # - FORCE_CPU=true              # Disable GPU, fall back to CPU

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "winnow"
-version = "0.6.0"
+version = "0.6.6"
 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"
@@ -29,6 +29,9 @@ dependencies = [
 gpu   = [
     "onnxruntime-gpu>=1.23.2; sys_platform == 'linux' and platform_machine == 'x86_64'",
     "nvidia-cudnn-cu12>=9.0.0; sys_platform == 'linux' and platform_machine == 'x86_64'",
+    "nvidia-cuda-runtime-cu12>=12.0; sys_platform == 'linux' and platform_machine == 'x86_64'",
+    "nvidia-cufft-cu12>=11.0; sys_platform == 'linux' and platform_machine == 'x86_64'",
+    "nvidia-curand-cu12>=10.0; sys_platform == 'linux' and platform_machine == 'x86_64'",
 ]
 rocm  = ["onnxruntime-rocm>=1.16.0; sys_platform == 'linux' and platform_machine == 'x86_64'"]
 intel = ["onnxruntime-openvino>=1.20.0; sys_platform == 'linux' and platform_machine == 'x86_64'"]

scheduler.py

@@ -49,11 +49,12 @@ def _run_scheduler() -> None:
             try:
                 main()
                 print("winnow run complete", flush=True)
-            except KeyboardInterrupt:
+            except (KeyboardInterrupt, SystemExit):
                 raise
             except Exception as e:
                 logger.error("winnow run failed: %s", e, exc_info=True)
                 print(f"winnow run failed: {e}", flush=True)
+            cron = croniter(schedule, time.time())
             next_run = cron.get_next(float)
             print(f"Next run: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(next_run))}", flush=True)
         time.sleep(min(60, max(1, next_run - time.time())))

tests/test_config.py

@@ -21,7 +21,7 @@ def test_config_loads_defaults(monkeypatch):
     assert cfg.MIN_FACE_COUNT == 3
     assert cfg.BLUR_THRESHOLD == 120.0
     assert cfg.MIN_CONFIDENCE == 0.7
-    assert cfg.MAX_AUTO_IMAGES == 20
+    assert cfg.MAX_AUTO_IMAGES == 5
     assert cfg.QUALITY_REPLACEMENT is True
     assert cfg.FACE_MARGIN == 0.15
     assert cfg.USE_FULL_RESOLUTION is True

uv.lock

@@ -340,6 +340,16 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/52/de/823919be3b9d0ccbf1f784035423c5f18f4267fb0123558d58b813c6ec86/nvidia_cuda_nvrtc_cu12-12.9.86-py3-none-win_amd64.whl", hash = "sha256:72972ebdcf504d69462d3bcd67e7b81edd25d0fb85a2c46d3ea3517666636349", size = 76408187, upload-time = "2025-06-05T20:12:27.819Z" },
 ]
 
+[[package]]
+name = "nvidia-cuda-runtime-cu12"
+version = "12.9.79"
+source = { registry = "https://pypi.org/simple" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/bc/e0/0279bd94539fda525e0c8538db29b72a5a8495b0c12173113471d28bce78/nvidia_cuda_runtime_cu12-12.9.79-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:83469a846206f2a733db0c42e223589ab62fd2fabac4432d2f8802de4bded0a4", size = 3515012, upload-time = "2025-06-05T20:00:35.519Z" },
+    { url = "https://files.pythonhosted.org/packages/bc/46/a92db19b8309581092a3add7e6fceb4c301a3fd233969856a8cbf042cd3c/nvidia_cuda_runtime_cu12-12.9.79-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:25bba2dfb01d48a9b59ca474a1ac43c6ebf7011f1b0b8cc44f54eb6ac48a96c3", size = 3493179, upload-time = "2025-06-05T20:00:53.735Z" },
+    { url = "https://files.pythonhosted.org/packages/59/df/e7c3a360be4f7b93cee39271b792669baeb3846c58a4df6dfcf187a7ffab/nvidia_cuda_runtime_cu12-12.9.79-py3-none-win_amd64.whl", hash = "sha256:8e018af8fa02363876860388bd10ccb89eb9ab8fb0aa749aaf58430a9f7c4891", size = 3591604, upload-time = "2025-06-05T20:11:17.036Z" },
+]
+
 [[package]]
 name = "nvidia-cudnn-cu12"
 version = "9.23.1.3"
@@ -353,6 +363,39 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/75/ec/62b56fc5e8219a268c6f62c4e9fb1369ebec049512328e650d1a9a28bcc8/nvidia_cudnn_cu12-9.23.1.3-py3-none-win_amd64.whl", hash = "sha256:b874af5bfab5e1010ae88bfead14bf8e9da6b20283582288f1c05f056090a398", size = 689996767, upload-time = "2026-06-09T19:44:25.343Z" },
 ]
 
+[[package]]
+name = "nvidia-cufft-cu12"
+version = "11.4.1.4"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "nvidia-nvjitlink-cu12", marker = "platform_machine != 's390x'" },
+]
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/9b/2b/76445b0af890da61b501fde30650a1a4bd910607261b209cccb5235d3daa/nvidia_cufft_cu12-11.4.1.4-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:1a28c9b12260a1aa7a8fd12f5ebd82d027963d635ba82ff39a1acfa7c4c0fbcf", size = 200822453, upload-time = "2025-06-05T20:05:27.889Z" },
+    { url = "https://files.pythonhosted.org/packages/95/f4/61e6996dd20481ee834f57a8e9dca28b1869366a135e0d42e2aa8493bdd4/nvidia_cufft_cu12-11.4.1.4-py3-none-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:c67884f2a7d276b4b80eb56a79322a95df592ae5e765cf1243693365ccab4e28", size = 200877592, upload-time = "2025-06-05T20:05:45.862Z" },
+    { url = "https://files.pythonhosted.org/packages/20/ee/29955203338515b940bd4f60ffdbc073428f25ef9bfbce44c9a066aedc5c/nvidia_cufft_cu12-11.4.1.4-py3-none-win_amd64.whl", hash = "sha256:8e5bfaac795e93f80611f807d42844e8e27e340e0cde270dcb6c65386d795b80", size = 200067309, upload-time = "2025-06-05T20:13:59.762Z" },
+]
+
+[[package]]
+name = "nvidia-curand-cu12"
+version = "10.3.10.19"
+source = { registry = "https://pypi.org/simple" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/14/1c/2a45afc614d99558d4a773fa740d8bb5471c8398eeed925fc0fcba020173/nvidia_curand_cu12-10.3.10.19-py3-none-manylinux_2_27_aarch64.whl", hash = "sha256:de663377feb1697e1d30ed587b07d5721fdd6d2015c738d7528a6002a6134d37", size = 68292066, upload-time = "2025-05-01T19:39:13.595Z" },
+    { url = "https://files.pythonhosted.org/packages/31/44/193a0e171750ca9f8320626e8a1f2381e4077a65e69e2fb9708bd479e34a/nvidia_curand_cu12-10.3.10.19-py3-none-manylinux_2_27_x86_64.whl", hash = "sha256:49b274db4780d421bd2ccd362e1415c13887c53c214f0d4b761752b8f9f6aa1e", size = 68295626, upload-time = "2025-05-01T19:39:38.885Z" },
+    { url = "https://files.pythonhosted.org/packages/e5/98/1bd66fd09cbe1a5920cb36ba87029d511db7cca93979e635fd431ad3b6c0/nvidia_curand_cu12-10.3.10.19-py3-none-win_amd64.whl", hash = "sha256:e8129e6ac40dc123bd948e33d3e11b4aa617d87a583fa2f21b3210e90c743cde", size = 68774847, upload-time = "2025-05-01T19:48:52.93Z" },
+]
+
+[[package]]
+name = "nvidia-nvjitlink-cu12"
+version = "12.9.86"
+source = { registry = "https://pypi.org/simple" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/46/0c/c75bbfb967457a0b7670b8ad267bfc4fffdf341c074e0a80db06c24ccfd4/nvidia_nvjitlink_cu12-12.9.86-py3-none-manylinux2010_x86_64.manylinux_2_12_x86_64.whl", hash = "sha256:e3f1171dbdc83c5932a45f0f4c99180a70de9bd2718c1ab77d14104f6d7147f9", size = 39748338, upload-time = "2025-06-05T20:10:25.613Z" },
+    { url = "https://files.pythonhosted.org/packages/97/bc/2dcba8e70cf3115b400fef54f213bcd6715a3195eba000f8330f11e40c45/nvidia_nvjitlink_cu12-12.9.86-py3-none-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:994a05ef08ef4b0b299829cde613a424382aff7efb08a7172c1fa616cc3af2ca", size = 39514880, upload-time = "2025-06-05T20:10:04.89Z" },
+    { url = "https://files.pythonhosted.org/packages/dd/7e/2eecb277d8a98184d881fb98a738363fd4f14577a4d2d7f8264266e82623/nvidia_nvjitlink_cu12-12.9.86-py3-none-win_amd64.whl", hash = "sha256:cc6fcec260ca843c10e34c936921a1c426b351753587fdd638e8cff7b16bb9db", size = 35584936, upload-time = "2025-06-05T20:16:08.525Z" },
+]
+
 [[package]]
 name = "onnx"
 version = "1.21.0"
@@ -862,7 +905,7 @@ wheels = [
 
 [[package]]
 name = "winnow"
-version = "0.6.0"
+version = "0.6.6"
 source = { editable = "." }
 dependencies = [
     { name = "croniter" },
@@ -880,7 +923,10 @@ cpu = [
     { name = "onnxruntime" },
 ]
 gpu = [
+    { name = "nvidia-cuda-runtime-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux'" },
     { name = "nvidia-cudnn-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux'" },
+    { name = "nvidia-cufft-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux'" },
+    { name = "nvidia-curand-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux'" },
     { name = "onnxruntime-gpu", marker = "platform_machine == 'x86_64' and sys_platform == 'linux'" },
 ]
 intel = [
@@ -901,7 +947,10 @@ requires-dist = [
     { name = "croniter", specifier = ">=5.0.2" },
     { name = "insightface", specifier = ">=0.7.3" },
     { name = "numpy", specifier = ">=2.2.6" },
+    { name = "nvidia-cuda-runtime-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux' and extra == 'gpu'", specifier = ">=12.0" },
     { name = "nvidia-cudnn-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux' and extra == 'gpu'", specifier = ">=9.0.0" },
+    { name = "nvidia-cufft-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux' and extra == 'gpu'", specifier = ">=11.0" },
+    { name = "nvidia-curand-cu12", marker = "platform_machine == 'x86_64' and sys_platform == 'linux' and extra == 'gpu'", specifier = ">=10.0" },
     { name = "onnxruntime", marker = "extra == 'cpu'", specifier = ">=1.23.2" },
     { name = "onnxruntime-gpu", marker = "platform_machine == 'x86_64' and sys_platform == 'linux' and extra == 'gpu'", specifier = ">=1.23.2" },
     { name = "onnxruntime-openvino", marker = "platform_machine == 'x86_64' and sys_platform == 'linux' and extra == 'intel'", specifier = ">=1.20.0" },

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:
@@ -103,8 +103,11 @@ class EmbeddingCache:
         count = 0
         for f in os.listdir(self.cache_dir):
             if f.endswith(".npy"):
-                os.remove(os.path.join(self.cache_dir, f))
+                try:
-                count += 1
+                    os.remove(os.path.join(self.cache_dir, f))
+                    count += 1
+                except OSError:
+                    pass
         logger.info("Cleared %s cached embeddings.", count)
 
 

winnow/cli.py

@@ -7,12 +7,13 @@ import sys
 from rich import print as rprint
 from rich.prompt import Confirm
 
+from . import __version__
 from .config import Config, _getenv_bool
 from .executor import execute_jobs, upload_to_frigate
 from .immich_api import get_immich_version, get_people, merge_people
 from .jobs import _show_preview, auto_configure, interactive_configure
 from .log_config import console, setup_logging
-from .upload_tracker import find_by_crop_dimension, get_person_summary, reset_person
+from .upload_tracker import find_by_crop_dimension, get_person_summary, reset_all_people, reset_person
 
 logger = logging.getLogger(__name__)
 
@@ -71,19 +72,33 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
     by_name: dict[str, list[dict]] = defaultdict(list)
     for p in people:
         name = (p.get("name") or "").strip()
-        if name:
+        if name and p.get("id"):
             by_name[name].append(p)
 
     duplicates = {name: ps for name, ps in by_name.items() if len(ps) > 1}
     if not duplicates:
         return people
 
+    def _smaller_duplicate_ids(groups: dict) -> set[str]:
+        """IDs of all but the largest person in each duplicate group."""
+        return {
+            pid
+            for ps in groups.values()
+            for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
+            if (pid := p.get("id"))
+        }
+
+    skip_ids = _smaller_duplicate_ids(duplicates)
+
+    def _excl(lst: list[dict]) -> list[dict]:
+        return [p for p in lst if p.get("id") not in skip_ids]
+
     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()):
             ordered = sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)
             entries = ", ".join(
-                f"[dim]{p['id'][:8]}…[/dim] ({p.get('assetCount', 0)} assets)"
+                f"[dim]{(p.get('id') or '?')[:8]}…[/dim] ({p.get('assetCount', 0)} assets)"
                 for p in ordered
             )
             rprint(f"  [yellow]{name}[/yellow] → {len(ps)} people: {entries}")
@@ -99,25 +114,21 @@ 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.
-        skip_ids = {
+        return _excl(people)
-            p["id"]
-            for ps in duplicates.values()
-            for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
-        }
-        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
     for name, ps in sorted(duplicates.items()):
         ordered = sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)
         survivor = ordered[0]
-        merge_ids = [p["id"] for p in ordered[1:]]
+        survivor_id = survivor.get("id")
+        merge_ids = [pid for p in ordered[1:] if (pid := p.get("id")) is not None]
         rprint(
             f"  [cyan]Merging {name!r} inside Immich:[/cyan] keeping "
-            f"[dim]{survivor['id'][:8]}…[/dim] ({survivor.get('assetCount', 0)} assets), "
+            f"[dim]{survivor_id[:8]}…[/dim] ({survivor.get('assetCount', 0)} assets), "
             f"absorbing {len(merge_ids)} smaller duplicate(s)..."
         )
-        if merge_people(survivor["id"], merge_ids):
+        if merge_people(survivor_id, merge_ids):
             rprint(f"  [green]✓ Merged {name!r}[/green]")
             merged_any = True
         else:
@@ -126,16 +137,22 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
     if merged_any:
         rprint("  [dim]Re-fetching people after merge...[/dim]")
         fresh = get_people()
+        if not fresh:
+            # Retry once: get_people() returns [] for both transient failures and
+            # auth errors (401); a second empty result strongly suggests a real failure.
+            fresh = get_people()
+        if not fresh:
+            logger.warning(
+                "Re-fetch after merge returned no people (tried twice)"
+                " — possible transient error or expired API key;"
+                " proceeding with pre-merge list. Check IMMICH_API_KEY if this recurs."
+            )
+            return _excl(people)
         # Filter out the smaller duplicate from any group whose merge failed — those
         # 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 = {
+        return _excl(fresh)
-            p["id"]
-            for ps in duplicates.values()
-            for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
-        }
-        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
     # downstream job creation never runs two jobs for the same Frigate folder.
@@ -143,12 +160,7 @@ def _handle_duplicate_people(people: list[dict]) -> list[dict]:
         "  [yellow]All merges failed — applying local deduplication"
         " to avoid overwriting output.[/yellow]"
     )
-    skip_ids = {
+    return _excl(people)
-        p["id"]
-        for ps in duplicates.values()
-        for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
-    }
-    return [p for p in people if p["id"] not in skip_ids]
 
 
 _UNSUPPORTED_VARS = [
@@ -173,12 +185,13 @@ def main() -> None:
         if trace_size:
             _handle_trace_crop(trace_size)
 
-        console.print(r"""
+        console.print(f"""
-    [bold blue]winnow[/bold blue]
+    [bold blue]winnow[/bold blue] [dim]v{__version__}[/dim]
     [dim]Immich -> Frigate Training Data Curator[/dim]
         """)
 
-        set_unsupported = [v for v in _UNSUPPORTED_VARS if os.environ.get(v)]
+        _FALSY = {"", "false", "0", "no", "off"}
+        set_unsupported = [v for v in _UNSUPPORTED_VARS if os.environ.get(v, "").strip().lower() not in _FALSY]
         if set_unsupported:
             console.print(
                 f"[bold yellow]⚠  Advanced tuning vars set: "
@@ -213,8 +226,7 @@ def main() -> None:
                         "and will be reset along with everyone else.[/yellow]"
                     )
                 if names:
-                    for name in names:
+                    reset_all_people()
-                        reset_person(name)
                     rprint(f"[bold yellow]Reset tracking data for all {len(names)} people.[/bold yellow]")
                 else:
                     rprint("[dim]No tracking data to reset.[/dim]")

winnow/config.py

@@ -127,12 +127,15 @@ class _Config:
         self.API_KEY = os.getenv("API_KEY")
         self.OUTPUT_DIR = os.getenv("OUTPUT_DIR", "./frigate_train")
         self.YEARS_FILTER = _getenv_int("YEARS_FILTER", 10)
+        if self.YEARS_FILTER < 0:
+            logging.warning("YEARS_FILTER=%s is negative — using default 10", self.YEARS_FILTER)
+            self.YEARS_FILTER = 10
         self.MIN_FACE_WIDTH = _getenv_int("MIN_FACE_WIDTH", 90)
         self.MIN_FACE_COUNT = _getenv_int("MIN_FACE_COUNT", 3)
         self.MERGE_DUPLICATE_PEOPLE = _getenv_bool("MERGE_DUPLICATE_PEOPLE", False)
         self.BLUR_THRESHOLD = _getenv_float("BLUR_THRESHOLD", 120.0)
         self.MIN_CONFIDENCE = _getenv_float("MIN_CONFIDENCE", 0.7)
-        self.MAX_AUTO_IMAGES = _getenv_int("MAX_AUTO_IMAGES", 20)
+        self.MAX_AUTO_IMAGES = _getenv_int("MAX_AUTO_IMAGES", 5)
         self.QUALITY_REPLACEMENT = _getenv_bool("QUALITY_REPLACEMENT", True)
         self.FRIGATE_SCORE_CEILING = _getenv_optional_float("FRIGATE_SCORE_CEILING")
         self.ENABLE_FRIGATE_SCORES = _getenv_bool("ENABLE_FRIGATE_SCORES", True)
@@ -173,7 +176,7 @@ class _Config:
                 data = json.loads(config_file.read_text())
                 if not self.IMMICH_URL:
                     self.IMMICH_URL = data.get("IMMICH_URL")
-                if os.getenv("OUTPUT_DIR") is None:
+                if not os.getenv("OUTPUT_DIR"):
                     self.OUTPUT_DIR = data.get("OUTPUT_DIR", self.OUTPUT_DIR)
             except (json.JSONDecodeError, OSError) as e:
                 logging.warning("Failed to load config file: %s", e)

winnow/diversity.py

@@ -57,9 +57,9 @@ def select_diverse_assets(
     Returns:
         List of selected assets
     """
-    # Fast path: fewer assets than limit
+    # Fast path: fewer assets than limit — sort for consistent ordering with other paths
     if limit != "auto" and len(assets) <= limit:
-        return assets
+        return sorted(assets, key=lambda x: x.get("fileCreatedAt", ""))
 
     # Sort by creation time
     assets = sorted(assets, key=lambda x: x.get("fileCreatedAt", ""))
@@ -181,6 +181,28 @@ def _crop_face_from_thumbnail(
     return crop
 
 
+def _scale_bbox_to_thumbnail(
+    bbox: tuple[float, float, float, float],
+    img: Image.Image,
+    asset: dict,
+    person_id: str | None = None,
+) -> tuple[float, float, float, float]:
+    """Scale a face bbox from original detection-image space to thumbnail-pixel space."""
+    x1, y1, x2, y2 = bbox
+    img_w, img_h = img.size
+    for person in asset.get("people", []):
+        if person_id and person.get("id") != person_id:
+            continue
+        faces = person.get("faces", [])
+        if faces:
+            meta_w = faces[0].get("imageWidth") or 0
+            meta_h = faces[0].get("imageHeight") or 0
+            scale_x = img_w / meta_w if meta_w else 1.0
+            scale_y = img_h / meta_h if meta_h else 1.0
+            return (x1 * scale_x, y1 * scale_y, x2 * scale_x, y2 * scale_y)
+    return bbox
+
+
 # =============================================================================
 # Embedding Collection
 # =============================================================================
@@ -258,9 +280,13 @@ def _select_by_embedding(
             confidence = _get_face_confidence(asset, person_id=person_id)
 
             face_bbox = _get_face_bbox(asset, person_id=person_id)
+            thumbnail_bbox = (
+                _scale_bbox_to_thumbnail(face_bbox, img, asset, person_id)
+                if face_bbox is not None else None
+            )
             quality = assess_quality(
                 img,
-                face_bbox=face_bbox,
+                face_bbox=thumbnail_bbox,
                 confidence=confidence,
                 blur_threshold=Config.BLUR_THRESHOLD,
                 min_face_px=Config.MIN_FACE_WIDTH,
@@ -277,6 +303,9 @@ def _select_by_embedding(
 
             emb = get_embedding(embed_img, asset_id=asset["id"])
             if emb is not None:
+                if np.linalg.norm(emb) < 1e-6:
+                    logger.debug("Zero-norm embedding for asset %s, skipping", asset["id"])
+                    continue
                 embeddings.append(emb)
                 valid_candidates.append(asset)
                 confidence_scores.append(confidence)
@@ -408,7 +437,8 @@ def _kmedoids(dist_matrix: np.ndarray, k: int, max_iter: int = 50) -> tuple[list
     for _ in range(max_iter):
         improved = False
         # Try swapping each medoid with a random non-medoid
-        non_medoids = [i for i in range(n) if i not in medoids]
+        medoid_set = set(medoids)
+        non_medoids = [i for i in range(n) if i not in medoid_set]
         if not non_medoids:
             break
 
@@ -483,7 +513,10 @@ def _cluster_aware_selection(
     norms = np.linalg.norm(emb_matrix, axis=1, keepdims=True)
     emb_normed = emb_matrix / np.maximum(norms, 1e-8)
 
-    # Build confidence weight array for hard example boosting
+    # Build confidence weight array for hard example boosting.
+    # Default to 1.0 for faces with no confidence score: treat as high-confidence
+    # (no boost) rather than hard-example territory. A missing score field should
+    # not cause these images to beat genuinely high-confidence detections in FPS.
     conf_array = np.ones(n)
     if confidence_scores:
         for i, c in enumerate(confidence_scores):
@@ -508,7 +541,6 @@ def _cluster_aware_selection(
 
     medoid_indices, cluster_labels = _kmedoids(dist_matrix, k)
     selected = list(medoid_indices)
-    selected_set = set(selected)
 
     logger.debug("Selected %s cluster medoids as initial picks.", len(selected))
 
@@ -522,10 +554,11 @@ def _cluster_aware_selection(
     for idx in selected:
         min_dists[idx] = -np.inf
 
+    # Hard example weighting: boost distance for low-confidence candidates.
+    # conf_array is constant after this point, so compute once outside the loop.
+    hard_weight = np.where(conf_array < 0.85, 1.0 + (0.85 - conf_array) * 2.0, 1.0)
+
     while len(selected) < target:
-        # Hard example weighting: boost distance for low-confidence candidates
-        # Confidence < 0.85 gets up to 1.5× distance boost
-        hard_weight = np.where(conf_array < 0.85, 1.0 + (0.85 - conf_array) * 2.0, 1.0)
         weighted_dists = min_dists * hard_weight
 
         best_idx = int(np.argmax(weighted_dists))
@@ -541,15 +574,19 @@ def _cluster_aware_selection(
             break
 
         selected.append(best_idx)
-        selected_set.add(best_idx)
 
         # Update min distances
         dists_to_new = dist_matrix[best_idx]
         min_dists = np.minimum(min_dists, dists_to_new)
         min_dists[best_idx] = -np.inf
 
-    selected_conf = [conf_array[i] for i in selected if conf_array[i] < 1.0]
+    hard_count = sum(
-    hard_count = sum(1 for c in selected_conf if c < 0.85)
+        1 for i in selected
+        if confidence_scores
+        and i < len(confidence_scores)
+        and confidence_scores[i] is not None
+        and confidence_scores[i] < 0.85
+    )
     logger.info("Selection complete: %s images (%s hard examples with confidence < 0.85).", len(selected), hard_count)
 
     # Slice to target: the while loop enforces this for non-auto mode, but
@@ -576,4 +613,4 @@ def _select_time_spread(assets: list, limit: int | str) -> list:
         return assets
 
     indices = np.linspace(0, len(assets) - 1, limit, dtype=int)
-    return [assets[i] for i in np.unique(indices)]
+    return [assets[i] for i in indices]

winnow/embeddings.py

@@ -26,22 +26,47 @@ logger = logging.getLogger(__name__)
 @contextmanager
 def _suppress_output():
     """Suppress stdout/stderr at the file-descriptor level, silencing C extension noise."""
-    devnull_fd = os.open(os.devnull, os.O_WRONLY)
+    devnull_fd = None
-    saved_out, saved_err = os.dup(1), os.dup(2)
+    saved_out = None
+    saved_err = None
     try:
+        devnull_fd = os.open(os.devnull, os.O_WRONLY)
+        saved_out = os.dup(1)
+        saved_err = os.dup(2)
         os.dup2(devnull_fd, 1)
         os.dup2(devnull_fd, 2)
         yield
     finally:
-        try:
+        # Each block is a separate sequential statement. A BaseException (e.g.
-            os.dup2(saved_out, 1)
+        # KeyboardInterrupt) raised inside block N would propagate past blocks N+1
-        finally:
+        # and N+2, leaving saved_err or devnull_fd unclosed. In CPython, KI is
+        # delivered between bytecodes, not mid-syscall; os.dup2 is a single C call
+        # and completes atomically, so this race is not realistically triggerable.
+        if saved_out is not None:
+            try:
+                os.dup2(saved_out, 1)
+            except OSError as e:
+                logger.debug("_suppress_output: failed to restore stdout fd: %s", e)
+            finally:
+                try:
+                    os.close(saved_out)
+                except OSError:
+                    pass
+        if saved_err is not None:
             try:
                 os.dup2(saved_err, 2)
+            except OSError as e:
+                logger.debug("_suppress_output: failed to restore stderr fd: %s", e)
             finally:
+                try:
+                    os.close(saved_err)
+                except OSError:
+                    pass
+        if devnull_fd is not None:
+            try:
                 os.close(devnull_fd)
-                os.close(saved_out)
+            except OSError:
-                os.close(saved_err)
+                pass
 
 
 # Lazy-loaded singleton
@@ -82,7 +107,6 @@ def get_insightface_app():
     global _insightface_app, _insightface_loaded
     if _insightface_loaded:
         return _insightface_app
-    _insightface_loaded = True
 
     ctx_id = -1
     insightface_home = os.environ.get("INSIGHTFACE_HOME", os.path.expanduser("~/.insightface"))
@@ -147,10 +171,12 @@ def get_insightface_app():
             _insightface_app.prepare(ctx_id=ctx_id, det_size=(640, 640))
 
         logger.info("InsightFace Buffalo_L: ready on %s (%.1fs)", device_str, time.time() - t0)
+        _insightface_loaded = True
         return _insightface_app
 
     except ImportError:
         logger.error("InsightFace not installed!")
+        _insightface_loaded = True
         return None
     except Exception as e:
         logger.error("Failed to load InsightFace: %s", e)
@@ -168,9 +194,11 @@ def get_insightface_app():
                     )
                     _insightface_app.prepare(ctx_id=-1, det_size=(640, 640))
                 logger.info("InsightFace Buffalo_L: ready on CPU (fallback, %.1fs)", time.time() - t0)
+                _insightface_loaded = True
                 return _insightface_app
             except Exception as ex:
                 logger.error("InsightFace CPU fallback failed: %s", ex)
+        _insightface_loaded = True
         return None
 
 
@@ -181,8 +209,9 @@ def get_face_embedding(img_pil: Image.Image) -> np.ndarray | None:
         return None
 
     try:
-        # InsightFace expects BGR cv2 image
+        # InsightFace expects BGR cv2 image; normalise mode first so RGBA/grayscale don't
-        img_bgr = cv2.cvtColor(np.asarray(img_pil), cv2.COLOR_RGB2BGR)
+        # raise a channel-count error inside cvtColor.
+        img_bgr = cv2.cvtColor(np.asarray(img_pil.convert("RGB")), cv2.COLOR_RGB2BGR)
 
         # Suppress scikit-image FutureWarning from InsightFace's face_align.py
         with warnings.catch_warnings():
@@ -192,9 +221,14 @@ def get_face_embedding(img_pil: Image.Image) -> np.ndarray | None:
         if not faces:
             return None
 
-        # Return embedding of largest face
+        # Return embedding of the face nearest the crop centre; a large margin can pull
-        largest = max(faces, key=lambda f: (f.bbox[2] - f.bbox[0]) * (f.bbox[3] - f.bbox[1]))
+        # a bigger neighbouring face into frame, and max-by-area would pick the wrong person.
-        return largest.embedding
+        cx, cy = img_pil.width / 2, img_pil.height / 2
+        nearest = min(
+            faces,
+            key=lambda f: ((f.bbox[0] + f.bbox[2]) / 2 - cx) ** 2 + ((f.bbox[1] + f.bbox[3]) / 2 - cy) ** 2,
+        )
+        return nearest.embedding
     except Exception as e:
         logger.error("Error getting face embedding: %s", e)
         return None

winnow/executor.py

@@ -1,6 +1,7 @@
 """Execution phase: image processing and Frigate upload."""
 
 import logging
+import operator
 import os
 import shutil
 from io import BytesIO
@@ -27,6 +28,10 @@ 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 (
+    REJECT_TRACKER_FILE,
+    UPLOAD_TRACKER_FILE,
+    begin_batch,
+    flush_batch,
     get_lowest_quality_mapped_file,
     get_most_redundant_mapped_file,
     get_tracked_frigate_file_count,
@@ -35,6 +40,7 @@ from .upload_tracker import (
     mark_rejected,
     mark_uploaded,
     remove_frigate_file,
+    remove_frigate_files_batch,
 )
 
 logger = logging.getLogger(__name__)
@@ -151,9 +157,9 @@ def execute_jobs(jobs: list[dict]) -> None:
                         if use_full_res:
                             img = fetch_full_image(asset["id"])
                             if img is None:
-                                # Both original and preview fallback failed — mark rejected
+                                # Full-res download failed — could be a transient network
-                                # so this asset isn't retried on every future run.
+                                # error, so don't mark rejected; it will be retried next run.
-                                mark_rejected(asset["id"], person_name=name)
+                                pass
                         else:
                             resp = requests.get(
                                 f"{Config.IMMICH_URL}/api/assets/{asset['id']}/thumbnail?size=preview&format=JPEG",
@@ -163,11 +169,11 @@ def execute_jobs(jobs: list[dict]) -> None:
                             if resp.ok:
                                 try:
                                     img = Image.open(BytesIO(resp.content))
-                                except PIL.UnidentifiedImageError:
+                                except (PIL.UnidentifiedImageError, OSError):
-                                    # Pillow cannot identify the format — genuinely corrupt
+                                    # Pillow cannot identify the format or the content is
-                                    # Immich thumbnail. Mark rejected so this asset isn't
+                                    # truncated. The download already succeeded (resp.ok),
-                                    # retried indefinitely. OSError/truncation errors are
+                                    # so this is a data problem, not a transient network
-                                    # transient and intentionally not caught here.
+                                    # error — mark rejected so it isn't retried forever.
                                     logger.warning("Invalid image data for asset %s — marking rejected", asset["id"])
                                     mark_rejected(asset["id"], person_name=name)
                                     img = None
@@ -180,12 +186,11 @@ def execute_jobs(jobs: list[dict]) -> None:
                             saved = process_face_mode(
                                 img, asset, person, person_dir, count, insightface_app=insightface_app
                             )
-                            if saved:
+                            if isinstance(saved, tuple):
                                 filename = f"{count}.jpg"
                                 asset_map[filename] = asset["id"]
                                 score_map[filename] = asset.get("quality_score")
-                                if isinstance(saved, tuple):
+                                dims_map[filename] = saved
-                                    dims_map[filename] = saved
                                 # Time-spread path: compute blur score from the downloaded
                                 # image. Capped at 1440px via blur_score_from_image() so the
                                 # scale matches the preview thumbnails the embedding path uses
@@ -196,8 +201,9 @@ def execute_jobs(jobs: list[dict]) -> None:
 
                                 count += 1
                             else:
+                                reason = saved if isinstance(saved, str) else "no usable face data"
                                 progress.console.print(
-                                    f"[yellow]Skipped {asset['id']} (no usable face data)[/yellow]"
+                                    f"[yellow]Skipped {asset['id']} ({reason})[/yellow]"
                                 )
                     except Exception as e:
                         logger.error("Failed to process asset %s: %s", asset.get("id", "<unknown>"), e)
@@ -345,252 +351,299 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                 # (manually deleted, or cleaned up outside winnow). This corrects the
                 # effective_count so those slots are available for new uploads.
                 stale = get_tracked_frigate_filenames(name) - known_frigate_files_at_start
-                for stale_fn in stale:
-                    remove_frigate_file(name, stale_fn)
                 if stale:
+                    remove_frigate_files_batch(name, list(stale))
                     progress.console.print(
                         f"  [dim]{name}: cleared {len(stale)} stale mapping(s)"
                         " (file(s) no longer in Frigate)[/dim]"
                     )
             effective_count = get_tracked_frigate_file_count(name)
-            pre_run_count = effective_count
             quality_replacement = job.get("config", {}).get("quality_replacement", False)
-            if Config.ENABLE_FRIGATE_SCORES and pre_run_count == 0:
+            if Config.ENABLE_FRIGATE_SCORES and effective_count == 0:
                 progress.console.print(
                     f"  [dim]{name}: first run — Frigate diversity scoring will apply from the next run[/dim]"
                 )
+            # Snapshot whether Frigate has a model before the upload loop starts.
+            # effective_count is incremented inside the loop on each successful upload,
+            # so using the live value would incorrectly trigger recognize_face calls
+            # mid-batch on the first run (after the first upload sets it to 1).
+            has_frigate_model = effective_count > 0
             actually_uploaded: list[tuple[str, str | None]] = []
             failed_deletes: set[str] = set()
             min_quality_score_for_slot: float | None = None
             person_has_fscores: bool = has_frigate_scores(name)
 
-            for fname in person_files:
+            begin_batch(UPLOAD_TRACKER_FILE)
-                fpath = os.path.join(person_dir, fname)
+            begin_batch(REJECT_TRACKER_FILE)
+            try:
+                for fname in person_files:
+                    fpath = os.path.join(person_dir, fname)
 
-                # If a previous replacement delete succeeded but that upload failed,
+                    # If a previous replacement delete succeeded but that upload failed,
-                # require the next candidate to beat the deleted file's score so the
+                    # require the next candidate to beat the deleted file's score so the
-                # freed slot isn't filled with something worse than what we removed.
+                    # freed slot isn't filled with something worse than what we removed.
-                if min_quality_score_for_slot is not None:
+                    if min_quality_score_for_slot is not None:
-                    file_score = score_map.get(fname)
+                        file_score = score_map.get(fname)
-                    if file_score is None or file_score <= min_quality_score_for_slot:
+                        if file_score is not None and file_score < min_quality_score_for_slot:
-                        score_str = f"{file_score:.3f}" if file_score is not None else "N/A"
+                            progress.console.print(
-                        progress.console.print(
+                                f"    [dim]⏭  {fname}: score {file_score:.3f} < freed slot floor"
-                            f"    [dim]⏭  {fname}: score {score_str} ≤ freed slot floor"
+                                f" {min_quality_score_for_slot:.3f}, skipping[/dim]"
-                            f" {min_quality_score_for_slot:.3f}, skipping[/dim]"
-                        )
-                        progress.advance(upload_task)
-                        continue
-
-                at_cap = effective_count >= Config.MAX_AUTO_IMAGES
-
-                # Pre-upload Frigate score — clean measurement (image not yet in training set).
-                # Called for all below-cap uploads (seeds frigate_scores for future at-cap
-                # replacement) and for at-cap uploads when scores already exist. Skipped on
-                # the first run (pre_run_count == 0) since Frigate has no model yet.
-                # recognize_face returns (face_name, score); we only use the score when the
-                # best match is for the correct person. Mismatches (or "unknown") are treated
-                # as None so a wrong-person score never drives a ceiling skip or replacement.
-                # Frigate rebuilds its model asynchronously after any delete (clear + background
-                # thread), so the first recognize call after a deletion returns None — our code
-                # handles this conservatively by skipping that candidate until the next run.
-                # LIMITATION — async rebuild during multi-replacement runs: each deletion in a
-                # single run triggers a background model rebuild in Frigate. Subsequent recognize
-                # calls in the same run may get None (rebuild in progress), causing later
-                # candidates to fall back to blur-score replacement or be skipped entirely.
-                # The more replacements that happen in one run, the worse the scoring gets.
-                # TODO(frigate-api): if Frigate exposes a model generation counter or a
-                # rebuild-complete signal, poll it between recognize calls during replacement
-                # sequences rather than accepting stale/None scores.
-                pre_fscore: float | None = None
-                if Config.ENABLE_FRIGATE_SCORES and pre_run_count > 0:
-                    if not at_cap or person_has_fscores:
-                        _result = recognize_face(fpath)
-                        if _result is not None and (_result[0] or "").casefold() == name.casefold():
-                            pre_fscore = _result[1]
-
-                # Below-cap novelty gate: skip candidates already covered by the Frigate model,
-                # including conditions learned from manually-added images winnow can't track.
-                # pre_fscore is None on the first run (pre_run_count == 0 skips recognize_face
-                # above), so this block never fires on the first run without an extra guard.
-                if not at_cap and pre_fscore is not None:
-                    _ceiling = Config.FRIGATE_SCORE_CEILING
-                    if _ceiling is None:
-                        # Dynamic default: bar = most-redundant tracked file's Frigate score.
-                        # Falls back to uploading freely when no tracked scores exist yet.
-                        _bar = get_most_redundant_mapped_file(name)
-                        _skip = _bar is not None and pre_fscore > _bar[2]
-                        _bar_str = f"most redundant tracked {_bar[2]:.2f}" if _bar else ""
-                    elif _ceiling == 0.0:
-                        _skip = False  # explicitly disabled
-                        _bar_str = ""
-                    else:
-                        _skip = pre_fscore > _ceiling
-                        _bar_str = f"ceiling {_ceiling:.2f}"
-                    if _skip:
-                        progress.console.print(
-                            f"    [dim]⏭  {fname}: Frigate score {pre_fscore:.2f}"
-                            f" > {_bar_str}, already covered[/dim]"
-                        )
-                        progress.advance(upload_task)
-                        continue
-
-                if at_cap:
-                    if not quality_replacement:
-                        progress.console.print(f"    [dim]⏭  {fname}: at cap, quality replacement disabled[/dim]")
-                        progress.advance(upload_task)
-                        continue
-
-                    using_fscore = person_has_fscores and Config.ENABLE_FRIGATE_SCORES
-                    if using_fscore:
-                        candidate_score = pre_fscore
-                        get_target = get_most_redundant_mapped_file
-                        score_label, better_note = "frigate", " (more novel)"
-                        no_score_msg = "Frigate recognize unavailable, skipping replacement"
-                    else:
-                        candidate_score = score_map.get(fname)
-                        get_target = get_lowest_quality_mapped_file
-                        score_label, better_note = "blur", ""
-                        no_score_msg = "no quality score, skipping replacement"
-
-                    if candidate_score is None:
-                        progress.console.print(f"    [dim]⏭  {fname}: {no_score_msg}[/dim]")
-                        progress.advance(upload_task)
-                        continue
-
-                    target = get_target(name, exclude=failed_deletes)
-                    not_better = target is None or (
-                        candidate_score >= target[2] if using_fscore else candidate_score <= target[2]
-                    )
-                    if not_better:
-                        target_str = f"{target[2]:.3f}" if target is not None else "N/A"
-                        op = "<" if using_fscore else ">"
-                        progress.console.print(
-                            f"    [dim]⏭  {fname}: {score_label} {candidate_score:.3f}"
-                            f" not {op} {target_str}, skipping[/dim]"
-                        )
-                        progress.advance(upload_task)
-                        continue
-
-                    target_frigate_file, _target_asset_id, target_score = target
-                    op = "<" if using_fscore else ">"
-                    progress.console.print(
-                        f"    🔄 {fname}: {score_label} {candidate_score:.3f} {op} {target_score:.3f},"
-                        f" replacing {target_frigate_file}{better_note}"
-                    )
-                    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:
-                        logger.warning("Failed to delete %s for %s, skipping replacement", target_frigate_file, name)
-                        failed_deletes.add(target_frigate_file)
-                        progress.advance(upload_task)
-                        continue
-
-                for attempt in range(1, max_retries + 1):
-                    try:
-                        with open(fpath, "rb") as f:
-                            resp = requests.post(
-                                f"{frigate_url}/api/faces/{encoded_name}/register",
-                                files={"file": (fname, f, "image/jpeg")},
-                                timeout=30,
                             )
-                        if resp.status_code == 200:
+                            progress.advance(upload_task)
-                            uploaded += 1
+                            continue
-                            person_uploaded += 1
-                            effective_count += 1
-                            min_quality_score_for_slot = None
 
-                            asset_id = asset_map.get(fname)
+                    at_cap = effective_count >= Config.MAX_AUTO_IMAGES
-                            if asset_id:
-                                try:
-                                    mark_uploaded(
-                                        asset_id,
-                                        person_name=name,
-                                        score=score_map.get(fname),
-                                        crop_dims=dims_map.get(fname),
-                                        frigate_score=pre_fscore,
-                                    )
-                                except Exception as tracker_exc:
-                                    # Upload to Frigate succeeded — don't retry on tracker
-                                    # failure or we'd upload a duplicate to Frigate.
-                                    logger.error(
-                                        "Tracker write failed for %s — upload succeeded"
-                                        " but asset may be re-selected next run: %s",
-                                        fname, tracker_exc,
-                                    )
-                                if pre_fscore is not None:
-                                    person_has_fscores = True
-                                actually_uploaded.append((fname, asset_id))
 
-                            break
+                    # Pre-upload Frigate score — clean measurement (image not yet in training set).
+                    # Called for all below-cap uploads (seeds frigate_scores for future at-cap
+                    # replacement) and for at-cap uploads when scores already exist.
+                    # Skipped when has_frigate_model is False (effective_count was 0 before the loop).
+                    # recognize_face returns (face_name, score); we only use the score when the
+                    # best match is for the correct person. Mismatches (or "unknown") are treated
+                    # as None so a wrong-person score never drives a ceiling skip or replacement.
+                    # Frigate rebuilds its model asynchronously after any delete (clear + background
+                    # thread), so the first recognize call after a deletion returns None — our code
+                    # handles this conservatively by skipping that candidate until the next run.
+                    # LIMITATION — async rebuild during multi-replacement runs: each deletion in a
+                    # single run triggers a background model rebuild in Frigate. Subsequent recognize
+                    # calls in the same run may get None (rebuild in progress), causing later
+                    # candidates to fall back to blur-score replacement or be skipped entirely.
+                    # The more replacements that happen in one run, the worse the scoring gets.
+                    # TODO(frigate-api): if Frigate exposes a model generation counter or a
+                    # rebuild-complete signal, poll it between recognize calls during replacement
+                    # sequences rather than accepting stale/None scores.
+                    pre_fscore: float | None = None
+                    if Config.ENABLE_FRIGATE_SCORES and has_frigate_model:
+                        if not at_cap or person_has_fscores:
+                            _result = recognize_face(fpath)
+                            if _result is not None and (_result[0] or "").casefold() == name.casefold():
+                                pre_fscore = _result[1]
+
+                    # Below-cap novelty gate: skip candidates already covered by the Frigate model,
+                    # including conditions learned from manually-added images winnow can't track.
+                    # pre_fscore is None when effective_count == 0 (no Frigate model yet),
+                    # so this block never fires on the first run without an extra guard.
+                    if not at_cap and pre_fscore is not None:
+                        _ceiling = Config.FRIGATE_SCORE_CEILING
+                        if _ceiling is None:
+                            # Dynamic default: bar = most-redundant tracked file's Frigate score.
+                            # Falls back to uploading freely when no tracked scores exist yet.
+                            _bar = get_most_redundant_mapped_file(name)
+                            _skip = _bar is not None and pre_fscore > _bar[2]
+                            _bar_str = f"most redundant tracked {_bar[2]:.2f}" if _bar else ""
+                        elif _ceiling == 0.0:
+                            _skip = False  # explicitly disabled
+                            _bar_str = ""
                         else:
+                            _skip = pre_fscore > _ceiling
+                            _bar_str = f"ceiling {_ceiling:.2f}"
+                        if _skip:
+                            progress.console.print(
+                                f"    [dim]⏭  {fname}: Frigate score {pre_fscore:.2f}"
+                                f" > {_bar_str}, already covered[/dim]"
+                            )
+                            progress.advance(upload_task)
+                            continue
+
+                    if at_cap:
+                        if not quality_replacement:
+                            progress.console.print(f"    [dim]⏭  {fname}: at cap, quality replacement disabled[/dim]")
+                            progress.advance(upload_task)
+                            continue
+
+                        using_fscore = person_has_fscores and Config.ENABLE_FRIGATE_SCORES
+                        if using_fscore:
+                            candidate_score = pre_fscore
+                            get_target = get_most_redundant_mapped_file
+                            score_label, better_note = "frigate", " (more novel)"
+                            no_score_msg = "Frigate recognize unavailable, skipping replacement"
+                            is_better_than = operator.lt
+                        else:
+                            candidate_score = score_map.get(fname)
+                            get_target = get_lowest_quality_mapped_file
+                            score_label, better_note = "blur", ""
+                            no_score_msg = "no quality score, skipping replacement"
+                            is_better_than = operator.gt
+
+                        if candidate_score is None:
+                            progress.console.print(f"    [dim]⏭  {fname}: {no_score_msg}[/dim]")
+                            progress.advance(upload_task)
+                            continue
+
+                        target = get_target(name, exclude=failed_deletes)
+                        not_better = target is None or not is_better_than(candidate_score, target[2])
+                        if not_better:
+                            target_str = f"{target[2]:.3f}" if target is not None else "N/A"
+                            cmp_op = "<" if using_fscore else ">"
+                            progress.console.print(
+                                f"    [dim]⏭  {fname}: {score_label} {candidate_score:.3f}"
+                                f" not {cmp_op} {target_str}, skipping[/dim]"
+                            )
+                            progress.advance(upload_task)
+                            continue
+
+                        target_frigate_file, _target_asset_id, target_score = target
+                        cmp_op = "<" if using_fscore else ">"
+                        progress.console.print(
+                            f"    🔄 {fname}: {score_label} {candidate_score:.3f} {cmp_op} {target_score:.3f},"
+                            f" replacing {target_frigate_file}{better_note}"
+                        )
+                        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:
+                            logger.warning(
+                                "Failed to delete %s for %s, skipping replacement",
+                                target_frigate_file, name,
+                            )
+                            failed_deletes.add(target_frigate_file)
+                            progress.advance(upload_task)
+                            continue
+
+                    for attempt in range(1, max_retries + 1):
+                        try:
+                            with open(fpath, "rb") as f:
+                                resp = requests.post(
+                                    f"{frigate_url}/api/faces/{encoded_name}/register",
+                                    files={"file": (fname, f, "image/jpeg")},
+                                    timeout=30,
+                                )
+                            if resp.status_code == 200:
+                                uploaded += 1
+                                person_uploaded += 1
+                                effective_count += 1
+                                min_quality_score_for_slot = None  # for/else rollback mirrors this pair
+                                asset_id = asset_map.get(fname)
+                                if asset_id:
+                                    try:
+                                        mark_uploaded(
+                                            asset_id,
+                                            person_name=name,
+                                            score=score_map.get(fname),
+                                            crop_dims=dims_map.get(fname),
+                                            frigate_score=pre_fscore,
+                                        )
+                                    except Exception as tracker_exc:
+                                        # Upload to Frigate succeeded — don't retry on tracker
+                                        # failure or we'd upload a duplicate to Frigate.
+                                        logger.error(
+                                            "Tracker write failed for %s — upload succeeded"
+                                            " but asset may be re-selected next run: %s",
+                                            fname, tracker_exc,
+                                        )
+                                    else:
+                                        if pre_fscore is not None:
+                                            person_has_fscores = True
+                                    # Always record for reconcile so the Frigate filename→asset_id
+                                    # mapping is created even when the tracker write fails.
+                                    # Trade-off: if mark_uploaded failed, asset_id is absent from
+                                    # asset_ids and scores. Consequences: (1) re-selected next run
+                                    # → Frigate duplicate; (2) excluded from quality-replacement
+                                    # candidates (_pick_mapped_file requires a scores entry);
+                                    # (3) counted toward MAX_AUTO_IMAGES cap (via frigate_files).
+                                    # The alternative — not appending — leaves the file permanently
+                                    # unmapped (reconcile never creates the frigate_files entry),
+                                    # making (2) and (3) permanent. Frigate duplicate is lesser.
+                                    actually_uploaded.append((fname, asset_id))
+
+                                break
+                            else:
+                                if attempt < max_retries:
+                                    logger.warning(
+                                        f"Upload attempt {attempt}/{max_retries} for {fname}:"
+                                        f" HTTP {resp.status_code}, retrying..."
+                                    )
+                                    continue
+                                failed += 1
+                                person_failed += 1
+                                progress.console.print(
+                                    f"    [red]✗ {fname}: HTTP {resp.status_code} (after {max_retries} attempts)[/red]"
+                                )
+                                full_body = resp.text
+                                try:
+                                    error_detail = resp.json().get("message", full_body[:100])
+                                except Exception:
+                                    error_detail = full_body[:100]
+                                if resp.status_code in (400, 500):
+                                    progress.console.print(f"      [dim]{error_detail}[/dim]")
+                                else:
+                                    logger.debug("%s HTTP %s: %s", fname, resp.status_code, error_detail)
+                                _is_permanent = (
+                                    (resp.status_code == 400 and "face" in full_body.lower())
+                                    or resp.status_code == 422
+                                    or (resp.status_code == 500 and "could not process" in full_body.lower())
+                                )
+                                if _is_permanent:
+                                    asset_id = asset_map.get(fname)
+                                    if asset_id:
+                                        mark_rejected(asset_id, person_name=name)
+                        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc:
                             if attempt < max_retries:
                                 logger.warning(
                                     f"Upload attempt {attempt}/{max_retries} for {fname}:"
-                                    f" HTTP {resp.status_code}, retrying..."
+                                    f" {type(exc).__name__}, retrying..."
+                                )
+                                continue
+                            failed += 1
+                            person_failed += 1
+                            label = (
+                                "Connection refused"
+                                if isinstance(exc, requests.exceptions.ConnectionError)
+                                else "Request timed out (30s)"
+                            )
+                            progress.console.print(
+                                f"    [red]✗ {fname}: {label} (after {max_retries} attempts)[/red]"
+                            )
+                        except Exception as e:
+                            if attempt < max_retries:
+                                logger.warning(
+                                    f"Upload attempt {attempt}/{max_retries} for {fname}:"
+                                    f" {type(e).__name__}, retrying..."
                                 )
                                 continue
                             failed += 1
                             person_failed += 1
                             progress.console.print(
-                                f"    [red]✗ {fname}: HTTP {resp.status_code} (after {max_retries} attempts)[/red]"
+                                f"    [red]✗ {fname}: {type(e).__name__} - {e} (after {max_retries} attempts)[/red]"
                             )
-                            full_body = resp.text
+                    else:
-                            try:
+                        # All retries exhausted without a successful upload.
-                                error_detail = resp.json().get("message", full_body[:100])
+                        # Restore the slot freed by the preceding delete so the next
-                            except Exception:
+                        # candidate still sees at_cap=True and must beat the replacement gate.
-                                error_detail = full_body[:100]
+                        # Also clear the quality floor — the deleted file's score no longer
-                            if resp.status_code == 400:
+                        # represents any live Frigate file, and leaving it blocks the next
-                                progress.console.print(f"      [dim]{error_detail}[/dim]")
+                        # candidate from filling the restored slot.
-                            else:
+                        if at_cap:
-                                logger.debug("%s HTTP %s: %s", fname, resp.status_code, error_detail)
+                            effective_count += 1
-                            _is_permanent = (
+                            min_quality_score_for_slot = None
-                                (resp.status_code == 400 and "face" in full_body.lower())
-                                or resp.status_code == 422
-                            )
-                            if _is_permanent:
-                                asset_id = asset_map.get(fname)
-                                if asset_id:
-                                    mark_rejected(asset_id, person_name=name)
-                    except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc:
-                        if attempt < max_retries:
-                            logger.warning(
-                                f"Upload attempt {attempt}/{max_retries} for {fname}:"
-                                f" {type(exc).__name__}, retrying..."
-                            )
-                            continue
-                        failed += 1
-                        person_failed += 1
-                        label = (
-                            "Connection refused"
-                            if isinstance(exc, requests.exceptions.ConnectionError)
-                            else "Request timed out (30s)"
-                        )
-                        progress.console.print(
-                            f"    [red]✗ {fname}: {label} (after {max_retries} attempts)[/red]"
-                        )
-                    except Exception as e:
-                        if attempt < max_retries:
-                            logger.warning(
-                                f"Upload attempt {attempt}/{max_retries} for {fname}:"
-                                f" {type(e).__name__}, retrying..."
-                            )
-                            continue
-                        failed += 1
-                        person_failed += 1
-                        progress.console.print(
-                            f"    [red]✗ {fname}: {type(e).__name__} - {e} (after {max_retries} attempts)[/red]"
-                        )
 
-                progress.advance(upload_task)
+                    progress.advance(upload_task)
 
-            if min_quality_score_for_slot is not None:
+                if min_quality_score_for_slot is not None:
-                logger.warning(
+                    logger.warning(
-                    f"{name}: freed replacement slot (floor {min_quality_score_for_slot:.3f})"
+                        f"{name}: freed replacement slot (floor {min_quality_score_for_slot:.3f})"
-                    " was not filled this run — will be available next run"
+                        " 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/frigate_api.py

@@ -146,8 +146,10 @@ def delete_frigate_person_files(person_name: str, filenames: list[str]) -> bool:
     Returns True on success, False if unreachable or the request fails.
     """
     frigate_url = _get_frigate_url()
-    if not frigate_url or not filenames:
+    if not frigate_url:
         return False
+    if not filenames:
+        return True
     from urllib.parse import quote
     encoded_name = quote(person_name, safe="")
     try:

winnow/image_processing.py

@@ -48,7 +48,9 @@ def align_face(img: Image.Image, landmarks: list[list[float]] | np.ndarray) -> I
         if lm.shape != (5, 2):
             logger.debug("Invalid landmark shape: %s, expected (5, 2)", lm.shape)
             return None
-        aligned = norm_crop(img_np, lm)
+        with warnings.catch_warnings():
+            warnings.filterwarnings("ignore", message=".*estimate.*is deprecated", category=FutureWarning)
+            aligned = norm_crop(img_np, lm)
         return Image.fromarray(aligned)
     except ImportError:
         logger.debug("InsightFace not available for face alignment")
@@ -66,10 +68,11 @@ def process_face_mode(
     count: int,
     min_width: int | None = None,
     insightface_app=None,
-) -> tuple[int, int] | None:
+) -> tuple[int, int] | str:
     """Crop face based on Immich metadata and save to output directory.
 
-    Returns (width, height) of the saved crop, or None if no crop was saved.
+    Returns (width, height) of the saved crop, or a skip-reason string if the
+    face was filtered out.
     When insightface_app is provided and ENABLE_FACE_ALIGNMENT is True,
     re-detects the face in the Immich bbox region using InsightFace to get
     precise landmarks for a proper 112x112 aligned crop. Falls back to
@@ -89,14 +92,17 @@ def process_face_mode(
 
     if not face_info:
         logger.debug("No face info for %s in asset %s", person.get("name"), asset.get("id"))
-        return None
+        return "no face metadata"
 
     img_w, img_h = img.size
-    meta_w = face_info.get("imageWidth") or img_w
+    meta_w = face_info.get("imageWidth") or 0
-    meta_h = face_info.get("imageHeight") or img_h
+    meta_h = face_info.get("imageHeight") or 0
 
-    # Scale bounding box to actual image dimensions
+    # Scale bounding box from detection-image space to actual image dimensions.
-    scale_x, scale_y = img_w / meta_w, img_h / meta_h
+    # Fall back to 1.0 if Immich omits the field — bbox is assumed to already
+    # be in image space (correct for thumbnails, wrong for full-res).
+    scale_x = img_w / meta_w if meta_w else 1.0
+    scale_y = img_h / meta_h if meta_h else 1.0
     x1 = face_info["boundingBoxX1"] * scale_x
     y1 = face_info["boundingBoxY1"] * scale_y
     x2 = face_info["boundingBoxX2"] * scale_x
@@ -105,7 +111,7 @@ def process_face_mode(
     face_w, face_h = x2 - x1, y2 - y1
     if face_w < min_width or face_h < min_width:
         logger.debug("Face too small (%.1fx%.1f)", face_w, face_h)
-        return None
+        return f"face too small ({face_w:.0f}x{face_h:.0f}px, min {min_width}px)"
 
     # Re-detect face with InsightFace for landmark-based alignment.
     # Immich's /api/faces endpoint does not include landmarks, so the

winnow/immich_api.py

@@ -39,7 +39,11 @@ def get_immich_version() -> tuple[int, int, int] | None:
         )
         if resp.ok:
             data = resp.json()
-            return (int(data["major"]), int(data["minor"]), int(data["patch"]))
+            major, minor, patch = data.get("major"), data.get("minor"), data.get("patch")
+            if major is None or minor is None or patch is None:
+                logger.debug("Unexpected Immich version schema: %s", data)
+                return None
+            return (int(major), int(minor), int(patch))
         return None
     except Exception:
         return None
@@ -57,8 +61,12 @@ def get_people() -> list[dict]:
             logger.error("Immich API key is invalid or expired (401 Unauthorized). Update API_KEY.")
             return []
         resp.raise_for_status()
-        return resp.json().get("people", [])
+        data = resp.json()
-    except (requests.RequestException, ValueError) as e:
+        if not isinstance(data, dict):
+            logger.error("Unexpected response shape from Immich /people: %r", type(data))
+            return []
+        return data.get("people") or []
+    except (requests.RequestException, ValueError, AttributeError) as e:
         logger.error("Failed to fetch people from Immich: %s", e)
         return []
 
@@ -118,11 +126,15 @@ def fetch_all_assets(person: dict) -> tuple[list[dict], int]:
                 logger.error("Error fetching assets for %s (page %s): %s", name, page, resp.status_code)
                 break
 
-            page_assets = resp.json().get("assets", [])
+            body = resp.json()
+            if not isinstance(body, dict):
+                logger.error("Unexpected response shape fetching assets for %s (page %s): %r", name, page, type(body))
+                break
+            page_assets = body.get("assets", [])
             # Immich ≥2.x returns {"assets": {"items": [...]}};
             # earlier versions returned {"assets": [...]} directly.
             if isinstance(page_assets, dict):
-                page_assets = page_assets.get("items", [])
+                page_assets = page_assets.get("items") or []
 
             page_count = len(page_assets)  # raw count for termination check before filtering
 
@@ -277,10 +289,11 @@ def filter_recent_assets(assets: list[dict], years: int | None = None) -> list[d
 
     logger.debug("Filtering assets older than %s years (%s)", years, cutoff)
 
-    recent, skipped = [], 0
+    recent, skipped, bad_timestamp = [], 0, 0
     for asset in assets:
         created_at_str = asset.get("fileCreatedAt")
         if not isinstance(created_at_str, str) or not created_at_str:
+            bad_timestamp += 1
             continue
 
         try:
@@ -290,9 +303,15 @@ def filter_recent_assets(assets: list[dict], years: int | None = None) -> list[d
                 recent.append(asset)
             else:
                 skipped += 1
-        except ValueError:
+        except (ValueError, TypeError):
+            bad_timestamp += 1
             continue
 
+    if bad_timestamp:
+        logger.warning(
+            "filter_recent_assets: %s asset(s) had missing or unparseable fileCreatedAt"
+            " and were excluded from the pool.", bad_timestamp
+        )
     logger.debug("Retained %s assets (filtered %s old assets).", len(recent), skipped)
     return recent
 

winnow/jobs.py

@@ -65,12 +65,20 @@ def _get_strategy_choice(has_embedding: bool) -> tuple[int | str, str]:
 
 def _resolve_strategy(strategy: str, has_embedding: bool) -> tuple[int | str, str]:
     """Resolve env var strategy to (limit, selection_mode) without prompts."""
+    if strategy == "skip":
+        return 0, "skip"
     if not has_embedding:
-        return _getenv_int("LIMIT", 30), "time"
+        limit = _getenv_int("LIMIT", 30)
+        if limit <= 0:
+            logger.warning("LIMIT=%s is invalid — ignoring and using default 30", limit)
+            limit = 30
+        return limit, "time"
 
     custom_limit = _getenv_optional_int("LIMIT")
     if custom_limit is not None:
-        return custom_limit, "smart"
+        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"),
@@ -78,7 +86,11 @@ def _resolve_strategy(strategy: str, has_embedding: bool) -> tuple[int | str, st
         "standard": (30, "smart"),
         "broad": (100, "smart"),
     }
-    return strategy_map.get(strategy, ("auto", "smart"))
+    result = strategy_map.get(strategy)
+    if result is None:
+        logger.warning("Unrecognised STRATEGY=%r — falling back to auto", strategy)
+        return ("auto", "smart")
+    return result
 
 
 def _perform_selection(
@@ -184,13 +196,20 @@ def _configure_person(person: dict, people: list[dict]) -> dict | None:
     return job
 
 
+def _valid_people(people: list[dict]) -> list[dict]:
+    return sorted(
+        [p for p in people if (p.get("name") or "").strip() and p.get("id")],
+        key=lambda x: x["name"],
+    )
+
+
 def interactive_configure(people: list[dict]) -> list[dict]:
     """Interactive phase: select person(s), mode, and configure training strategy.
 
     Supports multi-person batch mode — after configuring one person,
     prompts to add another.
     """
-    valid_people = sorted([p for p in people if p.get("name")], key=lambda x: x["name"])
+    valid_people = _valid_people(people)
 
     if not valid_people:
         rprint("[red]No people found with names in Immich.[/red]")
@@ -201,9 +220,9 @@ def interactive_configure(people: list[dict]) -> list[dict]:
     while True:
         # Select person
         console.print("\n[bold cyan]Select Person to Train:[/bold cyan]")
+        queued_ids = {j["person"]["id"] for j in jobs}
         for idx, p in enumerate(valid_people, 1):
-            # Mark already-queued people
+            marker = " [dim](queued)[/dim]" if p.get("id") in queued_ids else ""
-            marker = " [dim](queued)[/dim]" if any(j["person"]["id"] == p["id"] for j in jobs) else ""
             console.print(f"  [bold]{idx}.[/bold] {p['name']}{marker}")
 
         p_choice = IntPrompt.ask("Enter Number", choices=[str(i) for i in range(1, len(valid_people) + 1)])
@@ -222,20 +241,20 @@ def interactive_configure(people: list[dict]) -> list[dict]:
 
 def auto_configure(people: list[dict]) -> list[dict]:
     """Non-interactive: configure jobs for all named people automatically."""
-    valid_people = sorted([p for p in people if p.get("name")], key=lambda x: x["name"])
+    valid_people = _valid_people(people)
 
     if not valid_people:
         rprint("[red]No people found with names in Immich.[/red]")
         return []
 
     strategy = os.environ.get("STRATEGY", "auto")
-    skip = [s.strip() for s in os.environ.get("SKIP_PEOPLE", "").split(",") if s.strip()]
+    skip = {s.strip().casefold() for s in os.environ.get("SKIP_PEOPLE", "").split(",") if s.strip()}
-    only = [s.strip() for s in os.environ.get("ONLY_PEOPLE", "").split(",") if s.strip()]
+    only = {s.strip().casefold() for s in os.environ.get("ONLY_PEOPLE", "").split(",") if s.strip()}
 
     if only:
-        valid_people = [p for p in valid_people if p["name"] in only]
+        valid_people = [p for p in valid_people if p["name"].casefold() in only]
     if skip:
-        valid_people = [p for p in valid_people if p["name"] not in skip]
+        valid_people = [p for p in valid_people if p["name"].casefold() not in skip]
 
     min_face_count = Config.MIN_FACE_COUNT
 
@@ -293,10 +312,11 @@ def auto_configure(people: list[dict]) -> list[dict]:
         # decides per-image whether to swap; any candidate could be an improvement).
         if not quality_replacement_only:
             if limit == "auto":
-                # Switch from open-ended auto to a fixed budget at remaining capacity
-                # so the diversity selector itself stops at the right count instead of
-                # selecting MAX_AUTO_IMAGES and then discarding the excess by position.
                 if already_uploaded > 0:
+                    # Switch from open-ended auto to a fixed budget at remaining capacity
+                    # so the diversity selector stops at the right count instead of
+                    # selecting more than MAX_AUTO_IMAGES and overflowing the cap.
+                    # First runs keep limit="auto" so FPS adaptive early-stop can fire.
                     limit = capacity
             else:
                 limit = min(limit, capacity)

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 = _laplacian_var(img_np)
-    variance = cv2.Laplacian(gray, cv2.CV_64F).var()
     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 = _laplacian_var(img_np)
-    blur_score = float(cv2.Laplacian(gray, cv2.CV_64F).var())
 
     checks = [
         (
@@ -139,22 +142,24 @@ def assess_quality(
     return QualityResult(passed=len(reasons) == 0, reasons=reasons, blur_score=blur_score)
 
 
-def blur_score_from_image(img: Image.Image, max_dim: int = 1440) -> float:
+def blur_score_from_image(img: Image.Image, max_dim: int = 1440) -> float | None:
     """Compute Laplacian-variance blur score, capped at max_dim px to normalise scale.
 
     Caps resolution so full-res and thumbnail scores are comparable — Laplacian
     variance grows with pixel count, making uncapped full-res scores much larger
     than thumbnail scores for the same perceived sharpness.
 
-    Returns 0.0 on any error so callers can treat the result as lowest quality.
+    Returns None on error so callers can distinguish a failed measurement from a
+    legitimately low (near-zero) score.
     """
     try:
         score_img = img.convert("RGB") if img.mode != "RGB" else img
         if score_img.width > max_dim or score_img.height > max_dim:
-            score_img = score_img.copy()
+            if score_img is img:
+                score_img = score_img.copy()
             score_img.thumbnail((max_dim, max_dim), Image.LANCZOS)
-        return float(assess_quality(score_img).blur_score)
+        return _laplacian_var(np.array(score_img))
     except Exception as exc:
         logger.debug("blur_score_from_image failed: %s", exc)
-        return 0.0
+        return None
 

winnow/reconcile.py

@@ -61,7 +61,7 @@ def reconcile_frigate_mappings(
             try:
                 return float(fname.rsplit("_", 1)[-1].rsplit(".", 1)[0])
             except (ValueError, IndexError):
-                return 0.0
+                return float("inf")
 
         logger.debug(
             "%s: mapping %s file(s) by filename timestamp — assumes Frigate processes"

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__)
 
@@ -43,6 +43,8 @@ REJECT_TRACKER_FILE = "frigate_rejected_ids.json"
 # Reduces per-call JSON reads from O(calls) to O(1) after the first load.
 # 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:
@@ -69,20 +71,66 @@ def _load(filename: str) -> dict:
     return data
 
 
+def _write_to_disk(path: Path, data: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".tmp")
+    try:
+        with open(tmp, "w") as f:
+            json.dump(data, f, indent=2)
+        os.replace(tmp, path)
+    except Exception:
+        tmp.unlink(missing_ok=True)
+        raise
+
+
 def _save(filename: str, data: dict) -> None:
     path = _tracker_path(filename)
-    _cache[str(path)] = data  # keep cache consistent with what we write
+    key = str(path)
-    path.parent.mkdir(parents=True, exist_ok=True)
+    if key in _deferred:
-    with open(path, "w") as f:
+        _cache[key] = data  # accumulate in cache; disk write deferred until flush_batch()
-        json.dump(data, f, indent=2)
+        _dirty.add(key)
+        return
+    _write_to_disk(path, data)
+    _cache[key] = data  # update cache only after successful write
+
+
+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.
+
+    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)
+    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:
-    return "uploaded_asset_ids" if "uploaded" in filename else "rejected_asset_ids"
+    return "uploaded_asset_ids" if filename == UPLOAD_TRACKER_FILE else "rejected_asset_ids"
-
-
-def _load_flat(filename: str) -> set[str]:
-    return set(_load(filename).get(_flat_key(filename), []))
 
 
 def _get_ids(entry: list | dict) -> list[str]:
@@ -96,12 +144,14 @@ def _migrate_entry(entry: list | dict) -> dict:
     """Ensure by_person entry is in the current dict format."""
     if isinstance(entry, list):
         return {"asset_ids": sorted(entry), "scores": {}, "frigate_scores": {}, "frigate_files": {}, "crop_dims": {}}
-    entry.setdefault("asset_ids", [])
+    # Copy top-level and all nested dicts so callers' mutations never reach the cache.
-    entry.setdefault("scores", {})
+    result = dict(entry)
-    entry.setdefault("frigate_scores", {})
+    result["asset_ids"] = list(result.get("asset_ids", []))
-    entry.setdefault("frigate_files", {})
+    result["scores"] = dict(result.get("scores", {}))
-    entry.setdefault("crop_dims", {})
+    result["frigate_scores"] = dict(result.get("frigate_scores", {}))
-    return entry
+    result["frigate_files"] = dict(result.get("frigate_files", {}))
+    result["crop_dims"] = dict(result.get("crop_dims", {}))
+    return result
 
 
 def _mark(
@@ -112,35 +162,46 @@ 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)
-    flat_key = _flat_key(filename)
+    by_person = dict(data.get("by_person", {}))
-    flat = set(data.get(flat_key, []))
+    entry = _migrate_entry(by_person.get(person_name, {}))
-    flat.add(asset_id)
+    ids = set(entry["asset_ids"])
-    data[flat_key] = sorted(flat)
+    ids.add(asset_id)
-    if person_name:
+    entry["asset_ids"] = sorted(ids)
-        by_person = data.setdefault("by_person", {})
+    if score is not None:
-        entry = _migrate_entry(by_person.get(person_name, {}))
+        entry["scores"][asset_id] = round(score, 4)
-        ids = set(entry["asset_ids"])
+    if crop_dims is not None:
-        ids.add(asset_id)
+        entry["crop_dims"][asset_id] = [crop_dims[0], crop_dims[1]]
-        entry["asset_ids"] = sorted(ids)
+    if frigate_score is not None:
-        if score is not None:
+        entry["frigate_scores"][asset_id] = round(frigate_score, 4)
-            entry["scores"][asset_id] = round(score, 4)
+    by_person[person_name] = entry
-        if crop_dims is not None:
+    new_data = dict(data)
-            entry["crop_dims"][asset_id] = [crop_dims[0], crop_dims[1]]
+    new_data["by_person"] = by_person
-        if frigate_score is not None:
+    _save(filename, new_data)
-            entry["frigate_scores"][asset_id] = round(frigate_score, 4)
+    logger.debug("Marked %s in %s (%s)", asset_id, filename, person_name)
-        by_person[person_name] = entry
-    _save(filename, data)
 
 
 # ── Public API ────────────────────────────────────────────────────────────────
 
 def load_uploaded_ids() -> set[str]:
-    return _load_flat(UPLOAD_TRACKER_FILE)
+    """Return all asset IDs recorded as uploaded. Derives from by_person (primary)
+    plus any legacy flat list still present in old tracker files."""
+    data = _load(UPLOAD_TRACKER_FILE)
+    ids = {aid for e in data.get("by_person", {}).values() for aid in _get_ids(e)}
+    ids.update(data.get("uploaded_asset_ids", []))  # backward compat with pre-0.6.1 files
+    return ids
 
 
 def load_rejected_ids() -> set[str]:
-    return _load_flat(REJECT_TRACKER_FILE)
+    """Return all asset IDs recorded as rejected. Derives from by_person (primary)
+    plus any legacy flat list still present in old tracker files."""
+    data = _load(REJECT_TRACKER_FILE)
+    ids = {aid for e in data.get("by_person", {}).values() for aid in _get_ids(e)}
+    ids.update(data.get("rejected_asset_ids", []))  # backward compat with pre-0.6.1 files
+    return ids
 
 
 def mark_uploaded(
@@ -151,35 +212,30 @@ 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})")
+
 
 
 
 def record_frigate_file(person_name: str, frigate_filename: str, asset_id: str) -> None:
-    """Record the mapping from a Frigate training filename to an Immich asset ID."""
+    """Record a single Frigate filename → asset_id mapping."""
-    data = _load(UPLOAD_TRACKER_FILE)
+    record_frigate_files_batch(person_name, {frigate_filename: asset_id})
-    by_person = data.setdefault("by_person", {})
-    entry = _migrate_entry(by_person.get(person_name, {}))
-    entry["frigate_files"][frigate_filename] = asset_id
-    by_person[person_name] = entry
-    _save(UPLOAD_TRACKER_FILE, data)
-    logger.debug(f"Mapped Frigate file {frigate_filename} → {asset_id} ({person_name})")
 
 
 def record_frigate_files_batch(person_name: str, mappings: dict[str, str]) -> None:
     """Record multiple Frigate filename → asset_id mappings in a single load/save."""
     if not mappings:
         return
-    data = _load(UPLOAD_TRACKER_FILE)
+    src = _load(UPLOAD_TRACKER_FILE)
-    by_person = data.setdefault("by_person", {})
+    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}")
 
@@ -190,15 +246,26 @@ def remove_frigate_file(person_name: str, frigate_filename: str) -> None:
     Does NOT unmark the source asset_id — the deletion was deliberate and
     we don't want to re-upload the inferior image on the next run.
     """
-    data = _load(UPLOAD_TRACKER_FILE)
+    remove_frigate_files_batch(person_name, [frigate_filename])
-    by_person = data.get("by_person", {})
+
-    entry = _migrate_entry(by_person.get(person_name, {}))
+
-    asset_id = entry["frigate_files"].pop(frigate_filename, None)
+def remove_frigate_files_batch(person_name: str, frigate_filenames: list[str]) -> None:
-    if asset_id:
+    """Remove multiple Frigate filenames in a single load/save."""
-        entry["frigate_scores"].pop(asset_id, None)
+    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 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 Frigate file mapping {frigate_filename} ({person_name})")
+    logger.debug(f"Removed {len(frigate_filenames)} Frigate file mapping(s) for {person_name}")
 
 
 def get_tracked_frigate_file_count(person_name: str) -> int:
@@ -226,9 +293,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, {}))
+    raw = data.get("by_person", {}).get(person_name)
-    frigate_files = entry.get("frigate_files", {})
+    if not raw or isinstance(raw, list):
-    frigate_scores = entry.get("frigate_scores", {})
+        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())
 
 
@@ -238,11 +307,12 @@ def _pick_mapped_file(
     data = _load(UPLOAD_TRACKER_FILE)
     entry = _migrate_entry(data.get("by_person", {}).get(person_name, {}))
     scores = entry.get(score_key, {})
-    candidates = [
+    seen_assets: set[str] = set()
-        (ff, asset_id, scores[asset_id])
+    candidates = []
-        for ff, asset_id in entry.get("frigate_files", {}).items()
+    for ff, asset_id in entry.get("frigate_files", {}).items():
-        if (exclude is None or ff not in exclude) and asset_id in scores
+        if (exclude is None or ff not in exclude) and asset_id in scores and asset_id not in seen_assets:
-    ]
+            seen_assets.add(asset_id)
+            candidates.append((ff, asset_id, scores[asset_id]))
     if not candidates:
         return None
     return max(candidates, key=lambda x: x[2]) if highest else min(candidates, key=lambda x: x[2])
@@ -285,9 +355,13 @@ def find_by_crop_dimension(size: int) -> list[dict]:
         entry = _migrate_entry(raw_entry)
         scores = entry.get("scores", {})
         frigate_files = entry.get("frigate_files", {})
-        asset_to_frigate = {v: k for k, v in frigate_files.items()}
+        asset_to_frigate: dict[str, str] = {}
+        for fn, aid in frigate_files.items():
+            asset_to_frigate.setdefault(aid, fn)  # first-seen wins; plain inversion silently drops duplicates
         frigate_scores = entry.get("frigate_scores", {})
         for asset_id, dims in entry.get("crop_dims", {}).items():
+            if not isinstance(dims, (list, tuple)) or len(dims) < 2:
+                continue
             w, h = dims[0], dims[1]
             if w == size or h == size:
                 results.append({
@@ -305,11 +379,38 @@ def find_by_crop_dimension(size: int) -> list[dict]:
 def update_frigate_count(person_name: str, count: int) -> None:
     """Record Frigate's authoritative training image count for a person."""
     data = _load(UPLOAD_TRACKER_FILE)
-    by_person = data.setdefault("by_person", {})
+    by_person = dict(data.get("by_person", {}))
     entry = _migrate_entry(by_person.get(person_name, {}))
     entry["frigate_count"] = count
     by_person[person_name] = entry
-    _save(UPLOAD_TRACKER_FILE, data)
+    new_data = dict(data)
+    new_data["by_person"] = by_person
+    _save(UPLOAD_TRACKER_FILE, new_data)
+
+
+def reset_all_people() -> None:
+    """Reset all tracking data in two writes (O(P) Frigate API calls, O(1) disk writes).
+
+    Preferred over calling reset_person() in a loop when RESET_PERSON=* — that
+    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 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")
+    _save(UPLOAD_TRACKER_FILE, {})
+    _save(REJECT_TRACKER_FILE, {})
+    logger.info("Reset all tracking data")
 
 
 def reset_person(person_name: str) -> None:
@@ -324,7 +425,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}")
@@ -332,16 +433,23 @@ def reset_person(person_name: str) -> None:
             logger.warning(f"Could not delete Frigate files for {person_name} — tracker reset proceeding anyway")
 
     changed = False
-    tracker_files = ((UPLOAD_TRACKER_FILE, upload_data), (REJECT_TRACKER_FILE, _load(REJECT_TRACKER_FILE)))
+    for filename in (UPLOAD_TRACKER_FILE, REJECT_TRACKER_FILE):
-    for filename, data in tracker_files:
+        src = upload_data if filename == UPLOAD_TRACKER_FILE else _load(REJECT_TRACKER_FILE)
-        flat_key = _flat_key(filename)
+        by_person = dict(src.get("by_person", {}))  # copy so pop() does not mutate the cache
-        by_person = data.get("by_person", {})
         tracker_entry = by_person.pop(person_name, None)
         if tracker_entry is not None:
-            person_ids = set(_get_ids(tracker_entry))
+            data = dict(src)
-            flat = set(data.get(flat_key, [])) - person_ids
-            data[flat_key] = sorted(flat)
             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 and not isinstance(data[flat_key], list):
+                logger.warning(
+                    "reset_person: %s has unexpected type for %s (%s) — skipping flat-list cleanup;"
+                    " all persons' legacy IDs in this field are unaffected but unreadable",
+                    filename, flat_key, type(data[flat_key]).__name__,
+                )
+            elif person_ids and flat_key in data:
+                data[flat_key] = sorted(set(data[flat_key]) - person_ids)
             _save(filename, data)
             changed = True
     if changed:
@@ -357,14 +465,14 @@ def get_person_summary() -> dict[str, dict]:
     names = set(uploaded_data) | set(rejected_data)
     result = {}
     for name in sorted(names):
-        u_entry = uploaded_data.get(name, {})
+        u_entry = _migrate_entry(uploaded_data.get(name, {}))
-        r_entry = rejected_data.get(name, {})
+        r_entry = _migrate_entry(rejected_data.get(name, {}))
         result[name] = {
-            "uploaded": len(_get_ids(u_entry)),
+            "uploaded": len(u_entry["asset_ids"]),
-            "rejected": len(_get_ids(r_entry)),
+            "rejected": len(r_entry["asset_ids"]),
-            "frigate_count": u_entry.get("frigate_count") if isinstance(u_entry, dict) else None,
+            "frigate_count": u_entry.get("frigate_count"),
-            "scores": u_entry.get("scores", {}) if isinstance(u_entry, dict) else {},
+            "scores": u_entry["scores"],
-            "frigate_files": u_entry.get("frigate_files", {}) if isinstance(u_entry, dict) else {},
+            "frigate_files": u_entry["frigate_files"],
         }
     return result