Merge pull request 'Phase 3: confidence calibration fit + quality re-score' (#91) from feature/face-accuracy-phase3-recalibrate into master

.gitignore

@@ -108,3 +108,4 @@ viewer-frontend/.data/
 !viewer-frontend/public/logo.png
 !viewer-frontend/public/brand/
 !viewer-frontend/public/brand/**/*.png
+backend/data/confidence_calibration.json

ROADMAP.md

@@ -107,7 +107,7 @@ Living plan for product quality, auth/email reliability, and automation.
 
 - [x] **Phase 1 — quality sharpness** — Laplacian variance computed on the face crop (was kernel constant); unit tests in `tests/test_face_quality_score.py`; error-tag checklist in `docs/FACE_MATCH_ERROR_TAGGING.md`
 - [x] **Phase 2 — trusted refs + stricter Auto-Match + accept/reject log** — ref quality floor 0.5; browse/run tolerances 0.5/0.4; auto-accept default 85%; `match_decisions` table + logging on Save / auto-accept; tests in `tests/test_auto_match_phase2.py`
-- [ ] **Phase 3 — recalibrate confidence on labeled pairs; re-score corpus**
+- [x] **Phase 3 — recalibrate confidence + re-score quality** — fit distance→confidence knots from identified pairs / `match_decisions`; optional JSON overrides legacy curve; `scripts/fit_confidence_calibration.py` + `scripts/rescore_face_quality.py`; docs in `docs/FACE_CONFIDENCE_CALIBRATION.md`
 - [ ] **Phase 4 — multi-ref / ensemble embeddings**
 - [ ] **Phase 5 — harder reject of junk detections (tiny/blur/pose)**
 - [ ] **Phase 6 — multi-embedding per person**

backend/data/.gitkeep

@@ -0,0 +1 @@
+

backend/services/confidence_calibration.py

@@ -0,0 +1,203 @@
+"""Distance → confidence calibration (Phase 3).
+
+Default path keeps the legacy empirical piecewise curve. When a fitted
+knots file is present (from identified same/different pairs and/or
+``match_decisions``), ``calibrate_confidence`` interpolates those knots
+instead so displayed % tracks JRCC data.
+"""
+
+from __future__ import annotations
+
+import functools
+import json
+import os
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Iterable, Sequence
+
+import numpy as np
+
+# Default install path (overridable via CONFIDENCE_CALIBRATION_PATH)
+_DEFAULT_CALIBRATION_PATH = (
+    Path(__file__).resolve().parent.parent / "data" / "confidence_calibration.json"
+)
+
+# Minimum labeled distances before a fit may be applied (unless forced)
+MIN_SAME_PAIRS = 20
+MIN_DIFF_PAIRS = 20
+
+
+@dataclass(frozen=True)
+class CalibrationKnots:
+    """Monotone distance → confidence knots (distance ascending)."""
+
+    distances: tuple[float, ...]
+    confidences: tuple[float, ...]
+    source: str = "legacy"
+    n_same: int = 0
+    n_diff: int = 0
+    fitted_at: str | None = None
+
+    def __post_init__(self) -> None:
+        if len(self.distances) != len(self.confidences):
+            raise ValueError("distances and confidences must be same length")
+        if len(self.distances) < 2:
+            raise ValueError("need at least 2 knots")
+
+
+def default_calibration_path() -> Path:
+    override = os.getenv("CONFIDENCE_CALIBRATION_PATH")
+    if override:
+        return Path(override)
+    return _DEFAULT_CALIBRATION_PATH
+
+
+def interpolate_confidence(distance: float, knots: CalibrationKnots) -> float:
+    """Linear interpolate confidence for a cosine distance; clamp to knots ends."""
+    d = float(distance)
+    xs = np.asarray(knots.distances, dtype=np.float64)
+    ys = np.asarray(knots.confidences, dtype=np.float64)
+    if d <= xs[0]:
+        return float(ys[0])
+    if d >= xs[-1]:
+        return float(ys[-1])
+    conf = float(np.interp(d, xs, ys))
+    return max(1.0, min(100.0, conf))
+
+
+def legacy_empirical_confidence(distance: float, tolerance: float) -> float:
+    """Original desktop-parity piecewise curve (tolerance-relative bands)."""
+    if distance <= 0.12:
+        confidence = 100 * np.exp(-distance * 2.8)
+        return float(min(100, max(92, confidence)))
+    if distance <= tolerance * 0.5:
+        confidence = 100 * np.exp(-distance * 2.6)
+        return float(min(92, max(82, confidence)))
+    if distance <= tolerance:
+        normalized_distance = (distance - tolerance * 0.5) / (tolerance * 0.5)
+        confidence = 82 - (normalized_distance * 32)
+        return float(max(50, min(82, confidence)))
+    if distance <= tolerance * 1.5:
+        normalized_distance = (distance - tolerance) / (tolerance * 0.5)
+        confidence = 50 - (normalized_distance * 30)
+        return float(max(20, min(50, confidence)))
+    confidence = 20 * np.exp(-(distance - tolerance * 1.5) * 1.5)
+    return float(max(1, min(20, confidence)))
+
+
+def fit_knots_from_distances(
+    same_distances: Sequence[float],
+    different_distances: Sequence[float],
+    *,
+    n_bins: int = 16,
+    distance_max: float = 1.2,
+    source: str = "identified_pairs",
+) -> CalibrationKnots:
+    """Estimate P(same | distance) via Laplace-smoothed histograms → knots."""
+    same = np.asarray([float(x) for x in same_distances if x is not None], dtype=np.float64)
+    diff = np.asarray(
+        [float(x) for x in different_distances if x is not None], dtype=np.float64
+    )
+    if same.size == 0 or diff.size == 0:
+        raise ValueError("need at least one same and one different distance")
+
+    edges = np.linspace(0.0, distance_max, n_bins + 1)
+    mids = 0.5 * (edges[:-1] + edges[1:])
+    same_hist, _ = np.histogram(np.clip(same, 0, distance_max), bins=edges)
+    diff_hist, _ = np.histogram(np.clip(diff, 0, distance_max), bins=edges)
+    p_same = (same_hist + 1.0) / (same_hist + diff_hist + 2.0)
+    conf = p_same * 100.0
+
+    for i in range(1, len(conf)):
+        if conf[i] > conf[i - 1]:
+            conf[i] = conf[i - 1]
+
+    distances = (0.0, *tuple(float(x) for x in mids), float(distance_max))
+    confidences = (
+        float(min(100.0, max(float(conf[0]), 95.0))),
+        *tuple(float(max(1.0, min(100.0, c))) for c in conf),
+        float(max(1.0, min(float(conf[-1]), 10.0))),
+    )
+    conf_list = list(confidences)
+    for i in range(1, len(conf_list)):
+        if conf_list[i] > conf_list[i - 1]:
+            conf_list[i] = conf_list[i - 1]
+
+    return CalibrationKnots(
+        distances=distances,
+        confidences=tuple(conf_list),
+        source=source,
+        n_same=int(same.size),
+        n_diff=int(diff.size),
+        fitted_at=datetime.now(timezone.utc).isoformat(),
+    )
+
+
+def knots_ready_to_apply(
+    knots: CalibrationKnots,
+    *,
+    min_same: int = MIN_SAME_PAIRS,
+    min_diff: int = MIN_DIFF_PAIRS,
+) -> bool:
+    return knots.n_same >= min_same and knots.n_diff >= min_diff
+
+
+def save_calibration_knots(knots: CalibrationKnots, path: Path | None = None) -> Path:
+    dest = path or default_calibration_path()
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "distances": list(knots.distances),
+        "confidences": list(knots.confidences),
+        "source": knots.source,
+        "n_same": knots.n_same,
+        "n_diff": knots.n_diff,
+        "fitted_at": knots.fitted_at,
+    }
+    dest.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
+    clear_calibration_cache()
+    return dest
+
+
+@functools.lru_cache(maxsize=4)
+def _load_fitted_knots_cached(path_str: str) -> CalibrationKnots | None:
+    path = Path(path_str)
+    if not path.is_file():
+        return None
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        return CalibrationKnots(
+            distances=tuple(float(x) for x in data["distances"]),
+            confidences=tuple(float(x) for x in data["confidences"]),
+            source=str(data.get("source", "file")),
+            n_same=int(data.get("n_same", 0)),
+            n_diff=int(data.get("n_diff", 0)),
+            fitted_at=data.get("fitted_at"),
+        )
+    except Exception:
+        return None
+
+
+def load_calibration_knots(path: Path | None = None) -> CalibrationKnots | None:
+    """Load fitted knots from disk, or None if missing/invalid."""
+    return _load_fitted_knots_cached(str(path or default_calibration_path()))
+
+
+def clear_calibration_cache() -> None:
+    _load_fitted_knots_cached.cache_clear()
+
+
+def get_active_knots() -> CalibrationKnots | None:
+    """Fitted knots if present; otherwise None (caller uses legacy curve)."""
+    return load_calibration_knots()
+
+
+def merge_distance_lists(*groups: Iterable[float]) -> list[float]:
+    out: list[float] = []
+    for group in groups:
+        out.extend(float(x) for x in group if x is not None)
+    return out
+
+
+def knots_to_dict(knots: CalibrationKnots) -> dict:
+    return asdict(knots)

backend/services/face_service.py

@@ -1839,6 +1839,7 @@ def calibrate_confidence(distance: float, tolerance: float = None) -> float:
     """Convert distance to calibrated confidence percentage, matching desktop exactly.
     
     Uses empirical calibration method matching desktop _calibrate_confidence.
+    When a fitted knots file exists (Phase 3), interpolates those instead.
     Args:
         distance: Cosine distance (0 = identical, 2 = opposite)
         tolerance: Matching tolerance threshold (default: DEFAULT_FACE_TOLERANCE)
@@ -1863,43 +1864,17 @@ def calibrate_confidence(distance: float, tolerance: float = None) -> float:
         sigmoid_factor = 1 / (1 + np.exp(5 * (normalized_distance - 1)))
         return max(1, min(100, sigmoid_factor * 100))
     
-    else:  # "empirical" - default method (matching desktop exactly)
-        # Empirical calibration parameters for DeepFace ArcFace model
-        # These are derived from analysis of distance distributions for matching/non-matching pairs
-        # Moderate calibration: stricter than original but not too strict
-        
-        # For very close distances (< 0.12): very high confidence
-        if distance <= 0.12:
-            # Very close matches: exponential decay from 100%
-            confidence = 100 * np.exp(-distance * 2.8)
-            return min(100, max(92, confidence))
-        
-        # For distances well below threshold: high confidence
-        elif distance <= tolerance * 0.5:
-            # Close matches: exponential decay
-            confidence = 100 * np.exp(-distance * 2.6)
-            return min(92, max(82, confidence))
-        
-        # For distances near threshold: moderate confidence  
-        elif distance <= tolerance:
-            # Near-threshold matches: sigmoid-like curve
-            # Maps distance to probability based on empirical data
-            normalized_distance = (distance - tolerance * 0.5) / (tolerance * 0.5)
-            confidence = 82 - (normalized_distance * 32)  # 82% to 50% range
-            return max(50, min(82, confidence))
-        
-        # For distances above threshold: low confidence
-        elif distance <= tolerance * 1.5:
-            # Above threshold but not too far: rapid decay
-            normalized_distance = (distance - tolerance) / (tolerance * 0.5)
-            confidence = 50 - (normalized_distance * 30)  # 50% to 20% range
-            return max(20, min(50, confidence))
-        
-        # For very large distances: very low confidence
-        else:
-            # Very far matches: very low probability
-            confidence = 20 * np.exp(-(distance - tolerance * 1.5) * 1.5)
-            return max(1, min(20, confidence))
+    else:  # "empirical" - default method
+        from backend.services.confidence_calibration import (
+            get_active_knots,
+            interpolate_confidence,
+            legacy_empirical_confidence,
+        )
+
+        knots = get_active_knots()
+        if knots is not None:
+            return interpolate_confidence(distance, knots)
+        return legacy_empirical_confidence(distance, tolerance)
 
 
 def _calculate_face_size_ratio(face: Face, photo: Photo) -> float:

backend/services/quality_rescore.py

@@ -0,0 +1,249 @@
+"""Quality re-score and labeled-pair sampling for Phase 3 calibration."""
+
+from __future__ import annotations
+
+import json
+import random
+from pathlib import Path
+from typing import Callable, List, Optional, Tuple
+
+import numpy as np
+from PIL import Image
+from sqlalchemy.orm import Session, joinedload
+
+from backend.db.models import Face, MatchDecision, PersonEncoding, Photo
+from backend.services.confidence_calibration import (
+    CalibrationKnots,
+    fit_knots_from_distances,
+    merge_distance_lists,
+)
+from backend.services.face_service import (
+    calculate_cosine_distance,
+    calculate_face_quality_score,
+    load_face_encoding,
+)
+from src.utils.exif_utils import EXIFOrientationHandler
+
+
+def _parse_location(raw) -> dict:
+    if isinstance(raw, dict):
+        return raw
+    return json.loads(raw) if raw else {}
+
+
+def _load_photo_array(photo_path: Path) -> Optional[np.ndarray]:
+    if not photo_path.is_file():
+        return None
+    try:
+        corrected_image, _orientation = (
+            EXIFOrientationHandler.correct_image_orientation_from_path(str(photo_path))
+        )
+        if corrected_image is not None:
+            return np.array(corrected_image.convert("RGB"))
+        return np.array(Image.open(photo_path).convert("RGB"))
+    except Exception:
+        try:
+            return np.array(Image.open(photo_path).convert("RGB"))
+        except Exception:
+            return None
+
+
+def rescore_face_quality(
+    db: Session,
+    face: Face,
+    *,
+    commit: bool = False,
+) -> Tuple[float, float] | None:
+    """Recompute quality_score for one face from its crop (Phase 1 sharpness fix).
+
+    Returns (old_score, new_score) or None if the photo/crop cannot be loaded.
+    """
+    photo = face.photo
+    if photo is None:
+        photo = db.query(Photo).filter(Photo.id == face.photo_id).first()
+    if photo is None:
+        return None
+
+    image_np = _load_photo_array(Path(photo.path))
+    if image_np is None:
+        return None
+
+    loc = _parse_location(face.location)
+    h, w = image_np.shape[:2]
+    new_int = calculate_face_quality_score(image_np, loc, w, h)
+    new_score = new_int / 100.0
+    old_score = float(face.quality_score) if face.quality_score is not None else 0.0
+    face.quality_score = new_score
+    db.add(face)
+    db.query(PersonEncoding).filter(PersonEncoding.face_id == face.id).update(
+        {"quality_score": new_score},
+        synchronize_session=False,
+    )
+    if commit:
+        db.commit()
+    return old_score, new_score
+
+
+def rescore_face_qualities(
+    db: Session,
+    *,
+    limit: int | None = None,
+    offset: int = 0,
+    dry_run: bool = False,
+    commit_every: int = 50,
+    progress: Optional[Callable[[int, int], None]] = None,
+) -> dict:
+    """Batch re-score face quality. Does not re-detect or re-embed."""
+    query = (
+        db.query(Face)
+        .options(joinedload(Face.photo))
+        .order_by(Face.id.asc())
+        .offset(offset)
+    )
+    if limit is not None:
+        query = query.limit(limit)
+    faces: List[Face] = query.all()
+    total = len(faces)
+    updated = 0
+    skipped = 0
+    deltas: list[float] = []
+
+    for i, face in enumerate(faces, start=1):
+        result = rescore_face_quality(db, face, commit=False)
+        if result is None:
+            skipped += 1
+        else:
+            old, new = result
+            deltas.append(new - old)
+            updated += 1
+            if dry_run:
+                # Roll back in-memory change for dry-run
+                face.quality_score = old
+        if progress:
+            progress(i, total)
+        if not dry_run and updated and updated % commit_every == 0:
+            db.commit()
+
+    if dry_run:
+        db.rollback()
+    elif updated:
+        db.commit()
+
+    return {
+        "examined": total,
+        "updated": updated,
+        "skipped": skipped,
+        "dry_run": dry_run,
+        "mean_delta": float(np.mean(deltas)) if deltas else 0.0,
+        "max_abs_delta": float(np.max(np.abs(deltas))) if deltas else 0.0,
+    }
+
+
+def sample_identified_pair_distances(
+    db: Session,
+    *,
+    max_same: int = 500,
+    max_diff: int = 500,
+    min_quality: float = 0.3,
+    seed: int = 42,
+) -> Tuple[list[float], list[float]]:
+    """Cosine distances for same-person vs different-person identified faces."""
+    faces: List[Face] = (
+        db.query(Face)
+        .filter(Face.person_id.isnot(None))
+        .filter(Face.encoding.isnot(None))
+        .filter(Face.quality_score >= min_quality)
+        .all()
+    )
+    by_person: dict[int, list[Face]] = {}
+    for f in faces:
+        by_person.setdefault(int(f.person_id), []).append(f)
+
+    rng = random.Random(seed)
+    same: list[float] = []
+    # Same-person pairs
+    for person_faces in by_person.values():
+        if len(person_faces) < 2:
+            continue
+        for i in range(len(person_faces)):
+            for j in range(i + 1, len(person_faces)):
+                a = load_face_encoding(person_faces[i].encoding)
+                b = load_face_encoding(person_faces[j].encoding)
+                same.append(calculate_cosine_distance(a, b))
+    rng.shuffle(same)
+    same = same[:max_same]
+
+    # Different-person pairs (sample)
+    person_ids = [pid for pid, fl in by_person.items() if fl]
+    diff: list[float] = []
+    attempts = 0
+    max_attempts = max_diff * 20
+    while len(diff) < max_diff and attempts < max_attempts and len(person_ids) >= 2:
+        attempts += 1
+        p1, p2 = rng.sample(person_ids, 2)
+        f1 = rng.choice(by_person[p1])
+        f2 = rng.choice(by_person[p2])
+        a = load_face_encoding(f1.encoding)
+        b = load_face_encoding(f2.encoding)
+        diff.append(calculate_cosine_distance(a, b))
+
+    return same, diff
+
+
+def distances_from_match_decisions(
+    db: Session,
+) -> Tuple[list[float], list[float]]:
+    """Accept → same distances; reject → different (when distance was logged)."""
+    rows: List[MatchDecision] = (
+        db.query(MatchDecision).filter(MatchDecision.distance.isnot(None)).all()
+    )
+    same: list[float] = []
+    diff: list[float] = []
+    for row in rows:
+        d = float(row.distance) if row.distance is not None else None
+        if d is None:
+            continue
+        if row.decision == "accept":
+            same.append(d)
+        elif row.decision == "reject":
+            diff.append(d)
+    return same, diff
+
+
+def fit_calibration_from_db(
+    db: Session,
+    *,
+    max_same: int = 500,
+    max_diff: int = 500,
+    include_decisions: bool = True,
+) -> Tuple[CalibrationKnots, dict]:
+    """Fit confidence knots from identified pairs (+ optional match_decisions)."""
+    same_id, diff_id = sample_identified_pair_distances(
+        db, max_same=max_same, max_diff=max_diff
+    )
+    same_dec: list[float] = []
+    diff_dec: list[float] = []
+    if include_decisions:
+        same_dec, diff_dec = distances_from_match_decisions(db)
+
+    same = merge_distance_lists(same_id, same_dec)
+    diff = merge_distance_lists(diff_id, diff_dec)
+    stats = {
+        "same_identified": len(same_id),
+        "diff_identified": len(diff_id),
+        "same_decisions": len(same_dec),
+        "diff_decisions": len(diff_dec),
+        "same_total": len(same),
+        "diff_total": len(diff),
+    }
+    if not same or not diff:
+        raise ValueError(
+            "Not enough labeled distances to fit "
+            f"(same={len(same)}, diff={len(diff)}). "
+            "Identify more people or accumulate Auto-Match decisions."
+        )
+    source = "identified_pairs"
+    if same_dec or diff_dec:
+        source = "identified_pairs+match_decisions"
+    knots = fit_knots_from_distances(same, diff, source=source)
+    return knots, stats

docs/FACE_CONFIDENCE_CALIBRATION.md

@@ -0,0 +1,52 @@
+# Face confidence calibration & quality re-score (Phase 3)
+
+After Phase 1 (honest sharpness) and Phase 2 (decision log + stricter
+Auto-Match), Phase 3 makes displayed confidence track JRCC data and
+backfills `quality_score` without re-detecting faces.
+
+## Confidence calibration
+
+`calibrate_confidence` still uses the legacy empirical curve by default.
+
+When `backend/data/confidence_calibration.json` exists (or
+`CONFIDENCE_CALIBRATION_PATH`), it **interpolates fitted knots** instead.
+Knots are estimated from:
+
+1. **Identified same-person / different-person** embedding pairs in the DB
+2. **`match_decisions`** rows that include `distance` (Auto-Match accept/reject)
+
+### Fit on the app host
+
+```bash
+cd /opt/punimtag
+set -a && . ./.env && set +a
+./venv/bin/python scripts/fit_confidence_calibration.py --dry-run
+# when same≥20 and diff≥20 (or --force):
+./venv/bin/python scripts/fit_confidence_calibration.py --apply
+sudo -u appuser pm2 restart punimtag-api --update-env
+```
+
+Do **not** commit a production knots file with real corpus stats unless you
+intend every environment to share it. Prefer generating per host.
+
+## Quality re-score
+
+Existing `faces.quality_score` values were computed with the broken
+sharpness term. Re-score crops only (no DeepFace):
+
+```bash
+cd /opt/punimtag
+set -a && . ./.env && set +a
+./venv/bin/python scripts/rescore_face_quality.py --dry-run --limit 100
+./venv/bin/python scripts/rescore_face_quality.py   # all faces
+```
+
+Also updates `person_encodings.quality_score` for the same face ids.
+
+## Ops notes
+
+- DEV currently has few identified people — fit may need `--force` or more
+  Identify / Auto-Match Saves before knots are trustworthy.
+- Keep sampling misses in `FACE_MATCH_ERROR_TAGGING.md` (private sheet).
+- After quality re-score, Auto-Match reference picking (quality ≥ 0.5) uses
+  updated scores immediately (no API restart required for DB values).

scripts/fit_confidence_calibration.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+"""Fit confidence calibration knots from labeled face pairs (Phase 3).
+
+Usage (on DEV/PROD app host, from /opt/punimtag):
+
+  set -a && . ./.env && set +a
+  ./venv/bin/python scripts/fit_confidence_calibration.py --dry-run
+  ./venv/bin/python scripts/fit_confidence_calibration.py --apply
+
+Requires ~20+ same-person and different-person distances (or --force).
+After --apply, restart punimtag-api so the process reloads the knots file.
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from backend.db.session import SessionLocal  # noqa: E402
+from backend.services.confidence_calibration import (  # noqa: E402
+    default_calibration_path,
+    knots_ready_to_apply,
+    save_calibration_knots,
+)
+from backend.services.quality_rescore import fit_calibration_from_db  # noqa: E402
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Write knots JSON (default: print only)",
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Write even if below MIN_SAME/MIN_DIFF sample sizes",
+    )
+    parser.add_argument(
+        "--output",
+        type=Path,
+        default=None,
+        help="Override output path (default: backend/data/confidence_calibration.json)",
+    )
+    args = parser.parse_args()
+
+    db = SessionLocal()
+    try:
+        knots, stats = fit_calibration_from_db(db)
+    except ValueError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 1
+    finally:
+        db.close()
+
+    print("Sample counts:", stats)
+    print(f"Source: {knots.source}  fitted_at={knots.fitted_at}")
+    print(f"Knots ({len(knots.distances)}):")
+    for d, c in zip(knots.distances, knots.confidences):
+        print(f"  distance={d:.3f} → confidence={c:.1f}%")
+
+    ready = knots_ready_to_apply(knots)
+    if not ready and not args.force:
+        print(
+            "\nNot enough samples to apply "
+            f"(need same>={20}, diff>={20}; got same={knots.n_same}, diff={knots.n_diff}). "
+            "Re-run with --force to write anyway, or identify more faces / review Auto-Match.",
+            file=sys.stderr,
+        )
+        return 2
+
+    if args.apply:
+        path = save_calibration_knots(knots, args.output or default_calibration_path())
+        print(f"\nWrote {path}")
+        print("Restart punimtag-api (and worker if needed) to load the new calibration.")
+    else:
+        print("\nDry-run only. Pass --apply to write the knots file.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/rescore_face_quality.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+"""Re-score face quality_score using fixed Laplacian sharpness (Phase 3).
+
+Does not re-detect or re-embed — only updates quality_score on faces
+(and matching person_encodings rows).
+
+Usage (on DEV/PROD app host, from /opt/punimtag):
+
+  set -a && . ./.env && set +a
+  ./venv/bin/python scripts/rescore_face_quality.py --dry-run --limit 100
+  ./venv/bin/python scripts/rescore_face_quality.py --limit 500
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from backend.db.session import SessionLocal  # noqa: E402
+from backend.services.quality_rescore import rescore_face_qualities  # noqa: E402
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--limit", type=int, default=None, help="Max faces to examine")
+    parser.add_argument("--offset", type=int, default=0)
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Compute deltas but do not commit",
+    )
+    parser.add_argument("--commit-every", type=int, default=50)
+    args = parser.parse_args()
+
+    db = SessionLocal()
+
+    def progress(i: int, total: int) -> None:
+        if i == total or i % 100 == 0:
+            print(f"  {i}/{total}", flush=True)
+
+    try:
+        summary = rescore_face_qualities(
+            db,
+            limit=args.limit,
+            offset=args.offset,
+            dry_run=args.dry_run,
+            commit_every=args.commit_every,
+            progress=progress,
+        )
+    finally:
+        db.close()
+
+    print(summary)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_confidence_calibration_phase3.py

@@ -0,0 +1,102 @@
+"""Unit tests for Phase 3 confidence calibration + quality rescore helpers."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+
+from backend.services.confidence_calibration import (
+    CalibrationKnots,
+    clear_calibration_cache,
+    fit_knots_from_distances,
+    interpolate_confidence,
+    knots_ready_to_apply,
+    legacy_empirical_confidence,
+    load_calibration_knots,
+    save_calibration_knots,
+)
+from backend.services.face_service import calibrate_confidence
+
+
+class TestLegacyEmpirical:
+    def test_close_distance_is_high(self):
+        assert legacy_empirical_confidence(0.05, 0.5) >= 92
+
+    def test_near_threshold_mid(self):
+        c = legacy_empirical_confidence(0.4, 0.5)
+        assert 50 <= c <= 82
+
+    def test_far_is_low(self):
+        assert legacy_empirical_confidence(1.2, 0.5) <= 20
+
+
+class TestFitAndInterpolate:
+    def test_fit_separates_same_and_different(self):
+        rng = np.random.default_rng(0)
+        same = rng.normal(0.18, 0.05, size=80).clip(0.01, 0.45)
+        diff = rng.normal(0.65, 0.12, size=80).clip(0.3, 1.1)
+        knots = fit_knots_from_distances(same, diff)
+        assert knots_ready_to_apply(knots)
+        near = interpolate_confidence(0.15, knots)
+        far = interpolate_confidence(0.8, knots)
+        assert near > far + 20
+        assert near >= 70
+        assert far <= 40
+
+    def test_interpolate_monotone_nonincreasing(self):
+        knots = fit_knots_from_distances(
+            [0.1, 0.12, 0.15, 0.2],
+            [0.5, 0.6, 0.7, 0.8],
+        )
+        prev = 101.0
+        for d in np.linspace(0, 1.0, 21):
+            c = interpolate_confidence(float(d), knots)
+            assert c <= prev + 1e-6
+            prev = c
+
+
+class TestCalibrationFileRoundTrip:
+    def test_save_load_activates_calibrate_confidence(self, tmp_path: Path, monkeypatch):
+        knots = CalibrationKnots(
+            distances=(0.0, 0.3, 1.0),
+            confidences=(99.0, 70.0, 5.0),
+            source="test",
+            n_same=50,
+            n_diff=50,
+        )
+        path = tmp_path / "cal.json"
+        save_calibration_knots(knots, path)
+        monkeypatch.setenv("CONFIDENCE_CALIBRATION_PATH", str(path))
+        clear_calibration_cache()
+
+        loaded = load_calibration_knots()
+        assert loaded is not None
+        assert loaded.confidences[0] == 99.0
+
+        # Fitted path should be used by calibrate_confidence (empirical)
+        conf = calibrate_confidence(0.0, tolerance=0.5)
+        assert conf == 99.0
+        conf_far = calibrate_confidence(1.0, tolerance=0.5)
+        assert conf_far == 5.0
+
+        monkeypatch.delenv("CONFIDENCE_CALIBRATION_PATH", raising=False)
+        clear_calibration_cache()
+
+    def test_without_file_matches_legacy(self, monkeypatch, tmp_path: Path):
+        monkeypatch.setenv("CONFIDENCE_CALIBRATION_PATH", str(tmp_path / "missing.json"))
+        clear_calibration_cache()
+        for d in (0.05, 0.2, 0.4, 0.6, 1.0):
+            assert calibrate_confidence(d, 0.5) == legacy_empirical_confidence(d, 0.5)
+        monkeypatch.delenv("CONFIDENCE_CALIBRATION_PATH", raising=False)
+        clear_calibration_cache()
+
+
+class TestQualityScoreUsesCrop:
+    """Sanity: quality helper still differentiates sharp vs blur (Phase 1)."""
+
+    def test_import_rescore_module(self):
+        from backend.services import quality_rescore
+
+        assert callable(quality_rescore.rescore_face_qualities)
+        assert callable(quality_rescore.fit_calibration_from_db)