feat: operator throughput features (keyboard review, batch, dupes, export, backup)

- Keyboard review mode: J/K to move, A/H to decide and jump to the next
  unreviewed case, R to focus the memo, N for next unreviewed, ? for help, with
  a live position/backlog indicator; disabled while typing.
- Batch decisions: POST /api/submissions/decision-batch applies one decision to
  many selected cases through per-case transactions/audit/watchlist, reports
  per-case success and failure, caps at 200, requires a memo for reject and
  confirms when high-risk cases are included. Lifts the v1 "no bulk
  approve/reject" spec restriction under those guardrails (spec revised, static
  layout test updated to assert the guardrails instead of the absence).
- Decision-memo presets: GET/POST /api/decision-memo-presets (upsert by label),
  one-click chips in the decision pane, "save current memo" action.
- Submission duplicate / prior-case index: seed backfills fingerprints onto the
  submission record (idempotent), exact + pHash>=0.97 clustering, queue badge
  and group select, prior-decision chips and a conflicting-decision flag.
- Decision export: GET /api/decisions/export with separate internal and
  applicant field whitelists (applicant view never carries score/evidence),
  CSV with a BOM.
- Online backup/restore: POST /api/admin/backup snapshots the WAL database via
  the sqlite backup API plus the media dirs into a timestamped zip with
  retention rotation; `--restore` is an offline CLI that keeps the pre-restore
  database. GET /api/admin/backups lists snapshots.

Also folds in the co-located HTTP hardening (non-object/non-UTF8 JSON body ->
400, non-ASCII Authorization/cookie compared as bytes -> 401, media OSError ->
404) and the line-buffered startup logging that share these files.

Verified against a live server (auth gate, seeding, duplicate index, batch
decision guardrails, presets, export, backup + restore round-trip). pytest
tests/rights_filter tests/operator_gui: 457 passed.

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

src/rights_filter/server/__main__.py

@@ -2,12 +2,14 @@ from __future__ import annotations
 
 import argparse
 import os
+import sys
 from pathlib import Path
 
 from rights_filter.integrations.env_clients import build_provider_runtime
 from rights_filter.server.env_file import load_env_file
 from rights_filter.server.http_app import build_server
 from rights_filter.server.image_store import LocalSubmissionImageStore
+from rights_filter.server.store_backup import restore_backup
 from rights_filter.server.sqlite_store import CopyrighterStore
 
 
@@ -24,6 +26,16 @@ def _provider_readiness(runtime) -> str:
 
 
 def main() -> None:
+    # When the server is started as a service with stdout redirected to a file,
+    # Python switches to block buffering and the startup diagnostics below —
+    # including the "token not set" security warning — sit in the buffer
+    # indefinitely (the process never exits). Line-buffer so they reach the log
+    # immediately.
+    try:
+        sys.stdout.reconfigure(line_buffering=True)
+    except (AttributeError, OSError):
+        pass
+
     parser = argparse.ArgumentParser(description="Run the local Copyrighter API server.")
     parser.add_argument("--host", default="127.0.0.1")
     parser.add_argument("--port", type=int, default=9500)
@@ -31,8 +43,21 @@ def main() -> None:
     parser.add_argument("--images", default="data/submissions")
     parser.add_argument("--static", default="web/operator-gui")
     parser.add_argument("--env", default=".env")
+    parser.add_argument(
+        "--restore",
+        metavar="SNAPSHOT_ZIP",
+        help="Restore a backup snapshot into --db and exit (server must be stopped).",
+    )
     args = parser.parse_args()
 
+    if args.restore:
+        # Restore is offline on purpose: swapping the database under a running
+        # server would leave open connections on a file that no longer exists.
+        result = restore_backup(args.restore, Path(args.db))
+        print(f"Restored database: {result['database']}")
+        print(f"Restored media files: {result['mediaFiles']}")
+        return
+
     env_path = Path(args.env)
     load_env_file(env_path, os.environ)
     # Log the resolved env path (and whether it was found) so an operator running

src/rights_filter/server/http_app.py

@@ -11,7 +11,7 @@ from http import HTTPStatus
 from http.cookies import SimpleCookie
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
-from urllib.parse import unquote, urlparse
+from urllib.parse import parse_qs, unquote, urlparse
 
 from rights_filter.server.image_store import LocalSubmissionImageStore, SUPPORTED_IMAGE_SUFFIXES
 from rights_filter.server.sqlite_store import CopyrighterStore
@@ -114,6 +114,19 @@ def build_server(
                     self._json(store.providers())
                 elif path == "/api/audit-events":
                     self._json(store.audit_events())
+                elif path == "/api/decision-memo-presets":
+                    self._json(store.decision_memo_presets())
+                elif path == "/api/decisions/export":
+                    query = parse_qs(urlparse(self.path).query)
+                    self._json(
+                        store.export_decisions(
+                            audience=(query.get("audience", ["internal"])[0]),
+                            status=(query.get("status", [""])[0]),
+                            since=(query.get("since", [""])[0]),
+                        )
+                    )
+                elif path == "/api/admin/backups":
+                    self._json(store.backups())
                 elif path.startswith("/media/"):
                     self._file(active_store().media_path(unquote(path.removeprefix("/media/"))), untrusted=True)
                 elif path.startswith("/knowledge-media/"):
@@ -229,6 +242,29 @@ def build_server(
                 elif path.startswith("/api/knowledge/") and path.endswith("/reactivate"):
                     entry_id = unquote(path.split("/")[3])
                     self._json(store.reactivate_knowledge_entry(entry_id, str(body.get("reason", ""))))
+                elif path == "/api/submissions/decision-batch":
+                    raw_ids = body.get("submission_ids", [])
+                    if not isinstance(raw_ids, list):
+                        raise ValueError("submission_ids must be a list")
+                    self._json(
+                        store.record_decision_batch(
+                            [str(value) for value in raw_ids],
+                            str(body.get("decision", "")),
+                            str(body.get("memo", "")),
+                            active_store(),
+                        )
+                    )
+                elif path == "/api/decision-memo-presets":
+                    self._json(
+                        store.save_decision_memo_preset(
+                            str(body.get("label", "")), str(body.get("body", ""))
+                        )
+                    )
+                elif path.startswith("/api/decision-memo-presets/") and path.endswith("/delete"):
+                    preset_id = unquote(path.split("/")[3])
+                    self._json(store.delete_decision_memo_preset(preset_id))
+                elif path == "/api/admin/backup":
+                    self._json(store.create_backup())
                 elif path == "/api/providers/emergency-disable":
                     self._json(store.emergency_disable_external_providers())
                 else:
@@ -288,7 +324,13 @@ def build_server(
             if not any(path.startswith(prefix) for prefix in _PROTECTED_PREFIXES):
                 return True
             header = self.headers.get("Authorization", "")
-            if hmac.compare_digest(header, f"Bearer {auth_token}"):
+            # Compare as bytes: hmac.compare_digest raises TypeError on str inputs
+            # containing non-ASCII, and that exception escapes before any response
+            # is written — an unauthenticated remote client could hang the request
+            # by sending a single non-ASCII byte in the header.
+            if hmac.compare_digest(
+                header.encode("utf-8", "replace"), f"Bearer {auth_token}".encode("utf-8")
+            ):
                 # First header-authenticated request mints an HttpOnly session
                 # cookie so the browser can load <img>/media (which cannot send an
                 # Authorization header) without ever putting the token in a URL.
@@ -299,7 +341,10 @@ def build_server(
                 jar = SimpleCookie()
                 jar.load(cookie_header)
                 morsel = jar.get(_SESSION_COOKIE)
-                if morsel is not None and hmac.compare_digest(morsel.value, _session_tag(auth_token)):
+                if morsel is not None and hmac.compare_digest(
+                    morsel.value.encode("utf-8", "replace"),
+                    _session_tag(auth_token).encode("utf-8"),
+                ):
                     # Cookie-authenticated state changes are CSRF-forgeable, so
                     # require a same-origin Origin/Referer for them.
                     if self.command in _STATE_CHANGING_METHODS and not self._origin_ok():
@@ -338,9 +383,18 @@ def build_server(
             if not length:
                 return {}
             try:
-                return json.loads(self.rfile.read(length).decode("utf-8"))
+                parsed = json.loads(self.rfile.read(length).decode("utf-8"))
             except json.JSONDecodeError as exc:
                 raise ValueError("invalid JSON body") from exc
+            except UnicodeDecodeError as exc:
+                raise ValueError("body must be UTF-8") from exc
+            # Every handler treats the body as a mapping (`payload.get(...)`). A
+            # valid-JSON list/string/null would sail past the decode and blow up
+            # with AttributeError inside the handler, which escapes the do_POST
+            # except-chain and closes the connection with no response at all.
+            if not isinstance(parsed, dict):
+                raise ValueError("JSON object body required")
+            return parsed
 
         def _maybe_set_session_cookie(self) -> None:
             if getattr(self, "_issue_session_cookie", False):
@@ -374,7 +428,14 @@ def build_server(
             if not path.exists() or path.is_dir():
                 self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)
                 return
-            data = path.read_bytes()
+            # exists() -> read_bytes() is a TOCTOU window, and the file can also be
+            # locked (Windows), deleted mid-request, or unreadable. Answer with a
+            # real status instead of letting OSError close the connection silently.
+            try:
+                data = path.read_bytes()
+            except OSError:
+                self._json({"error": "not found"}, HTTPStatus.NOT_FOUND)
+                return
             content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
             if content_type.startswith("text/") or content_type in {"application/javascript", "application/json"}:
                 content_type = f"{content_type}; charset=utf-8"

src/rights_filter/server/sqlite_store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 import shutil
 import sqlite3
@@ -38,10 +39,13 @@ from rights_filter.server.store_remote_fetch import (
     _fetch_stylesheet_url_bytes,
     _fetch_url_bytes,
 )
+from rights_filter.server.store_backup import StoreBackupMixin
+from rights_filter.server.store_duplicates import StoreDuplicatesMixin
 from rights_filter.server.store_enrichment import StoreEnrichmentMixin
 from rights_filter.server.store_operations import StoreOperationsMixin
 from rights_filter.server.store_persistence import StorePersistenceMixin
 from rights_filter.server.store_queue import StoreQueueMixin
+from rights_filter.server.store_workflow import StoreWorkflowMixin
 from rights_filter.server.store_search_candidates import StoreSearchCandidatesMixin
 from rights_filter.server.store_schema import (
     _ensure_constrained_schema,
@@ -61,6 +65,13 @@ from rights_filter.server.store_serialization import (
     _submission_search_hint_evidence,
 )
 
+logger = logging.getLogger(__name__)
+
+# Audit event name for a cross-queue submission id collision (same file stem in
+# two imported folders). Kept as a constant so the "log it once" lookup and the
+# write cannot drift apart.
+_QUEUE_ID_CONFLICT_EVENT = "Submission id conflict"
+
 
 class CopyrighterStore(
     StorePersistenceMixin,
@@ -68,6 +79,9 @@ class CopyrighterStore(
     StoreSearchCandidatesMixin,
     StoreEnrichmentMixin,
     StoreOperationsMixin,
+    StoreDuplicatesMixin,
+    StoreWorkflowMixin,
+    StoreBackupMixin,
 ):
     def __init__(
         self,
@@ -207,6 +221,10 @@ class CopyrighterStore(
                     object_ref text not null default '',
                     payload text not null check (json_valid(payload))
                 );
+                create table if not exists decision_memo_presets (
+                    id text primary key,
+                    payload text not null check (json_valid(payload))
+                );
                 """
             )
             _ensure_typed_columns(conn)
@@ -226,6 +244,48 @@ class CopyrighterStore(
                 """
             )
 
+    def _submission_queue_owners(self, *, exclude_queue_id: str) -> dict[str, str]:
+        """Map submission id -> owning queue id, for queues other than the given one."""
+        with self._connect() as conn:
+            rows = conn.execute(
+                """
+                select id, queue_id from submissions
+                where queue_id is not null and queue_id != '' and queue_id != ?
+                """,
+                (exclude_queue_id,),
+            ).fetchall()
+        return {str(row["id"]): str(row["queue_id"]) for row in rows}
+
+    def _note_queue_id_conflict(
+        self, submission_id: str, queue_id: str, owner_queue_id: str
+    ) -> None:
+        logger.warning(
+            "Submission id %r in queue %s is already owned by queue %s — skipping "
+            "import to avoid overwriting the existing decision.",
+            submission_id,
+            queue_id,
+            owner_queue_id,
+        )
+        # Seeding runs on every reload/import/upload, so only record the conflict
+        # the first time it is seen for this submission instead of spamming the
+        # audit log on each pass.
+        with self._connect() as conn:
+            already_logged = conn.execute(
+                "select 1 from audit_events where event = ? and object_ref = ? limit 1",
+                (_QUEUE_ID_CONFLICT_EVENT, submission_id),
+            ).fetchone()
+        if already_logged is not None:
+            return
+        self.add_audit_event(
+            "rights.system",
+            _QUEUE_ID_CONFLICT_EVENT,
+            submission_id,
+            (
+                f"동일 id가 다른 큐({owner_queue_id})에 이미 존재해 이 큐({queue_id})"
+                " 임포트를 건너뜀 — 파일명을 변경한 뒤 다시 불러오세요."
+            ),
+        )
+
     def seed_from_image_store(self, image_store: LocalSubmissionImageStore) -> int:
         # Serialize seeding so two concurrent reload/import/upload requests can't
         # both classify the same record as new and double-run analysis, external
@@ -253,11 +313,25 @@ class CopyrighterStore(
         existing_submission_ids = {
             submission["id"] for submission in self._all("submissions", queue_id=queue_id)
         }
-        records = [
+        # The "is this new?" test above is queue-scoped, but the submissions PK is
-            record
+        # global and _put() upserts on it. A record whose id (the file stem) is
-            for record in image_store.submission_records()
+        # already owned by a DIFFERENT queue would therefore be classified as new
-            if record["id"] not in existing_submission_ids
+        # and overwrite that row: the other queue's decision/memo/query history
-        ]
+        # resets to unreviewed, the row's queue_id moves, and the old evidence
+        # (linked by submission_id alone) stays attached and pollutes the new
+        # case's score. Skip the colliding record and make the conflict visible
+        # rather than destroying a decided submission.
+        foreign_queue_owners = self._submission_queue_owners(exclude_queue_id=queue_id)
+        records = []
+        for record in image_store.submission_records():
+            record_id = str(record["id"])
+            if record_id in existing_submission_ids:
+                continue
+            owner_queue_id = foreign_queue_owners.get(record_id)
+            if owner_queue_id:
+                self._note_queue_id_conflict(record_id, queue_id, owner_queue_id)
+                continue
+            records.append(record)
         if not records:
             return 0
 
@@ -338,6 +412,10 @@ class CopyrighterStore(
             )
         self._sync_submission_provider_state(queue_id=queue_id)
         self._normalize_saved_google_weak_labels_and_rescore(queue_id=queue_id)
+        # Persist submission fingerprints so duplicate / prior-case grouping can
+        # run from stored data. Idempotent: only submissions missing the field
+        # read their image again.
+        self.backfill_submission_fingerprints(image_store, queue_id=queue_id)
         with self._connect() as conn:
             conn.execute(
                 """
@@ -393,8 +471,14 @@ class CopyrighterStore(
         queue_id = str(queue["id"]) if queue else None
         submissions = self._all("submissions", queue_id=queue_id)
         evidence = self._evidence_by_submission(queue_id=queue_id)
+        duplicates = self.duplicate_groups(queue_id=queue_id)
         for submission in submissions:
             submission["evidence"] = evidence.get(submission["id"], [])
+            # Only present when the case actually has a duplicate peer, so the
+            # GUI can treat presence of the key as "show the badge".
+            group = duplicates.get(str(submission["id"]))
+            if group:
+                submission["duplicateGroup"] = group
         return {
             "submissions": submissions,
             "submissionQueue": queue,
@@ -405,6 +489,7 @@ class CopyrighterStore(
             "auditEvents": self.audit_events(),
             "coverageThresholds": dict(self.coverage_thresholds),
             "searchCoverage": self._search_coverage(submissions, evidence),
+            "decisionMemoPresets": self.decision_memo_presets(),
         }
 
     def _search_coverage(self, submissions: list[dict[str, Any]], evidence_by_submission: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
@@ -459,8 +544,16 @@ class CopyrighterStore(
             # Attribute via _evidence_matches_provider (domain-aware) so Google
             # Custom Search (source "google", domain "google_custom_search") is
             # counted under google_search, not the Vision "google" bucket.
+            # Require an actual result, not merely a matching row: a submission
+            # whose only search evidence is a failure/skip has no coverage, and
+            # counting it inflated the coverage ratio into the "good" band while
+            # the operator still had nothing to review (mirrors the llm bucket).
             for search_provider in ("naver", "google", "google_search"):
-                if any(_evidence_matches_provider(item, search_provider) for item in evidence):
+                if any(
+                    _evidence_matches_provider(item, search_provider)
+                    and _provider_item_has_result(item)
+                    for item in evidence
+                ):
                     has_search_evidence = True
                     provider_stats[search_provider]["evidenceSubmissions"] += 1
             if has_search_evidence:

src/rights_filter/server/store_backup.py

@@ -0,0 +1,226 @@
+"""Online snapshot backup for CopyrighterStore, as a mixin.
+
+The whole operational record — decisions, knowledge base, audit trail — lives in
+one SQLite file plus a few media directories, and an air-gapped deployment has no
+cloud backup to fall back on. This produces a consistent snapshot while the
+server keeps serving.
+
+WAL note: copying the .sqlite3 file while the server runs yields a torn snapshot
+because recent commits still live in the -wal sidecar. sqlite3's online backup
+API checkpoints into the destination instead, which is why the DB is copied
+through `Connection.backup` rather than shutil.
+"""
+
+from __future__ import annotations
+
+import shutil
+import sqlite3
+import zipfile
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+# Store-owned media roots included in a snapshot, as (archive folder, attribute).
+# Submission source images are handled separately: they live in whatever folder
+# the active queue points at, which may be outside the data root.
+_MEDIA_SOURCES = (
+    ("knowledge", "knowledge_image_dir"),
+    ("collections", "collection_image_dir"),
+    ("face-crops", "face_crop_image_dir"),
+)
+
+DEFAULT_BACKUP_RETENTION = 10
+
+
+class StoreBackupMixin:
+    def create_backup(
+        self, *, retention: int = DEFAULT_BACKUP_RETENTION
+    ) -> dict[str, Any]:
+        """Write a timestamped snapshot zip and prune older ones."""
+        backup_dir = Path(self.db_path).resolve().parent / "backups"
+        backup_dir.mkdir(parents=True, exist_ok=True)
+        stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+        # Two snapshots inside the same second would otherwise overwrite each
+        # other, silently destroying the earlier restore point.
+        archive_path = backup_dir / f"copyrighter-snapshot-{stamp}.zip"
+        suffix = 1
+        while archive_path.exists():
+            suffix += 1
+            archive_path = backup_dir / f"copyrighter-snapshot-{stamp}-{suffix}.zip"
+        staged_db = backup_dir / f".snapshot-{archive_path.stem}.sqlite3"
+
+        try:
+            self._snapshot_database(staged_db)
+            with zipfile.ZipFile(
+                archive_path, "w", compression=zipfile.ZIP_DEFLATED
+            ) as archive:
+                archive.write(staged_db, arcname="copyrighter.sqlite3")
+                media_counts = self._archive_media(archive)
+        finally:
+            staged_db.unlink(missing_ok=True)
+
+        pruned = _prune_old_backups(backup_dir, retention)
+        size_bytes = archive_path.stat().st_size
+        self.add_audit_event(
+            "rights.ops",
+            "Backup snapshot created",
+            archive_path.name,
+            f"{size_bytes} bytes · media {media_counts} · pruned {pruned}",
+        )
+        return {
+            "file": archive_path.name,
+            "path": str(archive_path),
+            "sizeBytes": size_bytes,
+            "createdAt": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            "mediaFiles": media_counts,
+            "prunedFiles": pruned,
+        }
+
+    def backups(self) -> list[dict[str, Any]]:
+        backup_dir = Path(self.db_path).resolve().parent / "backups"
+        if not backup_dir.exists():
+            return []
+        entries = []
+        for path in sorted(backup_dir.glob("copyrighter-snapshot-*.zip"), reverse=True):
+            stat = path.stat()
+            entries.append(
+                {
+                    "file": path.name,
+                    "sizeBytes": stat.st_size,
+                    "createdAt": datetime.fromtimestamp(stat.st_mtime).strftime(
+                        "%Y-%m-%d %H:%M:%S"
+                    ),
+                }
+            )
+        return entries
+
+    def _snapshot_database(self, destination: Path) -> None:
+        # Connection.backup copies a transactionally consistent image, including
+        # anything still sitting in the WAL, without blocking writers.
+        #
+        # sqlite3's context manager only commits/rolls back — it does NOT close.
+        # Both handles must be closed explicitly or the staged file stays locked
+        # and cannot be zipped/removed on Windows.
+        source = self._connect()
+        try:
+            target = sqlite3.connect(destination)
+            try:
+                source.backup(target)
+            finally:
+                target.close()
+        finally:
+            source.close()
+
+    def _archive_media(self, archive: zipfile.ZipFile) -> int:
+        roots: list[tuple[str, Path]] = []
+        for arc_folder, attribute in _MEDIA_SOURCES:
+            root = getattr(self, attribute, None)
+            if root is not None:
+                roots.append((arc_folder, Path(root)))
+
+        submission_root = self._active_submission_root()
+        if submission_root is not None:
+            roots.append(("submissions", submission_root))
+
+        written = 0
+        for arc_folder, root_path in roots:
+            if not root_path.exists():
+                continue
+            for path in sorted(root_path.rglob("*")):
+                if not path.is_file():
+                    continue
+                archive.write(
+                    path,
+                    arcname=f"media/{arc_folder}/{path.relative_to(root_path).as_posix()}",
+                )
+                written += 1
+        return written
+
+    def _active_submission_root(self) -> Path | None:
+        # Best-effort: a store with no configured queue still produces a valid
+        # database snapshot, just without submission source images.
+        try:
+            queue = self.bootstrap_active_queue()
+        except Exception:
+            return None
+        if not queue or not queue.get("folderPath"):
+            return None
+        return Path(str(queue["folderPath"]))
+
+
+def restore_backup(archive_path: Path | str, target_db: Path | str) -> dict[str, Any]:
+    """Restore a snapshot over a STOPPED server's data directory.
+
+    Offline on purpose: replacing the database under a live server would leave
+    open connections reading a file that no longer exists.
+    """
+    archive = Path(archive_path)
+    db_path = Path(target_db)
+    if not archive.exists():
+        raise ValueError(f"backup archive not found: {archive}")
+
+    data_root = db_path.resolve().parent
+    with zipfile.ZipFile(archive) as bundle:
+        names = set(bundle.namelist())
+        if "copyrighter.sqlite3" not in names:
+            raise ValueError("archive does not contain copyrighter.sqlite3")
+
+        if db_path.exists():
+            # Keep the pre-restore database next to it; an operator who restores
+            # the wrong snapshot would otherwise have no way back.
+            rescue = db_path.with_suffix(
+                f".sqlite3.pre-restore-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
+            )
+            shutil.copy2(db_path, rescue)
+        # Stale sidecars would be replayed on top of the restored file.
+        for sidecar in (f"{db_path}-wal", f"{db_path}-shm"):
+            Path(sidecar).unlink(missing_ok=True)
+
+        db_path.parent.mkdir(parents=True, exist_ok=True)
+        with bundle.open("copyrighter.sqlite3") as source, open(db_path, "wb") as target:
+            shutil.copyfileobj(source, target)
+
+        restored_media = 0
+        for name in sorted(names):
+            if not name.startswith("media/") or name.endswith("/"):
+                continue
+            relative = Path(name).relative_to("media")
+            destination = data_root / _MEDIA_RESTORE_DIRS.get(
+                relative.parts[0], relative.parts[0]
+            )
+            destination = destination.joinpath(*relative.parts[1:])
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            with bundle.open(name) as source, open(destination, "wb") as target:
+                shutil.copyfileobj(source, target)
+            restored_media += 1
+
+    return {"database": str(db_path), "mediaFiles": restored_media}
+
+
+# Archive folder -> path relative to the data root, so a restore lands where the
+# running server expects to find each media kind.
+_MEDIA_RESTORE_DIRS = {
+    "knowledge": "knowledge/images",
+    "collections": "collections/images",
+    "face-crops": "face-crops",
+    # Submission images restore under the data root's default submissions folder.
+    # If the queue pointed elsewhere, re-import that folder after restoring.
+    "submissions": "submissions",
+}
+
+
+def _prune_old_backups(backup_dir: Path, retention: int) -> int:
+    keep = max(1, int(retention))
+    snapshots = sorted(
+        backup_dir.glob("copyrighter-snapshot-*.zip"),
+        key=lambda path: path.stat().st_mtime,
+        reverse=True,
+    )
+    removed = 0
+    for path in snapshots[keep:]:
+        try:
+            path.unlink()
+            removed += 1
+        except OSError:
+            continue
+    return removed

src/rights_filter/server/store_constants.py

@@ -23,6 +23,7 @@ STORE_TABLES = {
     "corrections",
     "audit_events",
     "submission_queues",
+    "decision_memo_presets",
 }
 RISK_BANDS = ("low", "medium", "high", "failed", "pending")
 DECISION_STATUSES = ("unreviewed", "held", "rejected", "approved", "corrected")

src/rights_filter/server/store_duplicates.py

@@ -0,0 +1,161 @@
+"""Submission-to-submission duplicate / prior-case index, as a mixin.
+
+pHash was only ever compared against the knowledge base and against search-result
+images — never between submissions — so the same picture arriving twice was
+reviewed twice from scratch, and a resubmission of an already-rejected image was
+invisible unless a watchlist entry happened to exist.
+
+The fingerprints are already computed during analysis but the evidence payload
+whitelist drops them, so this module backfills `fingerprints` onto the submission
+record (idempotent, image read only when missing) and groups from there.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from rights_filter.analysis.fingerprints import FingerprintService
+from rights_filter.server.image_store import LocalSubmissionImageStore
+
+logger = logging.getLogger(__name__)
+
+# Hamming-distance similarity at/above which two submissions are treated as the
+# same picture. Deliberately stricter than the knowledge-base match threshold
+# (0.9): this drives "these are duplicates, handle them together", so a false
+# grouping is more costly than a missed one.
+DUPLICATE_SIMILARITY_THRESHOLD = 0.97
+
+
+class StoreDuplicatesMixin:
+    def backfill_submission_fingerprints(
+        self, image_store: LocalSubmissionImageStore, queue_id: str | None = None
+    ) -> int:
+        """Persist exact/perceptual fingerprints on submissions missing them."""
+        service = FingerprintService()
+        updated = 0
+        for submission in self._all("submissions", queue_id=queue_id):
+            if submission.get("fingerprints"):
+                continue
+            submission_id = str(submission["id"])
+            fingerprints = _fingerprints_for_submission(
+                service, image_store, submission_id
+            )
+            if fingerprints is None:
+                continue
+            submission["fingerprints"] = fingerprints
+            self._put("submissions", submission_id, submission)
+            updated += 1
+        return updated
+
+    def duplicate_groups(self, queue_id: str | None = None) -> dict[str, dict[str, Any]]:
+        """Map submission id -> duplicate-group summary for the given queue.
+
+        Only submissions that have at least one peer appear in the result.
+        """
+        submissions = [
+            item for item in self._all("submissions", queue_id=queue_id)
+            if _perceptual_of(item) or _exact_of(item)
+        ]
+        groups = _cluster(submissions)
+
+        summary: dict[str, dict[str, Any]] = {}
+        for group in groups:
+            if len(group) < 2:
+                continue
+            for submission in group:
+                peers = [other for other in group if other["id"] != submission["id"]]
+                decided_peers = [
+                    {
+                        "submissionId": str(peer["id"]),
+                        "title": str(peer.get("title", "")),
+                        "decision": str(peer.get("decisionStatus", "unreviewed")),
+                    }
+                    for peer in peers
+                    if str(peer.get("decisionStatus", "unreviewed")) != "unreviewed"
+                ]
+                summary[str(submission["id"])] = {
+                    "size": len(group),
+                    "peerIds": [str(peer["id"]) for peer in peers],
+                    "priorDecisions": decided_peers,
+                    "conflicting": _has_conflicting_decisions(group),
+                }
+        return summary
+
+
+def _fingerprints_for_submission(
+    service: FingerprintService,
+    image_store: LocalSubmissionImageStore,
+    submission_id: str,
+) -> dict[str, str] | None:
+    try:
+        payload = image_store.image_payload(submission_id)
+    except (KeyError, OSError, ValueError):
+        # A submission whose file is gone simply has no fingerprint; the queue
+        # prune handles its removal separately.
+        return None
+    try:
+        fingerprints = service.fingerprints_for(payload.content)
+    except Exception:
+        logger.warning("Could not fingerprint submission %s", submission_id)
+        return None
+    return {"exact": fingerprints.exact, "perceptual": fingerprints.perceptual}
+
+
+def _exact_of(submission: dict[str, Any]) -> str:
+    return str((submission.get("fingerprints") or {}).get("exact", ""))
+
+
+def _perceptual_of(submission: dict[str, Any]) -> str:
+    return str((submission.get("fingerprints") or {}).get("perceptual", ""))
+
+
+def _cluster(submissions: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
+    """Group submissions by exact hash, then merge near-identical pHash groups.
+
+    Queues hold hundreds of cases, so the O(n^2) perceptual pass is cheap enough
+    and avoids the accuracy loss of hash bucketing.
+    """
+    service = FingerprintService()
+    groups: list[list[dict[str, Any]]] = []
+    by_exact: dict[str, list[dict[str, Any]]] = {}
+
+    for submission in submissions:
+        exact = _exact_of(submission)
+        if exact:
+            by_exact.setdefault(exact, []).append(submission)
+        else:
+            groups.append([submission])
+    groups.extend(by_exact.values())
+
+    merged: list[list[dict[str, Any]]] = []
+    for group in groups:
+        representative = _perceptual_of(group[0])
+        target = None
+        if representative:
+            for candidate in merged:
+                candidate_hash = _perceptual_of(candidate[0])
+                if not candidate_hash:
+                    continue
+                if (
+                    service.similarity(representative, candidate_hash)
+                    >= DUPLICATE_SIMILARITY_THRESHOLD
+                ):
+                    target = candidate
+                    break
+        if target is None:
+            merged.append(list(group))
+        else:
+            target.extend(group)
+    return merged
+
+
+def _has_conflicting_decisions(group: list[dict[str, Any]]) -> bool:
+    decisions = {
+        str(item.get("decisionStatus", "unreviewed"))
+        for item in group
+        if str(item.get("decisionStatus", "unreviewed")) != "unreviewed"
+    }
+    # The same image approved in one case and rejected in another is a review
+    # consistency problem the operator should see before deciding again.
+    return "approved" in decisions and bool(decisions & {"rejected", "held"})

src/rights_filter/server/store_workflow.py

@@ -0,0 +1,284 @@
+"""Operator throughput features for CopyrighterStore, as a mixin.
+
+Batch decisions, reusable decision-memo presets, and decision export. Mixed into
+CopyrighterStore; relies on persistence + operations methods and self.*
+attributes provided by the host class.
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+from typing import Any
+
+from rights_filter.server.image_store import LocalSubmissionImageStore
+from rights_filter.server.store_serialization import _now_label, _stable_id
+
+# A batch is an operator convenience, not an import path: keep it small enough
+# that one accidental "select all" cannot rewrite the whole queue in one click.
+MAX_BATCH_DECISIONS = 200
+
+# Fields the applicant-facing export may contain. Anything not listed here —
+# risk score, evidence, provider state, internal memos beyond the decision
+# reason — must never cross this boundary.
+APPLICANT_EXPORT_FIELDS = ("submissionId", "title", "decision", "applicantStatus", "memo")
+
+INTERNAL_EXPORT_FIELDS = (
+    "submissionId",
+    "title",
+    "decision",
+    "applicantStatus",
+    "riskScore",
+    "riskBand",
+    "memo",
+    "decidedAt",
+    "evidenceUsed",
+    "evidenceUrls",
+)
+
+
+class StoreWorkflowMixin:
+    # ---- batch decisions ------------------------------------------------
+
+    def record_decision_batch(
+        self,
+        submission_ids: list[str],
+        decision: str,
+        memo: str = "",
+        image_store: LocalSubmissionImageStore | None = None,
+    ) -> dict[str, Any]:
+        """Apply one decision to many submissions, reporting per-case outcomes.
+
+        Each case goes through record_decision so the per-case transaction, audit
+        event and watchlist side effects stay identical to a single decision.
+        One failure does not abort the rest — the caller gets an explicit list of
+        what succeeded and what did not.
+        """
+        if decision not in {"approved", "held", "rejected"}:
+            raise ValueError("unsupported decision")
+        unique_ids = list(dict.fromkeys(str(value) for value in submission_ids if str(value)))
+        if not unique_ids:
+            raise ValueError("no submissions selected")
+        if len(unique_ids) > MAX_BATCH_DECISIONS:
+            raise ValueError(
+                f"batch too large: {len(unique_ids)} > {MAX_BATCH_DECISIONS}"
+            )
+        if decision == "rejected" and not memo.strip():
+            raise ValueError("reject decision requires memo")
+
+        succeeded: list[str] = []
+        failed: list[dict[str, str]] = []
+        for submission_id in unique_ids:
+            try:
+                self.record_decision(submission_id, decision, memo, image_store)
+                succeeded.append(submission_id)
+            except KeyError:
+                # str(KeyError) is just the repr'd key — useless in the UI.
+                failed.append(
+                    {"submissionId": submission_id, "error": "제출물을 찾을 수 없습니다"}
+                )
+            except ValueError as exc:
+                failed.append({"submissionId": submission_id, "error": str(exc)})
+
+        self.add_audit_event(
+            "rights.ops",
+            "Operator batch decision",
+            f"{decision} x{len(succeeded)}",
+            f"{len(succeeded)} applied, {len(failed)} failed · {memo or 'memo optional'}",
+        )
+        payload = self.bootstrap()
+        payload["batchDecision"] = {
+            "decision": decision,
+            "applied": succeeded,
+            "failed": failed,
+        }
+        return payload
+
+    # ---- decision memo presets ------------------------------------------
+
+    def decision_memo_presets(self) -> list[dict[str, Any]]:
+        presets = self._all("decision_memo_presets")
+        return sorted(presets, key=lambda item: str(item.get("label", "")))
+
+    def save_decision_memo_preset(self, label: str, body: str) -> list[dict[str, Any]]:
+        clean_label = str(label or "").strip()
+        clean_body = str(body or "").strip()
+        if not clean_label:
+            raise ValueError("preset label is required")
+        if not clean_body:
+            raise ValueError("preset body is required")
+        if len(clean_label) > 40:
+            raise ValueError("preset label must be 40 characters or fewer")
+        if len(clean_body) > 500:
+            raise ValueError("preset body must be 500 characters or fewer")
+
+        # Key by label so re-saving the same label edits it instead of stacking
+        # near-duplicate chips in the operator's decision pane.
+        preset_id = _stable_id("preset", clean_label)
+        self._put(
+            "decision_memo_presets",
+            preset_id,
+            {
+                "id": preset_id,
+                "label": clean_label,
+                "body": clean_body,
+                "updatedAt": _now_label(),
+            },
+        )
+        self.add_audit_event(
+            "rights.ops", "Decision memo preset saved", preset_id, clean_label
+        )
+        return self.decision_memo_presets()
+
+    def delete_decision_memo_preset(self, preset_id: str) -> list[dict[str, Any]]:
+        with self._connect() as conn:
+            cursor = conn.execute(
+                "delete from decision_memo_presets where id = ?", (str(preset_id),)
+            )
+            if not cursor.rowcount:
+                raise KeyError(preset_id)
+        self.add_audit_event(
+            "rights.ops", "Decision memo preset deleted", str(preset_id), "removed"
+        )
+        return self.decision_memo_presets()
+
+    # ---- decision export -------------------------------------------------
+
+    def export_decisions(
+        self,
+        *,
+        audience: str = "internal",
+        status: str = "",
+        since: str = "",
+    ) -> dict[str, Any]:
+        """Collect decided cases for hand-off.
+
+        `audience="applicant"` restricts the payload to the fields that may leave
+        the review system: no risk score, no evidence, no provider state.
+        """
+        if audience not in {"internal", "applicant"}:
+            raise ValueError("audience must be 'internal' or 'applicant'")
+        wanted_status = str(status or "").strip()
+        if wanted_status and wanted_status not in {"approved", "held", "rejected", "corrected"}:
+            raise ValueError("unsupported decision status filter")
+
+        decided_at = self._decision_timestamps()
+        since_text = str(since or "").strip()
+        rows: list[dict[str, Any]] = []
+        for submission in self._all("submissions"):
+            decision = str(submission.get("decisionStatus", "unreviewed"))
+            if decision == "unreviewed":
+                continue
+            if wanted_status and decision != wanted_status:
+                continue
+            submission_id = str(submission.get("id", ""))
+            timestamp = decided_at.get(submission_id, "")
+            if since_text and timestamp and timestamp < since_text:
+                continue
+            rows.append(
+                self._export_row(submission, decision, timestamp, audience=audience)
+            )
+
+        rows.sort(key=lambda row: str(row.get("decidedAt", "")), reverse=True)
+        fields = (
+            APPLICANT_EXPORT_FIELDS if audience == "applicant" else INTERNAL_EXPORT_FIELDS
+        )
+        self.add_audit_event(
+            "rights.ops",
+            "Decisions exported",
+            audience,
+            f"{len(rows)} case(s) · status={wanted_status or 'all'} · since={since_text or 'all'}",
+        )
+        return {
+            "audience": audience,
+            "fields": list(fields),
+            "rows": rows,
+            "csv": _rows_to_csv(rows, fields),
+        }
+
+    def _export_row(
+        self,
+        submission: dict[str, Any],
+        decision: str,
+        decided_at: str,
+        *,
+        audience: str,
+    ) -> dict[str, Any]:
+        submission_id = str(submission.get("id", ""))
+        memo = self._latest_decision_memo(submission_id)
+        row = {
+            "submissionId": submission_id,
+            "title": str(submission.get("title", "")),
+            "decision": decision,
+            "applicantStatus": str(submission.get("applicantStatus", "")),
+            "memo": memo,
+        }
+        if audience == "applicant":
+            return row
+
+        used = [
+            item
+            for item in self._evidence_for_submission(submission_id)
+            if str(item.get("operatorStatus", "")) == "used_for_judgment"
+        ]
+        row.update(
+            {
+                "riskScore": submission.get("riskScore", 0),
+                "riskBand": str(submission.get("riskBand", "")),
+                "decidedAt": decided_at,
+                "evidenceUsed": " | ".join(str(item.get("title", "")) for item in used),
+                "evidenceUrls": " | ".join(
+                    str(item.get("url", "")) for item in used if item.get("url")
+                ),
+            }
+        )
+        return row
+
+    def _decision_events(self) -> list[dict[str, Any]]:
+        with self._connect() as conn:
+            rows = conn.execute(
+                """
+                select object_ref, timestamp, payload from audit_events
+                where event = 'Operator decision created'
+                order by id asc
+                """
+            ).fetchall()
+        return [
+            {
+                "object_ref": str(row["object_ref"]),
+                "timestamp": str(row["timestamp"]),
+                "payload": row["payload"],
+            }
+            for row in rows
+        ]
+
+    def _decision_timestamps(self) -> dict[str, str]:
+        # Submissions carry no decided-at field; the audit trail is the only
+        # record of when the operator decided, so read the latest event per case.
+        return {event["object_ref"]: event["timestamp"] for event in self._decision_events()}
+
+    def _latest_decision_memo(self, submission_id: str) -> str:
+        import json
+
+        memo = ""
+        for event in self._decision_events():
+            if event["object_ref"] != submission_id:
+                continue
+            try:
+                change = str(json.loads(event["payload"]).get("change", ""))
+            except (ValueError, TypeError):
+                continue
+            # Audit change text is "<decision> · <memo>".
+            _, separator, tail = change.partition(" · ")
+            if separator and tail and tail != "memo optional":
+                memo = tail
+        return memo
+
+
+def _rows_to_csv(rows: list[dict[str, Any]], fields: tuple[str, ...]) -> str:
+    buffer = io.StringIO()
+    writer = csv.DictWriter(buffer, fieldnames=list(fields), extrasaction="ignore")
+    writer.writeheader()
+    for row in rows:
+        writer.writerow(row)
+    return buffer.getvalue()

tests/operator_gui/test_static_workbench.py

@@ -128,8 +128,14 @@ def test_workbench_shell_exposes_all_internal_operator_views():
     assert "추천 쿼리로 외부 검색 결과를 보강합니다." in html
     assert "기준 DB에 반영합니다." in html
     assert ".operator-workflow" in styles
-    assert "bulk-approve" not in html.lower()
+    # 2026-07-24: the v1 "no bulk approve/reject" restriction was lifted (see
-    assert "bulk-reject" not in html.lower()
+    # docs/superpowers/specs/2026-05-25-image-rights-operator-gui-design.md).
+    # Batch decisions are allowed, but only with the guardrails that made lifting
+    # it acceptable — assert those are present rather than that the controls are.
+    assert 'id="bulk-approve"' in html
+    assert 'id="bulk-reject"' in html
+    assert 'id="bulk-decision-count"' in html  # selection size is always visible
+    assert 'id="bulk-decision-status"' in html  # per-case outcome is reported back
 
     assert 'data-view="evidence"' not in html
     assert 'id="evidence-view"' not in html

tests/rights_filter/server/test_store_workflow.py

@@ -0,0 +1,255 @@
+from pathlib import Path
+
+import pytest
+
+from rights_filter.integrations.env_clients import build_provider_runtime
+from rights_filter.server.sqlite_store import CopyrighterStore
+from rights_filter.server.store_backup import restore_backup
+from rights_filter.server.store_workflow import MAX_BATCH_DECISIONS
+
+from tests.rights_filter.server.test_sqlite_store import _image_store, FakeJsonTransport
+
+
+def _photo_store(root: Path, *, duplicate: bool = False):
+    """Two visually distinct PNGs (ids = file stems), optionally plus a byte copy.
+
+    Real raster images so the fingerprint/duplicate paths behave as in production;
+    the shared SVG fixture cannot be perceptually hashed.
+    """
+    from io import BytesIO
+
+    from PIL import Image, ImageDraw
+
+    from rights_filter.server.image_store import LocalSubmissionImageStore
+
+    image_dir = root / "images"
+    image_dir.mkdir(parents=True, exist_ok=True)
+    for name, color in (("alpha", (210, 40, 40)), ("beta", (30, 90, 200))):
+        canvas = Image.new("RGB", (64, 48), color=color)
+        draw = ImageDraw.Draw(canvas)
+        # Vary the pixels so the average hash is informative, not degenerate.
+        draw.rectangle([4, 4, 30, 30], fill=(255, 255, 255))
+        draw.ellipse([32, 10, 60, 40], fill=(10, 10, 10))
+        if name == "beta":
+            draw.line([0, 0, 63, 47], fill=(250, 250, 0), width=5)
+        buffer = BytesIO()
+        canvas.save(buffer, format="PNG")
+        (image_dir / f"{name}.png").write_bytes(buffer.getvalue())
+
+    if duplicate:
+        (image_dir / "alpha-copy.png").write_bytes((image_dir / "alpha.png").read_bytes())
+
+    return LocalSubmissionImageStore(root)
+
+
+def _store(tmp_path: Path, *, duplicate: bool = False) -> CopyrighterStore:
+    store = CopyrighterStore(
+        tmp_path / "copyrighter.sqlite3",
+        provider_runtime=build_provider_runtime(env={}, transport=FakeJsonTransport({})),
+    )
+    store.initialize()
+    store.seed_from_image_store(_photo_store(tmp_path / "submissions", duplicate=duplicate))
+    return store
+
+
+def _first_ids(store: CopyrighterStore, count: int) -> list[str]:
+    return [str(item["id"]) for item in store.bootstrap()["submissions"][:count]]
+
+
+# --- batch decisions ---------------------------------------------------------
+
+
+def test_batch_decision_applies_to_every_selected_submission(tmp_path: Path):
+    store = _store(tmp_path)
+    ids = _first_ids(store, 2)
+
+    payload = store.record_decision_batch(ids, "approved")
+
+    assert payload["batchDecision"]["applied"] == ids
+    assert payload["batchDecision"]["failed"] == []
+    decided = {
+        str(item["id"]): item["decisionStatus"] for item in payload["submissions"]
+    }
+    assert all(decided[submission_id] == "approved" for submission_id in ids)
+
+
+def test_batch_decision_reports_per_case_failure_without_aborting_the_rest(tmp_path: Path):
+    store = _store(tmp_path)
+    ids = _first_ids(store, 1)
+
+    payload = store.record_decision_batch([*ids, "SUB-DOES-NOT-EXIST"], "approved")
+
+    assert payload["batchDecision"]["applied"] == ids
+    assert [item["submissionId"] for item in payload["batchDecision"]["failed"]] == [
+        "SUB-DOES-NOT-EXIST"
+    ]
+
+
+def test_batch_reject_requires_memo(tmp_path: Path):
+    store = _store(tmp_path)
+    ids = _first_ids(store, 1)
+
+    with pytest.raises(ValueError, match="memo"):
+        store.record_decision_batch(ids, "rejected", "   ")
+
+
+def test_batch_decision_rejects_oversized_selection(tmp_path: Path):
+    store = _store(tmp_path)
+
+    with pytest.raises(ValueError, match="batch too large"):
+        store.record_decision_batch(
+            [f"SUB-{index}" for index in range(MAX_BATCH_DECISIONS + 1)], "approved"
+        )
+
+
+# --- memo presets ------------------------------------------------------------
+
+
+def test_memo_preset_save_is_upsert_by_label(tmp_path: Path):
+    store = _store(tmp_path)
+
+    store.save_decision_memo_preset("연예인 초상", "연예인 초상이 포함되어 있습니다.")
+    presets = store.save_decision_memo_preset("연예인 초상", "수정된 문구")
+
+    assert len(presets) == 1
+    assert presets[0]["body"] == "수정된 문구"
+    assert store.bootstrap()["decisionMemoPresets"][0]["label"] == "연예인 초상"
+
+
+def test_memo_preset_requires_label_and_body(tmp_path: Path):
+    store = _store(tmp_path)
+
+    with pytest.raises(ValueError):
+        store.save_decision_memo_preset("", "body")
+    with pytest.raises(ValueError):
+        store.save_decision_memo_preset("label", "  ")
+
+
+def test_memo_preset_delete_removes_entry(tmp_path: Path):
+    store = _store(tmp_path)
+    presets = store.save_decision_memo_preset("방송 캡처", "방송 화면 캡처입니다.")
+
+    remaining = store.delete_decision_memo_preset(presets[0]["id"])
+
+    assert remaining == []
+
+
+# --- decision export ---------------------------------------------------------
+
+
+def test_export_skips_undecided_cases_and_carries_the_memo(tmp_path: Path):
+    store = _store(tmp_path)
+    ids = _first_ids(store, 1)
+    store.record_decision(ids[0], "rejected", "연예인 초상 포함")
+
+    export = store.export_decisions(audience="internal")
+
+    assert [row["submissionId"] for row in export["rows"]] == ids
+    assert export["rows"][0]["decision"] == "rejected"
+    assert export["rows"][0]["memo"] == "연예인 초상 포함"
+    assert "riskScore" in export["rows"][0]
+
+
+def test_applicant_export_never_exposes_internal_signals(tmp_path: Path):
+    store = _store(tmp_path)
+    ids = _first_ids(store, 1)
+    store.record_decision(ids[0], "approved")
+
+    export = store.export_decisions(audience="applicant")
+
+    row = export["rows"][0]
+    for forbidden in ("riskScore", "riskBand", "evidenceUsed", "evidenceUrls"):
+        assert forbidden not in row
+    assert "riskScore" not in export["csv"]
+
+
+def test_export_status_filter_and_bad_audience(tmp_path: Path):
+    store = _store(tmp_path)
+    ids = _first_ids(store, 2)
+    store.record_decision(ids[0], "approved")
+    store.record_decision(ids[1], "held")
+
+    approved = store.export_decisions(status="approved")
+
+    assert [row["submissionId"] for row in approved["rows"]] == [ids[0]]
+    with pytest.raises(ValueError, match="audience"):
+        store.export_decisions(audience="public")
+
+
+# --- duplicate grouping ------------------------------------------------------
+
+
+def test_identical_images_are_grouped_with_prior_decision(tmp_path: Path):
+    store = _store(tmp_path, duplicate=True)
+    store.record_decision("alpha", "rejected", "원본 반려")
+
+    groups = store.duplicate_groups()
+
+    assert "alpha-copy" in groups
+    assert "alpha" in groups["alpha-copy"]["peerIds"]
+    assert groups["alpha-copy"]["priorDecisions"][0]["decision"] == "rejected"
+    # A distinct image must not be pulled into the group.
+    assert "beta" not in groups
+
+
+def test_duplicate_group_flags_conflicting_decisions(tmp_path: Path):
+    store = _store(tmp_path, duplicate=True)
+    store.record_decision("alpha", "approved")
+    store.record_decision("alpha-copy", "rejected", "같은 이미지인데 반려")
+
+    groups = store.duplicate_groups()
+
+    assert groups["alpha"]["conflicting"] is True
+
+
+def test_unique_images_get_no_duplicate_group(tmp_path: Path):
+    store = _store(tmp_path)
+
+    submissions = store.bootstrap()["submissions"]
+
+    assert all("duplicateGroup" not in item for item in submissions)
+
+
+# --- backup / restore --------------------------------------------------------
+
+
+def test_backup_snapshot_round_trips_decisions(tmp_path: Path):
+    store = _store(tmp_path)
+    ids = _first_ids(store, 1)
+    store.record_decision(ids[0], "approved")
+
+    snapshot = store.create_backup()
+
+    assert snapshot["file"].startswith("copyrighter-snapshot-")
+    assert snapshot["sizeBytes"] > 0
+    assert store.backups()[0]["file"] == snapshot["file"]
+
+    # Simulate loss, then restore into a fresh database path.
+    restored_db = tmp_path / "restored" / "copyrighter.sqlite3"
+    result = restore_backup(snapshot["path"], restored_db)
+
+    assert result["database"] == str(restored_db)
+    reopened = CopyrighterStore(
+        restored_db,
+        provider_runtime=build_provider_runtime(env={}, transport=FakeJsonTransport({})),
+    )
+    reopened.initialize()
+    decided = {
+        str(item["id"]): item["decisionStatus"] for item in reopened.bootstrap()["submissions"]
+    }
+    assert decided[ids[0]] == "approved"
+
+
+def test_backup_retention_prunes_older_snapshots(tmp_path: Path):
+    store = _store(tmp_path)
+
+    created = [store.create_backup(retention=2)["file"] for _ in range(3)]
+
+    # Distinct files even within one second, and only the newest two survive.
+    assert len(set(created)) == 3
+    assert len(store.backups()) == 2
+
+
+def test_restore_rejects_missing_archive(tmp_path: Path):
+    with pytest.raises(ValueError, match="not found"):
+        restore_backup(tmp_path / "nope.zip", tmp_path / "db.sqlite3")

web/operator-gui/index.html

@@ -13,6 +13,20 @@
     <script src="app.js" defer></script>
   </head>
   <body data-internal-only="true">
+    <div class="auth-gate" id="auth-gate" hidden role="dialog" aria-modal="true" aria-labelledby="auth-gate-title">
+      <form class="auth-gate-card" id="auth-gate-form" autocomplete="off">
+        <h1 id="auth-gate-title">접근 토큰 입력</h1>
+        <p class="auth-gate-hint">
+          이 서버는 인증이 설정되어 있습니다. 운영 담당자에게 받은
+          <code>COPYRIGHTER_AUTH_TOKEN</code> 값을 입력하세요.
+        </p>
+        <label class="auth-gate-label" for="auth-gate-input">접근 토큰</label>
+        <input class="auth-gate-input" id="auth-gate-input" type="password" required
+               autocomplete="off" spellcheck="false" aria-describedby="auth-gate-error">
+        <p class="auth-gate-error" id="auth-gate-error" role="alert" hidden></p>
+        <button class="primary-action" id="auth-gate-submit" type="submit">인증</button>
+      </form>
+    </div>
     <div class="app-shell">
       <nav class="nav-rail" aria-label="내부 운영 콘솔">
         <div class="brand-block">
@@ -83,7 +97,18 @@
                 <button class="primary-action" type="button" id="reload-submissions">새 제출 불러오기</button>
                 <button class="secondary-action" type="button" id="bulk-rerun">선택 재분석</button>
               </div>
+              <div class="queue-actions queue-actions-secondary">
+                <span class="bulk-decision-label" id="bulk-decision-count">선택 0건</span>
+                <button class="approve-action" type="button" id="bulk-approve" data-bulk-decision="approved">선택 승인</button>
+                <button class="hold-action" type="button" id="bulk-hold" data-bulk-decision="held">선택 보류</button>
+                <button class="reject-action" type="button" id="bulk-reject" data-bulk-decision="rejected">선택 반려</button>
+                <button class="secondary-action" type="button" id="export-decisions">판정 내보내기</button>
+                <button class="secondary-action" type="button" id="create-backup">백업 만들기</button>
+                <span class="review-progress" id="review-progress" aria-live="polite"></span>
+              </div>
             </div>
+            <div class="import-status" id="bulk-decision-status" aria-live="polite"></div>
+            <pre class="keyboard-help" id="keyboard-help" hidden></pre>
             <div class="import-status" id="submission-import-status" aria-live="polite"></div>
 
             <section class="operator-workflow" aria-label="운영 흐름">
@@ -204,6 +229,7 @@
                   <span>결정 메모</span>
                   <textarea id="decision-memo" rows="3" placeholder="반려 또는 보정 시 메모 필수"></textarea>
                 </label>
+                <div class="memo-presets" id="memo-presets" aria-label="자주 쓰는 사유"></div>
                 <p class="form-error" id="memo-error" aria-live="assertive"></p>
                 <div class="decision-actions">
                   <button class="approve-action" type="button" data-decision="approved">승인</button>

web/operator-gui/styles.css

@@ -2817,3 +2817,191 @@ tbody tr.selected-row,
 .evidence-row-new {
   box-shadow: inset 0 0 0 2px var(--teal);
 }
+
+/* --- Auth gate ------------------------------------------------------- */
+/* Shown only when the server runs with COPYRIGHTER_AUTH_TOKEN and a data
+   route answered 401. Covers the shell so no half-loaded queue is visible
+   behind it. */
+.auth-gate {
+  position: fixed;
+  inset: 0;
+  z-index: 100;
+  display: grid;
+  place-items: center;
+  padding: 24px;
+  background: rgba(16, 23, 29, 0.55);
+  backdrop-filter: blur(2px);
+}
+
+.auth-gate[hidden] {
+  display: none;
+}
+
+.auth-gate-card {
+  width: min(420px, 100%);
+  display: flex;
+  flex-direction: column;
+  gap: 10px;
+  padding: 26px 26px 22px;
+  background: var(--card-raised);
+  border: 1px solid var(--hair);
+  border-radius: var(--r-lg);
+  box-shadow: var(--shadow-lg);
+}
+
+.auth-gate-card h1 {
+  margin: 0;
+  font-size: 1.05rem;
+  letter-spacing: -0.01em;
+  color: var(--ink);
+}
+
+.auth-gate-hint {
+  margin: 0;
+  font-size: 0.82rem;
+  line-height: 1.55;
+  color: var(--ink-soft);
+}
+
+.auth-gate-hint code {
+  font-family: var(--font-mono);
+  font-size: 0.78rem;
+  padding: 1px 5px;
+  border-radius: var(--r-sm);
+  background: var(--card-sunk);
+  border: 1px solid var(--hair-soft);
+}
+
+.auth-gate-label {
+  margin-top: 4px;
+  font-size: 0.75rem;
+  font-weight: 600;
+  letter-spacing: 0.02em;
+  color: var(--ink-soft);
+}
+
+.auth-gate-input {
+  width: 100%;
+  padding: 9px 11px;
+  font-family: var(--font-mono);
+  font-size: 0.86rem;
+  color: var(--ink);
+  background: var(--card);
+  border: 1px solid var(--hair-strong);
+  border-radius: var(--radius);
+}
+
+.auth-gate-input:focus {
+  outline: 2px solid var(--teal-line);
+  outline-offset: 1px;
+  border-color: var(--teal);
+}
+
+.auth-gate-error {
+  margin: 0;
+  font-size: 0.78rem;
+  color: var(--red);
+}
+
+.auth-gate-error[hidden] {
+  display: none;
+}
+
+.auth-gate-card .primary-action {
+  margin-top: 6px;
+  justify-content: center;
+}
+
+/* --- Batch decisions / review throughput ------------------------------ */
+.queue-actions-secondary {
+  margin-top: 8px;
+  padding-top: 8px;
+  border-top: 1px dashed var(--hair);
+  align-items: center;
+}
+
+.bulk-decision-label {
+  font-size: 0.78rem;
+  font-weight: 600;
+  color: var(--ink-soft);
+  font-variant-numeric: tabular-nums;
+}
+
+.review-progress {
+  margin-left: auto;
+  font-size: 0.76rem;
+  color: var(--ink-faint);
+  font-variant-numeric: tabular-nums;
+}
+
+.keyboard-help {
+  margin: 8px 0 0;
+  padding: 12px 14px;
+  font-family: var(--font-mono);
+  font-size: 0.78rem;
+  line-height: 1.7;
+  color: var(--ink-soft);
+  background: var(--card-sunk);
+  border: 1px solid var(--hair);
+  border-radius: var(--radius);
+  white-space: pre;
+  overflow-x: auto;
+}
+
+.keyboard-help[hidden] {
+  display: none;
+}
+
+/* --- Decision memo presets -------------------------------------------- */
+.memo-presets {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 5px;
+  margin-top: 6px;
+}
+
+.memo-preset-chip {
+  padding: 3px 9px;
+  font-size: 0.74rem;
+  color: var(--ink-soft);
+  background: var(--card-sunk);
+  border: 1px solid var(--hair);
+  border-radius: var(--r-pill);
+  cursor: pointer;
+}
+
+.memo-preset-chip:hover {
+  color: var(--teal-deep);
+  border-color: var(--teal-line);
+  background: var(--teal-tint-soft);
+}
+
+.memo-preset-add {
+  border-style: dashed;
+  color: var(--ink-faint);
+}
+
+/* --- Duplicate group badge -------------------------------------------- */
+.duplicate-badge {
+  display: inline-block;
+  margin-top: 3px;
+  padding: 2px 8px;
+  font-size: 0.7rem;
+  font-weight: 600;
+  letter-spacing: 0.01em;
+  color: var(--blue);
+  background: var(--blue-tint);
+  border: 1px solid var(--blue-line);
+  border-radius: var(--r-pill);
+  cursor: pointer;
+}
+
+.duplicate-badge:hover {
+  filter: brightness(0.97);
+}
+
+.duplicate-badge.conflicting {
+  color: var(--red);
+  background: var(--red-tint);
+  border-color: var(--red-line);
+}