diff --git a/.gitignore b/.gitignore
index 0f1cec5..8e147b0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -38,3 +38,6 @@ tmp_dbg_img/
# OS
.DS_Store
Thumbs.db
+
+# PowerShell module-analysis cache written into cwd by some automation
+Microsoft/
diff --git a/src/rights_filter/admin/correction_handlers.py b/src/rights_filter/admin/correction_handlers.py
index 412f712..3578a80 100644
--- a/src/rights_filter/admin/correction_handlers.py
+++ b/src/rights_filter/admin/correction_handlers.py
@@ -1,3 +1,13 @@
+"""In-memory reference implementation of the operator review flow.
+
+NOT wired into the running server: the live operator console persists through
+``rights_filter.server.store_operations`` / ``sqlite_store``, which is the source
+of truth for decision semantics, audit records and knowledge-entry side effects.
+This module backs unit tests and documents the intended domain shape; changing
+behavior here does NOT change what operators see. Keep the two in step, or fold
+this into the store path, rather than treating this file as production logic.
+"""
+
from __future__ import annotations
from rights_filter.domain.records import (
diff --git a/src/rights_filter/admin/detailed_review_presenter.py b/src/rights_filter/admin/detailed_review_presenter.py
index fbdaac5..8cef38b 100644
--- a/src/rights_filter/admin/detailed_review_presenter.py
+++ b/src/rights_filter/admin/detailed_review_presenter.py
@@ -1,3 +1,13 @@
+"""In-memory reference implementation of the operator review flow.
+
+NOT wired into the running server: the live operator console persists through
+``rights_filter.server.store_operations`` / ``sqlite_store``, which is the source
+of truth for decision semantics, audit records and knowledge-entry side effects.
+This module backs unit tests and documents the intended domain shape; changing
+behavior here does NOT change what operators see. Keep the two in step, or fold
+this into the store path, rather than treating this file as production logic.
+"""
+
from __future__ import annotations
from typing import Any
@@ -54,6 +64,7 @@ def _group_evidence(evidence: list[Evidence]) -> dict[str, list[dict[str, Any]]]
"naver": [],
"google": [],
"llm": [],
+ "skipped": [],
"failures": [],
}
for item in evidence:
@@ -70,6 +81,11 @@ def _group_name(item: Evidence) -> str:
return "google"
if item.source == EvidenceSource.LLM_SUMMARY:
return "llm"
+ # A provider that was deliberately not called (quota reached, disabled) is
+ # operational status, not a failure — grouping the two together made a
+ # normally-configured case look like it had errored.
+ if item.source in {EvidenceSource.EXTERNAL_SKIPPED, EvidenceSource.SEARCH_SKIPPED}:
+ return "skipped"
return "failures"
diff --git a/src/rights_filter/admin/knowledge_base_handlers.py b/src/rights_filter/admin/knowledge_base_handlers.py
index 66c37ab..77cb8fe 100644
--- a/src/rights_filter/admin/knowledge_base_handlers.py
+++ b/src/rights_filter/admin/knowledge_base_handlers.py
@@ -1,3 +1,13 @@
+"""In-memory reference implementation of the operator review flow.
+
+NOT wired into the running server: the live operator console persists through
+``rights_filter.server.store_operations`` / ``sqlite_store``, which is the source
+of truth for decision semantics, audit records and knowledge-entry side effects.
+This module backs unit tests and documents the intended domain shape; changing
+behavior here does NOT change what operators see. Keep the two in step, or fold
+this into the store path, rather than treating this file as production logic.
+"""
+
from __future__ import annotations
from typing import Any
@@ -22,7 +32,14 @@ def register_manual_entry(
sample_fingerprints: list[str] | None = None,
**payload: Any,
) -> KnowledgeBaseEntry:
+ # Extra keys were screened for biometric templates and then silently dropped,
+ # so a caller passing a field this handler does not persist got a success
+ # response and no entry data. Reject instead of losing the input quietly.
assert_no_biometric_template(payload)
+ if payload:
+ raise ValueError(
+ "unsupported knowledge entry fields: " + ", ".join(sorted(payload))
+ )
entry = create_manual_knowledge_entry(
entry_type=entry_type,
name=name,
diff --git a/src/rights_filter/admin/review_handlers.py b/src/rights_filter/admin/review_handlers.py
index 1ad55d2..5b4376d 100644
--- a/src/rights_filter/admin/review_handlers.py
+++ b/src/rights_filter/admin/review_handlers.py
@@ -1,3 +1,13 @@
+"""In-memory reference implementation of the operator review flow.
+
+NOT wired into the running server: the live operator console persists through
+``rights_filter.server.store_operations`` / ``sqlite_store``, which is the source
+of truth for decision semantics, audit records and knowledge-entry side effects.
+This module backs unit tests and documents the intended domain shape; changing
+behavior here does NOT change what operators see. Keep the two in step, or fold
+this into the store path, rather than treating this file as production logic.
+"""
+
from __future__ import annotations
from typing import Any
diff --git a/src/rights_filter/analysis/face_person_detection.py b/src/rights_filter/analysis/face_person_detection.py
index 78e8557..46b0817 100644
--- a/src/rights_filter/analysis/face_person_detection.py
+++ b/src/rights_filter/analysis/face_person_detection.py
@@ -1,10 +1,17 @@
from __future__ import annotations
+import logging
from io import BytesIO
from dataclasses import dataclass
from rights_filter.analysis.preprocessing import ImagePayload
+logger = logging.getLogger(__name__)
+
+# Emit the "OpenCV build lacks CascadeClassifier" warning at most once per
+# process instead of on every image so operator logs stay readable.
+_warned_missing_cascade = False
+
FaceBox = tuple[int, int, int, int]
@@ -43,31 +50,52 @@ class HeuristicFacePersonDetector:
if decoded is None:
return FacePersonSignal(face_count=0, person_count=0)
+ # Some OpenCV builds (e.g. the 5.x preview wheels) drop the legacy Haar
+ # cascade API. Detect that up front and degrade to a presence-negative
+ # signal instead of letting an AttributeError escape into the seed/
+ # analysis pipeline and take down every ingest/re-analysis request.
+ if not (hasattr(cv2, "CascadeClassifier") and hasattr(cv2, "data")):
+ global _warned_missing_cascade
+ if not _warned_missing_cascade:
+ _warned_missing_cascade = True
+ logger.warning(
+ "OpenCV build %s lacks CascadeClassifier/data — face "
+ "presence detection disabled (install "
+ "opencv-python-headless==4.13.0.92).",
+ getattr(cv2, "__version__", "?"),
+ )
+ return FacePersonSignal(face_count=0, person_count=0)
+
try:
gray = cv2.cvtColor(decoded, cv2.COLOR_BGR2GRAY)
except Exception:
return FacePersonSignal(face_count=0, person_count=0)
face_boxes: list[FaceBox] = []
- for cascade_name in (
- "haarcascade_frontalface_default.xml",
- "haarcascade_profileface.xml",
- ):
- cascade = cv2.CascadeClassifier(f"{cv2.data.haarcascades}{cascade_name}")
- if hasattr(cascade, "empty") and cascade.empty():
- continue
- rects = cascade.detectMultiScale(
- gray,
- scaleFactor=1.1,
- minNeighbors=4,
- minSize=(30, 30),
- )
- for rect in rects:
- box = tuple(int(value) for value in rect)
- if len(box) == 4 and not any(
- _boxes_overlap(box, existing) for existing in face_boxes
- ):
- face_boxes.append(box)
+ try:
+ for cascade_name in (
+ "haarcascade_frontalface_default.xml",
+ "haarcascade_profileface.xml",
+ ):
+ cascade = cv2.CascadeClassifier(f"{cv2.data.haarcascades}{cascade_name}")
+ if hasattr(cascade, "empty") and cascade.empty():
+ continue
+ rects = cascade.detectMultiScale(
+ gray,
+ scaleFactor=1.1,
+ minNeighbors=4,
+ minSize=(30, 30),
+ )
+ for rect in rects:
+ box = tuple(int(value) for value in rect)
+ if len(box) == 4 and not any(
+ _boxes_overlap(box, existing) for existing in face_boxes
+ ):
+ face_boxes.append(box)
+ except Exception:
+ # Any cascade/runtime failure degrades to a presence-negative signal
+ # rather than crashing the caller.
+ return FacePersonSignal(face_count=0, person_count=0)
face_count = len(face_boxes)
return FacePersonSignal(
@@ -99,8 +127,10 @@ def _decode_with_pillow(content: bytes, cv2: object, np: object) -> object | Non
try:
from PIL import Image
- with Image.open(BytesIO(content)) as image:
- rgb = image.convert("RGB")
+ from rights_filter.analysis.preprocessing import apply_exif_orientation
+
+ with Image.open(BytesIO(content)) as opened:
+ rgb = apply_exif_orientation(opened).convert("RGB")
return cv2.cvtColor(np.array(rgb), cv2.COLOR_RGB2BGR)
except Exception:
return None
diff --git a/src/rights_filter/analysis/fingerprints.py b/src/rights_filter/analysis/fingerprints.py
index 12c8363..4ea8ec0 100644
--- a/src/rights_filter/analysis/fingerprints.py
+++ b/src/rights_filter/analysis/fingerprints.py
@@ -26,6 +26,19 @@ class FingerprintService:
if left_value.startswith("unavailable:") or right_value.startswith("unavailable:"):
return 0.0
if _looks_like_64bit_hash(left_value) and _looks_like_64bit_hash(right_value):
+ # A near-uniform image (blank, solid colour, very low detail) makes
+ # this 8x8 average hash collapse to nearly-all-ones or nearly-all-
+ # zeros. Two such hashes that differ by a bit or two would score
+ # ~0.98 and be reported as a knowledge-base match worth +80, even
+ # though neither hash carries usable information. Refuse to infer
+ # similarity from an uninformative hash.
+ #
+ # Identical hashes are handled above and still match: this cannot
+ # separate "same picture" from "two different blank images" — that
+ # needs the exact (SHA-256) fingerprint, which callers compare
+ # separately — but it does remove the near-miss false positives.
+ if _is_degenerate_hash(left_value) or _is_degenerate_hash(right_value):
+ return 0.0
distance = (int(left_value, 16) ^ int(right_value, 16)).bit_count()
return 1 - (distance / 64)
return 0.0
@@ -35,8 +48,13 @@ def _perceptual_hash(content: bytes, exact_digest: str) -> str:
try:
from PIL import Image
- with Image.open(BytesIO(content)) as image:
- thumbnail = image.convert("L").resize((8, 8))
+ from rights_filter.analysis.preprocessing import apply_exif_orientation
+
+ with Image.open(BytesIO(content)) as opened:
+ # Normalize orientation first: the same photo saved rotated with an
+ # EXIF tag would otherwise hash differently from its upright copy and
+ # never match as a duplicate.
+ thumbnail = apply_exif_orientation(opened).convert("L").resize((8, 8))
if hasattr(thumbnail, "get_flattened_data"):
pixels = list(thumbnail.get_flattened_data())
else:
@@ -51,6 +69,18 @@ def _perceptual_hash(content: bytes, exact_digest: str) -> str:
return f"{bits:016x}"
+# A 64-bit average hash is degenerate when nearly every bit agrees: the image had
+# almost no variation around its mean, so the bits describe the threshold, not the
+# picture. 4 of 64 bits is the practical cut-off — below it, unrelated images
+# routinely collide.
+_MIN_INFORMATIVE_BITS = 4
+
+
+def _is_degenerate_hash(value: str) -> bool:
+ set_bits = int(value, 16).bit_count()
+ return set_bits <= _MIN_INFORMATIVE_BITS or set_bits >= 64 - _MIN_INFORMATIVE_BITS
+
+
def _looks_like_64bit_hash(value: str) -> bool:
if len(value) != 16:
return False
diff --git a/src/rights_filter/analysis/preprocessing.py b/src/rights_filter/analysis/preprocessing.py
index 1dddccf..211018b 100644
--- a/src/rights_filter/analysis/preprocessing.py
+++ b/src/rights_filter/analysis/preprocessing.py
@@ -32,23 +32,19 @@ def build_external_derivative(
if original.width <= 0 or original.height <= 0:
raise PreprocessingError("image dimensions must be positive")
- longest_side = max(original.width, original.height)
- if longest_side > max_side:
- scale = max_side / longest_side
- width = max(1, round(original.width * scale))
- height = max(1, round(original.height * scale))
- else:
- width = original.width
- height = original.height
-
- if original.metadata.get("format", "").upper() == "AVIF":
- content = _jpeg_derivative(original.content, width, height)
- else:
- # Re-encode through PIL to drop embedded EXIF/GPS metadata from the image
- # bytes. Clearing only the metadata dict (below) left camera serials and
- # GPS coordinates inside the bytes that get shipped to external APIs.
- content = _reencode_without_metadata(original.content, width, height)
- if content is None:
+ # Re-encode through PIL to drop embedded EXIF/GPS metadata from the image
+ # bytes and to apply the EXIF orientation. Clearing only the metadata dict
+ # (below) left camera serials and GPS coordinates inside the bytes shipped to
+ # external APIs; skipping the orientation shipped rotated phone photos
+ # sideways (and hid faces from the detector, which has no EXIF awareness).
+ # The output size is derived from the UPRIGHT image, since a 90° rotation
+ # swaps width/height relative to the stored dimensions.
+ derived = _upright_derivative(
+ original.content,
+ max_side,
+ force_jpeg=original.metadata.get("format", "").upper() == "AVIF",
+ )
+ if derived is None:
# Re-encoding failed (Pillow missing or undecodable bytes): fall back to
# the original bytes, but warn — those bytes may still carry EXIF/GPS
# metadata that is about to be sent to an external API. On an air-gapped
@@ -58,6 +54,9 @@ def build_external_derivative(
"metadata may not have been stripped before external submission"
)
content = original.content
+ width, height = _scaled_size(original.width, original.height, max_side)
+ else:
+ content, width, height = derived
return ImagePayload(
content=content,
@@ -81,8 +80,10 @@ def build_face_crop_derivatives(
try:
from PIL import Image
- with Image.open(BytesIO(original.content)) as image:
- image = image.convert("RGB")
+ with Image.open(BytesIO(original.content)) as opened:
+ # Face boxes come from the detector, which sees the upright image —
+ # crop from the same orientation or the boxes address the wrong pixels.
+ image = apply_exif_orientation(opened).convert("RGB")
crops: list[ImagePayload] = []
image_width, image_height = image.size
for box in face_boxes[:max_crops]:
@@ -111,46 +112,69 @@ def build_face_crop_derivatives(
return []
-def _jpeg_derivative(content: bytes, width: int, height: int) -> bytes | None:
- try:
- from PIL import Image
-
- with Image.open(BytesIO(content)) as image:
- image = image.convert("RGB")
- if image.size != (width, height):
- image = image.resize((width, height))
- output = BytesIO()
- image.save(output, format="JPEG", quality=90, optimize=True)
- return output.getvalue()
- except Exception:
- return None
-
-
_REENCODABLE_FORMATS = {"JPEG", "PNG", "WEBP", "GIF", "BMP", "TIFF"}
-def _reencode_without_metadata(content: bytes, width: int, height: int) -> bytes | None:
- """Re-save the image (preserving format where possible) so embedded EXIF/GPS
- metadata is dropped. Returns None when PIL is unavailable or the bytes are
- not a decodable image, so the caller can fall back to the original content."""
+def _scaled_size(width: int, height: int, max_side: int) -> tuple[int, int]:
+ longest_side = max(width, height)
+ if longest_side <= max_side:
+ return width, height
+ scale = max_side / longest_side
+ return max(1, round(width * scale)), max(1, round(height * scale))
+
+
+def apply_exif_orientation(image: object) -> object:
+ """Rotate/flip an opened PIL image per its EXIF orientation tag.
+
+ Every consumer that reasons about pixels (external derivative, face crops,
+ perceptual hash, face detection) must call this: a phone photo stored
+ rotated with orientation=6 otherwise reaches the detector on its side, where
+ upright-trained cascades miss the face entirely.
+ """
+ try:
+ from PIL import ImageOps
+
+ return ImageOps.exif_transpose(image) or image
+ except Exception:
+ return image
+
+
+def _upright_derivative(
+ content: bytes, max_side: int, force_jpeg: bool = False
+) -> tuple[bytes, int, int] | None:
+ """Re-save the image upright and metadata-free, capped to `max_side`.
+
+ Returns (bytes, width, height) using the orientation-corrected dimensions, or
+ None when PIL is unavailable / the bytes are not a decodable image so the
+ caller can fall back to the original content.
+ """
try:
from PIL import Image
- with Image.open(BytesIO(content)) as image:
- source_format = (image.format or "").upper()
- save_format = source_format if source_format in _REENCODABLE_FORMATS else "JPEG"
+ with Image.open(BytesIO(content)) as opened:
+ source_format = (opened.format or "").upper()
+ image = apply_exif_orientation(opened)
+ save_format = (
+ "JPEG"
+ if force_jpeg or source_format not in _REENCODABLE_FORMATS
+ else source_format
+ )
if save_format in {"JPEG", "BMP"} and image.mode not in {"RGB", "L"}:
image = image.convert("RGB")
elif image.mode == "P":
image = image.convert("RGBA")
+ width, height = _scaled_size(int(image.width), int(image.height), max_side)
if image.size != (width, height):
image = image.resize((width, height))
# Rebuild from raw pixels so NO metadata (EXIF/GPS/ICC) survives —
# Pillow can otherwise re-emit image.info["exif"] when re-saving.
clean = Image.frombytes(image.mode, image.size, image.tobytes())
output = BytesIO()
- clean.save(output, format=save_format)
- return output.getvalue()
+ if save_format == "JPEG":
+ clean.save(output, format="JPEG", quality=90, optimize=True)
+ else:
+ clean.save(output, format=save_format)
+ return output.getvalue(), width, height
except Exception:
return None
diff --git a/src/rights_filter/analysis/risk_scoring.py b/src/rights_filter/analysis/risk_scoring.py
index 7ae0ba6..82952c3 100644
--- a/src/rights_filter/analysis/risk_scoring.py
+++ b/src/rights_filter/analysis/risk_scoring.py
@@ -9,6 +9,7 @@ class RiskScorer:
reasons: list[str] = []
counted_web_buckets: set[str] = set()
counted_naver_buckets: set[str] = set()
+ counted_face_person = False
for item in evidence:
if item.data.get("operator_status") in {"irrelevant", "false_positive"}:
@@ -24,6 +25,13 @@ class RiskScorer:
score += 30
reasons.append(item.reason)
elif item.source == EvidenceSource.FACE_PERSON:
+ # "A person is present" is one signal no matter how many rows
+ # assert it. Re-analysis appends a fresh FACE_PERSON row while the
+ # previous one remains, so without this bucket the same image
+ # accrued +35 per rerun and drifted into the high band on its own.
+ if counted_face_person:
+ continue
+ counted_face_person = True
score += 35
reasons.append(item.reason)
elif item.source == EvidenceSource.WEB_DETECTION:
@@ -85,6 +93,16 @@ class RiskScorer:
def _web_detection_points(item: Evidence) -> int:
if item.data.get("weak_hint"):
return 0
+ # Google custom IMAGE search results (match="search_image") are context-only:
+ # existence of a text-query image hit is not itself a rights signal (the
+ # matching pHash FINGERPRINT evidence carries the score). Fresh evidence has
+ # no bare "url" key so it scores 0, but the persistence round-trip aliases
+ # result_url -> url; without this explicit gate the round-trip would flip the
+ # same evidence to +45 via the url fallback below, violating the
+ # score-preservation invariant and inflating bands on any query that returns
+ # anything.
+ if item.data.get("match") == "search_image":
+ return 0
if item.data.get("match") == "full":
return 45
if item.data.get("match") in {"partial", "page"}:
diff --git a/src/rights_filter/analysis/search_result_promoter.py b/src/rights_filter/analysis/search_result_promoter.py
index 932ddad..13e0fe2 100644
--- a/src/rights_filter/analysis/search_result_promoter.py
+++ b/src/rights_filter/analysis/search_result_promoter.py
@@ -26,12 +26,14 @@ class SearchResultPromoter:
data={**item.data, "promoted": False, "promotion_reason": "no search results"},
)
- # Scan only the actual result fields — never the query. The query is our
- # own generated input; including it self-promotes any result whose query
- # happened to contain a marker word.
+ # Scan only the human-readable result fields — never the query and never
+ # the URL. The query is our own generated input (it would self-promote any
+ # result whose query contained a marker word), and a URL's host/path
+ # routinely contains marker substrings for unrelated reasons
+ # ("/game/", "album" in a CDN path, a "cover" directory), which promoted
+ # irrelevant results to 0.8 confidence and +40 points.
text = " ".join(
- str(item.data.get(key, ""))
- for key in ("title", "description", "result_url")
+ str(item.data.get(key, "")) for key in ("title", "description")
).lower()
promoted = _has_named_rights_signal(text)
data = {
diff --git a/src/rights_filter/domain/records.py b/src/rights_filter/domain/records.py
index e7adc54..502a677 100644
--- a/src/rights_filter/domain/records.py
+++ b/src/rights_filter/domain/records.py
@@ -1,6 +1,6 @@
from __future__ import annotations
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
from enum import StrEnum
from itertools import count
from typing import Any
@@ -242,9 +242,36 @@ class InMemoryRightsFilterRepository:
def create_rejected_image_entry(
self, decision_id: str, submission_id: str, fingerprints: list[str]
) -> KnowledgeBaseEntry:
+ name = f"rejected:{submission_id}"
+ # One submission has one rejection reference. Re-deciding the same case
+ # previously appended another entry each time, so a submission rejected
+ # N times seeded N duplicate reference images into every later similarity
+ # check. Refresh the existing entry in place instead.
+ incumbent = next(
+ (
+ entry
+ for entry in self._knowledge_entries.values()
+ if entry.active
+ and entry.entry_type == KnowledgeEntryType.REJECTED_IMAGE
+ and entry.name == name
+ ),
+ None,
+ )
+ if incumbent is not None:
+ merged = list(
+ dict.fromkeys([*incumbent.sample_fingerprints, *fingerprints])
+ )
+ entry = replace(
+ incumbent,
+ source_decision_id=decision_id,
+ sample_fingerprints=merged,
+ )
+ self.save_knowledge_entry(entry)
+ return entry
+
entry = KnowledgeBaseEntry.create(
entry_type=KnowledgeEntryType.REJECTED_IMAGE,
- name=f"rejected:{submission_id}",
+ name=name,
provenance=KnowledgeProvenance.AUTOMATIC_REJECTION,
source_decision_id=decision_id,
sample_fingerprints=fingerprints,
diff --git a/src/rights_filter/integrations/cloud_vision_web_detection.py b/src/rights_filter/integrations/cloud_vision_web_detection.py
index ed74a2e..79e9cd5 100644
--- a/src/rights_filter/integrations/cloud_vision_web_detection.py
+++ b/src/rights_filter/integrations/cloud_vision_web_detection.py
@@ -112,6 +112,10 @@ class CloudVisionWebDetectionAdapter:
source=EvidenceSource.EXTERNAL_SKIPPED,
reason=reason or "external API skipped",
confidence=1.0,
+ # Deliberately NOT provider-tagged: a policy refusal means the
+ # call never ran, and the provider badge derives "failed" from
+ # any attributed failure-source row. Leaving skips unattributed
+ # keeps the badge at "not run", which is what actually happened.
data={"submission_id": submission_id},
)
]
@@ -124,14 +128,27 @@ class CloudVisionWebDetectionAdapter:
source=EvidenceSource.FAILURE,
reason=f"External API failed: {exc}",
confidence=1.0,
- data={"submission_id": submission_id},
+ data={"submission_id": submission_id, "provider": "google"},
)
]
# Record the quota debit only after the request actually succeeds, so a
# failed/timed-out call does not permanently consume the daily limit.
policy.record_call()
- return _map_response(submission_id, response)
+ try:
+ return _map_response(submission_id, response)
+ except Exception as exc:
+ # A malformed/unexpected response shape must not abort the caller
+ # after the quota was already spent — surface it as failure evidence
+ # like a transport error does.
+ return [
+ Evidence(
+ source=EvidenceSource.FAILURE,
+ reason=f"External API response could not be parsed: {exc}",
+ confidence=1.0,
+ data={"submission_id": submission_id, "provider": "google"},
+ )
+ ]
def _map_response(submission_id: str, response: dict[str, Any]) -> list[Evidence]:
@@ -208,6 +225,9 @@ def _map_response(submission_id: str, response: dict[str, Any]) -> list[Evidence
)
for label in response.get("best_guess_labels", []):
+ # Tolerate a bare string / non-mapping entry instead of raising on .get.
+ if not isinstance(label, dict):
+ continue
text = label.get("label", "")
evidence.append(
Evidence(
diff --git a/src/rights_filter/integrations/external_policy.py b/src/rights_filter/integrations/external_policy.py
index 69e86b5..1725b86 100644
--- a/src/rights_filter/integrations/external_policy.py
+++ b/src/rights_filter/integrations/external_policy.py
@@ -1,6 +1,8 @@
from __future__ import annotations
-from dataclasses import dataclass
+import threading
+from dataclasses import dataclass, field
+from datetime import date
@dataclass
@@ -11,19 +13,37 @@ class ExternalApiPolicy:
allow_online_sync: bool = False
daily_limit: int | None = None
calls_made: int = 0
+ window_day: str = ""
+ _lock: threading.Lock = field(
+ default_factory=threading.Lock, repr=False, compare=False
+ )
+
+ def _roll_window_locked(self) -> None:
+ # daily_limit is a per-day budget. Reset the counter when the calendar
+ # day changes; otherwise the count only ever grows and, once the limit is
+ # hit, web detection stays disabled until the process restarts (it never
+ # re-windows for the next day). Mirrors SearchApiPolicy.
+ today = date.today().isoformat()
+ if self.window_day != today:
+ self.window_day = today
+ self.calls_made = 0
def can_call(self) -> tuple[bool, str | None]:
- if self.disabled:
- return False, "external API disabled"
- if not self.compliance_approved:
- return False, "external API compliance not approved"
- if not self.metadata_logging_accepted:
- return False, "metadata logging not accepted"
- if not self.allow_online_sync:
- return False, "online synchronous mode not allowed"
- if self.daily_limit is not None and self.calls_made >= self.daily_limit:
- return False, "external API usage limit reached"
- return True, None
+ with self._lock:
+ self._roll_window_locked()
+ if self.disabled:
+ return False, "external API disabled"
+ if not self.compliance_approved:
+ return False, "external API compliance not approved"
+ if not self.metadata_logging_accepted:
+ return False, "metadata logging not accepted"
+ if not self.allow_online_sync:
+ return False, "online synchronous mode not allowed"
+ if self.daily_limit is not None and self.calls_made >= self.daily_limit:
+ return False, "external API usage limit reached"
+ return True, None
def record_call(self) -> None:
- self.calls_made += 1
+ with self._lock:
+ self._roll_window_locked()
+ self.calls_made += 1
diff --git a/src/rights_filter/integrations/google_custom_search.py b/src/rights_filter/integrations/google_custom_search.py
index 4e823dd..cdd9b0f 100644
--- a/src/rights_filter/integrations/google_custom_search.py
+++ b/src/rights_filter/integrations/google_custom_search.py
@@ -102,7 +102,7 @@ class GoogleCustomSearchAdapter:
raise ValueError("Google custom search requires a text query")
requested_calls = _requested_calls(self.client, "image_pages")
- allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
+ allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@@ -120,8 +120,14 @@ class GoogleCustomSearchAdapter:
]
try:
- response = self.client.search_image(query)
- policy.record_call(requested_calls)
+ try:
+ response = self.client.search_image(query)
+ except Exception:
+ # The request never reached the provider — refund the
+ # reservation. A failure AFTER a successful fetch keeps the
+ # debit, since the provider did serve the call.
+ policy.release(requested_calls)
+ raise
# Map inside the try so a malformed payload becomes ENRICHMENT_FAILURE
# evidence instead of an uncaught exception that aborts the request.
return _map_image_response(submission_id, query, response)
@@ -148,7 +154,7 @@ class GoogleCustomSearchAdapter:
raise ValueError("Google custom search requires a text query")
requested_calls = _requested_calls(self.client, "web_pages")
- allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
+ allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@@ -166,8 +172,14 @@ class GoogleCustomSearchAdapter:
]
try:
- response = self.client.search_web(query)
- policy.record_call(requested_calls)
+ try:
+ response = self.client.search_web(query)
+ except Exception:
+ # The request never reached the provider — refund the
+ # reservation. A failure AFTER a successful fetch keeps the
+ # debit, since the provider did serve the call.
+ policy.release(requested_calls)
+ raise
return _map_web_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [
diff --git a/src/rights_filter/integrations/http_json.py b/src/rights_filter/integrations/http_json.py
index c81b98e..4b9cd03 100644
--- a/src/rights_filter/integrations/http_json.py
+++ b/src/rights_filter/integrations/http_json.py
@@ -5,7 +5,17 @@ import socket
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
-from urllib.request import Request, urlopen
+from urllib.request import Request, getproxies, proxy_bypass, urlopen
+
+
+def _proxy_applies(scheme: str, host: str) -> bool:
+ """True when the system proxy configuration would route `host` for `scheme`."""
+ try:
+ if scheme not in getproxies():
+ return False
+ return not proxy_bypass(host)
+ except Exception:
+ return False
class HttpJsonError(RuntimeError):
@@ -16,6 +26,10 @@ class HttpJsonError(RuntimeError):
# cannot exhaust memory. JSON API responses are small; 10 MiB is generous.
_MAX_RESPONSE_BYTES = 10 * 1024 * 1024
+# An HTTP error body only ever becomes part of an exception message, so keep it
+# small — the full 10 MiB cap would put a whole error page into a log line.
+_MAX_ERROR_DETAIL_BYTES = 4096
+
class UrllibJsonTransport:
def __init__(self, connect_timeout: float | None = None) -> None:
@@ -52,7 +66,9 @@ class UrllibJsonTransport:
raw = raw_bytes.decode("utf-8", errors="replace")
except HTTPError as exc:
try:
- detail = exc.read().decode("utf-8", errors="replace")
+ # Apply the same cap on the error path: an error page can be far
+ # larger than a JSON reply, and it is only used for a message.
+ detail = exc.read(_MAX_ERROR_DETAIL_BYTES).decode("utf-8", errors="replace")
except Exception:
detail = ""
raise HttpJsonError(f"HTTP {exc.code}: {detail}") from exc
@@ -73,6 +89,12 @@ class UrllibJsonTransport:
host = parts.hostname
if not host:
return
+ # A direct TCP probe bypasses the system proxy. On a proxy-only network
+ # the probe always fails while the actual (proxied) request would have
+ # succeeded, so skip the fast-fail check when a proxy applies to this host
+ # and let urlopen route through it.
+ if _proxy_applies(parts.scheme, host):
+ return
port = parts.port or (443 if parts.scheme == "https" else 80)
try:
with socket.create_connection((host, port), timeout=self.connect_timeout):
diff --git a/src/rights_filter/integrations/naver_search.py b/src/rights_filter/integrations/naver_search.py
index a6af5b9..0d601dc 100644
--- a/src/rights_filter/integrations/naver_search.py
+++ b/src/rights_filter/integrations/naver_search.py
@@ -162,7 +162,7 @@ class NaverSearchAdapter:
raise ValueError("Naver search requires a text query")
requested_calls = _requested_calls(self.client, "image_pages")
- allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
+ allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@@ -179,8 +179,14 @@ class NaverSearchAdapter:
]
try:
- response = self.client.search_image(query)
- policy.record_call(requested_calls)
+ try:
+ response = self.client.search_image(query)
+ except Exception:
+ # The request never reached the provider — refund the
+ # reservation. A failure AFTER a successful fetch keeps the
+ # debit, since the provider did serve the call.
+ policy.release(requested_calls)
+ raise
return _map_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [
@@ -204,7 +210,7 @@ class NaverSearchAdapter:
raise ValueError("Naver search requires a text query")
requested_calls = _requested_calls(self.client, "blog_pages")
- allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
+ allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@@ -222,8 +228,14 @@ class NaverSearchAdapter:
]
try:
- response = self.client.search_blog(query)
- policy.record_call(requested_calls)
+ try:
+ response = self.client.search_blog(query)
+ except Exception:
+ # The request never reached the provider — refund the
+ # reservation. A failure AFTER a successful fetch keeps the
+ # debit, since the provider did serve the call.
+ policy.release(requested_calls)
+ raise
return _map_blog_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [
@@ -248,7 +260,7 @@ class NaverSearchAdapter:
raise ValueError("Naver search requires a text query")
requested_calls = _requested_calls(self.client, "web_pages")
- allowed, reason = policy.can_call(self.provider, requested_calls=requested_calls)
+ allowed, reason = policy.try_reserve(self.provider, requested_calls=requested_calls)
if not allowed:
return [
Evidence(
@@ -266,8 +278,14 @@ class NaverSearchAdapter:
]
try:
- response = self.client.search_web(query)
- policy.record_call(requested_calls)
+ try:
+ response = self.client.search_web(query)
+ except Exception:
+ # The request never reached the provider — refund the
+ # reservation. A failure AFTER a successful fetch keeps the
+ # debit, since the provider did serve the call.
+ policy.release(requested_calls)
+ raise
return _map_web_response(submission_id, query, response)
except Exception as exc: # pragma: no cover - defensive adapter boundary
return [
diff --git a/src/rights_filter/integrations/search_policy.py b/src/rights_filter/integrations/search_policy.py
index 2e09082..fc79f70 100644
--- a/src/rights_filter/integrations/search_policy.py
+++ b/src/rights_filter/integrations/search_policy.py
@@ -27,7 +27,36 @@ class SearchApiPolicy:
self.window_day = today
self.calls_made = 0
+ def try_reserve(self, provider: str, requested_calls: int = 1) -> tuple[bool, str | None]:
+ """Atomically check the budget and debit it in one locked step.
+
+ Checking and debiting separately let concurrent operator requests both
+ pass the check before either recorded, overshooting daily_limit by the
+ number of racing callers. Callers refund with release() when the request
+ never actually reached the provider.
+ """
+ with self._lock:
+ self._roll_window_locked()
+ if self.disabled:
+ return False, "search API disabled"
+ if not self.compliance_approved:
+ return False, "search API compliance not approved"
+ if provider not in self.allowed_providers:
+ return False, "search provider not allowed"
+ calls = max(1, int(requested_calls or 1))
+ if self.daily_limit is not None and self.calls_made + calls > self.daily_limit:
+ return False, "search API usage limit reached"
+ self.calls_made += calls
+ return True, None
+
+ def release(self, count: int = 1) -> None:
+ """Return an unused reservation (the request never reached the provider)."""
+ with self._lock:
+ self._roll_window_locked()
+ self.calls_made = max(0, self.calls_made - max(1, int(count or 1)))
+
def can_call(self, provider: str, requested_calls: int = 1) -> tuple[bool, str | None]:
+ """Non-reserving budget check, for callers that only need to look."""
with self._lock:
self._roll_window_locked()
if self.disabled:
diff --git a/src/rights_filter/jobs/batch_analyzer.py b/src/rights_filter/jobs/batch_analyzer.py
index 60a0b89..ebd982d 100644
--- a/src/rights_filter/jobs/batch_analyzer.py
+++ b/src/rights_filter/jobs/batch_analyzer.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import logging
from dataclasses import dataclass
from rights_filter.analysis.internal_analyzer import InternalAnalyzer
@@ -14,6 +15,8 @@ from rights_filter.domain.records import (
from rights_filter.integrations.cloud_vision_web_detection import CloudVisionWebDetectionAdapter
from rights_filter.integrations.external_policy import ExternalApiPolicy
+logger = logging.getLogger(__name__)
+
@dataclass(frozen=True)
class SubmissionImage:
@@ -86,6 +89,17 @@ class BatchAnalyzer:
# has_analysis_run() skip the submission permanently on re-runs.
failed += 1
continue
+ except Exception:
+ # Isolate any unexpected per-submission failure (e.g. an OpenCV
+ # build missing CascadeClassifier, a decoder edge case) so one
+ # bad image cannot abort the whole batch/seed. Log with the
+ # submission id for diagnosis instead of swallowing silently.
+ logger.exception(
+ "Unexpected analysis failure for submission %s; skipping.",
+ submission.submission_id,
+ )
+ failed += 1
+ continue
external_skipped += sum(
1 for item in evidence if item.source == EvidenceSource.EXTERNAL_SKIPPED
diff --git a/src/rights_filter/server/env_file.py b/src/rights_filter/server/env_file.py
index 10dd3e2..0ca8224 100644
--- a/src/rights_filter/server/env_file.py
+++ b/src/rights_filter/server/env_file.py
@@ -24,10 +24,25 @@ def load_env_file(path: Path | str, environ: MutableMapping[str, str]) -> None:
key = key.strip()
if not key or key in environ:
continue
- environ[key] = _strip_quotes(value.strip())
+ environ[key] = _parse_value(value.strip())
-def _strip_quotes(value: str) -> str:
- if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
+def _parse_value(value: str) -> str:
+ if _is_quoted(value):
+ # Quoted values are verbatim — a '#' inside quotes is part of the secret.
return value[1:-1]
- return value
+ return _strip_inline_comment(value)
+
+
+def _is_quoted(value: str) -> bool:
+ return len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}
+
+
+def _strip_inline_comment(value: str) -> str:
+ # ` #` starts a trailing comment on an unquoted value (dotenv convention).
+ # Without this, `TOKEN=abc123 # staging key` silently loaded the comment as
+ # part of the token and every authenticated request failed with no clue why.
+ marker = value.find(" #")
+ if marker == -1:
+ return value
+ return value[:marker].rstrip()
diff --git a/src/rights_filter/server/image_store.py b/src/rights_filter/server/image_store.py
index 046adeb..9d46db2 100644
--- a/src/rights_filter/server/image_store.py
+++ b/src/rights_filter/server/image_store.py
@@ -1,11 +1,14 @@
from __future__ import annotations
import json
+import logging
from pathlib import Path
from typing import Any
from rights_filter.analysis.preprocessing import ImagePayload
+logger = logging.getLogger(__name__)
+
SUPPORTED_IMAGE_SUFFIXES = {
".avif",
".bmp",
@@ -30,13 +33,23 @@ class LocalSubmissionImageStore:
manifest = self.root / "submissions.json"
if manifest.exists():
records = json.loads(manifest.read_text(encoding="utf-8"))
- normalized = [
- self._normalize_manifest_record(record)
- for record in records
- if isinstance(record, dict)
- and record.get("file")
- and self._safe_path(str(record["file"])).exists()
- ]
+ normalized: list[dict[str, Any]] = []
+ for record in records:
+ if not isinstance(record, dict) or not record.get("file"):
+ continue
+ # Skip a malformed row (e.g. a path escaping the image root)
+ # instead of letting its ValueError abort the whole listing —
+ # one bad manifest line previously paralyzed loading and
+ # re-analysis for the entire queue.
+ try:
+ if not self._safe_path(str(record["file"])).exists():
+ continue
+ normalized.append(self._normalize_manifest_record(record))
+ except (ValueError, OSError):
+ logger.warning(
+ "Skipping malformed submissions.json record: file=%r",
+ record.get("file"),
+ )
return self._with_unlisted_files(normalized)
return self._scan_records()
@@ -45,8 +58,8 @@ class LocalSubmissionImageStore:
path = self._safe_path(record["file"])
return ImagePayload(
content=path.read_bytes(),
- width=int(record["width"]),
- height=int(record["height"]),
+ width=_safe_dimension(record.get("width")),
+ height=_safe_dimension(record.get("height")),
metadata={
"submission_id": submission_id,
"title": str(record.get("title", submission_id)),
@@ -127,6 +140,15 @@ class LocalSubmissionImageStore:
return path
+def _safe_dimension(value: Any) -> int:
+ # A manifest may carry a non-numeric or missing width/height; fall back to
+ # the same placeholder the normalizer uses rather than raising mid-analysis.
+ try:
+ return int(value)
+ except (TypeError, ValueError):
+ return 1
+
+
def _image_size(path: Path) -> tuple[int, int]:
try:
from PIL import Image
diff --git a/src/rights_filter/server/store_enrichment.py b/src/rights_filter/server/store_enrichment.py
index c31d891..2986b2e 100644
--- a/src/rights_filter/server/store_enrichment.py
+++ b/src/rights_filter/server/store_enrichment.py
@@ -395,12 +395,21 @@ class StoreEnrichmentMixin:
if not crops:
continue
crop = crops[0]
- call_count += 1
crop_evidence = self.provider_runtime.google_adapter.detect(
f"{submission_id}:face-crop-{crop_index}",
crop,
self.provider_runtime.external_policy,
)
+ # Only count a crop as transmitted when the adapter actually issued a
+ # request. A policy refusal returns EXTERNAL_SKIPPED without sending
+ # anything, so counting it inflated provider usage and — worse —
+ # produced a false "personal image sent to external provider" audit
+ # record for biometric data that never left the system. A FAILURE
+ # still counts: the request may have been sent before it failed.
+ if any(
+ item.source != EvidenceSource.EXTERNAL_SKIPPED for item in crop_evidence
+ ):
+ call_count += 1
for item in crop_evidence:
face_web_evidence = _face_crop_web_evidence(submission_id, crop_index, item)
evidence.append(face_web_evidence)
diff --git a/src/rights_filter/server/store_operations.py b/src/rights_filter/server/store_operations.py
index a65ebe4..7c297f0 100644
--- a/src/rights_filter/server/store_operations.py
+++ b/src/rights_filter/server/store_operations.py
@@ -10,12 +10,14 @@ and self.* attributes provided by the host class. Behavior unchanged.
from __future__ import annotations
import json
+import logging
import sqlite3
from typing import Any
from rights_filter.analysis.fingerprints import FingerprintService
from rights_filter.analysis.search_result_promoter import SearchResultPromoter
-from rights_filter.domain.records import Evidence, EvidenceSource
+from rights_filter.domain.records import DataClass, Evidence, EvidenceSource
+from rights_filter.governance.policies import assert_operator_evidence_payload_allowed
from rights_filter.server.image_store import LocalSubmissionImageStore
from rights_filter.server.store_serialization import (
_domain_evidence_from_ui,
@@ -32,6 +34,8 @@ from rights_filter.server.store_serialization import (
)
from rights_filter.server.store_text import _text_list, _unique_texts
+logger = logging.getLogger(__name__)
+
class StoreOperationsMixin:
def promote_watchlist_entry(self, entry_id: str) -> dict[str, Any]:
@@ -228,15 +232,31 @@ class StoreOperationsMixin:
}
self._put("evidence", evidence["id"], {**evidence, "submission_id": submission_id})
- if image_store is not None:
- self._rerun_internal_analysis(submission_id, image_store)
- google_evidence = self._rerun_google_image_search(submission_id, image_store)
- query_source_evidence = [
- *google_evidence,
- *_submission_search_hint_evidence(submission),
- ]
- self._auto_naver_search(submission_id, query_source_evidence, image_store)
- self._auto_google_custom_search(submission_id, query_source_evidence, image_store)
+ rerun_error = ""
+ try:
+ if image_store is not None:
+ self._rerun_internal_analysis(submission_id, image_store)
+ google_evidence = self._rerun_google_image_search(submission_id, image_store)
+ query_source_evidence = [
+ *google_evidence,
+ *_submission_search_hint_evidence(submission),
+ ]
+ self._auto_naver_search(submission_id, query_source_evidence, image_store)
+ self._auto_google_custom_search(submission_id, query_source_evidence, image_store)
+ except Exception as exc:
+ # A submission whose image file went missing (or any other enrichment
+ # failure) must not escape as an unhandled exception: the handler
+ # would close the connection with no response while this method has
+ # already written lastAnalysis and the "queued" marker, leaving the
+ # case permanently marked as re-analyzing. Record the failure, keep
+ # the rescore/return path intact, and let the marker cleanup below run.
+ rerun_error = str(exc) or exc.__class__.__name__
+ logger.exception("Rerun enrichment failed for submission %s", submission_id)
+ finally:
+ # The rerun is synchronous, so by the time we return there is nothing
+ # "queued" — drop the marker instead of accumulating one dead row per
+ # rerun (previously they piled up in the operator evidence list).
+ self._delete_evidence_by_id(evidence["id"])
self._ensure_llm_summary(submission_id)
@@ -268,6 +288,7 @@ class StoreOperationsMixin:
refreshed = self._get("submissions", submission_id)
refreshed["lastRerunDiff"] = {
"at": _now_label(),
+ "error": rerun_error,
"scoreBefore": score_before,
"scoreAfter": int(refreshed.get("riskScore", 0) or 0),
"addedEvidenceIds": added_ids,
@@ -356,7 +377,18 @@ class StoreOperationsMixin:
item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
for item in domain_evidence
):
- provider_payload["lastFailure"] = domain_evidence[0].reason if domain_evidence else f"{provider} search failed"
+ # Report the reason of the item that actually failed. Indexing
+ # [0] picked whatever came first — usually a successful result —
+ # so the operator saw a success string labelled as the failure.
+ provider_payload["lastFailure"] = next(
+ (
+ item.reason
+ for item in domain_evidence
+ if item.source
+ in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
+ ),
+ f"{provider} search failed",
+ )
else:
provider_payload["lastSuccess"] = _now_label()
provider_payload["lastFailure"] = "없음"
@@ -403,6 +435,25 @@ class StoreOperationsMixin:
submission_id,
source_evidence,
)
+ # Enforce the LLM-summary governance rule on the real persistence path.
+ # The policy existed but no production caller invoked it, so an
+ # ungrounded summary (no source URLs and no source evidence ids) could be
+ # stored and shown as if it were source-linked. A summary that fails the
+ # check is dropped rather than persisted.
+ if llm_evidence.source == EvidenceSource.LLM_SUMMARY:
+ try:
+ assert_operator_evidence_payload_allowed(
+ DataClass.LLM_SUMMARY, llm_evidence.data
+ )
+ except ValueError as exc:
+ logger.warning(
+ "Discarding ungrounded LLM summary for submission %s: %s",
+ submission_id,
+ exc,
+ )
+ llm_provider["lastFailure"] = f"LLM assistance failed: {exc}"
+ self._apply_provider_usage_delta("llm", 1, llm_provider)
+ return False
self._delete_llm_summary_evidence(submission_id)
self._put(
"evidence",
@@ -417,6 +468,10 @@ class StoreOperationsMixin:
self._apply_provider_usage_delta("llm", 1, llm_provider)
return True
+ def _delete_evidence_by_id(self, evidence_id: str) -> None:
+ with self._connect() as conn:
+ conn.execute("delete from evidence where id = ?", (evidence_id,))
+
def _delete_llm_summary_evidence(self, submission_id: str) -> None:
with self._connect() as conn:
conn.execute(
@@ -657,15 +712,42 @@ class StoreOperationsMixin:
item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
for item in domain_evidence
):
- provider_payload["lastFailure"] = domain_evidence[0].reason if domain_evidence else f"{provider} collection failed"
+ provider_payload["lastFailure"] = next(
+ (
+ item.reason
+ for item in domain_evidence
+ if item.source
+ in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
+ ),
+ f"{provider} collection failed",
+ )
else:
provider_payload["lastSuccess"] = _now_label()
provider_payload["lastFailure"] = "없음"
+ # The cascade stages can surface the same candidate id more than once
+ # (e.g. an image result that also appears in the page sweep). Persisting
+ # is keyed by id, so counting every append reported more "collected"
+ # candidates than rows actually stored.
+ deduped_candidates = list({candidate["id"]: candidate for candidate in pending}.values())
+ collected = len(deduped_candidates)
+
+ # Every search was refused before it ran (quota/disabled provider), so we
+ # learned nothing. Wiping the previous collection here would destroy the
+ # operator's working set as punishment for exhausted quota.
+ if not deduped_candidates and provider_call_count == 0:
+ # Still persist why it was refused (zero delta merges lastFailure
+ # without charging quota) so the provider panel explains the no-op.
+ self._apply_provider_usage_delta(provider, 0, provider_payload)
+ payload = self.bootstrap()
+ payload["collected"] = 0
+ payload["collectionSkipped"] = True
+ return payload
+
with self._write_lock:
with self._transaction() as conn:
conn.execute("delete from collection_candidates")
- for candidate in pending:
+ for candidate in deduped_candidates:
self._put("collection_candidates", candidate["id"], candidate, conn=conn)
self.add_audit_event(
"rights.ops",
diff --git a/src/rights_filter/server/store_serialization.py b/src/rights_filter/server/store_serialization.py
index 586412d..4471608 100644
--- a/src/rights_filter/server/store_serialization.py
+++ b/src/rights_filter/server/store_serialization.py
@@ -186,6 +186,13 @@ def _evidence_matches_provider(evidence: dict[str, Any], provider_id: str) -> bo
if provider_id == "naver":
return source == "naver" or domain in {"naver", "naver_blog", "naver_web"}
if provider_id == "google":
+ # Vision failure/skip rows serialize with source "failure", so matching on
+ # source == "google" alone left them unattributed: the badge showed "not
+ # run" for a provider that had actually failed, while the score still
+ # carried the failure reason. Rows written before the adapter tagged the
+ # provider fall back to domain "internal" and are matched by title.
+ if source == "failure" and domain == "google":
+ return True
return source == "google" and domain != "google_custom_search"
if provider_id == "google_search":
return domain == "google_custom_search"
@@ -478,11 +485,29 @@ def _validate_payload(table: str, id_value: str, payload: dict[str, Any]) -> Non
raise ValueError(f"{table} payload id does not match storage id")
+# A quota-skip (SEARCH_SKIPPED) or transient failure (ENRICHMENT_FAILURE) means
+# the query never actually executed / produced no evidence, so it must NOT enter
+# the dedup signature set — otherwise auto-search treats it as "already done" and
+# never retries it, even after quota resets or the network recovers. Genuine
+# "returned no results" rows keep their real provider source (naver/google) and
+# stay deduped correctly (they did run).
+_RETRYABLE_SKIP_DOMAIN_SOURCES = {
+ EvidenceSource.SEARCH_SKIPPED.value,
+ EvidenceSource.ENRICHMENT_FAILURE.value,
+}
+
+
+def _is_retryable_skip_evidence(item: dict[str, Any]) -> bool:
+ return str(item.get("domainSource", "")) in _RETRYABLE_SKIP_DOMAIN_SOURCES
+
+
def _existing_naver_query_signatures(evidence: list[dict[str, Any]]) -> set[str]:
return {
str(item.get("querySignature") or _naver_query_signature(str(item.get("query", ""))))
for item in evidence
- if item.get("query") and item.get("source") in {"naver", "failure"}
+ if item.get("query")
+ and item.get("source") in {"naver", "failure"}
+ and not _is_retryable_skip_evidence(item)
}
@@ -494,6 +519,7 @@ def _existing_google_custom_query_signatures(evidence: list[dict[str, Any]]) ->
and item.get("source") in {"google", "failure"}
and str(item.get("domain", "")) == "google_custom_search"
and item.get("querySignature")
+ and not _is_retryable_skip_evidence(item)
}
diff --git a/tests/rights_filter/admin/test_detailed_review_presenter.py b/tests/rights_filter/admin/test_detailed_review_presenter.py
index 78663af..82f5fca 100644
--- a/tests/rights_filter/admin/test_detailed_review_presenter.py
+++ b/tests/rights_filter/admin/test_detailed_review_presenter.py
@@ -65,7 +65,10 @@ def test_detailed_review_groups_operator_evidence_and_actions():
assert len(review["evidence_groups"]["naver"]) == 1
assert len(review["evidence_groups"]["google"]) == 1
assert len(review["evidence_groups"]["llm"]) == 1
- assert len(review["evidence_groups"]["failures"]) == 1
+ # A provider that was skipped by policy is operational status, not an error:
+ # it must not be presented alongside genuine failures.
+ assert len(review["evidence_groups"]["skipped"]) == 1
+ assert "failures" not in review["evidence_groups"]
def test_missing_analysis_returns_unavailable_review_model():
diff --git a/tests/rights_filter/server/test_image_store.py b/tests/rights_filter/server/test_image_store.py
index c26fbdf..a0d6c15 100644
--- a/tests/rights_filter/server/test_image_store.py
+++ b/tests/rights_filter/server/test_image_store.py
@@ -101,7 +101,16 @@ def test_local_image_store_skips_manifest_records_with_missing_files(tmp_path: P
assert [record["id"] for record in records] == ["copied"]
-def test_local_image_store_rejects_manifest_paths_outside_root(tmp_path: Path):
+def test_local_image_store_skips_manifest_paths_outside_root_without_losing_valid_records(
+ tmp_path: Path,
+):
+ # An out-of-root path must never be surfaced as a submission, but it also must
+ # not abort the whole listing: a single malformed manifest row previously
+ # raised ValueError and paralyzed loading/re-analysis for the entire queue.
+ image_dir = tmp_path / "images"
+ image_dir.mkdir()
+ (image_dir / "good.svg").write_text("", encoding="utf-8")
+ (tmp_path.parent / "secret.svg").write_text("", encoding="utf-8")
(tmp_path / "submissions.json").write_text(
"""
[
@@ -111,17 +120,22 @@ def test_local_image_store_rejects_manifest_paths_outside_root(tmp_path: Path):
"file": "../secret.svg",
"width": 100,
"height": 100
+ },
+ {
+ "id": "SUB-GOOD",
+ "title": "Good sample",
+ "file": "images/good.svg",
+ "width": 640,
+ "height": 480
}
]
""",
encoding="utf-8",
)
- store = LocalSubmissionImageStore(tmp_path)
+ records = LocalSubmissionImageStore(tmp_path).submission_records()
- try:
- store.submission_records()
- except ValueError as exc:
- assert "outside image store" in str(exc)
- else:
- raise AssertionError("expected unsafe manifest path to be rejected")
+ ids = [record["id"] for record in records]
+ assert "SUB-BAD" not in ids
+ assert "SUB-GOOD" in ids
+ assert all("secret" not in str(record["file"]) for record in records)
diff --git a/tests/rights_filter/test_2026_07_24_review_fixes.py b/tests/rights_filter/test_2026_07_24_review_fixes.py
new file mode 100644
index 0000000..172c1e7
--- /dev/null
+++ b/tests/rights_filter/test_2026_07_24_review_fixes.py
@@ -0,0 +1,311 @@
+"""Regression tests for the 2026-07-24 multi-agent review remediation.
+
+One test per confirmed finding, named after the behaviour it locks in.
+"""
+
+from pathlib import Path
+
+import pytest
+
+from rights_filter.analysis.fingerprints import FingerprintService
+from rights_filter.analysis.risk_scoring import RiskScorer
+from rights_filter.analysis.search_result_promoter import SearchResultPromoter
+from rights_filter.domain.records import Evidence, EvidenceSource
+from rights_filter.integrations.external_policy import ExternalApiPolicy
+from rights_filter.server.env_file import load_env_file
+from rights_filter.server.image_store import LocalSubmissionImageStore
+from rights_filter.server.store_serialization import (
+ _domain_evidence_from_ui,
+ _evidence_matches_provider,
+ _evidence_payload,
+ _existing_naver_query_signatures,
+)
+
+
+# --- H1: an OpenCV build without CascadeClassifier must not crash analysis ----
+
+
+def test_face_detection_degrades_when_opencv_lacks_cascade_classifier(monkeypatch):
+ import rights_filter.analysis.face_person_detection as detection
+
+ class _Cv2WithoutCascades:
+ IMREAD_COLOR = 1
+ COLOR_BGR2GRAY = 6
+
+ def imdecode(self, *_args):
+ return [[0]]
+
+ def cvtColor(self, *_args):
+ return [[0]]
+
+ monkeypatch.setitem(__import__("sys").modules, "cv2", _Cv2WithoutCascades())
+ detection._warned_missing_cascade = False
+
+ signal = detection.HeuristicFacePersonDetector()._detect_with_opencv(b"bytes")
+
+ assert signal.face_count == 0
+ assert signal.present is False
+
+
+def test_batch_analyzer_isolates_an_unexpected_per_submission_failure():
+ from rights_filter.analysis.preprocessing import ImagePayload
+ from rights_filter.domain.records import InMemoryRightsFilterRepository
+ from rights_filter.integrations.cloud_vision_web_detection import (
+ CloudVisionWebDetectionAdapter,
+ FakeWebDetectionClient,
+ )
+ from rights_filter.jobs.batch_analyzer import BatchAnalyzer, SubmissionImage
+
+ class _ExplodingAnalyzer:
+ def analyze(self, submission_id, image):
+ if submission_id == "BAD":
+ raise AttributeError("module 'cv2' has no attribute 'CascadeClassifier'")
+ return []
+
+ batch = BatchAnalyzer(
+ repository=InMemoryRightsFilterRepository(),
+ internal_analyzer=_ExplodingAnalyzer(),
+ external_adapter=CloudVisionWebDetectionAdapter(FakeWebDetectionClient()),
+ external_policy=ExternalApiPolicy(disabled=True),
+ scorer=RiskScorer(),
+ )
+ image = ImagePayload(content=b"x", width=10, height=10, metadata={})
+
+ summary = batch.run(
+ [SubmissionImage("BAD", image), SubmissionImage("GOOD", image)]
+ )
+
+ # The bad submission is counted, and the good one still gets analysed.
+ assert summary.failed == 1
+ assert summary.processed == 1
+
+
+# --- H2: the persistence round-trip must not invent risk points --------------
+
+
+def test_custom_image_search_result_scores_the_same_before_and_after_round_trip():
+ fresh = Evidence(
+ source=EvidenceSource.WEB_DETECTION,
+ reason="Google custom image search result found",
+ confidence=0.45,
+ data={
+ "submission_id": "SUB-1",
+ "provider": "google_custom_search",
+ "search_type": "image",
+ "query": "test",
+ "result_url": "https://example.test/page",
+ "match": "search_image",
+ },
+ )
+ scorer = RiskScorer()
+
+ before = scorer.score([fresh]).score
+ restored = _domain_evidence_from_ui(_evidence_payload("SUB-1", fresh))
+ after = scorer.score([restored]).score
+
+ assert before == 0
+ assert after == before
+
+
+# --- H3: a quota-skipped query must stay retryable ---------------------------
+
+
+def test_quota_skipped_query_is_not_registered_as_already_searched():
+ skipped = Evidence(
+ source=EvidenceSource.SEARCH_SKIPPED,
+ reason="search API usage limit reached",
+ confidence=1.0,
+ data={
+ "submission_id": "SUB-1",
+ "provider": "naver",
+ "query": "IU album",
+ "query_signature": "naver:iu album",
+ },
+ )
+ executed = Evidence(
+ source=EvidenceSource.NAVER_SEARCH,
+ reason="Naver search returned no results",
+ confidence=0.0,
+ data={
+ "submission_id": "SUB-1",
+ "provider": "naver",
+ "query": "other query",
+ "query_signature": "naver:other query",
+ },
+ )
+ payloads = [
+ _evidence_payload("SUB-1", skipped),
+ _evidence_payload("SUB-1", executed),
+ ]
+
+ signatures = _existing_naver_query_signatures(payloads)
+
+ assert "naver:iu album" not in signatures
+ # A query that DID run (even with zero results) stays deduped.
+ assert "naver:other query" in signatures
+
+
+# --- H5: the external daily limit must re-window each day --------------------
+
+
+def test_external_policy_resets_its_daily_budget_on_a_new_day():
+ policy = ExternalApiPolicy(
+ compliance_approved=True,
+ metadata_logging_accepted=True,
+ allow_online_sync=True,
+ daily_limit=1,
+ )
+
+ policy.record_call()
+ assert policy.can_call()[0] is False
+
+ policy.window_day = "2000-01-01" # simulate the calendar day rolling over
+ assert policy.can_call()[0] is True
+
+
+# --- H6: one malformed manifest row must not paralyse the queue --------------
+
+
+def test_manifest_with_a_bad_row_still_lists_the_valid_submissions(tmp_path: Path):
+ (tmp_path / "images").mkdir()
+ (tmp_path / "images" / "ok.svg").write_text("", encoding="utf-8")
+ (tmp_path / "submissions.json").write_text(
+ '[{"id":"BAD","file":"../outside.svg","width":1,"height":1},'
+ '{"id":"OK","file":"images/ok.svg","width":1,"height":1}]',
+ encoding="utf-8",
+ )
+
+ records = LocalSubmissionImageStore(tmp_path).submission_records()
+
+ assert [record["id"] for record in records] == ["OK"]
+
+
+def test_non_numeric_manifest_dimensions_do_not_raise(tmp_path: Path):
+ (tmp_path / "images").mkdir()
+ (tmp_path / "images" / "ok.svg").write_text("", encoding="utf-8")
+ (tmp_path / "submissions.json").write_text(
+ '[{"id":"OK","file":"images/ok.svg","width":"wide","height":null}]',
+ encoding="utf-8",
+ )
+
+ payload = LocalSubmissionImageStore(tmp_path).image_payload("OK")
+
+ assert payload.width == 1
+ assert payload.height == 1
+
+
+# --- MEDIUM/LOW --------------------------------------------------------------
+
+
+def test_vision_failure_is_attributed_to_the_google_provider():
+ failure = Evidence(
+ source=EvidenceSource.FAILURE,
+ reason="External API failed: timeout",
+ confidence=1.0,
+ data={"submission_id": "SUB-1", "provider": "google"},
+ )
+
+ payload = _evidence_payload("SUB-1", failure)
+
+ assert _evidence_matches_provider(payload, "google") is True
+
+
+def test_face_person_evidence_counts_once_however_many_rows_exist():
+ rows = [
+ Evidence(
+ source=EvidenceSource.FACE_PERSON,
+ reason="Face or person detected",
+ confidence=1.0,
+ data={"submission_id": "SUB-1"},
+ )
+ for _ in range(3)
+ ]
+
+ assert RiskScorer().score(rows).score == 35
+
+
+def test_degenerate_perceptual_hashes_do_not_produce_a_near_match():
+ service = FingerprintService()
+ # Two "blank image" hashes differing by one bit: 0.98 similarity under a raw
+ # Hamming comparison, which used to register as a knowledge-base match.
+ assert service.similarity("phash:ffffffffffffffff", "phash:fffffffffffffffe") == 0.0
+ # Identical fingerprints still match, and informative hashes still compare.
+ assert service.similarity("phash:ffffffffffffffff", "phash:ffffffffffffffff") == 1.0
+ assert service.similarity("phash:0f0f0f0f0f0f0f0f", "phash:0f0f0f0f0f0f0f0f") == 1.0
+
+
+def test_promoter_ignores_marker_words_that_only_appear_in_the_url():
+ promoter = SearchResultPromoter()
+ item = Evidence(
+ source=EvidenceSource.NAVER_SEARCH,
+ reason="Naver search result found",
+ confidence=0.5,
+ data={
+ "submission_id": "SUB-1",
+ "title": "동네 산책 사진",
+ "description": "주말 나들이 기록",
+ "result_url": "https://cdn.example.test/game/album/cover/img.jpg",
+ },
+ )
+
+ promoted = promoter.promote([item])[0]
+
+ assert promoted.data["promoted"] is False
+
+
+def test_env_file_strips_a_trailing_inline_comment(tmp_path: Path):
+ env_path = tmp_path / ".env"
+ env_path.write_text(
+ 'COPYRIGHTER_AUTH_TOKEN=abc123 # staging key\nQUOTED="keep # this"\n',
+ encoding="utf-8",
+ )
+ environ: dict[str, str] = {}
+
+ load_env_file(env_path, environ)
+
+ assert environ["COPYRIGHTER_AUTH_TOKEN"] == "abc123"
+ assert environ["QUOTED"] == "keep # this"
+
+
+def test_search_policy_reservation_is_atomic_and_refundable():
+ from rights_filter.integrations.search_policy import SearchApiPolicy
+
+ policy = SearchApiPolicy(
+ compliance_approved=True, allowed_providers={"naver"}, daily_limit=1
+ )
+
+ assert policy.try_reserve("naver")[0] is True
+ # The budget is consumed at reservation time, so a racing caller is refused.
+ assert policy.try_reserve("naver")[0] is False
+
+ policy.release()
+ assert policy.try_reserve("naver")[0] is True
+
+
+def test_rejected_image_knowledge_entry_is_reused_across_repeat_rejections():
+ from rights_filter.domain.records import InMemoryRightsFilterRepository
+
+ repository = InMemoryRightsFilterRepository()
+
+ first = repository.create_rejected_image_entry("dec-1", "SUB-1", ["phash:aaaa"])
+ second = repository.create_rejected_image_entry("dec-2", "SUB-1", ["phash:bbbb"])
+
+ assert first.id == second.id
+ assert second.source_decision_id == "dec-2"
+ assert set(second.sample_fingerprints) == {"phash:aaaa", "phash:bbbb"}
+
+
+def test_manual_knowledge_registration_rejects_unknown_fields():
+ from rights_filter.admin.knowledge_base_handlers import register_manual_entry
+ from rights_filter.domain.records import (
+ InMemoryRightsFilterRepository,
+ KnowledgeEntryType,
+ )
+
+ with pytest.raises(ValueError, match="unsupported"):
+ register_manual_entry(
+ InMemoryRightsFilterRepository(),
+ KnowledgeEntryType.CELEBRITY,
+ "name",
+ surprise="dropped silently before",
+ )