Re-audit round 4: 5 blind reviewers over the new surface — 24 fixes, +28 tests

The findings clustered exactly where prediction said: the unreviewed web
layer. The big ones: decisions made while an extract/resolve job runs
are now refused with a 409 (the job's end-of-run rewrite from a
start-of-run snapshot would silently revert them); a cross-origin guard
blocks preflight-free mutations from hostile webpages (bodyless run
triggers, cross-site photo form posts); the JobRunner sets terminal
status in a finally catching BaseException (a greenlet death could
wedge every future run behind 409s) and writes tracebacks into the
visible job log; and a boot token lets clients accept the revision
reset after a server restart instead of freezing forever.

Even the thrice-audited core yielded one HIGH: an unvetoed bare
typo-read sibling of a confident row duplicated its add when the game
wasn't in the collection — diff now treats it as satisfied. Second-copy
adds carry a flag through to_add.csv and the upload log so verify
honestly reports them unverifiable instead of OK. Also: merged_into
chains collapse transitively; diff/enrich treat a BGG queue timeout
like a missing token; enrich prunes orphaned games.json keys; the
wizard shell-quotes .env values and creates the file 0600 from the
first byte; fsio stats the tmp inode before replace and uses unique tmp
names; an explicit missing --config errors; storage state is
owner-only; extract re-extracts corrupt caches, aborts on 3 identical
failures, and exits nonzero when nothing succeeded; torn JSON artifacts
degrade with in-browser warnings instead of 500ing every page; photo
uploads are atomic with cache-invalidation ordered first; the pipeline
page computes `running` before the buttons that depend on it; the
photo dropzone alerts on network failure; and lost-contact banners
clear on recovery everywhere.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/bggpipe/config.py

@@ -95,6 +95,11 @@ class Config:
 def load_config(path: Path | None = None) -> Config:
     cfg = Config()
     p = path or DEFAULT_CONFIG_PATH
+    if path is not None and not path.exists():
+        # an EXPLICIT --config that doesn't exist must not silently fall
+        # back to defaults — that reads photos/ while the user believes
+        # their prod config is active
+        raise FileNotFoundError(f"--config {path} does not exist")
     if p.exists():
         raw = tomllib.loads(p.read_text())
         known = {

src/bggpipe/diff.py

@@ -19,12 +19,18 @@ Outputs both artifacts:
 from __future__ import annotations
 
 import os
+from collections import Counter
 from dataclasses import dataclass, field
 from pathlib import Path
 
 import typer
 
-from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
+from bggpipe.bgg_client import (
+    BGGAuthError,
+    BGGClient,
+    BGGQueueTimeout,
+    client_for,
+)
 from bggpipe.config import Config
 from bggpipe.fsio import atomic_write_csv
 from bggpipe.models import (
@@ -44,6 +50,7 @@ TO_ADD_COLUMNS = [
     "version_name",
     "title_raw",
     "source_photos",
+    "second_copy",  # "1": game already had copies — verify can't confirm it
 ]
 TO_UPDATE_COLUMNS = ["collid", "bgg_id", "bgg_name", "version_id", "version_name"]
 
@@ -100,7 +107,10 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
                 p for p in row["source_photos"].split(";") if p
             )
 
-    def add_row(row: dict, confident: bool) -> dict:
+    queued_adds: Counter[int] = Counter()  # per game, across all passes
+
+    def add_row(row: dict, confident: bool, second_copy: bool = False) -> dict:
+        queued_adds[int(row["bgg_id"])] += 1
         photos = {p for p in row["source_photos"].split(";") if p}
         photos |= merged_photos.get(row["title_raw"], set())
         return {
@@ -112,6 +122,7 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
             "version_name": row["version_name"] if confident else "",
             "title_raw": row["title_raw"],
             "source_photos": ";".join(sorted(photos)),
+            "second_copy": "1" if second_copy else "",
         }
 
     recognized: list[dict] = []
@@ -212,7 +223,7 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
                 "left untouched"
             )
         else:
-            result.to_add.append(add_row(row, True))
+            result.to_add.append(add_row(row, True, second_copy=True))
             result.second_copies.append(
                 f"{row['title_raw']}: adding as a NEW copy with version "
                 f"{row['version_name']!r} ({row['version_id']}) — every "
@@ -226,14 +237,22 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
         bgg_id = int(row["bgg_id"])
         copies = by_object.get(bgg_id, [])
         if not copies:
-            result.to_add.append(add_row(row, False))
+            if queued_adds[bgg_id] and not row.get("dedupe_veto"):
+                # an unvetoed bare row is a typo-read sibling: a pass-1 add
+                # for the same absent game already covers the physical box —
+                # queueing it again would upload a duplicate entry
+                result.already_owned.append(row["title_raw"])
+            else:
+                result.to_add.append(
+                    add_row(row, False, second_copy=bool(queued_adds[bgg_id]))
+                )
             continue
         remaining = unconsumed(bgg_id)
         if remaining:
             consumed_collids.add(remaining[0].coll_id)
             result.already_owned.append(row["title_raw"])
         elif row.get("dedupe_veto"):
-            result.to_add.append(add_row(row, False))
+            result.to_add.append(add_row(row, False, second_copy=True))
             result.second_copies.append(
                 f"{row['title_raw']}: adding as a NEW version-less copy — "
                 "human-vetoed duplicate, every existing entry claimed by "
@@ -263,7 +282,7 @@ def run_diff(cfg: Config, *, client: BGGClient | None = None) -> DiffResult:
         client = client or client_for(cfg)
         try:
             collection = client.collection_full(cfg.bgg_username, refresh=True)
-        except BGGAuthError as err:
+        except (BGGAuthError, BGGQueueTimeout) as err:
             # a present-but-invalid token must not traceback when the
             # snapshot fallback is sitting right there
             typer.echo(f"Live fetch failed ({err}) — using snapshot files.")

src/bggpipe/enrich.py

@@ -19,7 +19,12 @@ import json
 
 import typer
 
-from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
+from bggpipe.bgg_client import (
+    BGGAuthError,
+    BGGClient,
+    BGGQueueTimeout,
+    client_for,
+)
 from bggpipe.config import Config
 from bggpipe.fsio import atomic_write_text
 from bggpipe.models import is_confident_version, is_recognized
@@ -80,7 +85,7 @@ def run_enrich(
         try:
             for thing in client.things_full(batch, refresh=refresh):
                 fetched[thing["bgg_id"]] = thing
-        except BGGAuthError:
+        except (BGGAuthError, BGGQueueTimeout):
             blocked = True
             break
 
@@ -90,6 +95,20 @@ def run_enrich(
             games[key] = {**fetched[bgg_id], "version": version}
             updated += 1
 
+    # prune keys no current target claims: a row whose version was approved
+    # after a bare-key run (or was later rejected) must not leave an orphan
+    # entry in the frontend seed data. Only safe when nothing was blocked —
+    # a token-less run knows too little to declare anything stale.
+    if not blocked:
+        current = {key for key, _, _ in targets}
+        stale = [k for k in games if k not in current]
+        for k in stale:
+            del games[k]
+        if stale:
+            typer.echo(
+                f"  pruned {len(stale)} stale entr{'y' if len(stale) == 1 else 'ies'}"
+            )
+
     games_path.parent.mkdir(parents=True, exist_ok=True)
     atomic_write_text(
         games_path, json.dumps(games, indent=2, ensure_ascii=False) + "\n"

src/bggpipe/extract.py

@@ -265,7 +265,13 @@ def rebuild_artifacts(
     entries: list[dict] = []
     unidentified: dict[str, list[dict]] = {}
     for raw_file in sorted(raw_dir.glob("*.json")):
+        try:
             data = json.loads(raw_file.read_text())
+        except json.JSONDecodeError as err:
+            raise ValueError(
+                f"{raw_file} is corrupt ({err}) — delete it (or re-upload "
+                "the photo) and run extract again"
+            ) from err
         if isinstance(data, list):  # bare-array shape
             entries.extend(data)
             continue
@@ -315,17 +321,37 @@ def run_extract(
     vision = vision or default_vision(cfg.model)
 
     failed: list[str] = []
+    consecutive: tuple[str, int] = ("", 0)
+    attempted = 0
     for photo in photos:
         raw_path = raw_dir / f"{photo.name}.json"
         if raw_path.exists() and not only and not force:
+            try:
+                json.loads(raw_path.read_text())
                 typer.echo(f"  {photo.name}: already extracted, skipping")
                 continue
+            except json.JSONDecodeError:
+                typer.echo(f"  {photo.name}: cache corrupt — re-extracting")
+        attempted += 1
         try:
             result = extract_photo(photo, vision)
         except Exception as err:  # one bad photo must not block the rest
             failed.append(photo.name)
             typer.echo(f"  {photo.name}: FAILED ({err}) — continuing")
+            kind = type(err).__name__
+            consecutive = (
+                kind,
+                consecutive[1] + 1 if kind == consecutive[0] else 1,
+            )
+            if consecutive[1] >= 3:
+                typer.echo(
+                    "  aborting — 3 identical consecutive failures look "
+                    "systemic (bad API key? wrong model in config.toml?), "
+                    "not per-photo; nothing more will be attempted"
+                )
+                break
             continue
+        consecutive = ("", 0)
         atomic_write_text(
             raw_path, json.dumps(result, indent=2, ensure_ascii=False) + "\n"
         )
@@ -352,6 +378,10 @@ def run_extract(
             f"\n{len(failed)} photo(s) failed extraction (re-run to retry): "
             + ", ".join(failed)
         )
+        if len(failed) == attempted:
+            # every attempted photo failed: this run accomplished nothing
+            # and a wrapper (or the web job runner) must not report success
+            raise typer.Exit(code=1)
 
     if unidentified:
         typer.echo(

src/bggpipe/fsio.py

@@ -9,12 +9,26 @@ from __future__ import annotations
 
 import csv
 import os
+import uuid
 from pathlib import Path
 
 
+def _tmp_for(path: Path) -> Path:
+    # unique per writer: a FIXED tmp name lets two concurrent processes
+    # interleave writes into one inode and replace garbage into place
+    return path.with_name(f"{path.name}.{uuid.uuid4().hex[:8]}.tmp")
+
+
+def atomic_write_bytes(path: Path, data: bytes) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = _tmp_for(path)
+    tmp.write_bytes(data)
+    os.replace(tmp, path)
+
+
 def atomic_write_text(path: Path, text: str) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
-    tmp = path.with_name(path.name + ".tmp")
+    tmp = _tmp_for(path)
     tmp.write_text(text)
     os.replace(tmp, path)
 
@@ -23,12 +37,16 @@ def atomic_write_csv(path: Path, columns: list[str], rows: list[dict]) -> int:
     """Atomic CSV rewrite (tmp + os.replace). Returns the written file's
     mtime_ns so callers tracking their own writes avoid a re-stat race."""
     path.parent.mkdir(parents=True, exist_ok=True)
-    tmp = path.with_name(path.name + ".tmp")
+    tmp = _tmp_for(path)
     with tmp.open("w", newline="") as f:
         writer = csv.DictWriter(
             f, fieldnames=columns, extrasaction="ignore", restval=""
         )
         writer.writeheader()
         writer.writerows(rows)
+    # stat the TMP inode before the replace: statting the destination after
+    # could hand back a FOREIGN writer's mtime landing in the gap, and the
+    # caller would record someone else's write as its own
+    mtime = tmp.stat().st_mtime_ns
     os.replace(tmp, path)
-    return path.stat().st_mtime_ns
+    return mtime

src/bggpipe/init_wizard.py

@@ -77,20 +77,34 @@ class InitReport:
 
 
 def _env_file_keys(env_path: Path) -> set[str]:
-    """Key names with non-empty values in .env. Values are never read into
-    variables that outlive this parse and are never printed."""
+    """Key names with non-empty values in .env. Handles `export KEY=v` and
+    quoted values; a quoted-empty value ("" / '') counts as NOT set. Values
+    never outlive this parse and are never printed."""
     if not env_path.exists():
         return set()
     present = set()
     for line in env_path.read_text().splitlines():
         line = line.strip()
-        if "=" in line and not line.startswith("#"):
+        if "=" not in line or line.startswith("#"):
+            continue
         key, _, value = line.partition("=")
-            if value.strip():
-                present.add(key.strip())
+        key = key.strip()
+        if key.startswith("export "):
+            key = key.removeprefix("export ").strip()
+        value = value.strip()
+        if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"":
+            value = value[1:-1]
+        if value:
+            present.add(key)
     return present
 
 
+def _quote_env_value(value: str) -> str:
+    """Single-quote for `source`/direnv safety: spaces, $, backslashes and
+    quotes must survive the shell verbatim."""
+    return "'" + value.replace("'", "'\\''") + "'"
+
+
 def _default_secret_prompt(label: str) -> str:
     return typer.prompt(label, default="", show_default=False, hide_input=True)
 
@@ -162,11 +176,14 @@ def run_init(
             report.keys_missing.append(key)
             continue
         is_new_file = not env_path.exists()
-        with env_path.open("a") as f:
+        # owner-only from the FIRST byte — chmod-after-write leaves a
+        # world-readable window holding a credential
+        fd = os.open(env_path, os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
+        with os.fdopen(fd, "a") as f:
             if is_new_file:
                 f.write(ENV_HEADER)
-            f.write(f"{key}={value}\n")
-        env_path.chmod(0o600)  # credentials: owner-only, every time
+            f.write(f"{key}={_quote_env_value(value)}\n")
+        env_path.chmod(0o600)  # older files created by hand tighten up too
         report.keys_written.append(key)
         typer.echo(f"  {key}: saved to {env_path}")
 

src/bggpipe/jobs.py

@@ -13,11 +13,14 @@ from __future__ import annotations
 import io
 import threading
 import time
+import traceback
 from collections.abc import Callable
 from contextlib import redirect_stdout
 
 import typer
 
+MAX_LOG_LINES = 1000  # buffer cap; snapshots serve the last 200
+
 
 class _LineBuffer(io.TextIOBase):
     def __init__(self) -> None:
@@ -30,6 +33,8 @@ class _LineBuffer(io.TextIOBase):
             self._partial += text
             *complete, self._partial = self._partial.split("\n")
             self._lines.extend(complete)
+            if len(self._lines) > MAX_LOG_LINES:  # bound memory on long runs
+                del self._lines[: len(self._lines) - MAX_LOG_LINES]
         return len(text)
 
     def lines(self) -> list[str]:
@@ -67,6 +72,7 @@ class JobRunner:
 
     def _run(self, fn: Callable[[], object]) -> None:
         status, error = "done", ""
+        try:
             try:
                 with redirect_stdout(self._buffer):
                     fn()
@@ -76,8 +82,14 @@ class JobRunner:
             except SystemExit as exc:
                 if exc.code:
                     status, error = "failed", f"exited with code {exc.code}"
-        except Exception as exc:  # surfaced in the UI, never swallowed
+            except BaseException as exc:  # incl. GreenletExit et al: the
+                # runner must NEVER stay "running" — that wedges every
+                # future stage behind a 409 until a server restart
                 status, error = "failed", f"{type(exc).__name__}: {exc}"
+                # the short error names the exception; the log carries the
+                # traceback, or a failure is undiagnosable from the UI
+                self._buffer.write("\n" + traceback.format_exc())
+        finally:
             with self._lock:
                 self._status = status
                 self._error = error

src/bggpipe/resolve.py

@@ -475,6 +475,23 @@ def dedupe_matches(rows: list[dict], titles: list[TitleEntry]) -> list[MergeEven
                     bgg_id=bgg_id,
                 )
             )
+    # A survivor can itself lose in a later run; rows pointing at it would
+    # form a chain diff's one-level photo hop can't follow. Rewrite every
+    # merged row to its terminal survivor.
+    status_by_title = {r["title_raw"]: r for r in rows}
+    for row in rows:
+        if row["match_status"] != "merged":
+            continue
+        target, hops = row["merged_into"], 0
+        while (
+            hops < 10
+            and (nxt := status_by_title.get(target)) is not None
+            and nxt["match_status"] == "merged"
+            and nxt["merged_into"]
+        ):
+            target = nxt["merged_into"]
+            hops += 1
+        row["merged_into"] = target
     return events
 
 

src/bggpipe/static/app.js

@@ -16,8 +16,11 @@ function showBanner(html) {
 }
 
 function errorBanner(detail) {
-  showBanner(`<div class="banner error">lost contact with the server ` +
-             `(${esc(detail)}) — check its terminal</div>`);
+  // an HTTP status means the server answered — the problem is its data
+  const httpish = /^\d{3}$/.test(String(detail));
+  showBanner(`<div class="banner error">${httpish
+    ? `the server hit an error (HTTP ${esc(detail)}) — a data file may be broken`
+    : `lost contact with the server (${esc(detail)})`} — check its terminal</div>`);
 }
 
 async function fetchJSON(url) {

src/bggpipe/templates/pages/catalog.html

@@ -51,5 +51,5 @@ async function refresh() {
 
 document.getElementById("catsearch").addEventListener("input", render);
 refresh().catch(err => errorBanner(err.message || err));
-pollLoop(refresh, 5000);
+pollLoop(refresh, 5000, () => showBanner(""));
 </script>

src/bggpipe/templates/pages/library.html

@@ -57,5 +57,5 @@ async function refresh() {
 
 document.getElementById("libsearch").addEventListener("input", render);
 refresh().catch(err => errorBanner(err.message || err));
-pollLoop(refresh, 10000);
+pollLoop(refresh, 10000, () => showBanner(""));
 </script>

src/bggpipe/templates/pages/photos.html

@@ -87,15 +87,21 @@ async function sendPhotos(files) {
   if (!files.length) return;
   const form = new FormData();
   for (const f of files) form.append("files", f);
-  const res = await fetch("/api/photos", {method: "POST", body: form});
+  let res;
+  try {
+    res = await fetch("/api/photos", {method: "POST", body: form});
+  } catch (err) {
+    alert("Upload failed (no response from the server): " + err);
+    return;
+  }
   if (!res.ok) {
     const detail = await res.json().then(d => d.detail).catch(() => null);
     alert("Upload failed: " + (detail ?? res.statusText));
     return;
   }
-  refresh();
+  refresh().catch(() => {});  // the next poll self-heals a refresh hiccup
 }
 
 refresh().catch(err => errorBanner(err.message || err));
-pollLoop(refresh, 5000);
+pollLoop(refresh, 5000, () => showBanner(""));
 </script>

src/bggpipe/templates/pages/pipeline.html

@@ -11,7 +11,7 @@ let running = false;
 
 async function runStage(stage, body) {
   const res = await apiPost(`/api/run/${stage}`, body);
-  if (res) refresh();
+  if (res) refresh().catch(() => {});  // the poll self-heals a hiccup
 }
 
 function stageCard(num, name, facts, actions) {
@@ -28,6 +28,7 @@ function runBtn(stage, label) {
 
 function render() {
   const m = P.matches, log = P.upload_log;
+  running = P.job.status === "running";  // buttons below depend on this
 
   const banners = [];
   if (P.stub_data) banners.push(
@@ -67,7 +68,6 @@ function render() {
   });
 
   const job = P.job;
-  running = job.status === "running";
   const state = document.getElementById("jobstate");
   if (job.status === "idle") state.textContent = "idle";
   else state.innerHTML = `<span class="${esc(job.status)}">${esc(job.stage)}: ${esc(job.status)}</span>` +

src/bggpipe/templates/pages/queue.html

@@ -53,5 +53,5 @@ async function refresh() {
 }
 
 refresh().catch(err => errorBanner(err.message || err));
-pollLoop(refresh, 5000);
+pollLoop(refresh, 5000, () => showBanner(""));
 </script>

src/bggpipe/templates/pages/review.html

@@ -267,7 +267,10 @@ refresh().catch(err => errorBanner(err.message || err));
 let lastGood = null;
 pollLoop(async () => {
   const fresh = await fetchJSON("/api/state");
-  if (STATE && fresh.revision < STATE.revision) return;
+  // discard stale responses — but only within one server lifetime: a
+  // restart resets revision to 0 (boot changes), and refusing forever
+  // would freeze the page
+  if (STATE && fresh.boot === STATE.boot && fresh.revision < STATE.revision) return;
   if (JSON.stringify(fresh) !== JSON.stringify(STATE)) {
     STATE = fresh;
     render();

src/bggpipe/upload.py

@@ -48,6 +48,7 @@ UPLOAD_LOG_COLUMNS = [
     "collid",
     "name",
     "version_id",
+    "second_copy",
     "status",
     "timestamp",
     "error",
@@ -64,6 +65,7 @@ class UploadJob:
     version_id: str = ""
     version_name: str = ""
     collid: str = ""
+    second_copy: bool = False  # game had prior copies: verify can't confirm
 
     @property
     def key(self) -> tuple[str, str, str]:
@@ -135,6 +137,7 @@ def build_queue(
             name=row["bgg_name"],
             version_id=row.get("version_id", ""),
             version_name=row.get("version_name", ""),
+            second_copy=bool(row.get("second_copy")),
         )
         for row in to_add
     ] + [
@@ -161,11 +164,19 @@ def build_queue(
     deferred: list[UploadJob] = []
     update_game_seen: set[str] = set()
     seen: Counter[tuple[str, str, str]] = Counter()
+    queued_versions: dict[str, set[str]] = {}
+    for job in candidates:
+        if job.action == "add":
+            queued_versions.setdefault(job.bgg_id, set()).add(job.version_id)
     for job in candidates:
         prior = done_versions.get(
             ("update", job.collid) if job.action == "update" else ("add", job.bgg_id),
             set(),
         )
+        if job.action == "add" and prior & queued_versions.get(job.bgg_id, set()):
+            # the done version is still queued alongside this one: a
+            # multi-edition game partway through, not a re-review drift
+            prior = set()
         if prior and job.version_id not in prior:
             typer.echo(
                 f"  note: {job.name} was previously {job.action}ed with a "
@@ -303,6 +314,7 @@ class PlaywrightUploader:
                     "page changed — see docs/bgg-upload-flow.md)"
                 ) from err
             self._context.storage_state(path=str(self._storage_state))
+            self._storage_state.chmod(0o600)  # session cookies: owner-only
         self._authed = True
 
     def _open_dialog(self, opener):
@@ -444,6 +456,7 @@ def _process(
             "collid": job.collid,
             "name": job.name,
             "version_id": job.version_id,
+            "second_copy": "1" if job.second_copy else "",
             "status": status,
             "timestamp": now(),
             # the column carries degradation notes on successes too
@@ -488,6 +501,15 @@ def verify_uploads(log_rows: list[dict], collection: list[CollectionItem]) -> li
             added_copies[int(row["bgg_id"])] += 1
     for row in latest.values():
         if row["status"] in ("added", "added_no_version"):
+            if row.get("second_copy"):
+                # the game had copies before this add: collection state can't
+                # distinguish "new entry created" from "existing entry
+                # edited" (the failure the unverified dialog could produce)
+                problems.append(
+                    f"{row['name']}: second-copy add can't be verified from "
+                    "collection counts — confirm by eye on BGG"
+                )
+                continue
             copies = by_object.get(int(row["bgg_id"]), [])
             if not copies:
                 problems.append(f"{row['name']}: logged added but not in collection")

src/bggpipe/webreview.py

@@ -18,6 +18,7 @@ import io
 import json
 import os
 import threading
+import time
 import warnings
 import webbrowser
 import xml.etree.ElementTree as ET
@@ -26,17 +27,18 @@ from collections.abc import Callable
 from functools import partial
 from importlib import resources
 from pathlib import Path
+from urllib.parse import urlsplit
 
 import typer
 from defusedxml.ElementTree import fromstring as _safe_fromstring
 from fastapi import FastAPI, HTTPException, UploadFile
-from fastapi.responses import FileResponse, HTMLResponse, Response
+from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, Response
 from pydantic import BaseModel
 from rich.console import Console
 
 from bggpipe.bgg_client import BGGClient, cached_paths
 from bggpipe.config import DEFAULT_REVIEW_PORT, Config
-from bggpipe.fsio import atomic_write_text
+from bggpipe.fsio import atomic_write_bytes, atomic_write_text
 from bggpipe.jobs import JobRunner
 from bggpipe.models import (
     CONFIDENT_VERSION_STATUSES,
@@ -219,13 +221,39 @@ def create_app(
         input_fn=lambda prompt: "",
         client=client,
     )
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
         thumbnails = load_thumbnails(cfg.cache_dir)
         dismissed = DismissStore(cfg.dismissed_path)
+    startup_notes = [str(w.message) for w in caught]
     # FastAPI runs sync endpoints in a threadpool: without this, a polled
     # GET /api/state can freshen()-swap session.rows out from under a
     # concurrent decision POST, silently dropping the decision.
     lock = threading.Lock()
     revision = {"n": 0}  # bumped on every mutation and reload
+    boot = time.time()  # clients detect restarts (revision resets with us)
+    # server-side degradations, shown in-browser (quarantined dismiss file,
+    # unreadable thumbnails, torn artifacts)
+    app_warnings: list[str] = list(startup_notes)
+
+    ALLOWED_HOSTS = {"127.0.0.1", "localhost", "testserver"}
+
+    @app.middleware("http")
+    async def origin_guard(request, call_next):
+        # A hostile webpage can fire preflight-free cross-origin POSTs at a
+        # localhost server (bodyless run triggers, multipart photo posts).
+        # Mutations must come from us: same-host, and no foreign Origin.
+        if request.method not in ("GET", "HEAD", "OPTIONS"):
+            host = (request.headers.get("host") or "").split(":")[0]
+            origin = request.headers.get("origin")
+            origin_host = urlsplit(origin).hostname if origin else None
+            if host not in ALLOWED_HOSTS or (
+                origin_host is not None and origin_host not in ALLOWED_HOSTS
+            ):
+                return JSONResponse(
+                    {"detail": "cross-origin request refused"}, status_code=403
+                )
+        return await call_next(request)
 
     def freshen() -> None:
         """Serve every request from the current file state: an extract or
@@ -239,7 +267,18 @@ def create_app(
         available = photo_names()
         sightings = []
         if cfg.unidentified_path.exists():
-            for photo, entries in json.loads(cfg.unidentified_path.read_text()).items():
+            try:
+                data = json.loads(cfg.unidentified_path.read_text())
+            except (json.JSONDecodeError, OSError) as err:
+                # one torn file must not 500 every page and badge at once
+                note = (
+                    f"{cfg.unidentified_path.name} unreadable ({err}) — "
+                    "reshoot list unavailable"
+                )
+                if note not in app_warnings:
+                    app_warnings.append(note)
+                return []
+            for photo, entries in data.items():
                 for s in entries:
                     if _sighting_key(photo, s) in dismissed.keys:
                         continue
@@ -248,6 +287,17 @@ def create_app(
                     )
         return sightings
 
+    def read_games() -> dict:
+        if not cfg.games_path.exists():
+            return {}
+        try:
+            return json.loads(cfg.games_path.read_text())
+        except (json.JSONDecodeError, OSError) as err:
+            note = f"{cfg.games_path.name} unreadable ({err}) — library unavailable"
+            if note not in app_warnings:
+                app_warnings.append(note)
+            return {}
+
     def find_row(title_raw: str, source_photos: str, row_ix: int | None = None) -> dict:
         freshen()
         row = session.find_row(title_raw, source_photos, row_ix)
@@ -345,7 +395,8 @@ def create_app(
         ]
         return {
             "revision": revision["n"],
-            "warnings": session.warnings[-10:],
+            "boot": boot,
+            "warnings": (session.warnings + app_warnings)[-10:],
             "pending": [row_payload(r) for r in session.pending_rows()],
             "versions": [version_payload(r) for r in session.version_rows()],
             "merges": merges,
@@ -453,9 +504,7 @@ def create_app(
 
     @app.get("/api/library")
     def api_library() -> list[dict]:
-        if not cfg.games_path.exists():
-            return []
-        games = json.loads(cfg.games_path.read_text())
+        games = read_games()
         return sorted(games.values(), key=lambda g: (g.get("name") or "").casefold())
 
     def _csv_count(path: Path) -> int:
@@ -476,10 +525,9 @@ def create_app(
                     import csv as _csv
 
                     log_counts = Counter(row["status"] for row in _csv.DictReader(f))
-            games = 0
-            if cfg.games_path.exists():
-                games = len(json.loads(cfg.games_path.read_text()))
+            games = len(read_games())
             return {
+                "boot": boot,
                 # key NAMES only — never values (credentials stay out of
                 # every payload and log)
                 "env": {
@@ -532,12 +580,13 @@ def create_app(
                 raise HTTPException(400, f"not a photo: {name or '(unnamed)'}")
             target = cfg.photos_dir / name
             data = await upload_file.read()
-            target.write_bytes(data)
-            # a re-uploaded photo means "re-extract this one": drop its raw
-            # cache (regenerable) so the next extract run picks it up
+            # a re-uploaded photo means "re-extract this one": the raw cache
+            # goes FIRST, so a crash can never leave the new photo paired
+            # with the old photo's extraction
             stale = cfg.extract_raw_dir / f"{name}.json"
             if stale.exists():
                 stale.unlink()
+            atomic_write_bytes(target, data)
             saved.append(name)
         return {"saved": saved, "photos": len(photo_names())}
 
@@ -552,7 +601,20 @@ def create_app(
             revision["n"] += 1
             return _decide(body)
 
+    def _refuse_if_rewriting() -> None:
+        snap = jobs.snapshot()
+        if snap["status"] == "running" and snap["stage"] in ("extract", "resolve"):
+            # the job holds a start-of-run snapshot and will atomically
+            # rewrite matches.csv/titles.json at the end — a decision saved
+            # now would be silently reverted by that rewrite
+            raise HTTPException(
+                409,
+                f"a {snap['stage']} run is rewriting the pipeline data — "
+                "wait for it to finish, then decide",
+            )
+
     def _decide(body: DecisionBody) -> dict:
+        _refuse_if_rewriting()
         row = find_row(body.title_raw, body.source_photos, body.row_ix)
         if body.action == "pick":
             candidates = json.loads(row["candidates_json"] or "[]")
@@ -579,6 +641,7 @@ def create_app(
             return _version(body)
 
     def _version(body: VersionBody) -> dict:
+        _refuse_if_rewriting()
         row = find_row(body.title_raw, body.source_photos, body.row_ix)
         if body.action == "unknown":
             session.decide_version(row, None)
@@ -595,6 +658,7 @@ def create_app(
     def api_veto_merge(body: VetoBody) -> dict:
         with lock:
             revision["n"] += 1
+            _refuse_if_rewriting()
             row = find_row(body.title_raw, body.source_photos, body.row_ix)
             if row["match_status"] != "merged":
                 raise HTTPException(400, "row is not merged")

tests/test_config.py

@@ -8,8 +8,11 @@ from bggpipe.config import Config, load_config
 
 
 def test_defaults_when_no_file(tmp_path, monkeypatch):
+    # no config.toml in cwd and no explicit path -> defaults (an EXPLICIT
+    # missing path errors instead; see the dedicated test)
     monkeypatch.delenv("BGG_USERNAME", raising=False)
-    cfg = load_config(tmp_path / "missing.toml")
+    monkeypatch.chdir(tmp_path)
+    cfg = load_config()
     assert cfg == Config()
     assert cfg.cache_dir == Path("data/bgg_cache")
 
@@ -44,3 +47,10 @@ def test_unknown_toml_keys_warn(tmp_path, monkeypatch):
     p.write_text('photo_dir = "oops"\n')
     with pytest.warns(UserWarning, match="photo_dir"):
         load_config(p)
+
+
+def test_explicit_missing_config_errors_instead_of_silent_defaults(tmp_path):
+    import pytest
+
+    with pytest.raises(FileNotFoundError, match="does not exist"):
+        load_config(tmp_path / "nope.toml")

tests/test_diff.py

@@ -364,3 +364,36 @@ def test_live_diff_falls_back_to_snapshots_on_auth_failure(tmp_path, monkeypatch
     write_matches(cfg.matches_path, [_match("5 MINUTE DUNGEON", "207830")])
     result = run_diff(cfg, client=_LiveClient([], fail_auth=True))
     assert result.already_owned == ["5 MINUTE DUNGEON"]  # snapshots served
+
+
+def test_bare_sibling_of_confident_add_is_not_a_duplicate_upload():
+    # photo A reads "Wingspan" (confident version), photo B misreads
+    # "Wingspam" (bare) resolving to the same absent game: ONE add, not two
+    rows = [
+        _match("Wingspan", "266192", vstatus="version_auto", vid="1", vname="1st"),
+        _match("Wingspam", "266192"),
+    ]
+    result = compute_diff(rows, [])
+    assert len(result.to_add) == 1
+    assert result.already_owned == ["Wingspam"]
+
+
+def test_vetoed_bare_sibling_still_adds_for_absent_game():
+    vetoed = {**_match("Wingspan", "266192"), "dedupe_veto": "1"}
+    rows = [
+        _match("Wingspan", "266192", vstatus="version_auto", vid="1", vname="1st"),
+        vetoed,
+    ]
+    result = compute_diff(rows, [])
+    assert len(result.to_add) == 2
+    assert result.to_add[1]["second_copy"] == "1"
+
+
+def test_second_copy_flag_travels_on_exhausted_adds():
+    rows = [
+        _match("Catan", "13", vstatus="version_auto", vid="123", vname="3rd"),
+        _match("Catan", "13", vstatus="version_auto", vid="123", vname="3rd"),
+    ]
+    result = compute_diff(rows, [_item(13, 900, version_id=123)])
+    (added,) = result.to_add
+    assert added["second_copy"] == "1"

tests/test_extract.py

@@ -317,3 +317,35 @@ def test_low_confidence_reads_are_surfaced(tmp_path, capsys):
     assert "Low-confidence reads" in out
     assert "'Patchwork' (medium)" in out
     assert "'Catan'" not in out.split("Low-confidence reads")[1]
+
+
+def test_corrupt_raw_cache_is_reextracted_not_skipped(tmp_path):
+    cfg = _cfg(tmp_path)
+    cfg.photos_dir.mkdir()
+    _write_image(cfg.photos_dir / "shelf.jpg")
+    vision, calls = _vision_stub('[{"title_raw": "Catan", "confidence": "high"}]')
+    run_extract(cfg, vision=vision)
+    raw = cfg.extract_raw_dir / "shelf.jpg.json"
+    raw.write_text("{torn")  # corrupt the cache
+    run_extract(cfg, vision=vision)
+    assert len(calls) == 2  # re-extracted, not skipped
+    json.loads(raw.read_text())  # cache healed
+
+
+def test_systemic_failures_abort_and_exit_nonzero(tmp_path):
+    import pytest
+    import typer as _typer
+
+    calls = []
+
+    def broken_vision(image_b64, media_type):
+        calls.append(1)
+        raise RuntimeError("invalid x-api-key")
+
+    cfg = _cfg(tmp_path)
+    cfg.photos_dir.mkdir()
+    for i in range(6):
+        _write_image(cfg.photos_dir / f"p{i}.jpg")
+    with pytest.raises(_typer.Exit):
+        run_extract(cfg, vision=broken_vision)
+    assert len(calls) == 3  # aborted after 3 identical failures

tests/test_fsio.py

@@ -0,0 +1,49 @@
+"""fsio carries the project's central resumability promise: a kill or
+crash mid-write must never leave a torn file. These pin that promise
+directly — six modules depend on it."""
+
+from __future__ import annotations
+
+import os
+import stat
+
+import pytest
+
+from bggpipe.fsio import atomic_write_bytes, atomic_write_csv, atomic_write_text
+
+
+def test_failed_write_leaves_original_intact(tmp_path, monkeypatch):
+    target = tmp_path / "artifact.json"
+    target.write_text("precious")
+    # a read-only directory makes the tmp-file write fail
+    os.chmod(tmp_path, stat.S_IRUSR | stat.S_IXUSR)
+    try:
+        with pytest.raises(OSError):
+            atomic_write_text(target, "replacement")
+    finally:
+        os.chmod(tmp_path, 0o755)
+    assert target.read_text() == "precious"
+    assert not list(tmp_path.glob("*.tmp"))  # no leftovers
+
+
+def test_returned_mtime_matches_the_written_file(tmp_path):
+    target = tmp_path / "rows.csv"
+    mtime = atomic_write_csv(target, ["a", "b"], [{"a": "1", "b": "2"}])
+    # ReviewSession records this as "my own write" — it must be the mtime
+    # the file actually carries, or external-change detection breaks
+    assert mtime == target.stat().st_mtime_ns
+
+
+def test_bytes_variant_round_trips(tmp_path):
+    target = tmp_path / "photo.jpg"
+    atomic_write_bytes(target, b"\xff\xd8jpeg")
+    assert target.read_bytes() == b"\xff\xd8jpeg"
+    atomic_write_bytes(target, b"\xff\xd8jpeg2")  # overwrite is atomic too
+    assert target.read_bytes() == b"\xff\xd8jpeg2"
+
+
+def test_concurrent_writers_use_distinct_tmp_names(tmp_path):
+    from bggpipe.fsio import _tmp_for
+
+    target = tmp_path / "x.csv"
+    assert _tmp_for(target) != _tmp_for(target)  # no shared-inode interleave

tests/test_init.py

@@ -43,8 +43,8 @@ def test_creates_dirs_config_and_env_from_nothing(tmp_path, monkeypatch):
     assert (tmp_path / "photos").is_dir() and (tmp_path / "data").is_dir()
     assert (tmp_path / "config.toml").exists()
     env = (tmp_path / ".env").read_text()
-    assert "ANTHROPIC_API_KEY=sk-test-123" in env
-    assert "BGG_USERNAME=eric" in env
+    assert "ANTHROPIC_API_KEY='sk-test-123'" in env
+    assert "BGG_USERNAME='eric'" in env
     assert report.keys_written == ["ANTHROPIC_API_KEY", "BGG_USERNAME"]
     assert set(report.keys_missing) == {"BGG_PASSWORD", "BGG_API_TOKEN"}
 
@@ -117,3 +117,33 @@ def test_secret_values_never_appear_in_output(tmp_path, monkeypatch, capsys):
     _clear_env(monkeypatch)
     _run(tmp_path, answers={"BGG_PASSWORD": "s3cret-value-xyz"})
     assert "s3cret-value-xyz" not in capsys.readouterr().out
+
+
+def test_values_are_shell_quoted_for_source(tmp_path, monkeypatch):
+    # a password with spaces, $, and quotes must survive `source .env`
+    _clear_env(monkeypatch)
+    _run(tmp_path, answers={"BGG_PASSWORD": "pa$s wo'rd"})
+    env = (tmp_path / ".env").read_text()
+    assert "BGG_PASSWORD='pa$s wo'\\''rd'" in env
+
+
+def test_env_parsing_negatives(tmp_path, monkeypatch):
+    # commented, quoted-empty, and export-prefixed lines must parse sanely
+    _clear_env(monkeypatch)
+    (tmp_path / ".env").write_text(
+        "# BGG_PASSWORD=commented-out\n"
+        'ANTHROPIC_API_KEY=""\n'
+        "export BGG_USERNAME='eric'\n"
+    )
+    report = _run(tmp_path)
+    assert "BGG_USERNAME" in report.keys_ready  # export form recognized
+    assert "ANTHROPIC_API_KEY" in report.keys_missing  # quoted-empty ≠ set
+    assert "BGG_PASSWORD" in report.keys_missing  # comments don't count
+
+
+def test_env_file_is_owner_only_from_creation(tmp_path, monkeypatch):
+    import os as _os
+
+    _clear_env(monkeypatch)
+    _run(tmp_path, answers={"BGG_API_TOKEN": "tok-123"})
+    assert _os.stat(tmp_path / ".env").st_mode & 0o777 == 0o600

tests/test_resolve.py

@@ -660,3 +660,15 @@ def test_reordered_titles_json_cannot_mispair_editions(client, tmp_path):
     assert by_photos["b.jpg"]["match_status"] == "auto"  # kept its resolution
     assert "a.jpg" in by_photos  # newcomer resolved as its own row
     assert len(rows) == 2
+
+
+def test_merged_into_chains_resolve_to_terminal_survivor():
+    # X merged into Y in a prior run; this run merges Y into W — X must
+    # point at W, or diff's one-level photo hop loses X's provenance
+    x = _mrow("Wingspam", "266192", "x.jpg", status="merged")
+    x["merged_into"] = "Wingspan Typo"
+    y = _mrow("Wingspan Typo", "266192", "y.jpg", name="Wingspan")
+    w = _mrow("Wingspan", "266192", "w.jpg", name="Wingspan")
+    dedupe_matches([x, y, w], [])
+    assert y["match_status"] == "merged" and y["merged_into"] == "Wingspan"
+    assert x["merged_into"] == "Wingspan"  # chain collapsed

tests/test_upload.py

@@ -466,3 +466,28 @@ def test_run_upload_verify_wiring(tmp_path, capsys):
     )
     assert client.calls == [{"username": "tester", "refresh": True}]
     assert "Verification OK" in capsys.readouterr().out
+
+
+def test_verify_marks_second_copy_adds_unverifiable():
+    log = [
+        {**_log_row(action="add", bgg_id="13", status="added"), "second_copy": "1"},
+    ]
+    problems = verify_uploads(log, [_item(13, 900, version_id=7)])
+    (problem,) = problems
+    assert "can't be verified" in problem and "confirm by eye" in problem
+
+
+def test_drift_warning_suppressed_for_multi_edition_partial_run(tmp_path, capsys):
+    # run 1 added edition A; edition B is still queued alongside A's row:
+    # that's a two-edition game mid-way, not a re-review drift
+    cfg = _cfg(tmp_path)
+    _seed_data(
+        tmp_path,
+        to_add=[
+            _add_row(bgg_id="13", name="Catan", version_id="1", version_name="A"),
+            _add_row(bgg_id="13", name="Catan", version_id="2", version_name="B"),
+        ],
+        log=[_log_row(action="add", bgg_id="13", version_id="1", status="added")],
+    )
+    run_upload(cfg, uploader=FakeUploader(), sleep=lambda s: None, now=_now)
+    assert "manual correction" not in capsys.readouterr().out

tests/test_web_dashboard.py

@@ -4,6 +4,7 @@ stage functions are injected so nothing slow or networked ever runs."""
 from __future__ import annotations
 
 import threading
+import time
 
 import typer
 from fastapi.testclient import TestClient
@@ -315,3 +316,308 @@ def test_non_photo_files_are_invisible(tmp_path):
     web = _app(cfg)
     assert [p["name"] for p in web.get("/api/photos-list").json()] == ["shelf.jpg"]
     assert web.get("/photos/.DS_Store").status_code == 404
+
+
+# -- jobs: live-progress and termination contracts ----------------------
+
+
+def test_running_snapshot_shows_partial_line_then_finishes():
+    import typer as _typer
+
+    runner = JobRunner()
+    release = threading.Event()
+
+    def stage():
+        print("progress: 40%", end="", flush=True)  # no newline yet
+        release.wait()
+        _typer.echo(" done")
+
+    runner.start("extract", stage)
+    for _ in range(100):
+        snap = runner.snapshot()
+        if snap["log"]:
+            break
+        time.sleep(0.01)
+    assert snap["status"] == "running"
+    assert snap["log"] == ["progress: 40%"]
+    release.set()
+    runner.wait()
+    assert runner.snapshot()["log"] == ["progress: 40% done"]
+
+
+def test_log_serves_last_200_lines_and_buffer_is_bounded():
+    import typer as _typer
+
+    from bggpipe.jobs import MAX_LOG_LINES
+
+    runner = JobRunner()
+
+    def stage():
+        for i in range(MAX_LOG_LINES + 300):
+            _typer.echo(f"line {i}")
+
+    runner.start("extract", stage)
+    runner.wait()
+    log = runner.snapshot()["log"]
+    assert len(log) == 200
+    assert log[-1] == f"line {MAX_LOG_LINES + 299}"
+
+
+def test_zero_exit_codes_count_as_done():
+    import typer as _typer
+
+    for exc in (_typer.Exit(), SystemExit(0)):
+        runner = JobRunner()
+        runner.start("diff", lambda exc=exc: (_ for _ in ()).throw(exc))
+        runner.wait()
+        assert runner.snapshot()["status"] == "done"
+
+
+def test_nonzero_systemexit_is_failed():
+    runner = JobRunner()
+    runner.start("diff", lambda: (_ for _ in ()).throw(SystemExit(2)))
+    runner.wait()
+    snap = runner.snapshot()
+    assert snap["status"] == "failed" and "2" in snap["error"]
+
+
+def test_base_exception_cannot_wedge_the_runner():
+    class Rude(BaseException):
+        pass
+
+    runner = JobRunner()
+    runner.start("upload", lambda: (_ for _ in ()).throw(Rude("greenlet died")))
+    runner.wait()
+    snap = runner.snapshot()
+    assert snap["status"] == "failed"
+    assert "Rude" in snap["error"]
+    assert any("Traceback" in line for line in snap["log"])  # diagnosable
+    assert runner.start("extract", lambda: None)  # slot is free again
+    runner.wait()
+
+
+# -- concurrency and restart contracts ----------------------------------
+
+
+def test_revision_bumps_on_mutations_not_reads(tmp_path):
+    from bggpipe.resolve import write_matches
+
+    cfg = _cfg(tmp_path)
+    write_matches(
+        cfg.matches_path,
+        [
+            {
+                "title_raw": "Mystery",
+                "bgg_id": "",
+                "bgg_name": "",
+                "year": "",
+                "type": "",
+                "match_status": "unmatched",
+                "version_id": "",
+                "version_name": "",
+                "version_status": "",
+                "candidates_json": "[]",
+                "version_candidates_json": "[]",
+                "source_photos": "x.jpg",
+            }
+        ],
+    )
+    web = _app(cfg)
+    first = web.get("/api/state").json()
+    assert "boot" in first  # clients detect restarts by boot change
+    second = web.get("/api/state").json()
+    assert second["revision"] == first["revision"]  # reads never bump
+
+    res = web.post(
+        "/api/decision",
+        json={"title_raw": "Mystery", "source_photos": "x.jpg", "action": "reject"},
+    )
+    assert res.json()["revision"] > first["revision"]  # mutations bump
+
+    # external rewrite bumps on the next read
+    rows = []
+    write_matches(cfg.matches_path, rows)
+    bumped = web.get("/api/state").json()
+    assert bumped["revision"] > res.json()["revision"]
+
+
+def test_torn_data_files_degrade_with_warnings_not_500(tmp_path):
+    cfg = _cfg(tmp_path)
+    cfg.unidentified_path.write_text("{torn")
+    cfg.games_path.write_text("[not even close")
+    web = _app(cfg)
+    state = web.get("/api/state")
+    pipeline = web.get("/api/pipeline")
+    library = web.get("/api/library")
+    assert state.status_code == pipeline.status_code == library.status_code == 200
+    assert library.json() == []
+    assert any("unreadable" in w for w in state.json()["warnings"])
+
+
+def test_decisions_locked_out_while_matches_rewriting_job_runs(tmp_path):
+    from bggpipe.resolve import read_matches, write_matches
+
+    cfg = _cfg(tmp_path)
+    write_matches(
+        cfg.matches_path,
+        [
+            {
+                "title_raw": "Mystery",
+                "bgg_id": "",
+                "bgg_name": "",
+                "year": "",
+                "type": "",
+                "match_status": "unmatched",
+                "version_id": "",
+                "version_name": "",
+                "version_status": "",
+                "candidates_json": "[]",
+                "version_candidates_json": "[]",
+                "source_photos": "x.jpg",
+            }
+        ],
+    )
+    release = threading.Event()
+    jobs = JobRunner()
+    web = _app(cfg, stages={"resolve": release.wait}, jobs=jobs)
+    web.post("/api/run/resolve")
+    res = web.post(
+        "/api/decision",
+        json={"title_raw": "Mystery", "source_photos": "x.jpg", "action": "reject"},
+    )
+    assert res.status_code == 409
+    assert "resolve run is rewriting" in res.json()["detail"]
+    # the decision was NOT applied
+    assert read_matches(cfg.matches_path)[0]["match_status"] == "unmatched"
+    release.set()
+    jobs.wait()
+
+
+def test_cross_origin_mutations_are_refused(tmp_path):
+    web = _app(_cfg(tmp_path))
+    # foreign Origin on a same-host request: refused
+    res = web.post("/api/run/extract", headers={"origin": "https://evil.example"})
+    assert res.status_code == 403
+    # foreign Host (DNS rebinding): refused
+    res = web.post(
+        "/api/photos",
+        headers={"host": "evil.example"},
+        files={"files": ("a.jpg", b"x", "image/jpeg")},
+    )
+    assert res.status_code == 403
+    # reads are unaffected
+    assert (
+        web.get("/api/pipeline", headers={"origin": "https://evil.example"}).status_code
+        == 200
+    )
+
+
+def test_concurrent_decisions_and_polls_never_lose_a_decision(tmp_path):
+    from bggpipe.resolve import read_matches, write_matches
+
+    cfg = _cfg(tmp_path)
+    base = {
+        "bgg_id": "",
+        "bgg_name": "",
+        "year": "",
+        "type": "",
+        "match_status": "unmatched",
+        "version_id": "",
+        "version_name": "",
+        "version_status": "",
+        "candidates_json": "[]",
+        "version_candidates_json": "[]",
+        "source_photos": "x.jpg",
+    }
+    rows = [{**base, "title_raw": f"Game{i}"} for i in range(8)]
+    write_matches(cfg.matches_path, rows)
+    web = _app(cfg)
+
+    def decide(i):
+        web.post(
+            "/api/decision",
+            json={
+                "title_raw": f"Game{i}",
+                "source_photos": "x.jpg",
+                "action": "reject",
+            },
+        )
+
+    def poll():
+        web.get("/api/state")
+
+    threads = [threading.Thread(target=decide, args=(i,)) for i in range(8)]
+    threads += [threading.Thread(target=poll) for _ in range(8)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+    saved = read_matches(cfg.matches_path)
+    assert all(r["match_status"] == "rejected" for r in saved)
+
+
+def test_decision_for_vanished_row_is_404_and_touches_nothing(tmp_path):
+    from bggpipe.resolve import read_matches, write_matches
+
+    cfg = _cfg(tmp_path)
+    row = {
+        "title_raw": "Kept",
+        "bgg_id": "",
+        "bgg_name": "",
+        "year": "",
+        "type": "",
+        "match_status": "unmatched",
+        "version_id": "",
+        "version_name": "",
+        "version_status": "",
+        "candidates_json": "[]",
+        "version_candidates_json": "[]",
+        "source_photos": "x.jpg",
+    }
+    write_matches(cfg.matches_path, [row])
+    web = _app(cfg)
+    res = web.post(
+        "/api/decision",
+        json={"title_raw": "Gone", "source_photos": "x.jpg", "action": "reject"},
+    )
+    assert res.status_code == 404
+    assert read_matches(cfg.matches_path)[0]["match_status"] == "unmatched"
+
+
+def test_pipeline_reports_badge_fields(tmp_path):
+    from bggpipe.resolve import write_matches
+
+    cfg = _cfg(tmp_path)
+    base = {
+        "bgg_id": "13",
+        "bgg_name": "Catan",
+        "year": "",
+        "type": "boardgame",
+        "version_id": "",
+        "version_name": "",
+        "candidates_json": "[]",
+        "version_candidates_json": "[]",
+        "source_photos": "x.jpg",
+    }
+    write_matches(
+        cfg.matches_path,
+        [
+            {
+                **base,
+                "title_raw": "A",
+                "match_status": "ambiguous",
+                "version_status": "",
+            },
+            {
+                **base,
+                "title_raw": "B",
+                "match_status": "auto",
+                "version_status": "version_ambiguous",
+                "version_id": "1",
+            },
+        ],
+    )
+    cfg.to_add_path.write_text("bgg_id,bgg_name\n1,X\n2,Y\n")
+    p = _app(cfg).get("/api/pipeline").json()
+    assert p["pending_review"] == 2  # one match + one edition decision
+    assert p["to_add"] == 2  # header excluded