fix: serialize store read-modify-write and harden seed/queue paths

Under ThreadingHTTPServer several store writers ran unlocked read-modify-write
cycles, so concurrent operator actions could silently lose each other's
updates or revert an in-flight change.

Concurrency (_write_lock; external I/O kept OUTSIDE the lock):
- record_decision now takes _write_lock so a decision isn't reverted by a
  concurrent rescore/sweep.
- manual_search runs the network search first, then re-reads the submission
  under the lock before the queryHistory RMW (was clobbering concurrent edits).
- _apply_provider_usage_delta merges only usage/status onto the fresh row
  instead of rewriting a stale pre-search snapshot (kept an in-flight emergency
  disable from being reverted).
- _sync_provider_payloads preserves an operator's disable via an
  operatorDisabled flag (connectivity-only disables still reconnect).
- set_provider_enabled / emergency_disable / _sync_submission_provider_state /
  collect_keyword_candidates now lock (+ single transaction where applicable).

Robustness:
- seed: a zero-byte/corrupt image now yields a 'failed' submission + audit
  instead of an IndexError that aborted the whole import.
- ensure_queue: key the UPDATE/SELECT on the existing row id on a folder_path
  match (was matching zero rows and returning None -> seed TypeError); audit
  queue activation / legacy migration.
- store_enrichment: lastFailure uses the first failed item; auto-search dedup
  reads only the submission's evidence; log a swallowed face-crop load error.
- image_store: documents the pre-existing same-stem id limitation (a prior
  sibling-dependent fix was reverted — it resurrected decided submissions).

Tests: operator-disable survives reseed, usage delta keeps an in-flight
disable, seed survives an unanalyzable image, ensure_queue mismatched-id,
dismiss persistence.

src/rights_filter/server/image_store.py

@@ -78,6 +78,14 @@ class LocalSubmissionImageStore:
         return normalized
 
     def _scan_records(self) -> list[dict[str, Any]]:
+        # NOTE: a scanned file's id is its bare stem. Two same-stem files
+        # (a.jpg + a.png) collide and one silently merges into the other — a
+        # known, pre-existing limitation. A prior fix that appended the extension
+        # on collision was reverted because the id then depended on sibling files,
+        # so adding a same-stem sibling later re-derived ids and resurrected
+        # already-decided submissions as duplicates. A correct fix must preserve
+        # each incumbent id at seed time (where existing ids are known) — not in
+        # this pure scan — and carry a one-time asset-path migration.
         if not self.root.exists():
             return []
         records: list[dict[str, Any]] = []

src/rights_filter/server/sqlite_store.py

@@ -282,7 +282,30 @@ class CopyrighterStore(
             batch.run(submissions)
 
         for record in records:
-            run = repository.analysis_runs_for_submission(record["id"])[-1]
+            runs = repository.analysis_runs_for_submission(record["id"])
+            if not runs:
+                # BatchAnalyzer saves no run when preprocessing fails (an
+                # unreadable / zero-byte / corrupt image). Persist a 'failed'
+                # submission with an audit trail and move on, instead of raising
+                # IndexError here — which previously aborted the whole seed after
+                # a partial commit and re-failed on every subsequent reload.
+                failed_submission = _submission_payload(
+                    record,
+                    0,
+                    "failed",
+                    ["이미지 분석 실패 — 파일이 손상되었거나 비어 있습니다."],
+                    self.provider_runtime.provider_payloads,
+                )
+                failed_submission["queueId"] = queue_id
+                self._put("submissions", failed_submission["id"], failed_submission)
+                self.add_audit_event(
+                    "rights.system",
+                    "Analysis run created",
+                    str(record["id"]),
+                    "이미지 분석 실패 — 손상되었거나 빈 파일",
+                )
+                continue
+            run = runs[-1]
             score = run.score
             submission = _submission_payload(
                 record,
@@ -566,6 +589,10 @@ class CopyrighterStore(
         if decision == "rejected" and not memo.strip():
             raise ValueError("reject decision requires memo")
 
+        # Serialize with the other submission writers (rescore, provider sweeps)
+        # so an operator decision is not silently reverted by a concurrent
+        # lock-holding re-put of a stale submission payload.
+        with self._write_lock:
             with self._transaction() as conn:
                 submission = self._get("submissions", submission_id, conn=conn)
                 submission["decisionStatus"] = decision
@@ -627,6 +654,7 @@ class CopyrighterStore(
         return self._all("providers")
 
     def set_provider_enabled(self, provider_id: str, enabled: bool) -> list[dict[str, Any]]:
+        with self._write_lock:
             provider = self._get("providers", provider_id)
             if enabled and provider_id != "internal":
                 runtime_provider = self.provider_runtime.provider_payloads.get(provider_id, {})
@@ -634,28 +662,47 @@ class CopyrighterStore(
                     reason = str(runtime_provider.get("lastFailure", "")) or f"{provider_id} provider not connected"
                     raise ValueError(reason)
             provider["enabled"] = bool(enabled)
-        self._put("providers", provider_id, provider)
+            # Record that this on/off was an explicit operator action so a later
-        self._sync_submission_provider_state()
+            # reload does not silently flip it back to the connectivity default.
+            provider["operatorDisabled"] = not bool(enabled)
+            with self._transaction() as conn:
+                self._put("providers", provider_id, provider, conn=conn)
                 self.add_audit_event(
                     actor="admin.ops",
                     event="Provider setting changed",
                     object_id=provider["name"],
                     change="enabled" if enabled else "disabled",
+                    conn=conn,
                 )
+            self._sync_submission_provider_state()
         return self.providers()
 
     def emergency_disable_external_providers(self) -> list[dict[str, Any]]:
-        for provider in self.providers():
+        with self._write_lock:
+            providers = self.providers()
+            external_provider_ids = [
+                provider["id"] for provider in providers if provider["id"] != "internal"
+            ]
+            submissions = self._all("submissions")
+            with self._transaction() as conn:
+                for provider in providers:
                     if provider["id"] != "internal":
                         provider["enabled"] = False
-                self._put("providers", provider["id"], provider)
+                        # Deliberate admin stop — must persist across reloads.
-        external_provider_ids = [provider["id"] for provider in self.providers() if provider["id"] != "internal"]
+                        provider["operatorDisabled"] = True
-        for submission in self._all("submissions"):
+                        self._put("providers", provider["id"], provider, conn=conn)
+                for submission in submissions:
                     provider_state = submission.setdefault("providerState", {})
                     for provider_id in external_provider_ids:
                         provider_state[provider_id] = "disabled"
-            self._put("submissions", submission["id"], submission)
+                    self._put("submissions", submission["id"], submission, conn=conn)
-        self.add_audit_event("admin.ops", "Provider setting changed", "external providers", "emergency disabled")
+                self.add_audit_event(
+                    "admin.ops",
+                    "Provider setting changed",
+                    "external providers",
+                    "emergency disabled",
+                    conn=conn,
+                )
         return self.providers()
 
     def audit_events(self, limit: int = 500) -> list[dict[str, Any]]:

src/rights_filter/server/store_enrichment.py

@@ -10,10 +10,13 @@ rescoring provided by the host class. Behavior unchanged.
 from __future__ import annotations
 
 import json
+import logging
 from dataclasses import replace
 from datetime import datetime
 from typing import Any
 
+logger = logging.getLogger(__name__)
+
 from rights_filter.analysis.face_person_detection import HeuristicFacePersonDetector
 from rights_filter.analysis.fingerprints import FingerprintService
 from rights_filter.analysis.internal_analyzer import InternalAnalyzer
@@ -44,16 +47,39 @@ from rights_filter.server.store_serialization import (
 
 class StoreEnrichmentMixin:
     def _sync_provider_payloads(self) -> None:
+        with self._write_lock:
             existing = {provider["id"]: provider for provider in self._all("providers")}
             for provider_id, payload in self.provider_runtime.provider_payloads.items():
+                prior = existing.get(provider_id)
+                runtime_enabled = bool(payload.get("enabled", False))
+                # `operatorDisabled` distinguishes a deliberate operator/admin
+                # disable from a connectivity disable. Only the former survives a
+                # reconnect; a provider that was merely missing credentials is
+                # re-enabled once the adapter connects (e.g. env change adds keys).
+                operator_disabled = bool(prior.get("operatorDisabled", False)) if prior else False
+                if not runtime_enabled:
+                    # Adapter not connected — cannot be enabled regardless.
+                    enabled = False
+                elif operator_disabled:
+                    # Operator deliberately disabled a connected provider; do not
+                    # let reload/collection silently re-enable it (no audit trail).
+                    enabled = False
+                else:
+                    enabled = runtime_enabled
                 merged = {
                     **payload,
                     "usage": existing.get(provider_id, {}).get("usage", payload["usage"]),
+                    "enabled": enabled,
+                    "operatorDisabled": operator_disabled,
                 }
                 self._put("providers", provider_id, merged)
             self._sync_submission_provider_state()
 
     def _sync_submission_provider_state(self, queue_id: str | None = None) -> None:
+        # Full-table read-modify-write sweep; serialize it so a concurrent
+        # operator write to an unrelated submission is not reverted by a stale
+        # copy read at the top of this loop.
+        with self._write_lock:
             provider_payloads = {provider["id"]: provider for provider in self._all("providers")}
             evidence_by_submission = self._evidence_by_submission(queue_id=queue_id)
             for submission in self._all("submissions", queue_id=queue_id):
@@ -150,6 +176,9 @@ class StoreEnrichmentMixin:
         try:
             original = image_store.image_payload(submission_id)
         except Exception:
+            # 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)
             return
 
         signal = HeuristicFacePersonDetector().detect(original)
@@ -308,7 +337,14 @@ class StoreEnrichmentMixin:
             item.source in {EvidenceSource.FAILURE, EvidenceSource.EXTERNAL_SKIPPED}
             for item in domain_evidence
         ):
-            google_provider["lastFailure"] = domain_evidence[0].reason if domain_evidence else "Google image search failed"
+            google_provider["lastFailure"] = next(
+                (
+                    item.reason
+                    for item in domain_evidence
+                    if item.source in {EvidenceSource.FAILURE, EvidenceSource.EXTERNAL_SKIPPED}
+                ),
+                "Google image search failed",
+            )
         else:
             google_provider["lastSuccess"] = _now_label()
             google_provider["lastFailure"] = "없음"
@@ -408,7 +444,7 @@ class StoreEnrichmentMixin:
             return
 
         existing_signatures = _existing_google_custom_query_signatures(
-            self._evidence_by_submission().get(submission_id, [])
+            self._evidence_for_submission(submission_id)
         )
         can_compare_search_images = self._can_compare_search_result_images(
             submission_id,
@@ -542,7 +578,14 @@ class StoreEnrichmentMixin:
             item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
             for item in all_domain_evidence
         ):
-            google_search_provider["lastFailure"] = all_domain_evidence[0].reason if all_domain_evidence else "Google custom search failed"
+            google_search_provider["lastFailure"] = next(
+                (
+                    item.reason
+                    for item in all_domain_evidence
+                    if item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
+                ),
+                "Google custom search failed",
+            )
         else:
             google_search_provider["lastSuccess"] = _now_label()
             google_search_provider["lastFailure"] = "없음"
@@ -574,7 +617,7 @@ class StoreEnrichmentMixin:
             return
 
         existing_signatures = _existing_naver_query_signatures(
-            self._evidence_by_submission().get(submission_id, [])
+            self._evidence_for_submission(submission_id)
         )
         can_compare_search_images = self._can_compare_search_result_images(
             submission_id,
@@ -777,7 +820,14 @@ class StoreEnrichmentMixin:
             item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
             for item in all_domain_evidence
         ):
-            naver_provider["lastFailure"] = all_domain_evidence[0].reason if all_domain_evidence else "Naver search failed"
+            naver_provider["lastFailure"] = next(
+                (
+                    item.reason
+                    for item in all_domain_evidence
+                    if item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE}
+                ),
+                "Naver search failed",
+            )
         else:
             naver_provider["lastSuccess"] = _now_label()
             naver_provider["lastFailure"] = "없음"

src/rights_filter/server/store_operations.py

@@ -318,22 +318,29 @@ class StoreOperationsMixin:
         query: str,
         image_store: LocalSubmissionImageStore | None = None,
     ) -> dict[str, Any]:
-        submission = self._get("submissions", submission_id)
         provider_payload = self._get("providers", provider)
         if not provider_payload["enabled"]:
             raise ValueError(f"{provider} provider disabled")
 
+        # External search runs OUTSIDE the write lock (it can take seconds); only
+        # the read-modify-write of the submission is serialized below.
         domain_evidence, provider_call_count = self._manual_search_evidence(
             submission_id,
             provider,
             query,
             image_store,
         )
+
+        with self._write_lock:
             for item in domain_evidence:
                 evidence = _evidence_payload(submission_id, item)
                 evidence["status"] = "manual"
                 self._put("evidence", evidence["id"], evidence)
 
+            # Re-read the submission AFTER the search so an operator decision or
+            # other update made while the search was in flight is not clobbered
+            # by a stale snapshot.
+            submission = self._get("submissions", submission_id)
             submission.setdefault("queryHistory", []).insert(
                 0,
                 {
@@ -354,7 +361,11 @@ class StoreOperationsMixin:
                 provider_payload["lastSuccess"] = _now_label()
                 provider_payload["lastFailure"] = "없음"
             self._apply_provider_usage_delta(provider, provider_call_count, provider_payload)
+
+        # LLM narration may call out to Ollama — keep it outside the lock.
         self._ensure_llm_summary(submission_id)
+
+        with self._write_lock:
             if image_store is not None:
                 self._rescore_submission(submission_id)
             self._sync_submission_provider_state()
@@ -581,7 +592,11 @@ class StoreOperationsMixin:
         if provider == "google_search" and self.provider_runtime.google_custom_search_adapter is None:
             raise ValueError("google_search provider not connected")
 
-        self._clear_collection_candidates()
+        # Run the search cascade OUTSIDE the write lock and accumulate the
+        # candidates in memory. The clear + repopulate is then committed under the
+        # lock in one transaction, so it can't interleave with a concurrent
+        # promote (which would 404 on cleared rows) or a second collect (which
+        # would resurrect the previous query's rows).
         if provider == "google_search":
             domain_evidence = self.provider_runtime.google_custom_search_adapter.search_images(
                 "candidate-collection",
@@ -595,9 +610,10 @@ class StoreOperationsMixin:
                 self.provider_runtime.search_policy,
             )
         collected = 0
+        pending: list[dict[str, Any]] = []
         provider_call_count = 1 if any(item.source != EvidenceSource.SEARCH_SKIPPED for item in domain_evidence) else 0
         for candidate in self._collection_candidates_from_evidence(query, domain_evidence, provider):
-            self._put("collection_candidates", candidate["id"], candidate)
+            pending.append(candidate)
             collected += 1
 
         if provider == "google_search" and collected == 0:
@@ -610,7 +626,7 @@ class StoreOperationsMixin:
                 provider_call_count += 1
             domain_evidence.extend(web_evidence)
             for candidate in self._collection_candidates_from_evidence(query, web_evidence, provider):
-                self._put("collection_candidates", candidate["id"], candidate)
+                pending.append(candidate)
                 collected += 1
         if provider == "naver" and collected == 0:
             page_evidence = self.provider_runtime.naver_adapter.search_pages(
@@ -622,7 +638,7 @@ class StoreOperationsMixin:
                 provider_call_count += 1
             domain_evidence.extend(page_evidence)
             for candidate in self._collection_candidates_from_evidence(query, page_evidence, provider):
-                self._put("collection_candidates", candidate["id"], candidate)
+                pending.append(candidate)
                 collected += 1
         if provider == "naver" and collected == 0:
             web_evidence = self.provider_runtime.naver_adapter.search_web_pages(
@@ -634,7 +650,7 @@ class StoreOperationsMixin:
                 provider_call_count += 1
             domain_evidence.extend(web_evidence)
             for candidate in self._collection_candidates_from_evidence(query, web_evidence, provider):
-                self._put("collection_candidates", candidate["id"], candidate)
+                pending.append(candidate)
                 collected += 1
 
         if any(
@@ -645,8 +661,20 @@ class StoreOperationsMixin:
         else:
             provider_payload["lastSuccess"] = _now_label()
             provider_payload["lastFailure"] = "없음"
+
+        with self._write_lock:
+            with self._transaction() as conn:
+                conn.execute("delete from collection_candidates")
+                for candidate in pending:
+                    self._put("collection_candidates", candidate["id"], candidate, conn=conn)
+                self.add_audit_event(
+                    "rights.ops",
+                    "Keyword candidates collected",
+                    provider,
+                    f"{query} · {collected} candidates",
+                    conn=conn,
+                )
             self._apply_provider_usage_delta(provider, provider_call_count, provider_payload)
-        self.add_audit_event("rights.ops", "Keyword candidates collected", provider, f"{query} · {collected} candidates")
         payload = self.bootstrap()
         payload["collected"] = collected
         return payload
@@ -691,6 +719,31 @@ class StoreOperationsMixin:
                 )
         return self.bootstrap()
 
+    def dismiss_collection_candidate(self, candidate_id: str) -> dict[str, Any]:
+        # Persist the dismissal so the candidate stays gone across refreshes
+        # instead of resurrecting on the next bootstrap. Idempotent: a repeat /
+        # double-click on an already-removed candidate is a no-op.
+        with self._write_lock:
+            try:
+                candidate = self._get("collection_candidates", candidate_id)
+            except KeyError:
+                return self.bootstrap()
+            if candidate.get("status") == "promoted":
+                raise ValueError("promoted candidate cannot be dismissed")
+            query = str(candidate.get("query", ""))
+            with self._transaction() as conn:
+                conn.execute(
+                    "delete from collection_candidates where id = ?", (candidate_id,)
+                )
+                self.add_audit_event(
+                    "rights.ops",
+                    "Keyword candidate dismissed",
+                    candidate_id,
+                    f"dismissed candidate for query: {query}",
+                    conn=conn,
+                )
+        return self.bootstrap()
+
     def promote_collection_candidates(self, payload: dict[str, Any]) -> dict[str, Any]:
         candidate_ids = _unique_texts(_text_list(payload.get("candidate_ids", payload.get("candidateIds"))))
         if not candidate_ids:

src/rights_filter/server/store_persistence.py

@@ -231,18 +231,19 @@ class StorePersistenceMixin:
     def _apply_provider_usage_delta(
         self, provider_id: str, count: int, payload: dict[str, Any]
     ) -> None:
-        # Re-read current usage under the write lock and apply the delta so
+        # Start from the freshly-read row and merge only the fields this call
-        # concurrent provider calls don't lose increments. The rest of `payload`
+        # actually produced (usage delta + call status). The caller's `payload`
-        # (lastSuccess / lastFailure) is last-writer-wins, which is acceptable for
+        # was snapshotted before a multi-second external search, so writing it
-        # status fields.
+        # wholesale would revert any operator change made meanwhile — in
+        # particular an emergency disable (`enabled`). Merging keeps that change.
         with self._write_lock:
             current = self._get("providers", provider_id)
-            payload["usage"] = int(current.get("usage", 0) or 0) + int(count)
+            merged = dict(current)
-            self._put("providers", provider_id, payload)
+            for key in ("lastSuccess", "lastFailure", "lastRun"):
-
+                if key in payload:
-    def _clear_collection_candidates(self) -> None:
+                    merged[key] = payload[key]
-        with self._connect() as conn:
+            merged["usage"] = int(current.get("usage", 0) or 0) + int(count)
-            conn.execute("delete from collection_candidates")
+            self._put("providers", provider_id, merged)
 
     def _all(self, table: str, queue_id: str | None = None) -> list[dict[str, Any]]:
         _validate_table(table)

src/rights_filter/server/store_queue.py

@@ -109,14 +109,31 @@ class StoreQueueMixin:
                     """,
                     (queue_id, str(folder), queue_label, now.isoformat(" ", "seconds"), int(now.timestamp())),
                 )
+                effective_id = queue_id
             else:
+                # A row matched by folder_path may carry a different (older) id
+                # than the freshly-computed queue_id — keep the existing row's id
+                # as the effective key so the UPDATE, the deactivate-others sweep,
+                # and the final SELECT all target the real row instead of silently
+                # matching zero rows and returning None.
+                effective_id = existing["id"]
                 conn.execute(
                     "update submission_queues set is_active = 1, folder_path = ?, label = ? where id = ?",
-                    (str(folder), queue_label, queue_id),
+                    (str(folder), queue_label, effective_id),
+                )
+            conn.execute("update submission_queues set is_active = 0 where id != ?", (effective_id,))
+            migrated_legacy = self._has_queueless_submissions(conn=conn)
+            if migrated_legacy:
+                self._migrate_legacy_submissions_to_queue(effective_id, conn=conn)
+            # Audit only meaningful transitions (a brand-new queue or a legacy
+            # migration), not every reload, so the audit trail isn't spammed.
+            if existing is None or migrated_legacy:
+                change = f"folder {folder}" + (
+                    "; legacy queueless submissions migrated" if migrated_legacy else "; new queue"
+                )
+                self.add_audit_event(
+                    "rights.system", "Submission queue activated", effective_id, change, conn=conn
                 )
-            conn.execute("update submission_queues set is_active = 0 where id != ?", (queue_id,))
-            if self._has_queueless_submissions(conn=conn):
-                self._migrate_legacy_submissions_to_queue(queue_id, conn=conn)
             queue_row = conn.execute(
                 """
                 select id, folder_path, label, is_active, created_at, created_epoch,
@@ -124,10 +141,10 @@ class StoreQueueMixin:
                 from submission_queues
                 where id = ?
                 """,
-                (queue_id,),
+                (effective_id,),
             ).fetchone()
             if queue_row is None:
-                return self._queue_row_by_id(queue_id)
+                return self._queue_row_by_id(effective_id)
             return {
                 "id": queue_row["id"],
                 "folderPath": queue_row["folder_path"],

tests/rights_filter/server/test_sqlite_store.py

@@ -1228,6 +1228,150 @@ def test_sqlite_store_emergency_disable_marks_google_custom_search_state_disable
     assert submission["providerState"]["llm"] == "disabled"
 
 
+_CONNECTED_PROVIDER_ENV = {
+    "NAVER_CLIENT_ID": "naver-id",
+    "NAVER_CLIENT_SECRET": "naver-secret",
+    "GOOGLE_CLOUD_VISION_API_KEY": "google-key",
+    "GOOGLE_CUSTOM_SEARCH_API_KEY": "google-search-key",
+    "GOOGLE_CUSTOM_SEARCH_CX": "search-engine-id",
+    "OLLAMA_MODEL": "qwen2.5:0.5b-instruct",
+}
+
+
+def test_ensure_queue_handles_folder_match_with_mismatched_id(tmp_path: Path):
+    # 2026-07-11 #13: a stored queue row whose id differs from the freshly
+    # computed queue_id (e.g. after a path-normalization change) must still be
+    # found by folder_path and updated in place — the old code keyed the UPDATE
+    # and final SELECT on the computed id, matched zero rows, and returned None
+    # (which then crashed seed_from_image_store on str(queue["id"])).
+    store = CopyrighterStore(
+        tmp_path / "copyrighter.sqlite3",
+        provider_runtime=build_provider_runtime(env={}, transport=FakeJsonTransport({})),
+    )
+    store.initialize()
+    folder = (tmp_path / "queue-folder").resolve()
+    folder.mkdir()
+    with store._connect() as conn:
+        conn.execute(
+            """
+            insert into submission_queues
+            (id, folder_path, label, is_active, created_at, created_epoch, last_imported_epoch, last_imported_at)
+            values (?, ?, ?, 1, '', 0, 0, '')
+            """,
+            ("queue-LEGACYID", str(folder), "legacy"),
+        )
+        conn.commit()
+
+    result = store.ensure_queue(folder)
+    assert result is not None
+    assert result["id"] == "queue-LEGACYID"
+    assert result["isActive"] is True
+
+
+def test_dismiss_collection_candidate_persists_and_is_idempotent(tmp_path: Path):
+    # 2026-07-11 H9: dismissing a candidate must persist (delete the row) so it
+    # does not resurrect on the next bootstrap; a repeat dismiss is a no-op.
+    store = CopyrighterStore(
+        tmp_path / "copyrighter.sqlite3",
+        provider_runtime=build_provider_runtime(env={}, transport=FakeJsonTransport({})),
+    )
+    store.initialize()
+    candidate = {
+        "id": "cand-1",
+        "provider": "naver",
+        "query": "IU",
+        "title": "IU",
+        "status": "candidate",
+        "sourceUrl": "https://example.test/iu",
+        "collectedEpoch": 0,
+        "sampleFingerprints": [],
+    }
+    store._put("collection_candidates", "cand-1", candidate)
+    assert any(c["id"] == "cand-1" for c in store.bootstrap()["collectionCandidates"])
+
+    store.dismiss_collection_candidate("cand-1")
+    assert all(c["id"] != "cand-1" for c in store.bootstrap()["collectionCandidates"])
+
+    store.dismiss_collection_candidate("cand-1")  # idempotent no-op
+
+
+def test_seed_survives_unanalyzable_image(tmp_path: Path):
+    # 2026-07-11 H5: a single zero-byte/corrupt image must not abort the whole
+    # seed with IndexError (which left a partial commit that re-failed on every
+    # reload). It becomes a 'failed' submission; the good ones still seed.
+    root = tmp_path / "store"
+    image_dir = root / "images"
+    image_dir.mkdir(parents=True)
+    (image_dir / "good.svg").write_text("<svg><!-- ok --></svg>", encoding="utf-8")
+    (image_dir / "bad.png").write_bytes(b"")
+    (root / "submissions.json").write_text(
+        json.dumps(
+            [
+                {"id": "SUB-GOOD", "title": "good", "file": "images/good.svg", "width": 800, "height": 600, "submitted_at": "2026-07-11 10:00"},
+                {"id": "SUB-BAD", "title": "bad", "file": "images/bad.png", "width": 800, "height": 600, "submitted_at": "2026-07-11 10:00"},
+            ]
+        ),
+        encoding="utf-8",
+    )
+    image_store = LocalSubmissionImageStore(root)
+    store = CopyrighterStore(
+        tmp_path / "copyrighter.sqlite3",
+        provider_runtime=build_provider_runtime(env={}, transport=FakeJsonTransport({})),
+    )
+    store.initialize()
+    store.seed_from_image_store(image_store)  # must not raise
+
+    subs = {s["id"]: s for s in store.bootstrap()["submissions"]}
+    assert "SUB-GOOD" in subs
+    assert subs["SUB-BAD"]["riskBand"] == "failed"
+
+
+def test_operator_disabled_provider_survives_reseed(tmp_path: Path):
+    # 2026-07-11 H3: a provider an operator deliberately disabled must not be
+    # silently re-enabled by the next reload/seed just because the adapter is
+    # connected. A provider disabled only for connectivity still reconnects.
+    runtime = build_provider_runtime(
+        env=_CONNECTED_PROVIDER_ENV, transport=FakeJsonTransport({"items": []})
+    )
+    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3", provider_runtime=runtime)
+    store.initialize()
+    image_store = _image_store(tmp_path / "submissions")
+    store.seed_from_image_store(image_store)
+
+    store.set_provider_enabled("naver", False)
+    store.seed_from_image_store(image_store)  # reload
+
+    naver = next(p for p in store.providers() if p["id"] == "naver")
+    assert naver["enabled"] is False
+    google = next(p for p in store.providers() if p["id"] == "google")
+    assert google["enabled"] is True
+
+
+def test_provider_usage_delta_does_not_revert_inflight_disable(tmp_path: Path):
+    # 2026-07-11 H2: _apply_provider_usage_delta must merge only usage/status onto
+    # the fresh row. If it rewrote a stale pre-search snapshot, an emergency
+    # disable that landed while the search was in flight would be reverted.
+    runtime = build_provider_runtime(
+        env=_CONNECTED_PROVIDER_ENV, transport=FakeJsonTransport({"items": []})
+    )
+    store = CopyrighterStore(tmp_path / "copyrighter.sqlite3", provider_runtime=runtime)
+    store.initialize()
+    store.seed_from_image_store(_image_store(tmp_path / "submissions"))
+
+    stale = next(p for p in store.providers() if p["id"] == "naver")
+    assert stale["enabled"] is True
+
+    store.set_provider_enabled("naver", False)  # operator disables mid-search
+
+    stale["lastSuccess"] = "2026-07-11 00:00:00"
+    store._apply_provider_usage_delta("naver", 1, stale)  # search finishes
+
+    naver = next(p for p in store.providers() if p["id"] == "naver")
+    assert naver["enabled"] is False
+    assert naver["usage"] >= 1
+    assert naver["lastSuccess"] == "2026-07-11 00:00:00"
+
+
 def test_sqlite_store_manual_naver_search_uses_real_env_client(tmp_path: Path):
     transport = FakeJsonTransport(
         {