fix: 저장·복구 정합성 5종 — 호스트 재직렬화, 고아 크롭, assetMissing 생애주기

1) `_result_host` 가 자기가 만든 host 를 안 읽어 재직렬화 시 퇴행 (잠재)

   라운드트립은 provider 를 버리고 domain(= provider 이름)만 남긴다. 그래서
   복구된 행을 다시 직렬화하면 `explicit != provider` 가 성립해 호스트가
   "google_custom_search" / "google" / "internal" 로 되돌아갔다 — 847f7a8 이
   고친 바로 그 버그가, 테스트 전부 초록인 채로. host 우선 반환으로 고쳤고
   3회 왕복 안정성을 테스트로 고정했다. 현재 `_evidence_payload` 호출부는
   모두 신선한 증거라 잠재였지만, 증거 편집 엔드포인트나 마이그레이션이
   하나 생기면 바로 라이브가 된다.

   `_hostname_of` 도 함께 고쳤다: urlparse("//google") 이 "google" 을 호스트로
   보고하므로 점 없는 토큰은 거부한다. 잘못 저장된 host="google" 행은 URL
   파싱으로 자동 복구된다.

2) 복구가 남긴 고아 faceCrops 참조

   백업 복구는 보존기간 대상 미디어를 의도적으로 스킵하는데 DB 는 그대로
   복사되므로, 행은 존재하지 않는 크롭 썸네일을 계속 광고한다. 재생성 게이트가
   `"faceCrops" not in submission` 이라 복구된 행은 리로드에서도 건너뛰고,
   purge 는 존재하는 파일만 순회하므로 참조도 못 지운다 — 영구히 깨진 생체
   썸네일에 회수 경로가 없었다. 게이트를 파일 존재 확인으로 바꾸고(원본이
   있으면 재생성), 원본까지 없으면 참조를 정리하고 감사 이벤트를 남긴다.

3) assetMissing 이 한 번 켜지면 안 꺼졌다

   파일이 돌아와도 486행에서 조기 continue 하므로 플래그가 영구 거짓이 되고,
   1회 마킹 가드가 이후의 진짜 소실에 대한 감사 이벤트까지 막았다. 복귀 시
   해제 + "Submission asset restored" 이벤트를 추가해, 두 번째 소실도 감사에
   남는다.

4) 플래그와 감사 이벤트가 트랜잭션 밖 2회 쓰기였다

   사이에서 죽으면 플래그만 남고 이벤트는 없으며, 가드 때문에 이후 어떤
   리로드도 이벤트를 쓰지 않는다 — 감사 로그가 유일한 기록인 납품에서 판정
   보존의 근거가 사라진다. purge_expired_face_crops / set_provider_enabled 와
   같은 방식으로 한 트랜잭션에 묶었다.

5) 스코어러 마이그레이션이 감사 흔적 없이 밴드를 바꿨다

   가중치 변경은 다음 리로드에서 전 행을 리스코어한다. 운영자가 high 로
   반려한 케이스가 medium 으로 표시되는데 "새 근거" 와 "스코어러 교체" 를
   구분할 방법이 없었다. 라이브 스탬프 `scoredVersion` 을 도입해 버전이
   바뀌면서 밴드가 움직인 행에만 "Risk rescored" 이벤트를 1회 남긴다 —
   동일 버전의 일상 리스코어는 조용하므로 500행 감사 캡을 쓰지 않는다.
   as-created 스탬프(scorerVersion)는 그대로 보존된다.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BStHFrjQj2VypecSG9ex1V

src/rights_filter/server/sqlite_store.py

@@ -338,7 +338,7 @@ class CopyrighterStore(
         self._refresh_existing_submission_file_facts(image_store, queue_id=queue_id)
         self._refresh_existing_local_face_evidence(image_store, queue_id=queue_id)
         for submission in self._all("submissions", queue_id=queue_id):
-            if "faceCrops" not in submission:
+            if self._face_crops_need_sync(submission):
                 self._sync_face_crops(str(submission["id"]), image_store)
         self._rescore_all_submissions(queue_id=queue_id)
         self._sync_submission_provider_state(queue_id=queue_id)
@@ -477,12 +477,19 @@ class CopyrighterStore(
         public_prefix = image_store.public_prefix.rstrip("/") + "/"
         missing_ids: list[str] = []
         retained: list[dict[str, Any]] = []
+        recovered: list[dict[str, Any]] = []
         for submission in self._all("submissions", queue_id=queue_id):
             asset = str(submission.get("asset", ""))
             if not asset.startswith(public_prefix):
                 continue
             relative_path = asset.removeprefix(public_prefix)
             if image_store.media_path(relative_path).exists():
+                # The file came back - ordinary housekeeping is reversible. The
+                # marker has to come off with it: left set it becomes a permanent
+                # lie in the console, and the mark-once guard below would then
+                # suppress the audit event for a later, genuine disappearance.
+                if submission.get("assetMissing"):
+                    recovered.append(submission)
                 continue
             # A decided case IS the record of that decision, and evidence rows
             # cascade with the submission. Moving a reviewed image out of the
@@ -494,17 +501,36 @@ class CopyrighterStore(
                 continue
             missing_ids.append(str(submission["id"]))
 
+        for submission in recovered:
+            with self._transaction() as conn:
+                submission["assetMissing"] = False
+                self._put("submissions", submission["id"], submission, conn=conn)
+                self.add_audit_event(
+                    actor="system",
+                    event="Submission asset restored",
+                    object_id=str(submission["id"]),
+                    change="source image file present again",
+                    conn=conn,
+                )
+
         for submission in retained:
             # Mark once. This sweep runs on every reload, so an unmarked retain
-            # would re-emit the same audit event until it filled the log.
-            submission["assetMissing"] = True
-            self._put("submissions", submission["id"], submission)
-            self.add_audit_event(
-                actor="system",
-                event="Submission retained",
-                object_id=str(submission["id"]),
-                change="source image file missing; decided case kept",
-            )
+            # would re-emit the same audit event until it filled the log. Marker
+            # and event go in ONE transaction: a crash between two separate
+            # commits would persist the flag without the event, and the guard
+            # above then means no later reload ever writes it - leaving the
+            # retention of a decided case with no audit trail at all, in a
+            # delivery whose audit log is the only record.
+            with self._transaction() as conn:
+                submission["assetMissing"] = True
+                self._put("submissions", submission["id"], submission, conn=conn)
+                self.add_audit_event(
+                    actor="system",
+                    event="Submission retained",
+                    object_id=str(submission["id"]),
+                    change="source image file missing; decided case kept",
+                    conn=conn,
+                )
 
         if not missing_ids:
             return 0

src/rights_filter/server/store_enrichment.py

@@ -13,6 +13,7 @@ import json
 import logging
 from dataclasses import replace
 from datetime import datetime
+from pathlib import Path
 from typing import Any
 
 from rights_filter.analysis.face_person_detection import (
@@ -187,6 +188,56 @@ class StoreEnrichmentMixin:
             )
             self._rescore_submission(submission_id)
 
+    def _face_crop_path(self, submission_id: str, crop: dict[str, Any]) -> Path:
+        return self.face_crop_image_dir / submission_id / f"crop-{crop.get('index')}.jpg"
+
+    def _face_crops_need_sync(self, submission: dict[str, Any]) -> bool:
+        """Whether a reload has to rebuild (or clear) this row's face crops.
+
+        A missing key means the row predates the feature. A key whose files are
+        gone means the crops were taken out from under the DB: archive restore
+        skips retention-bound media on purpose (store_backup.py), and nothing
+        reconciled the references it left behind - the rebuild used to be gated
+        on the key's absence alone, so a restored row kept advertising
+        thumbnails the server would 404 forever.
+        """
+        if "faceCrops" not in submission:
+            return True
+        submission_id = str(submission.get("id", ""))
+        return any(
+            not self._face_crop_path(submission_id, crop).exists()
+            for crop in submission.get("faceCrops", [])
+            if isinstance(crop, dict)
+        )
+
+    def _prune_missing_face_crop_refs(self, submission_id: str) -> None:
+        """Drop crop references whose file is gone when no rebuild is possible.
+
+        Reached when the original image is unavailable too (a restore that
+        skipped both, or a deleted upload). purge_expired_face_crops only walks
+        files that still exist, so without this the row is stuck advertising
+        biometric thumbnails that cannot be served or regenerated.
+        """
+        try:
+            submission = self._get("submissions", submission_id)
+        except KeyError:
+            return
+        crops = [crop for crop in submission.get("faceCrops", []) if isinstance(crop, dict)]
+        remaining = [crop for crop in crops if self._face_crop_path(submission_id, crop).exists()]
+        if len(remaining) == len(crops):
+            return
+        dropped = len(crops) - len(remaining)
+        with self._transaction() as conn:
+            submission["faceCrops"] = remaining
+            self._put("submissions", submission_id, submission, conn=conn)
+            self.add_audit_event(
+                actor="system",
+                event="Face crop reference cleared",
+                object_id=submission_id,
+                change=f"{dropped} crop reference(s) had no file and no original image to rebuild from",
+                conn=conn,
+            )
+
     def _sync_face_crops(
         self,
         submission_id: str,
@@ -200,6 +251,7 @@ class StoreEnrichmentMixin:
             # A missing/unreadable image is expected for some rows, but log it so
             # a real disk/permission failure is not invisible.
             logger.warning("face-crop sync skipped: could not load image for %s", submission_id, exc_info=True)
+            self._prune_missing_face_crop_refs(submission_id)
             return
 
         signal = HeuristicFacePersonDetector().detect(original)

src/rights_filter/server/store_search_candidates.py

@@ -13,7 +13,7 @@ from pathlib import Path
 from typing import Any
 
 from rights_filter.analysis.fingerprints import FingerprintService
-from rights_filter.analysis.risk_scoring import RiskScorer
+from rights_filter.analysis.risk_scoring import SCORER_VERSION, RiskScorer
 from rights_filter.domain.records import (
     Evidence,
     EvidenceSource,
@@ -725,6 +725,14 @@ class StoreSearchCandidatesMixin:
             for item in self._evidence_for_submission(submission_id)
         ]
         score = RiskScorer().score(evidence)
+        previous_band = str(submission.get("riskBand", ""))
+        previous_score = submission.get("riskScore", "")
+        # The version the LIVE score was last produced by, as opposed to the
+        # as-created stamp below. A weight change rescores every row on the next
+        # folder reload, so without this a case an operator rejected as "high"
+        # could display as "medium" with different reasons and no trace that a
+        # scorer migration - rather than new evidence - moved it.
+        scored_version = str(submission.get("scoredVersion", ""))
         # Only the live score moves. initialRiskScore / initialRiskBand /
         # scorerVersion are the as-created stamp and must survive every rescore —
         # they are the other half of the comparison that makes score-preservation
@@ -738,7 +746,30 @@ class StoreSearchCandidatesMixin:
             # mark them instead of fabricating one, so a later calibration can
             # exclude them rather than silently trust a round-tripped value.
             submission["scorerVersion"] = "legacy"
-        self._put("submissions", submission_id, submission)
+        submission["scoredVersion"] = SCORER_VERSION
+        migrated_band = (
+            scored_version != SCORER_VERSION
+            and bool(previous_band)
+            and previous_band != score.band
+        )
+        if not migrated_band:
+            self._put("submissions", submission_id, submission)
+            return
+        # One event per row per migration: scoredVersion is updated in the same
+        # write, so an ordinary rescore on the same version stays silent and the
+        # 500-row audit cap is not spent on repeat reloads.
+        with self._transaction() as conn:
+            self._put("submissions", submission_id, submission, conn=conn)
+            self.add_audit_event(
+                actor="system",
+                event="Risk rescored",
+                object_id=submission_id,
+                change=(
+                    f"scorer {scored_version or 'pre-stamp'} -> {SCORER_VERSION}: "
+                    f"{previous_band} {previous_score} -> {score.band} {score.score}"
+                ),
+                conn=conn,
+            )
 
     def _rescore_all_submissions(self, queue_id: str | None = None) -> None:
         for submission in self._all("submissions", queue_id=queue_id):

src/rights_filter/server/store_serialization.py

@@ -277,6 +277,15 @@ def _result_host(evidence: Evidence) -> str:
     — and until now neither the operator's evidence card nor the LLM prompt could
     see it: every row read "domain=google".
     """
+    # A row that already carries a resolved host keeps it. Re-serializing a
+    # restored row used to walk straight past this: the round trip drops
+    # `provider` but keeps `domain` (which holds the provider NAME), so
+    # `explicit != provider` held and the host reverted to "google_custom_search"
+    # / "google" / "internal" - the exact bug this function was written to fix,
+    # silently, with every test still green.
+    restored = _hostname_of(str(evidence.data.get("host", "")))
+    if restored:
+        return restored
     provider = str(evidence.data.get("provider", "")).strip()
     explicit = str(evidence.data.get("domain", "")).strip()
     if explicit and explicit != provider:
@@ -294,6 +303,12 @@ def _hostname_of(value: str) -> str:
     if not text:
         return ""
     if "://" not in text:
+        # urlparse("//google") happily reports "google" as a hostname, so a bare
+        # token with no dot would pass an API name off as a publishing host. A
+        # hostname we can show an operator has a label separator; anything else
+        # is "no host", which is what the caller already handles.
+        if "." not in text.split("/")[0]:
+            return ""
         text = f"//{text}"
     try:
         host = urlparse(text).hostname or ""

src/rights_filter/server/store_workflow.py

@@ -78,6 +78,10 @@ class StoreWorkflowMixin:
             "initialRiskScore": submission.get("initialRiskScore"),
             "initialRiskBand": submission.get("initialRiskBand"),
             "scorerVersion": submission.get("scorerVersion", "legacy"),
+            # The version that produced the score shown above. It diverges from
+            # scorerVersion (the as-created stamp) after a weight change, which
+            # is exactly when "why is this case medium now?" gets asked.
+            "scoredVersion": submission.get("scoredVersion", ""),
             "providerState": dict(submission.get("providerState") or {}),
             "evidenceCount": len(shown),
             "contributedCount": sum(1 for item in shown if item["contributed"]),

tests/rights_filter/server/test_sqlite_store.py

@@ -13,6 +13,7 @@ from rights_filter.analysis.face_person_detection import (
     face_person_reason,
 )
 from rights_filter.analysis.preprocessing import ImagePayload, build_face_crop_derivatives
+from rights_filter.analysis.risk_scoring import SCORER_VERSION
 from rights_filter.domain.records import Evidence, EvidenceSource
 from rights_filter.server.image_store import LocalSubmissionImageStore
 from rights_filter.server import sqlite_store as sqlite_store_module
@@ -141,6 +142,27 @@ def _image_store(tmp_path: Path) -> LocalSubmissionImageStore:
     return LocalSubmissionImageStore(tmp_path)
 
 
+def _append_submission_record(image_root: Path, record_id: str, file_name: str) -> None:
+    """Drop one more image + manifest row into an existing queue folder."""
+    (image_root / "images" / file_name).write_text(
+        "<svg><!-- FACE PERSON --></svg>",
+        encoding="utf-8",
+    )
+    manifest_path = image_root / "submissions.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest.append(
+        {
+            "id": record_id,
+            "title": "Late arrival",
+            "file": f"images/{file_name}",
+            "width": 800,
+            "height": 600,
+            "submitted_at": "2026-05-26 11:00",
+        }
+    )
+    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+
+
 def _png_image_store(tmp_path: Path) -> LocalSubmissionImageStore:
     from PIL import Image
 
@@ -1070,9 +1092,15 @@ def test_sqlite_store_emergency_disable_stops_automatic_external_calls(tmp_path:
     calls_at_stop = len(transport.calls)
 
     submission_id = str(store.bootstrap()["submissions"][0]["id"])
-    store.seed_from_image_store(LocalSubmissionImageStore(image_root))
+    # A NEW file has to arrive for the seed path to reach its provider gate.
+    # Re-seeding the same folder returns at the "nothing new to analyze" guard
+    # eleven lines earlier, so the gate could be deleted outright and a
+    # re-seed-only assertion would still pass.
+    _append_submission_record(image_root, "SUB-LATE", "late.svg")
+    seeded = store.seed_from_image_store(LocalSubmissionImageStore(image_root))
     store.rerun_enrichment(submission_id, image_store)
 
+    assert seeded == 1, "the new record must actually reach the seed path"
     assert len(transport.calls) == calls_at_stop
 
 
@@ -8465,3 +8493,157 @@ def test_purge_expired_face_crops_keeps_recent_crops(tmp_path: Path):
 
     assert store.purge_expired_face_crops() == 0
     assert crop_file.exists()
+
+
+def test_asset_missing_marker_is_cleared_when_the_source_file_comes_back(tmp_path: Path):
+    # Archiving a decided image out of the queue folder is reversible, so the
+    # marker has to be too. Left permanently set it became a lie in the console,
+    # and the mark-once guard then suppressed the audit event for a later,
+    # genuine disappearance - the second loss was invisible in the audit trail.
+    image_root = tmp_path / "submissions"
+    image_store = _image_store(image_root)
+    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3")
+    store.initialize()
+    store.seed_from_image_store(image_store)
+    submission_id = str(store.bootstrap()["submissions"][0]["id"])
+    store.record_decision(submission_id, "rejected", memo="third-party photo")
+
+    image_path = image_root / "images" / "sample.svg"
+    archived = image_path.read_text(encoding="utf-8")
+    image_path.unlink()
+    store.seed_from_image_store(LocalSubmissionImageStore(image_root))
+    assert store._get("submissions", submission_id)["assetMissing"] is True
+
+    image_path.write_text(archived, encoding="utf-8")
+    store.seed_from_image_store(LocalSubmissionImageStore(image_root))
+    assert store._get("submissions", submission_id)["assetMissing"] is False
+    assert any(event["event"] == "Submission asset restored" for event in store.audit_events())
+
+    image_path.unlink()
+    store.seed_from_image_store(LocalSubmissionImageStore(image_root))
+    retained = [event for event in store.audit_events() if event["event"] == "Submission retained"]
+    assert len(retained) == 2, "a second, genuine loss has to be audited too"
+
+
+def _submission_with_face_crop_reference(store: CopyrighterStore, submission_id: str) -> None:
+    store._put(
+        "submissions",
+        submission_id,
+        {
+            "id": submission_id,
+            "title": "t",
+            "asset": "",
+            "riskScore": 0,
+            "riskBand": "low",
+            "decisionStatus": "unreviewed",
+            "providerState": {},
+            "fileFacts": {},
+            "evidence": [],
+            "faceCrops": [
+                {"index": 1, "url": f"/face-crop-media/{submission_id}/crop-1.jpg", "box": [0, 0, 1, 1]}
+            ],
+        },
+    )
+
+
+def test_reload_detects_face_crop_references_whose_files_are_gone(tmp_path: Path):
+    # Archive restore skips retention-bound media on purpose, so the DB comes
+    # back listing biometric thumbnails whose files are not there. The rebuild
+    # used to be gated on the key being absent, and purge_expired_face_crops
+    # only walks files that exist - so nothing reconciled the reference and the
+    # console rendered a permanently broken crop with no recovery path.
+    store = CopyrighterStore(tmp_path / "c.sqlite3")
+    store.initialize()
+    _submission_with_face_crop_reference(store, "SUB-CROP")
+
+    assert store._face_crops_need_sync(store._get("submissions", "SUB-CROP")) is True
+
+    crop_dir = store.face_crop_image_dir / "SUB-CROP"
+    crop_dir.mkdir(parents=True)
+    (crop_dir / "crop-1.jpg").write_bytes(b"jpegdata")
+    assert store._face_crops_need_sync(store._get("submissions", "SUB-CROP")) is False
+
+
+def test_face_crop_references_are_cleared_when_there_is_nothing_to_rebuild_from(tmp_path: Path):
+    # Reached when the original image is gone too. Clearing is the only honest
+    # outcome: the row cannot serve the crop and cannot regenerate it.
+    store = CopyrighterStore(tmp_path / "c.sqlite3")
+    store.initialize()
+    _submission_with_face_crop_reference(store, "SUB-CROP")
+
+    store._sync_face_crops("SUB-CROP", _image_store(tmp_path / "submissions"))
+
+    assert store._get("submissions", "SUB-CROP")["faceCrops"] == []
+    assert any(event["event"] == "Face crop reference cleared" for event in store.audit_events())
+
+
+def test_rescore_audits_a_band_change_caused_by_a_scorer_migration(tmp_path: Path):
+    # A weight change rescores every row on the next folder reload. Without a
+    # trace, a case an operator rejected as "high" displays as "medium" with
+    # different reasons and nothing says a scorer migration - rather than new
+    # evidence - moved it.
+    image_root = tmp_path / "submissions"
+    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3")
+    store.initialize()
+    store.seed_from_image_store(_image_store(image_root))
+    submission_id = str(store.bootstrap()["submissions"][0]["id"])
+
+    submission = store._get("submissions", submission_id)
+    submission["scoredVersion"] = "v1"
+    submission["riskBand"] = "high"
+    submission["riskScore"] = 100
+    store._put("submissions", submission_id, submission)
+
+    store._rescore_submission(submission_id)
+
+    events = [event for event in store.audit_events() if event["event"] == "Risk rescored"]
+    assert len(events) == 1
+    assert f"v1 -> {SCORER_VERSION}" in events[0]["change"]
+    assert store._get("submissions", submission_id)["scoredVersion"] == SCORER_VERSION
+
+    # An ordinary rescore on the same version stays silent, so repeat reloads do
+    # not spend the audit cap.
+    store._rescore_submission(submission_id)
+    assert len([event for event in store.audit_events() if event["event"] == "Risk rescored"]) == 1
+
+
+def test_evidence_host_survives_repeated_persistence_round_trips():
+    # The round trip drops `provider` but keeps `domain` (which holds the
+    # provider NAME), so re-serializing a restored row used to resolve the host
+    # back to "google_custom_search" - the bug this field was added to fix,
+    # reintroduced silently with every test still green.
+    fresh = Evidence(
+        source=EvidenceSource.WEB_DETECTION,
+        reason="Google custom web search result found",
+        confidence=0.35,
+        data={
+            "provider": "google_custom_search",
+            "domain": "blog.example.test",
+            "search_type": "web",
+            "match": "page",
+            "result_url": "https://blog.example.test/post/1",
+        },
+    )
+
+    first = store_serialization_module._evidence_payload("SUB-1", fresh)
+    second = store_serialization_module._evidence_payload(
+        "SUB-1", store_serialization_module._domain_evidence_from_ui(first)
+    )
+    third = store_serialization_module._evidence_payload(
+        "SUB-1", store_serialization_module._domain_evidence_from_ui(second)
+    )
+
+    assert first["host"] == "blog.example.test"
+    assert second["host"] == "blog.example.test"
+    assert third["host"] == "blog.example.test"
+
+
+def test_provider_names_are_not_accepted_as_publishing_hosts():
+    # urlparse("//google") reports "google" as a hostname, which is how an API
+    # name ended up rendered as the place an image was published.
+    hostname_of = store_serialization_module._hostname_of
+
+    assert hostname_of("google_custom_search") == ""
+    assert hostname_of("internal") == ""
+    assert hostname_of("biz.chosun.com") == "biz.chosun.com"
+    assert hostname_of("https://www.Example.co.kr/a.jpg") == "example.co.kr"