Add quality replacement for Frigate face training images

When a person is at MAX_AUTO_IMAGES, winnow now replaces the
lowest-quality mapped training image in Frigate if a higher-confidence
candidate is available, keeping the training set always optimised.

Only files winnow uploaded (tracked via frigate_files mapping) are ever
replaced — manually added Frigate training images are never touched.
A concurrent-upload race condition is detected per-file: if N>1 new
files appear after one upload, the mapping is skipped rather than
guessed, logging at INFO level. The per-file snapshot approach is
retained over a batch approach because wrong mappings (which a batch
approach risks on race) are worse than no mapping.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **Quality replacement**: when a person is at `MAX_AUTO_IMAGES`, winnow now checks each new candidate against the lowest-quality image already in Frigate and swaps it in if the new image scores higher. Only images winnow uploaded (tracked in `frigate_files`) are ever replaced — files added manually through Frigate's UI are left untouched permanently. Enabled by default; set `QUALITY_REPLACEMENT=false` to revert to the previous behaviour of skipping people at cap.
+- **Frigate filename mapping**: each successful upload now records the mapping from Frigate's assigned filename to the originating Immich asset ID and face confidence score in the tracker (`frigate_files` field). This is the foundation for quality replacement and future management of the Frigate training set.
+- **`QUALITY_REPLACEMENT` env var** (default `true`): controls whether at-cap people are eligible for quality replacement. When disabled, people at `MAX_AUTO_IMAGES` are skipped as before.
+
 ## [0.2.12] - 2026-06-13
 
 ### Added

README.md

@@ -177,7 +177,8 @@ In scheduled mode the process (and loaded models) stays resident between runs. T
 | `USE_FULL_RESOLUTION` | `true` | Download full-resolution originals rather than preview thumbnails |
 | `MIN_CONFIDENCE` | `0.7` | Minimum Immich face detection confidence |
 | `BLUR_THRESHOLD` | `100.0` | Laplacian variance threshold — lower accepts more blur |
-| `MAX_AUTO_IMAGES` | `80` | Maximum images selected in auto mode |
+| `MAX_AUTO_IMAGES` | `80` | Maximum training images per person in Frigate |
+| `QUALITY_REPLACEMENT` | `true` | When at cap, replace the lowest-quality mapped training image if a better candidate is found. Only affects images winnow uploaded — manually added Frigate training files are never touched. Set `false` to disable and skip people already at cap |
 
 ### GPU & Models
 

tests/test_upload_tracker.py

@@ -69,3 +69,73 @@ def test_duplicate_marks_are_idempotent():
     mark_uploaded("dup", person_name="Alice")
     mark_uploaded("dup", person_name="Alice")
     assert filter_already_uploaded(["dup", "new"]) == ["new"]
+
+
+# ── frigate_files mapping ─────────────────────────────────────────────────────
+
+def test_record_and_remove_frigate_file():
+    from winnow.upload_tracker import get_person_summary, record_frigate_file, remove_frigate_file
+    record_frigate_file("Alice", "Alice-1000.webp", "asset-a1")
+    assert "Alice-1000.webp" in get_person_summary()["Alice"]["frigate_files"]
+    remove_frigate_file("Alice", "Alice-1000.webp")
+    assert "Alice-1000.webp" not in get_person_summary()["Alice"]["frigate_files"]
+
+
+def test_remove_nonexistent_frigate_file_is_safe():
+    from winnow.upload_tracker import remove_frigate_file
+    # Should not raise even if the file was never recorded
+    remove_frigate_file("Alice", "Alice-ghost.webp")
+
+
+def test_remove_frigate_file_does_not_unmark_asset():
+    """Deleting a Frigate file should not re-expose the source asset for upload."""
+    from winnow.upload_tracker import (
+        filter_already_uploaded,
+        mark_uploaded,
+        record_frigate_file,
+        remove_frigate_file,
+    )
+    mark_uploaded("asset-a1", person_name="Alice")
+    record_frigate_file("Alice", "Alice-1000.webp", "asset-a1")
+    remove_frigate_file("Alice", "Alice-1000.webp")
+    # Asset must still be excluded — it was deliberately replaced, not lost
+    assert filter_already_uploaded(["asset-a1"]) == []
+
+
+def test_get_lowest_quality_mapped_file_none_when_empty():
+    from winnow.upload_tracker import get_lowest_quality_mapped_file
+    assert get_lowest_quality_mapped_file("Alice") is None
+
+
+def test_get_lowest_quality_mapped_file_returns_lowest():
+    from winnow.upload_tracker import (
+        get_lowest_quality_mapped_file,
+        mark_uploaded,
+        record_frigate_file,
+    )
+    mark_uploaded("asset-hi", person_name="Alice", score=0.95)
+    mark_uploaded("asset-lo", person_name="Alice", score=0.71)
+    record_frigate_file("Alice", "Alice-1000.webp", "asset-hi")
+    record_frigate_file("Alice", "Alice-1001.webp", "asset-lo")
+    result = get_lowest_quality_mapped_file("Alice")
+    assert result is not None
+    frigate_filename, asset_id, score = result
+    assert frigate_filename == "Alice-1001.webp"
+    assert asset_id == "asset-lo"
+    assert score == pytest.approx(0.71, abs=0.001)
+
+
+def test_get_lowest_quality_mapped_file_skips_unscored():
+    """Files mapped without a score should not be returned as candidates."""
+    from winnow.upload_tracker import (
+        get_lowest_quality_mapped_file,
+        mark_uploaded,
+        record_frigate_file,
+    )
+    mark_uploaded("asset-scored", person_name="Alice", score=0.85)
+    mark_uploaded("asset-noscr", person_name="Alice")
+    record_frigate_file("Alice", "Alice-1000.webp", "asset-scored")
+    record_frigate_file("Alice", "Alice-1001.webp", "asset-noscr")
+    result = get_lowest_quality_mapped_file("Alice")
+    assert result is not None
+    assert result[1] == "asset-scored"  # only scored file is a candidate

winnow/config.py

@@ -30,6 +30,7 @@ class _Config:
     BLUR_THRESHOLD: float = 100.0
     MIN_CONFIDENCE: float = 0.7
     MAX_AUTO_IMAGES: int = 80
+    QUALITY_REPLACEMENT: bool = True
 
     # People filtering
     MIN_FACE_COUNT: int = 0
@@ -60,6 +61,7 @@ class _Config:
         self.BLUR_THRESHOLD = float(os.getenv("BLUR_THRESHOLD", "100.0"))
         self.MIN_CONFIDENCE = float(os.getenv("MIN_CONFIDENCE", "0.7"))
         self.MAX_AUTO_IMAGES = int(os.getenv("MAX_AUTO_IMAGES", "80"))
+        self.QUALITY_REPLACEMENT = os.getenv("QUALITY_REPLACEMENT", "true").lower() in ("true", "1", "yes")
         self.FACE_MARGIN = float(os.getenv("FACE_MARGIN", "0.15"))
         self.USE_FULL_RESOLUTION = os.getenv("USE_FULL_RESOLUTION", "true").lower() in ("true", "1", "yes")
         self.ENABLE_FACE_ALIGNMENT = os.getenv("ENABLE_FACE_ALIGNMENT", "true").lower() in ("true", "1", "yes")

winnow/executor.py

@@ -15,7 +15,14 @@ from .config import Config, get_headers
 from .image_processing import process_face_mode, process_full_mode, process_object_mode
 from .immich_api import fetch_face_data, fetch_full_image
 from .log_config import console
-from .upload_tracker import mark_rejected, mark_uploaded
+from .frigate_api import delete_frigate_person_files, get_frigate_person_files
+from .upload_tracker import (
+    get_lowest_quality_mapped_file,
+    mark_rejected,
+    mark_uploaded,
+    record_frigate_file,
+    remove_frigate_file,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -246,8 +253,49 @@ def upload_to_frigate(jobs: list[dict]) -> None:
             person_uploaded = 0
             person_failed = 0
 
+            # Snapshot current Frigate filenames so we can identify which file
+            # each upload produces (Frigate assigns its own filename on ingest).
+            known_frigate_files: set[str] = set(get_frigate_person_files(name) or [])
+            quality_replacement = job.get("config", {}).get("quality_replacement", False)
+
             for fname in person_files:
                 fpath = os.path.join(person_dir, fname)
+
+                # Quality replacement gate: when at cap, only upload if this image
+                # scores higher than the worst mapped file already in Frigate.
+                at_cap = len(known_frigate_files) >= Config.MAX_AUTO_IMAGES
+                if at_cap:
+                    if not quality_replacement:
+                        progress.console.print(f"    [dim]⏭  {fname}: at cap, quality replacement disabled[/dim]")
+                        progress.advance(upload_task)
+                        continue
+                    new_score = score_map.get(fname)
+                    if new_score is None:
+                        progress.console.print(f"    [dim]⏭  {fname}: no confidence score, skipping replacement[/dim]")
+                        progress.advance(upload_task)
+                        continue
+                    worst = get_lowest_quality_mapped_file(name)
+                    if worst is None or new_score <= worst[2]:
+                        progress.console.print(
+                            f"    [dim]⏭  {fname}: score {new_score:.3f} ≤ worst mapped"
+                            f" {worst[2]:.3f if worst else 'N/A'}, skipping[/dim]"
+                        )
+                        progress.advance(upload_task)
+                        continue
+                    # Delete the worst mapped file to make room for the better one
+                    worst_frigate_file, _worst_asset_id, worst_score = worst
+                    progress.console.print(
+                        f"    🔄 {fname}: score {new_score:.3f} > {worst_score:.3f},"
+                        f" replacing {worst_frigate_file}"
+                    )
+                    if delete_frigate_person_files(name, [worst_frigate_file]):
+                        remove_frigate_file(name, worst_frigate_file)
+                        known_frigate_files.discard(worst_frigate_file)
+                    else:
+                        logger.warning(f"Failed to delete {worst_frigate_file} for {name}, skipping replacement")
+                        progress.advance(upload_task)
+                        continue
+
                 for attempt in range(1, max_retries + 1):
                     try:
                         with open(fpath, "rb") as f:
@@ -265,6 +313,19 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                             if asset_id:
                                 mark_uploaded(asset_id, person_name=name, score=score_map.get(fname))
 
+                            # Identify the Frigate filename assigned to this upload
+                            # and record the mapping for future quality management.
+                            current_files = set(get_frigate_person_files(name) or known_frigate_files)
+                            new_files = current_files - known_frigate_files
+                            if len(new_files) == 1 and asset_id:
+                                record_frigate_file(name, next(iter(new_files)), asset_id)
+                            elif len(new_files) > 1:
+                                logger.info(
+                                    f"{name}: {len(new_files)} new Frigate files after uploading {fname}"
+                                    f" (concurrent upload detected) — skipping file mapping"
+                                )
+                            known_frigate_files = current_files
+
                             break
                         else:
                             if attempt < max_retries:

winnow/frigate_api.py

@@ -8,26 +8,73 @@ import requests
 logger = logging.getLogger(__name__)
 
 
-def get_frigate_face_counts() -> dict[str, int] | None:
-    """Return {person_name: training_image_count} from Frigate's train directory.
-
-    Returns None if FRIGATE_URL is not set or the API is unreachable, so callers
-    can distinguish "API unavailable" from "person has 0 images."
-    """
+def _get_faces_data() -> dict | None:
+    """Fetch raw GET /api/faces response. Returns None if unavailable."""
     frigate_url = os.environ.get("FRIGATE_URL", "").rstrip("/")
     if not frigate_url:
         return None
     try:
         resp = requests.get(f"{frigate_url}/api/faces", timeout=10)
         resp.raise_for_status()
-        data = resp.json()
-        # Response: {person_name: [file, ...], "train": [...], ...}
-        # "train" is a flat pending list, not a person — skip it.
-        return {
-            name: len(files)
-            for name, files in data.items()
-            if name != "train" and isinstance(files, list)
-        }
+        return resp.json()
     except Exception as e:
-        logger.warning(f"Could not query Frigate face counts: {e}")
+        logger.warning(f"Could not query Frigate faces API: {e}")
         return None
+
+
+def get_frigate_face_counts() -> dict[str, int] | None:
+    """Return {person_name: training_image_count} from Frigate's train directory.
+
+    Returns None if FRIGATE_URL is not set or the API is unreachable, so callers
+    can distinguish "API unavailable" from "person has 0 images."
+    """
+    data = _get_faces_data()
+    if data is None:
+        return None
+    # Response: {person_name: [file, ...], "train": [...], ...}
+    # "train" is a flat pending list, not a person — skip it.
+    return {
+        name: len(files)
+        for name, files in data.items()
+        if name != "train" and isinstance(files, list)
+    }
+
+
+def get_frigate_person_files(person_name: str) -> list[str] | None:
+    """Return the list of training filenames for a person in Frigate.
+
+    Returns None if the API is unreachable. Returns an empty list if the
+    person exists but has no training images yet.
+    """
+    data = _get_faces_data()
+    if data is None:
+        return None
+    files = data.get(person_name)
+    return files if isinstance(files, list) else []
+
+
+def delete_frigate_person_files(person_name: str, filenames: list[str]) -> bool:
+    """Delete specific training files for a person from Frigate.
+
+    Uses POST /api/faces/{name}/delete with body {"ids": [filename, ...]}.
+    Returns True on success, False if unreachable or the request fails.
+    """
+    frigate_url = os.environ.get("FRIGATE_URL", "").rstrip("/")
+    if not frigate_url or not filenames:
+        return False
+    from urllib.parse import quote
+    encoded = quote(person_name, safe="")
+    try:
+        resp = requests.post(
+            f"{frigate_url}/api/faces/{encoded}/delete",
+            json={"ids": filenames},
+            timeout=10,
+        )
+        if resp.ok:
+            logger.debug(f"Deleted {len(filenames)} Frigate file(s) for {person_name}")
+            return True
+        logger.warning(f"Frigate delete returned {resp.status_code} for {person_name}")
+        return False
+    except Exception as e:
+        logger.warning(f"Failed to delete Frigate files for {person_name}: {e}")
+        return False

winnow/jobs.py

@@ -293,24 +293,34 @@ def auto_configure(people: list[dict]) -> list[dict]:
             already_uploaded = fc if fc is not None else person_summary.get("uploaded", 0)
         capacity = Config.MAX_AUTO_IMAGES - already_uploaded
         if capacity <= 0:
+            if not Config.QUALITY_REPLACEMENT:
+                rprint(
+                    f"  [dim]Skipping {name} (at cap:"
+                    f" {already_uploaded}/{Config.MAX_AUTO_IMAGES}, quality replacement disabled).[/dim]"
+                )
+                continue
             rprint(
-                f"  [dim]Skipping {name} (at lifetime cap:"
-                f" {already_uploaded}/{Config.MAX_AUTO_IMAGES} trained).[/dim]"
+                f"  [cyan]{name}: at cap ({already_uploaded}/{Config.MAX_AUTO_IMAGES}),"
+                f" checking for quality improvements...[/cyan]"
             )
-            continue
+            quality_replacement_only = True
+        else:
+            quality_replacement_only = False
+
+        config["quality_replacement"] = quality_replacement_only or Config.QUALITY_REPLACEMENT
 
         has_embedding = is_embedding_available(entity_type)
         limit, selection_mode = _resolve_strategy(strategy, has_embedding)
 
-        # Cap selection to remaining capacity.
-        # For auto mode with partial training, keep "auto" so adaptive stopping
-        # still runs — just trim the result to the remaining capacity afterward.
+        # Cap selection to remaining capacity (no cap when replacement-only — executor
+        # decides per-image whether to swap; any candidate could be an improvement).
         auto_cap = None
-        if limit == "auto":
-            if already_uploaded > 0:
-                auto_cap = capacity
-        else:
-            limit = min(limit, capacity)
+        if not quality_replacement_only:
+            if limit == "auto":
+                if already_uploaded > 0:
+                    auto_cap = capacity
+            else:
+                limit = min(limit, capacity)
 
         if selection_mode == "skip":
             continue

winnow/upload_tracker.py

@@ -11,10 +11,14 @@ Both are excluded from future candidate pools. To reset:
 
 by_person schema (frigate_uploaded_ids.json):
   {
-    "asset_ids":    ["immich-id-1", ...],   # all assets we attempted to upload
-    "scores":       {"immich-id-1": 0.953}, # Immich face confidence at upload time
-    "frigate_count": 42                     # last known Frigate training image count
+    "asset_ids":     ["immich-id-1", ...],                    # all assets we attempted to upload
+    "scores":        {"immich-id-1": 0.953},                  # Immich face confidence at upload time
+    "frigate_files": {"PersonName-123.webp": "immich-id-1"},  # Frigate filename → asset ID
+    "frigate_count": 42                                       # last known Frigate training image count
   }
+
+frigate_files only contains files winnow uploaded — files added manually through
+Frigate's UI are never mapped here and are never touched by quality replacement.
 """
 
 import json
@@ -72,9 +76,10 @@ def _get_ids(entry: list | dict) -> list[str]:
 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": {}}
+        return {"asset_ids": sorted(entry), "scores": {}, "frigate_files": {}}
     entry.setdefault("asset_ids", [])
     entry.setdefault("scores", {})
+    entry.setdefault("frigate_files", {})
     return entry
 
 
@@ -116,6 +121,49 @@ def mark_rejected(asset_id: str, person_name: str | None = None) -> None:
     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."""
+    data = _load(UPLOAD_TRACKER_FILE)
+    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 remove_frigate_file(person_name: str, frigate_filename: str) -> None:
+    """Remove a Frigate filename from the mapping after it has been deleted.
+
+    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)
+    by_person = data.get("by_person", {})
+    entry = _migrate_entry(by_person.get(person_name, {}))
+    entry["frigate_files"].pop(frigate_filename, None)
+    by_person[person_name] = entry
+    _save(UPLOAD_TRACKER_FILE, data)
+    logger.debug(f"Removed Frigate file mapping {frigate_filename} ({person_name})")
+
+
+def get_lowest_quality_mapped_file(person_name: str) -> tuple[str, str, float] | None:
+    """Return (frigate_filename, asset_id, score) for the mapped file with the lowest
+    confidence score, or None if no mapped files with known scores exist."""
+    data = _load(UPLOAD_TRACKER_FILE)
+    entry = _migrate_entry(data.get("by_person", {}).get(person_name, {}))
+    frigate_files = entry.get("frigate_files", {})
+    scores = entry.get("scores", {})
+    candidates = [
+        (frigate_filename, asset_id, scores[asset_id])
+        for frigate_filename, asset_id in frigate_files.items()
+        if asset_id in scores
+    ]
+    if not candidates:
+        return None
+    return min(candidates, key=lambda x: x[2])
+
+
 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)
@@ -143,7 +191,7 @@ def reset_person(person_name: str) -> None:
 
 
 def get_person_summary() -> dict[str, dict]:
-    """Return {person_name: {uploaded, rejected, frigate_count, scores}} for display/capacity."""
+    """Return {person_name: {uploaded, rejected, frigate_count, scores, frigate_files}} for display/capacity."""
     uploaded_data = _load(UPLOAD_TRACKER_FILE).get("by_person", {})
     rejected_data = _load(REJECT_TRACKER_FILE).get("by_person", {})
     names = set(uploaded_data) | set(rejected_data)
@@ -156,6 +204,7 @@ def get_person_summary() -> dict[str, dict]:
             "rejected": len(_get_ids(r_entry)),
             "frigate_count": u_entry.get("frigate_count") if isinstance(u_entry, dict) else None,
             "scores": u_entry.get("scores", {}) if isinstance(u_entry, dict) else {},
+            "frigate_files": u_entry.get("frigate_files", {}) if isinstance(u_entry, dict) else {},
         }
     return result