fix: Phase 1 HIGH review remediation (5 of 6)

Backend (data-integrity / audit-trail):
- mark_evidence_status: make the read-modify-write + audit atomic under
  _write_lock + _transaction (was 4 unlocked connections, last-writer-wins,
  audit could diverge from DB state).
- promote_collection_candidates: acquire _write_lock, use a deterministic
  _stable_id, wrap writes in _transaction, and guard already-promoted batches
  so a retry/double-click cannot mint duplicate knowledge entries.
- risk_scoring: bucket web-detection entities by name so two distinct entities
  in one image both count (was collapsed to one 'entity' bucket, under-scoring
  high->medium). Identical entities still dedup.

Frontend (silent loss of operator actions):
- runManualSearch: in-flight guard; rerunHistoricalQuery voids the call,
  preventing duplicate/racing search POSTs.
- deactivateCorrectionEntry: persist via the real knowledge-deactivate
  endpoint instead of a local-only mutation + fake audit event.
- provider-retry button: report honestly (no backend endpoint) instead of
  writing a fake 'Provider called' audit event.

Tests: +4 regression tests (mark_evidence_status rollback, batch-promote
idempotency, two distinct entities -> high, identical entity dedup).
Remaining Phase 1 item #6 (_auto_naver/_auto_google refactor) deferred.

Backend suite: 361 passed. app.js: node --check OK, 41 static tests pass.

src/rights_filter/analysis/risk_scoring.py

@@ -93,10 +93,15 @@ def _web_detection_points(item: Evidence) -> int:
 
 def _web_detection_bucket(item: Evidence) -> str:
     match = str(item.data.get("match", ""))
-    if match in {"full", "partial", "page", "visual", "entity"}:
+    if match in {"full", "partial", "page", "visual"}:
         return match
-    if item.data.get("entity"):
+    # Distinct named entities are independent risk signals (e.g. two different
-        return "entity"
+    # celebrities/characters in one image), so they must not share one "entity"
+    # bucket and dedup each other away. Key the bucket by the entity name; only
+    # genuinely identical entities collapse.
+    entity = str(item.data.get("entity", "")).strip()
+    if match == "entity" or entity:
+        return f"entity:{entity.casefold()}" if entity else "entity"
     if item.data.get("url"):
         return "url"
     return ""

src/rights_filter/server/sqlite_store.py

@@ -593,27 +593,34 @@ class CopyrighterStore(
     ) -> dict[str, Any]:
         if status not in EVIDENCE_OPERATOR_STATUSES:
             raise ValueError("unsupported evidence status")
-        payload = self._get("evidence", evidence_id)
+        # Serialize the read-modify-write so two operators marking the same item
-        if str(payload.get("submission_id", "")) != submission_id:
+        # cannot lose each other's update, and persist the evidence change + its
-            raise ValueError("evidence does not belong to submission")
+        # audit event atomically (mirrors record_decision). The rescore is
-
+        # derived state: it runs after the commit but still under the write lock,
-        payload["operatorStatus"] = status
+        # so no other writer interleaves between the change and the rescore.
-        payload["operatorStatusLabel"] = EVIDENCE_OPERATOR_STATUSES[status]
+        with self._write_lock:
-        payload["operatorStatusNote"] = note.strip()
+            with self._transaction() as conn:
-        payload["operatorStatusAt"] = _now_label()
+                payload = self._get("evidence", evidence_id, conn=conn)
-        payload["operatorStatusActor"] = "rights.ops"
+                if str(payload.get("submission_id", "")) != submission_id:
-        if status in NON_CONTRIBUTING_OPERATOR_STATUSES:
+                    raise ValueError("evidence does not belong to submission")
-            payload["contributed"] = False
+                payload["operatorStatus"] = status
-        else:
+                payload["operatorStatusLabel"] = EVIDENCE_OPERATOR_STATUSES[status]
-            payload["contributed"] = _default_evidence_contribution(payload)
+                payload["operatorStatusNote"] = note.strip()
-        self._put("evidence", evidence_id, payload)
+                payload["operatorStatusAt"] = _now_label()
-        self._rescore_submission(submission_id)
+                payload["operatorStatusActor"] = "rights.ops"
-        self.add_audit_event(
+                if status in NON_CONTRIBUTING_OPERATOR_STATUSES:
-            actor="rights.ops",
+                    payload["contributed"] = False
-            event="Evidence status changed",
+                else:
-            object_id=evidence_id,
+                    payload["contributed"] = _default_evidence_contribution(payload)
-            change=f"{EVIDENCE_OPERATOR_STATUSES[status]} · {note or 'note optional'}",
+                self._put("evidence", evidence_id, payload, conn=conn)
-        )
+                self.add_audit_event(
+                    actor="rights.ops",
+                    event="Evidence status changed",
+                    object_id=evidence_id,
+                    change=f"{EVIDENCE_OPERATOR_STATUSES[status]} · {note or 'note optional'}",
+                    conn=conn,
+                )
+            self._rescore_submission(submission_id)
         return self.review(submission_id)
 
     def providers(self) -> list[dict[str, Any]]:

src/rights_filter/server/store_operations.py

@@ -696,65 +696,78 @@ class StoreOperationsMixin:
         if not candidate_ids:
             raise ValueError("candidate_ids required")
 
-        candidates = [self._get("collection_candidates", candidate_id) for candidate_id in candidate_ids]
+        with self._write_lock:
-        sample_fingerprints = _unique_texts(
+            candidates = [self._get("collection_candidates", candidate_id) for candidate_id in candidate_ids]
-            fingerprint
+            sample_fingerprints = _unique_texts(
-            for candidate in candidates
+                fingerprint
-            for fingerprint in _text_list(candidate.get("sampleFingerprints"))
-        )
-        if not sample_fingerprints:
-            raise ValueError("selected candidates have no sample fingerprints")
-
-        entry_id = f"kb-candidate-{_timestamp_id()}"
-        queries = _unique_texts(str(candidate.get("query", "")) for candidate in candidates)
-        image_assets = _unique_texts(str(candidate.get("imageAsset", "")) for candidate in candidates)
-        image_facts = [
-            candidate.get("imageFacts", {})
-            for candidate in candidates
-            if isinstance(candidate.get("imageFacts", {}), dict) and candidate.get("imageFacts", {})
-        ]
-        fallback_name = next(
-            (
-                str(candidate.get("title", "")).strip() or str(candidate.get("query", "")).strip()
                 for candidate in candidates
-                if str(candidate.get("title", "")).strip() or str(candidate.get("query", "")).strip()
+                for fingerprint in _text_list(candidate.get("sampleFingerprints"))
-            ),
+            )
-            "Collected reference",
+            if not sample_fingerprints:
-        )
+                raise ValueError("selected candidates have no sample fingerprints")
-        name = str(payload.get("name", "")).strip() or fallback_name
+
-        memo = str(payload.get("memo", "")).strip() or f"Promoted from collected candidates: {', '.join(queries)}"
+            # Deterministic id so a racing retry / double-click upserts the same
-        entry = {
+            # knowledge entry instead of minting a duplicate timestamped row.
-            "id": entry_id,
+            entry_id = _stable_id("kb-candidate-batch", *sorted(candidate_ids))
-            "name": name,
+            # Idempotent: if this exact batch was already promoted, do nothing.
-            "type": _knowledge_type_value(str(payload.get("type", "rejected_reference"))),
+            if all(
-            "aliases": _text_list(payload.get("aliases")) or queries,
+                candidate.get("status") == "promoted"
-            "keywords": _text_list(payload.get("keywords")) or queries,
+                and candidate.get("promotedKnowledgeId") == entry_id
-            "memo": memo,
+                for candidate in candidates
-            "provenance": "manual",
+            ):
-            "active": True,
+                return self.bootstrap()
-            "entryStatus": "confirmed",
+
-            "sourceDecision": "",
+            queries = _unique_texts(str(candidate.get("query", "")) for candidate in candidates)
-            "sourceCandidate": candidate_ids[0] if len(candidate_ids) == 1 else "",
+            image_assets = _unique_texts(str(candidate.get("imageAsset", "")) for candidate in candidates)
-            "sourceCandidates": candidate_ids,
+            image_facts = [
-            "sampleFingerprints": sample_fingerprints,
+                candidate.get("imageFacts", {})
-            "imageAsset": image_assets[0] if image_assets else "",
+                for candidate in candidates
-            "imageAssets": image_assets,
+                if isinstance(candidate.get("imageFacts", {}), dict) and candidate.get("imageFacts", {})
-            "imageFacts": {
+            ]
-                "samples": len(candidates),
+            fallback_name = next(
-                "queries": queries,
+                (
-                "items": image_facts,
+                    str(candidate.get("title", "")).strip() or str(candidate.get("query", "")).strip()
-            },
+                    for candidate in candidates
-        }
+                    if str(candidate.get("title", "")).strip() or str(candidate.get("query", "")).strip()
-        self._put("knowledge_entries", entry_id, entry)
+                ),
-        for candidate in candidates:
+                "Collected reference",
-            candidate["status"] = "promoted"
+            )
-            candidate["promotedKnowledgeId"] = entry_id
+            name = str(payload.get("name", "")).strip() or fallback_name
-            self._put("collection_candidates", str(candidate["id"]), candidate)
+            memo = str(payload.get("memo", "")).strip() or f"Promoted from collected candidates: {', '.join(queries)}"
-        self.add_audit_event(
+            entry = {
-            "rights.ops",
+                "id": entry_id,
-            "Knowledge entry manually created",
+                "name": name,
-            name,
+                "type": _knowledge_type_value(str(payload.get("type", "rejected_reference"))),
-            f"promoted {len(candidates)} collected candidates",
+                "aliases": _text_list(payload.get("aliases")) or queries,
-        )
+                "keywords": _text_list(payload.get("keywords")) or queries,
+                "memo": memo,
+                "provenance": "manual",
+                "active": True,
+                "entryStatus": "confirmed",
+                "sourceDecision": "",
+                "sourceCandidate": candidate_ids[0] if len(candidate_ids) == 1 else "",
+                "sourceCandidates": candidate_ids,
+                "sampleFingerprints": sample_fingerprints,
+                "imageAsset": image_assets[0] if image_assets else "",
+                "imageAssets": image_assets,
+                "imageFacts": {
+                    "samples": len(candidates),
+                    "queries": queries,
+                    "items": image_facts,
+                },
+            }
+            with self._transaction() as conn:
+                self._put("knowledge_entries", entry_id, entry, conn=conn)
+                for candidate in candidates:
+                    candidate["status"] = "promoted"
+                    candidate["promotedKnowledgeId"] = entry_id
+                    self._put("collection_candidates", str(candidate["id"]), candidate, conn=conn)
+                self.add_audit_event(
+                    "rights.ops",
+                    "Knowledge entry manually created",
+                    name,
+                    f"promoted {len(candidates)} collected candidates",
+                    conn=conn,
+                )
         return self.bootstrap()
 

tests/rights_filter/analysis/test_risk_scoring.py

@@ -229,3 +229,54 @@ def test_score_is_clamped_to_100():
     )
     assert result.score == 100
     assert result.band == "high"
+
+
+def test_two_distinct_web_entities_stack_into_high_risk():
+    # Two different named entities in one image are independent signals and must
+    # not dedup each other away. Regression: both collapsed into one "entity"
+    # bucket, dropping the second +45 and under-scoring a high image to medium.
+    result = RiskScorer().score(
+        [
+            Evidence(
+                source=EvidenceSource.WEB_DETECTION,
+                reason="Web entity matched character A",
+                confidence=0.94,
+                data={"entity": "캐릭터 A", "category": "character"},
+            ),
+            Evidence(
+                source=EvidenceSource.WEB_DETECTION,
+                reason="Web entity matched celebrity B",
+                confidence=0.94,
+                data={"entity": "유명인 B", "category": "celebrity"},
+            ),
+        ]
+    )
+
+    assert result.score >= 70
+    assert result.band == "high"
+    assert "Web entity matched character A" in result.reasons
+    assert "Web entity matched celebrity B" in result.reasons
+
+
+def test_same_web_entity_repeated_does_not_double_count():
+    # The same entity reported twice IS redundant and must still dedup to one
+    # signal — guards the fix from over-counting genuinely identical entities.
+    result = RiskScorer().score(
+        [
+            Evidence(
+                source=EvidenceSource.WEB_DETECTION,
+                reason="Web entity matched character A",
+                confidence=0.94,
+                data={"entity": "캐릭터 A", "category": "character"},
+            ),
+            Evidence(
+                source=EvidenceSource.WEB_DETECTION,
+                reason="Web entity matched character A (again)",
+                confidence=0.94,
+                data={"entity": "캐릭터 A", "category": "character"},
+            ),
+        ]
+    )
+
+    assert result.score == 45
+    assert result.band == "medium"

tests/rights_filter/server/test_sqlite_store.py

@@ -8044,6 +8044,56 @@ def test_record_decision_rolls_back_when_audit_fails(tmp_path: Path, monkeypatch
         store._get("knowledge_entries", store_serialization_module._stable_id("kb-watchlist", "SUB-1"))
 
 
+def test_mark_evidence_status_rolls_back_when_audit_fails(tmp_path: Path, monkeypatch):
+    image_store = _image_store(tmp_path / "submissions")
+    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3")
+    store.initialize()
+    store.seed_from_image_store(image_store)
+    face_evidence = next(item for item in store.review("SUB-DB1")["evidence"] if item["source"] == "face")
+
+    def boom(*args, **kwargs):
+        raise RuntimeError("audit down")
+
+    monkeypatch.setattr(store, "add_audit_event", boom)
+    with pytest.raises(RuntimeError):
+        store.mark_evidence_status("SUB-DB1", face_evidence["id"], "false_positive", "오탐")
+
+    # The evidence status change must roll back atomically with the failed audit.
+    after = next(item for item in store.review("SUB-DB1")["evidence"] if item["id"] == face_evidence["id"])
+    assert after.get("operatorStatus") != "false_positive"
+
+
+def test_batch_promote_collection_candidates_is_idempotent_under_retry(tmp_path: Path):
+    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3")
+    store.initialize()
+    for index in (1, 2):
+        store._put(
+            "collection_candidates",
+            f"CAND-{index}",
+            {
+                "id": f"CAND-{index}",
+                "provider": "naver",
+                "query": f"query-{index}",
+                "title": f"Candidate {index}",
+                "status": "candidate",
+                "sourceUrl": f"https://example.test/{index}",
+                "collectedEpoch": index,
+                "sampleFingerprints": [f"fp-{index}"],
+            },
+        )
+    request = {"candidate_ids": ["CAND-1", "CAND-2"], "name": "Batch entry"}
+
+    store.promote_collection_candidates(request)
+    # A double-click / retry must NOT mint a second knowledge entry.
+    store.promote_collection_candidates(request)
+
+    entries = store.bootstrap()["knowledgeEntries"]
+    assert len(entries) == 1
+    candidate = store._get("collection_candidates", "CAND-1")
+    assert candidate["status"] == "promoted"
+    assert candidate["promotedKnowledgeId"] == entries[0]["id"]
+
+
 @pytest.mark.parametrize(
     "url",
     [

web/operator-gui/app.js

@@ -2381,10 +2381,14 @@ async function setDecision(decision, requiresMemo = false) {
 
 
 
+let isManualSearchActive = false;
+
 async function runManualSearch(event) {
 
   event.preventDefault();
 
+  if (isManualSearchActive) return;
+
   const submission = getSelectedCase();
 
   const provider = document.getElementById("manual-query-provider").value;
@@ -2429,6 +2433,8 @@ async function runManualSearch(event) {
 
 
 
+  isManualSearchActive = true;
+
   try {
 
     await apiJson("/api/search/manual", {
@@ -2449,6 +2455,10 @@ async function runManualSearch(event) {
 
     status.textContent = errorValue.message;
 
+  } finally {
+
+    isManualSearchActive = false;
+
   }
 
 }
@@ -2475,7 +2485,7 @@ function rerunHistoricalQuery(button) {
 
   providerSelect.value = provider;
 
-  runManualSearch(new Event("submit"));
+  void runManualSearch(new Event("submit"));
 
 }
 
@@ -3218,34 +3228,35 @@ function addKnowledgeEntryFromCase() {
 
 
 
-function deactivateCorrectionEntry(correctionId) {
+async function deactivateCorrectionEntry(correctionId) {
-
   const correction = corrections.find((item) => item.id === correctionId);
-
   if (!correction) return;
 
-  correction.derivedEntries.forEach((entry) => {
+  // Deactivate the derived knowledge entries through the real backend endpoint
-
+  // so the change is persisted and audited server-side. The previous version
-    entry.active = false;
+  // only mutated local state and appended a fake audit event, so the action was
-
+  // silently discarded on the next refreshFromApi().
-  });
+  const entryIds = (correction.derivedEntries || [])
-
+    .map((entry) => entry && entry.id)
-  auditEvents.unshift({
+    .filter(Boolean);
-
+  if (!entryIds.length) {
-    timestamp: "방금",
+    showApiError("이 보정 항목에는 서버에 반영할 수 있는 파생 지식 항목 ID가 없습니다.");
-
+    return;
-    actor: "rights.ops",
+  }
-
-    event: "Knowledge entry deactivated",
-
-    object: correction.decisionId,
-
-    change: "all derived entries inactive",
-
-  });
-
-  renderAll();
 
+  try {
+    let payload;
+    for (const entryId of entryIds) {
+      payload = await apiJson(`/api/knowledge/${encodeURIComponent(entryId)}/deactivate`, {
+        method: "POST",
+        body: JSON.stringify({ reason: "보정 파생 항목 일괄 비활성화" }),
+      });
+    }
+    if (payload) applyBootstrap(payload);
+    await refreshFromApi();
+  } catch (errorValue) {
+    showApiError(errorValue.message);
+  }
 }
 
 
@@ -3665,25 +3676,11 @@ function bindEvents() {
     const retryButton = event.target.closest("[data-provider-retry]");
 
     if (retryButton) {
-
+      // No provider-wide retry endpoint exists. Tell the operator honestly
-      auditEvents.unshift({
+      // instead of writing a fake "Provider called" audit event that never
-
+      // re-ran anything (which falsely implied the failures were retried).
-        timestamp: "방금",
+      showApiError("제공자 단위 '실패 재시도'는 아직 서버에 구현되지 않았습니다. 개별 케이스의 '재분석'을 사용하세요.");
-
-        actor: "admin.ops",
-
-        event: "Provider called",
-
-        object: retryButton.dataset.providerRetry,
-
-        change: "retry failed enrichments",
-
-      });
-
-      renderAll();
-
       return;
-
     }
 
 
@@ -3692,7 +3689,7 @@ function bindEvents() {
 
     if (correctionButton) {
 
-      deactivateCorrectionEntry(correctionButton.dataset.deactivateCorrection);
+      void deactivateCorrectionEntry(correctionButton.dataset.deactivateCorrection);
 
       return;