Re-audit round 3: 5 blind reviewers, 15 fixes, +18 tests — converging

Round 3's two HIGHs: _fill_version resolved versions with the LAST
same-title entry's cues (photo-aware lookup existed since round 1 but
this caller never used it), and the round-2 diff rework let an earlier
row's disagreement consume the exact-version copy a later row matched.
Diff claims now settle strongest-first across all rows (exact matches,
then versionless upgrades, then disagreement/second-copy), unvetoed
bare duplicates stay owned per spec, and updates are withheld with a
manual-fix note whenever any copy of the game already carries a version
(the row edit targets by name and could hit the wrong copy).

Also: entry-to-row pairing matches by photo overlap before position
(titles.json order churn from reshoot filenames could swap editions);
BGGQueueTimeout defers a title like a missing token; DismissStore
writes atomically, mutates memory only after the write, and
quarantines a torn file instead of bricking the server; version-picker
page-limit exhaustion stays retryable; verify's copy-count shortfall
reports once per game (the old guard was dead code); the upload log
header is created atomically; transient version-lookup failures record
a retryable version_error, not terminal version_unknown; extract
isolates per-photo failures and salvages JSON followed by prose; a
state revision counter stops stale poll responses reverting decisions;
plus the shared-predicate/fsio/docstring consolidation and CLI wiring,
live-diff, verify-wiring, and search-guard tests.

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

src/bggpipe/bgg_client.py

@@ -196,3 +196,9 @@ class BGGClient:
 def client_for(cfg: Config) -> BGGClient:
     """The standard injection fallback: every stage's `client or client_for(cfg)`."""
     return BGGClient(cfg.cache_dir, cfg.rate_limit_seconds)
+
+
+def cached_paths(cache_dir: Path, endpoint: str) -> list[Path]:
+    """Cache files for one endpoint — the ONLY sanctioned way to glob the
+    cache, so the filename layout stays private to cache_key."""
+    return sorted(cache_dir.glob(f"{endpoint}_*.xml"))

src/bggpipe/config.py

@@ -70,6 +70,19 @@ class Config:
     def dismissed_path(self) -> Path:
         return self.data_dir / "unidentified_dismissed.json"
 
+    @property
+    def snapshot_paths(self) -> tuple[Path, Path]:
+        return (
+            self.data_dir / "collection_snapshot_base.xml",
+            self.data_dir / "collection_snapshot_expansions.xml",
+        )
+
+    @property
+    def storage_state_path(self) -> Path:
+        # cwd-relative on purpose (credential-adjacent, gitignored) but
+        # centralized here with every other artifact path
+        return Path("storage_state.json")
+
     @property
     def stub_marker_paths(self) -> tuple[Path, Path]:
         # gitignored (travels with the stub XML) + committed (guards clones)

src/bggpipe/diff.py

@@ -18,7 +18,6 @@ Outputs both artifacts:
 
 from __future__ import annotations
 
-import csv
 import os
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -27,10 +26,11 @@ import typer
 
 from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
 from bggpipe.config import Config
+from bggpipe.fsio import atomic_write_csv
 from bggpipe.models import (
-    CONFIDENT_VERSION_STATUSES,
     RECOGNIZED_MATCH_STATUSES,
     CollectionItem,
+    is_confident_version,
     parse_collection,
 )
 from bggpipe.resolve import read_matches
@@ -69,8 +69,7 @@ def load_snapshot_collection(data_dir: Path) -> list[CollectionItem]:
     same physical copy can appear in both responses)."""
     items: list[CollectionItem] = []
     seen: set[int] = set()
-    for name in SNAPSHOT_FILES:
-        path = data_dir / name
+    for path in Config(data_dir=data_dir).snapshot_paths:
         if not path.exists():
             raise FileNotFoundError(
                 f"{path} not found — pull your collection while logged in "
@@ -137,30 +136,50 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
             c for c in by_object.get(bgg_id, []) if c.coll_id not in consumed_collids
         ]
 
-    def is_confident(row: dict) -> bool:
-        return bool(
-            row["version_status"] in CONFIDENT_VERSION_STATUSES and row["version_id"]
-        )
+    # Ordered sub-passes over the confident rows. Greedy per-row handling
+    # let an EARLIER row's disagreement consume the exact-version copy a
+    # LATER row matched — producing a duplicate upload. Claims must settle
+    # strongest-first across ALL rows: exact version matches, then
+    # versionless upgrades, then disagreement/second-copy handling.
+    confident_rows = [r for r in recognized if is_confident_version(r)]
+    leftover: list[dict] = []
 
-    # Pass 1 — confident-version rows claim copies first (an exact version
-    # match, then a versionless copy to upgrade). Bare rows must not steal
-    # a versionless copy a confident row would have upgraded.
-    for row in (r for r in recognized if is_confident(r)):
+    # 1a — exact (bgg_id, version) matches consume first
+    for row in confident_rows:
         bgg_id = int(row["bgg_id"])
-        version_id = int(row["version_id"])
-        remaining = unconsumed(bgg_id)
         if not by_object.get(bgg_id):
             result.to_add.append(add_row(row, True))
             continue
-        matching = [c for c in remaining if c.version_id == version_id]
+        matching = [
+            c for c in unconsumed(bgg_id) if c.version_id == int(row["version_id"])
+        ]
         if matching:
-            # exact (bgg_id, version) pair: consume, so a SECOND row with
-            # the same version (a vetoed duplicate = a real second copy)
-            # falls through to the branches below instead of vanishing
+            # consume, so a SECOND row with the same version (a vetoed
+            # duplicate = a real second copy) falls through to 1b/1c
             consumed_collids.add(matching[0].coll_id)
             result.already_owned.append(row["title_raw"])
+        else:
+            leftover.append(row)
+
+    # 1b — versionless copies get upgraded. Guard: upload's row-edit flow
+    # targets rows by game NAME (a collid can't drive the UI), so an update
+    # is only safe when EVERY copy of the game is versionless — otherwise
+    # the browser could open the versioned copy and overwrite it.
+    still_left: list[dict] = []
+    for row in leftover:
+        bgg_id = int(row["bgg_id"])
+        versionless = [c for c in unconsumed(bgg_id) if c.version_id is None]
+        any_versioned = any(c.version_id is not None for c in by_object.get(bgg_id, []))
+        if versionless and any_versioned:
+            consumed_collids.add(versionless[0].coll_id)
+            result.already_owned.append(row["title_raw"])
+            result.disagreements.append(
+                f"{row['title_raw']}: a versionless copy could take version "
+                f"{row['version_name']!r}, but another copy already carries "
+                "a version — set it by hand on BGG (the automated row edit "
+                "can't safely target a specific copy)"
+            )
             continue
-        versionless = [c for c in remaining if c.version_id is None]
         if versionless:
             target = versionless[0]
             consumed_collids.add(target.coll_id)
@@ -174,8 +193,14 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
                     "version_name": row["version_name"],
                 }
             )
-        elif remaining:
-            # an unclaimed copy exists but carries a DIFFERENT version:
+        else:
+            still_left.append(row)
+
+    # 1c — what remains disagrees with an unclaimed copy (report-only) or
+    # is an additional physical copy (every copy claimed by another row)
+    for row in still_left:
+        remaining = unconsumed(int(row["bgg_id"]))
+        if remaining:
             # most likely the same physical box mis-scored — report, never
             # touch, never duplicate (spec: report the disagreement)
             consumed_collids.add(remaining[0].coll_id)
@@ -187,9 +212,6 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
                 "left untouched"
             )
         else:
-            # every copy is claimed by other match rows: this row is an
-            # additional physical copy (spec: a pair is owned only when a
-            # collection item matches both ids)
             result.to_add.append(add_row(row, True))
             result.second_copies.append(
                 f"{row['title_raw']}: adding as a NEW copy with version "
@@ -197,25 +219,31 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
                 "existing entry of this game keeps its current version"
             )
 
-    # Pass 2 — bare (version-unknown) rows: owned while unclaimed copies
-    # remain; extras beyond the owned count (vetoed duplicates) are added
-    # as version-less new entries.
-    for row in (r for r in recognized if not is_confident(r)):
+    # Pass 2 — bare (version-unknown) rows. Spec: a bare id is owned if ANY
+    # copy exists — only a human veto (dedupe_veto) makes an extra bare row
+    # a genuine additional copy.
+    for row in (r for r in recognized if not is_confident_version(r)):
         bgg_id = int(row["bgg_id"])
-        if not by_object.get(bgg_id):
+        copies = by_object.get(bgg_id, [])
+        if not copies:
             result.to_add.append(add_row(row, False))
             continue
         remaining = unconsumed(bgg_id)
         if remaining:
             consumed_collids.add(remaining[0].coll_id)
             result.already_owned.append(row["title_raw"])
-        else:
+        elif row.get("dedupe_veto"):
             result.to_add.append(add_row(row, False))
             result.second_copies.append(
                 f"{row['title_raw']}: adding as a NEW version-less copy — "
-                "every existing entry of this game is claimed by another "
-                "match row"
+                "human-vetoed duplicate, every existing entry claimed by "
+                "another match row"
             )
+        else:
+            # unvetoed bare row, all copies claimed: per spec still owned
+            # (a typo-read sibling of a confident row must not become a
+            # spurious upload)
+            result.already_owned.append(row["title_raw"])
 
     result.unseen = [
         item for item in collection if item.object_id not in seen_object_ids
@@ -224,13 +252,7 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
 
 
 def _write_csv(path: Path, columns: list[str], rows: list[dict]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    tmp = path.with_name(path.name + ".tmp")
-    with tmp.open("w", newline="") as f:
-        writer = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore")
-        writer.writeheader()
-        writer.writerows(rows)
-    os.replace(tmp, path)  # atomic: a killed diff never leaves a torn queue
+    atomic_write_csv(path, columns, rows)  # a killed diff never tears the queue
 
 
 def run_diff(cfg: Config, *, client: BGGClient | None = None) -> DiffResult:

src/bggpipe/enrich.py

@@ -22,17 +22,14 @@ import typer
 from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
 from bggpipe.config import Config
 from bggpipe.fsio import atomic_write_text
-from bggpipe.models import (
-    CONFIDENT_VERSION_STATUSES,
-    RECOGNIZED_MATCH_STATUSES,
-)
+from bggpipe.models import is_confident_version, is_recognized
 from bggpipe.resolve import read_matches
 
 BATCH_SIZE = 20
 
 
 def _version_info(row: dict) -> dict | None:
-    if row["version_status"] not in CONFIDENT_VERSION_STATUSES or not row["version_id"]:
+    if not is_confident_version(row):
         return None
     version_id = int(row["version_id"])
     for cand in json.loads(row["version_candidates_json"] or "[]"):
@@ -64,7 +61,7 @@ def run_enrich(
 
     targets: list[tuple[str, int, dict | None]] = []
     for row in rows:
-        if row["match_status"] not in RECOGNIZED_MATCH_STATUSES or not row["bgg_id"]:
+        if not is_recognized(row):
             continue
         version = _version_info(row)
         key = f"{row['bgg_id']}:{version['version_id']}" if version else row["bgg_id"]

src/bggpipe/extract.py

@@ -111,6 +111,11 @@ def parse_vision_response(text: str) -> tuple[list[dict], list[dict], int]:
     A bare JSON array (the pre-unidentified response shape) still parses —
     it's all titles."""
     cleaned = _CODE_FENCE.sub("", text).strip()
+    if cleaned[:1] in ("[", "{"):
+        # trim trailing prose after a leading JSON payload ("{...}\nNote:")
+        end = cleaned.rfind("]" if cleaned[0] == "[" else "}")
+        if end != -1:
+            cleaned = cleaned[: end + 1]
     if cleaned[:1] not in ("[", "{"):
         starts = [i for i in (cleaned.find("["), cleaned.find("{")) if i != -1]
         if not starts:
@@ -309,12 +314,18 @@ def run_extract(
     raw_dir.mkdir(parents=True, exist_ok=True)
     vision = vision or default_vision(cfg.model)
 
+    failed: list[str] = []
     for photo in photos:
         raw_path = raw_dir / f"{photo.name}.json"
         if raw_path.exists() and not only and not force:
             typer.echo(f"  {photo.name}: already extracted, skipping")
             continue
-        result = extract_photo(photo, vision)
+        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")
+            continue
         atomic_write_text(
             raw_path, json.dumps(result, indent=2, ensure_ascii=False) + "\n"
         )
@@ -336,6 +347,11 @@ def run_extract(
         raw_dir, cfg.titles_path, cfg.unidentified_path
     )
     typer.echo(f"Wrote {len(deduped)} unique title(s) to {cfg.titles_path}.")
+    if failed:
+        typer.echo(
+            f"\n{len(failed)} photo(s) failed extraction (re-run to retry): "
+            + ", ".join(failed)
+        )
 
     if unidentified:
         typer.echo(

src/bggpipe/fsio.py

@@ -7,6 +7,7 @@ available to JSON artifacts and the XML response cache.
 
 from __future__ import annotations
 
+import csv
 import os
 from pathlib import Path
 
@@ -16,3 +17,18 @@ def atomic_write_text(path: Path, text: str) -> None:
     tmp = path.with_name(path.name + ".tmp")
     tmp.write_text(text)
     os.replace(tmp, path)
+
+
+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")
+    with tmp.open("w", newline="") as f:
+        writer = csv.DictWriter(
+            f, fieldnames=columns, extrasaction="ignore", restval=""
+        )
+        writer.writeheader()
+        writer.writerows(rows)
+    os.replace(tmp, path)
+    return path.stat().st_mtime_ns

src/bggpipe/models.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-import xml.etree.ElementTree as ET  # element types only; parsing goes via defusedxml
+import warnings
+import xml.etree.ElementTree as ET  # parsing itself goes via defusedxml
 from dataclasses import dataclass, field
 
 from defusedxml.ElementTree import fromstring as _safe_fromstring
@@ -17,6 +18,20 @@ class BGGResponseError(Exception):
 # human approved it, and a row reaches diff/enrich only when its match did.
 CONFIDENT_VERSION_STATUSES = ("version_auto", "version_approved")
 RECOGNIZED_MATCH_STATUSES = ("auto", "approved")
+PENDING_MATCH_STATUSES = ("ambiguous", "unmatched")
+UNDECIDED_MATCH_STATUSES = ("ambiguous", "unmatched", "merged")
+
+
+def is_recognized(row: dict) -> bool:
+    """A row that reaches diff/enrich: matched and carrying a real id."""
+    return row["match_status"] in RECOGNIZED_MATCH_STATUSES and bool(row["bgg_id"])
+
+
+def is_confident_version(row: dict) -> bool:
+    """A version trusted for upload: auto-scored or human-approved, with id."""
+    return row["version_status"] in CONFIDENT_VERSION_STATUSES and bool(
+        row["version_id"]
+    )
 
 
 @dataclass(frozen=True)
@@ -94,6 +109,11 @@ def parse_search(xml_text: str) -> list[SearchResult]:
                 type=item.get("type", "boardgame"),
             )
         )
+    if skipped and results:
+        warnings.warn(
+            f"search: {skipped} unparseable item(s) tolerated — schema drift?",
+            stacklevel=2,
+        )
     if skipped and not results:
         raise BGGResponseError(
             f"search response had {skipped} item(s), none parseable — "

src/bggpipe/resolve.py

@@ -2,15 +2,16 @@
 
 Reads data/titles.json, queries BGG search (+ thing stats for tie-breaks,
 + versions once a game is settled), classifies each title auto/ambiguous/
-unmatched, and appends rows to data/matches.csv. Re-runs skip titles
-already present in matches.csv unless --force.
+unmatched, then post-dedupes rows resolving to the same physical game
+(losers become match_status="merged"; review can veto). The whole file is
+rewritten atomically each run; re-runs pair entries to their existing rows
+(photo overlap, then position) unless --force starts over.
 """
 
 from __future__ import annotations
 
 import csv
 import json
-import os
 import re
 from collections import Counter
 from dataclasses import dataclass, field
@@ -19,13 +20,14 @@ from pathlib import Path
 import typer
 from rapidfuzz import fuzz
 
-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.extract import cues_conflict
+from bggpipe.fsio import atomic_write_csv
 from bggpipe.models import (
-    CONFIDENT_VERSION_STATUSES,
     RECOGNIZED_MATCH_STATUSES,
     GameVersion,
+    is_confident_version,
 )
 from bggpipe.normalize import normalize_title
 
@@ -442,10 +444,10 @@ def dedupe_matches(rows: list[dict], titles: list[TitleEntry]) -> list[MergeEven
             # re-running resolve must never overturn that (spec: re-runs
             # lose no work, least of all review decisions)
             continue
-        confident = (
-            row["version_status"] in CONFIDENT_VERSION_STATUSES and row["version_id"]
+        key = (
+            row["bgg_id"],
+            row["version_id"] if is_confident_version(row) else "",
         )
-        key = (row["bgg_id"], row["version_id"] if confident else "")
         groups.setdefault(key, []).append(row)
 
     events: list[MergeEvent] = []
@@ -487,17 +489,11 @@ def read_matches(path: Path) -> list[dict[str, str]]:
     return rows
 
 
-def write_matches(path: Path, rows: list[dict[str, str]]) -> None:
-    """Atomic full rewrite — review updates rows in place decision by decision."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    tmp = path.with_name(path.name + ".tmp")
-    with tmp.open("w", newline="") as f:
-        writer = csv.DictWriter(
-            f, fieldnames=MATCH_COLUMNS, extrasaction="ignore", restval=""
-        )
-        writer.writeheader()
-        writer.writerows(rows)
-    os.replace(tmp, path)
+def write_matches(path: Path, rows: list[dict[str, str]]) -> int:
+    """Atomic full rewrite — review updates rows in place decision by
+    decision. Returns the written file's mtime_ns so the caller can record
+    its own write without a re-stat race."""
+    return atomic_write_csv(path, MATCH_COLUMNS, rows)
 
 
 def run_resolve(
@@ -509,15 +505,43 @@ def run_resolve(
     existing_rows = read_matches(cfg.matches_path)
     client = client or client_for(cfg)
 
-    # Pair entries with existing rows BY TITLE, positionally, not by exact
-    # (title, photos) key: extract unions a new photo of an already-resolved
-    # game into its entry, and that must update the row's provenance — not
-    # re-resolve the game as a duplicate row. Same-title entries only stay
-    # separate when their cues conflict (two editions), and those pair up
-    # in stable file order on both sides.
+    # Pair entries with existing rows BY TITLE: photo-overlap first, then
+    # position. Pure position breaks when titles.json order churns (a
+    # reshoot photo sorting earlier reorders same-title entries); overlap
+    # keeps each edition glued to its own row, and position only settles
+    # entries with no photo history.
     rows_by_title: dict[str, list[dict]] = {}
     for row in existing_rows:
         rows_by_title.setdefault(row["title_raw"], []).append(row)
+
+    claimed_rows: set[int] = set()
+
+    def pair_row(entry: TitleEntry) -> dict | None:
+        candidates = [
+            r
+            for r in rows_by_title.get(entry.title_raw, [])
+            if id(r) not in claimed_rows
+        ]
+        if not candidates:
+            return None
+        photos = set(entry.source_photos)
+        for r in candidates:
+            if photos & set(r["source_photos"].split(";")):
+                claimed_rows.add(id(r))
+                return r
+        return None
+
+    def pair_row_positional(entry: TitleEntry) -> dict | None:
+        candidates = [
+            r
+            for r in rows_by_title.get(entry.title_raw, [])
+            if id(r) not in claimed_rows
+        ]
+        if candidates:
+            claimed_rows.add(id(candidates[0]))
+            return candidates[0]
+        return None
+
     seen_per_title: dict[str, int] = {}
 
     new_rows: list[MatchRow] = []
@@ -525,32 +549,48 @@ def run_resolve(
     photos_updated = False
     blocked: list[str] = []
     blocked_titles: set[str] = set()
+    # overlap pass first so a reordered titles.json can't mispair editions
+    paired_by_id: dict[int, dict] = {}
     for entry in entries:
-        ix = seen_per_title.get(entry.title_raw, 0)
-        seen_per_title[entry.title_raw] = ix + 1
-        paired = rows_by_title.get(entry.title_raw, [])
-        if entry.title_raw in blocked_titles and ix >= len(paired):
-            # an earlier same-title entry is waiting on the token: resolving
-            # this one now would append a row at the wrong position and
-            # corrupt next run's positional pairing — defer the whole group
-            blocked.append(entry.title_raw)
-            continue
-        if ix < len(paired):
-            row_dict = paired[ix]
+        row_dict = pair_row(entry)
+        if row_dict is not None:
+            paired_by_id[id(entry)] = row_dict
+    for entry in entries:
+        if id(entry) not in paired_by_id:
+            row_dict = pair_row_positional(entry)
+            if row_dict is not None:
+                paired_by_id[id(entry)] = row_dict
+
+    for entry in entries:
+        seen_per_title[entry.title_raw] = seen_per_title.get(entry.title_raw, 0) + 1
+        row_dict = paired_by_id.get(id(entry))
+        if row_dict is not None:
             photos = ";".join(entry.source_photos)
             if row_dict["source_photos"] != photos:
                 row_dict["source_photos"] = photos
                 photos_updated = True
             skipped += 1
             continue
+        if entry.title_raw in blocked_titles:
+            # an earlier same-title entry is waiting on the token: resolving
+            # this one now would claim the wrong pairing slot on the next
+            # run — defer the whole group
+            blocked.append(entry.title_raw)
+            continue
         try:
             row = resolve_entry(client, entry)
-        except BGGAuthError:
-            # No API token yet: cached titles still resolve; the rest wait.
-            # No row is written, so a future run picks them up untouched.
+        except (BGGAuthError, BGGQueueTimeout) as err:
+            # No token / BGG still queueing: cached titles still resolve;
+            # the rest wait. No row is written, so a future run picks them
+            # up untouched.
             blocked.append(entry.title_raw)
             blocked_titles.add(entry.title_raw)
-            typer.echo(f"  {entry.title_raw!r} -> waiting on BGG API token")
+            reason = (
+                "waiting on BGG API token"
+                if isinstance(err, BGGAuthError)
+                else "BGG still queueing — re-run in a minute"
+            )
+            typer.echo(f"  {entry.title_raw!r} -> {reason}")
             continue
         new_rows.append(row)
         detail = f"{row.bgg_name} ({row.bgg_id})" if row.bgg_id else "-"

src/bggpipe/review.py

@@ -23,7 +23,11 @@ from rich.table import Table
 
 from bggpipe.bgg_client import BGGAuthError, BGGClient, BGGQueueTimeout, client_for
 from bggpipe.config import Config
-from bggpipe.models import BGGResponseError
+from bggpipe.models import (
+    PENDING_MATCH_STATUSES,
+    UNDECIDED_MATCH_STATUSES,
+    BGGResponseError,
+)
 from bggpipe.resolve import (
     MatchRow,
     TitleEntry,
@@ -128,8 +132,16 @@ class ReviewSession:
         undecided = [
             i
             for i in candidates
-            if self.rows[i]["match_status"] in ("ambiguous", "unmatched", "merged")
+            if self.rows[i]["match_status"] in UNDECIDED_MATCH_STATUSES
         ]
+        if not undecided:
+            # a VERSION decision targets an approved row: prefer the slot
+            # still awaiting its edition, not the sibling already versioned
+            undecided = [
+                i
+                for i in candidates
+                if self.rows[i]["version_status"] == "version_ambiguous"
+            ]
         self.rows[(undecided or candidates)[0]] = row
         return True
 
@@ -147,11 +159,13 @@ class ReviewSession:
             )
             return
         try:
-            write_matches(self.cfg.matches_path, self.rows)
+            own_mtime = write_matches(self.cfg.matches_path, self.rows)
         except OSError:
             self._load()  # memory must never claim what disk doesn't hold
             raise
-        self._loaded_mtimes = self._data_mtimes()  # own writes aren't "changes"
+        # record the mtime write_matches itself observed: re-statting later
+        # could adopt a foreign rewrite landing in the gap as "own write"
+        self._loaded_mtimes = (own_mtime, self._data_mtimes()[1])
         self.decisions += 1
 
     def _apply_choice(self, row: dict, candidate: dict) -> None:
@@ -166,7 +180,10 @@ class ReviewSession:
     def _fill_version(self, row: dict) -> None:
         """Try version resolution for a just-approved row. Degrades gracefully:
         no cues, no token, or API trouble all leave version_unknown."""
-        entry = self._titles.get(row["title_raw"])
+        # photo-aware lookup: two same-title entries are two EDITIONS with
+        # different cues — the title-only dict would hand every row the last
+        # edition's cues and score the wrong version to version_auto
+        entry = self.cues_for(row["title_raw"], row["source_photos"])
         if entry is None or not row["bgg_id"]:
             row["version_status"] = row["version_status"] or "version_unknown"
             return
@@ -174,8 +191,13 @@ class ReviewSession:
         try:
             resolve_version(self.client, entry, shim)
         except _BGG_ERRORS as err:
-            self._warn(f"version lookup unavailable ({err}) — recorded version_unknown")
-            row["version_status"] = "version_unknown"
+            self._warn(
+                f"version lookup unavailable ({err}) — re-approve this row "
+                "to retry the edition lookup"
+            )
+            # distinct from version_unknown ("no cues"): a transient failure
+            # stays visibly retryable instead of terminal
+            row["version_status"] = "version_error"
             return
         row["version_status"] = shim.version_status
         row["version_id"] = str(shim.version_id or "")
@@ -208,6 +230,27 @@ class ReviewSession:
         row["dedupe_veto"] = "1"  # persists: resolve re-runs must not re-merge
         self._save(row)
 
+    def find_row(
+        self, title_raw: str, source_photos: str, row_ix: int | None = None
+    ) -> dict | None:
+        """Locate a row by ordinal (duplicate two-edition rows) or by key,
+        preferring a still-undecided slot on key collisions. None = gone."""
+        if row_ix is not None and 0 <= row_ix < len(self.rows):
+            row = self.rows[row_ix]
+            if row["title_raw"] == title_raw and row["source_photos"] == source_photos:
+                return row
+        matches = [
+            row
+            for row in self.rows
+            if row["title_raw"] == title_raw and row["source_photos"] == source_photos
+        ]
+        undecided = [
+            row for row in matches if row["match_status"] in UNDECIDED_MATCH_STATUSES
+        ]
+        if undecided or matches:
+            return (undecided or matches)[0]
+        return None
+
     def cues_for(
         self, title_raw: str, source_photos: str | None = None
     ) -> TitleEntry | None:
@@ -299,7 +342,7 @@ class ReviewSession:
                 self.decide_pick(row, candidates[int(answer) - 1])
                 return
             if lowered.startswith("m ") and answer[2:].strip().isdigit():
-                self._manual_id(row, int(answer[2:].strip()))
+                self.decide_manual(row, int(answer[2:].strip()))
                 return
             if lowered.startswith("f ") and answer[2:].strip():
                 candidates = self._research(answer[2:].strip()) or candidates
@@ -387,7 +430,7 @@ class ReviewSession:
             self.console.print("[dim]stopping — progress is saved[/dim]")
 
         remaining = sum(
-            1 for r in self.rows if r["match_status"] in ("ambiguous", "unmatched")
+            1 for r in self.rows if r["match_status"] in PENDING_MATCH_STATUSES
         )
         self.console.print(
             f"Recorded {self.decisions} decision(s); "

src/bggpipe/templates/review.html

@@ -575,6 +575,7 @@ setInterval(async () => {
   }
   if (pollMisses >= 3) render();  // recovered: rebuild banners from state
   pollMisses = 0;
+  if (STATE && fresh.revision < STATE.revision) return;  // stale poll response
   if (JSON.stringify(fresh) !== JSON.stringify(STATE)) {
     STATE = fresh;
     render();

src/bggpipe/upload.py

@@ -7,9 +7,10 @@ browser session. Etiquette (spec + bgg-api skill):
 - browser storage state persists locally (gitignored) so login is rare;
 - every attempt is appended to data/upload_log.csv immediately, so a killed
   run loses nothing and re-runs skip completed work;
-- refuses to touch the site while data/bgg_cache/STUB_FIXTURES.marker exists
-  (stub-resolved version ids must never reach BGG); --dry-run still works,
-  loudly labeled as synthetic.
+- refuses to touch the site while either provenance marker exists —
+  data/bgg_cache/STUB_FIXTURES.marker (gitignored) or data/STUB_DATA.marker
+  (committed, so fresh clones stay guarded); --dry-run still works, loudly
+  labeled as synthetic.
 
 Cloudflare: BGG fronts the site with a Turnstile check that blocks headless
 browsers outright (verified 2026-08-01 — headless shell never gets past
@@ -37,10 +38,10 @@ import typer
 
 from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
 from bggpipe.config import Config
+from bggpipe.fsio import atomic_write_text
 from bggpipe.models import CollectionItem
 
 BGG = "https://boardgamegeek.com"
-STORAGE_STATE_PATH = Path("storage_state.json")  # gitignored, credential-adjacent
 UPLOAD_LOG_COLUMNS = [
     "action",
     "bgg_id",
@@ -94,14 +95,14 @@ def _read_csv(path: Path) -> list[dict]:
 
 
 def append_log_row(path: Path, row: dict) -> None:
-    """Append one attempt, creating the file with a header on first write.
-    One row per attempt, flushed immediately — the log is the resume point."""
-    new = not path.exists()
-    path.parent.mkdir(parents=True, exist_ok=True)
+    """Append one attempt. One row per attempt, flushed immediately — the
+    log is the resume point, so its header is created ATOMICALLY first (a
+    torn header line would become DictReader's fieldnames and misparse
+    every logged success on the next run)."""
+    if not path.exists():
+        atomic_write_text(path, ",".join(UPLOAD_LOG_COLUMNS) + "\r\n")
     with path.open("a", newline="") as f:
         writer = csv.DictWriter(f, fieldnames=UPLOAD_LOG_COLUMNS, extrasaction="ignore")
-        if new:
-            writer.writeheader()
         writer.writerow(row)
 
 
@@ -148,12 +149,30 @@ def build_queue(
         for row in to_update
     ]
 
+    done_versions: dict[tuple[str, str], set[str]] = {}
+    for row in log_rows:
+        if row["status"] in DONE_STATUSES:
+            done_versions.setdefault(
+                (row["action"], row["collid"] or row["bgg_id"]), set()
+            ).add(row["version_id"])
+
     jobs: list[UploadJob] = []
     skipped_done = skipped_failed = 0
     deferred: list[UploadJob] = []
     update_game_seen: set[str] = set()
     seen: Counter[tuple[str, str, str]] = Counter()
     for job in candidates:
+        prior = done_versions.get(
+            ("update", job.collid) if job.action == "update" else ("add", job.bgg_id),
+            set(),
+        )
+        if prior and job.version_id not in prior:
+            typer.echo(
+                f"  note: {job.name} was previously {job.action}ed with a "
+                f"different version ({', '.join(sorted(prior)) or 'none'}) — "
+                "if re-review changed the version, the BGG entry needs a "
+                "manual correction (additive-only rule)"
+            )
         occurrence = seen[job.key]
         seen[job.key] += 1
         if not job.name:
@@ -215,11 +234,11 @@ class PlaywrightUploader:
     def __init__(
         self,
         username: str,
-        storage_state: Path = STORAGE_STATE_PATH,
+        storage_state: Path | None = None,
         headless: bool = False,
     ) -> None:
         self._username = username
-        self._storage_state = storage_state
+        self._storage_state = storage_state or Config().storage_state_path
         self._headless = headless
         self._authed = False
 
@@ -286,15 +305,16 @@ class PlaywrightUploader:
             self._context.storage_state(path=str(self._storage_state))
         self._authed = True
 
-    def _open_dialog(self, opener) -> object:
+    def _open_dialog(self, opener):
         """Click an opener that may no-op right after page load (hydration
-        race) and wait for the dialog to actually show."""
+        race) and wait for the dialog to actually show. (Params/return are
+        Playwright Locators; untyped because the import is lazy.)"""
         dialog = self._page.get_by_role("dialog")
         for attempt in (1, 2):
             opener.click()
             try:
                 dialog.wait_for(state="visible", timeout=5_000)
-                return dialog
+                break
             except self._timeout_error:
                 if attempt == 2:
                     raise
@@ -326,9 +346,16 @@ class PlaywrightUploader:
             # best guess is a next-page button, stopping when absent/disabled.
             nxt = dialog.get_by_role("button", name=re.compile("next|›|»", re.I)).first
             if nxt.count() == 0 or nxt.is_disabled():
-                break
+                break  # genuine end of list: added_no_version is honest
             nxt.click()
             self._page.wait_for_timeout(2_000)  # etiquette: paginating hits BGG too
+        else:
+            # never saw the end of the list: "not in picker" would be a
+            # false verdict frozen into DONE_STATUSES — stay retryable
+            raise RuntimeError(
+                f"hit MAX_VERSION_PAGES ({MAX_VERSION_PAGES}) without "
+                "finding the version or the end of the list — retryable"
+            )
         # Two-level dismissal: the sub-view has its own Cancel distinct from
         # the main dialog's.
         dialog.get_by_role("button", name="Cancel").first.click()
@@ -454,7 +481,7 @@ def verify_uploads(log_rows: list[dict], collection: list[CollectionItem]) -> li
 
     problems = []
     added_copies: Counter[int] = Counter()
-    seen_add_keys: set[tuple[str, str, str]] = set()
+    shortfall_reported: set[int] = set()
     latest: dict[tuple[str, str, str], dict] = {}
     for row in log_rows:
         k = _job_key(row)
@@ -467,13 +494,13 @@ def verify_uploads(log_rows: list[dict], collection: list[CollectionItem]) -> li
             if not copies:
                 problems.append(f"{row['name']}: logged added but not in collection")
             elif (
-                _job_key(row) not in seen_add_keys
+                int(row["bgg_id"]) not in shortfall_reported
                 and len(copies) < added_copies[int(row["bgg_id"])]
             ):
                 # the unverified second-copy dialog may EDIT the existing
                 # entry instead of creating one — a count shortfall is the
-                # only externally visible symptom
-                seen_add_keys.add(_job_key(row))
+                # only externally visible symptom; report once per GAME
+                shortfall_reported.add(int(row["bgg_id"]))
                 problems.append(
                     f"{row['name']}: {added_copies[int(row['bgg_id'])]} "
                     f"add(s) logged but only {len(copies)} cop"
@@ -515,7 +542,7 @@ def run_upload(
     headless: bool = False,
     uploader: Uploader | None = None,
     client: BGGClient | None = None,
-    storage_state: Path = STORAGE_STATE_PATH,
+    storage_state: Path | None = None,
     sleep: Callable[[float], None] = time.sleep,
     rng: random.Random | None = None,
     now: Callable[[], str] | None = None,
@@ -597,7 +624,9 @@ def run_upload(
                 )
                 raise typer.Exit(code=1)
             with PlaywrightUploader(
-                cfg.bgg_username, storage_state=storage_state, headless=headless
+                cfg.bgg_username,
+                storage_state=storage_state or cfg.storage_state_path,
+                headless=headless,
             ) as real:
                 results = _process(real, jobs, log_path, sleep=sleep, rng=rng, now=now)
         else:

src/bggpipe/webreview.py

@@ -16,6 +16,8 @@ import io
 import json
 import os
 import threading
+import warnings
+import xml.etree.ElementTree as ET
 from collections import Counter
 from importlib import resources
 from pathlib import Path
@@ -27,8 +29,9 @@ from fastapi.responses import FileResponse, HTMLResponse, Response
 from pydantic import BaseModel
 from rich.console import Console
 
-from bggpipe.bgg_client import BGGClient
+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.models import (
     CONFIDENT_VERSION_STATUSES,
     RECOGNIZED_MATCH_STATUSES,
@@ -42,15 +45,24 @@ def load_thumbnails(cache_dir: Path) -> dict[int, str]:
     thumbnails: dict[int, str] = {}
     if not cache_dir.is_dir():
         return thumbnails
-    for path in cache_dir.glob("thing_*.xml"):
+    unreadable = []
+    for path in cached_paths(cache_dir, "thing"):
         try:
             root = _safe_fromstring(path.read_text())
-        except Exception:  # a corrupt cache file must not kill the UI
-            continue
-        for item in root.findall("item"):
-            thumb = (item.findtext("thumbnail") or "").strip()
-            if thumb and item.get("id"):
-                thumbnails[int(item.get("id"))] = thumb
+            for item in root.findall("item"):
+                thumb = (item.findtext("thumbnail") or "").strip()
+                if thumb and item.get("id"):
+                    thumbnails[int(item.get("id"))] = thumb
+        except (ET.ParseError, OSError, ValueError):
+            # cosmetic degradation is fine — eating the evidence is not:
+            # this same file will crash resolve/review later if served
+            unreadable.append(path.name)
+    if unreadable:
+        warnings.warn(
+            f"{len(unreadable)} unreadable cache file(s) — thumbnails "
+            f"missing; delete to refetch: {', '.join(unreadable[:3])}",
+            stacklevel=2,
+        )
     return thumbnails
 
 
@@ -78,14 +90,26 @@ class DismissStore:
 
     def __init__(self, path: Path) -> None:
         self.path = path
-        self.keys: set[str] = (
-            set(json.loads(path.read_text())) if path.exists() else set()
-        )
+        self.keys: set[str] = set()
+        if path.exists():
+            try:
+                self.keys = set(json.loads(path.read_text()))
+            except (json.JSONDecodeError, OSError) as err:
+                # a torn write must not brick the server; quarantine and go on
+                quarantine = path.with_name(path.name + ".corrupt")
+                path.rename(quarantine)
+                warnings.warn(
+                    f"{path} was unreadable ({err}) — moved to {quarantine}; "
+                    "previously dismissed tickets will reappear",
+                    stacklevel=2,
+                )
 
     def add(self, key: str) -> None:
+        # write first, mutate after: a failed write must leave the ticket
+        # visible (memory never claims what disk doesn't hold)
+        updated = sorted(self.keys | {key})
+        atomic_write_text(self.path, json.dumps(updated, indent=2) + "\n")
         self.keys.add(key)
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(json.dumps(sorted(self.keys), indent=2) + "\n")
 
 
 class DecisionBody(BaseModel):
@@ -131,35 +155,22 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
     # 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
 
     def freshen() -> None:
         """Serve every request from the current file state: an extract or
         resolve run in another terminal must show up without a restart."""
         nonlocal thumbnails
         if session.reload_if_changed():
+            revision["n"] += 1
             thumbnails = load_thumbnails(cfg.cache_dir)
 
     def find_row(title_raw: str, source_photos: str, row_ix: int | None = None) -> dict:
         freshen()
-        # ordinal first: (title_raw, source_photos) is not unique when one
-        # photo holds two editions of the same game
-        if row_ix is not None and 0 <= row_ix < len(session.rows):
-            row = session.rows[row_ix]
-            if row["title_raw"] == title_raw and row["source_photos"] == source_photos:
-                return row
-        matches = [
-            row
-            for row in session.rows
-            if row["title_raw"] == title_raw and row["source_photos"] == source_photos
-        ]
-        undecided = [
-            row
-            for row in matches
-            if row["match_status"] in ("ambiguous", "unmatched", "merged")
-        ]
-        if undecided or matches:
-            return (undecided or matches)[0]
-        raise HTTPException(404, "row not found — matches.csv changed underneath?")
+        row = session.find_row(title_raw, source_photos, row_ix)
+        if row is None:
+            raise HTTPException(404, "row not found — matches.csv changed underneath?")
+        return row
 
     def photo_names() -> set[str]:
         if not cfg.photos_dir.is_dir():
@@ -201,12 +212,18 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
         freshen()
         counts = Counter(row["match_status"] for row in session.rows)
         resolved_titles = {r["title_raw"] for r in session.rows}
-        rows_by_title: dict[str, dict] = {}
+        rows_by_title: dict[str, list[dict]] = {}
         for r in session.rows:
-            rows_by_title.setdefault(r["title_raw"], r)
+            rows_by_title.setdefault(r["title_raw"], []).append(r)
+        title_seen: Counter[str] = Counter()
         catalog = []
         for entry in session.titles:
-            row = rows_by_title.get(entry.title_raw)
+            same_title = rows_by_title.get(entry.title_raw, [])
+            ix = title_seen[entry.title_raw]
+            title_seen[entry.title_raw] += 1
+            # positional pairing, same rule as run_resolve: the ix-th entry
+            # of a title reports the ix-th row of that title
+            row = same_title[ix] if ix < len(same_title) else None
             catalog.append(
                 {
                     "title_raw": entry.title_raw,
@@ -249,6 +266,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
             for r in session.merged_rows()
         ]
         return {
+            "revision": revision["n"],
             "warnings": session.warnings[-10:],
             "pending": [row_payload(r) for r in session.pending_rows()],
             "versions": [version_payload(r) for r in session.version_rows()],
@@ -280,6 +298,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
     @app.post("/api/decision")
     def api_decision(body: DecisionBody) -> dict:
         with lock:
+            revision["n"] += 1
             return _decide(body)
 
     def _decide(body: DecisionBody) -> dict:
@@ -305,6 +324,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
     @app.post("/api/version")
     def api_version(body: VersionBody) -> dict:
         with lock:
+            revision["n"] += 1
             return _version(body)
 
     def _version(body: VersionBody) -> dict:
@@ -323,6 +343,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
     @app.post("/api/veto-merge")
     def api_veto_merge(body: VetoBody) -> dict:
         with lock:
+            revision["n"] += 1
             row = find_row(body.title_raw, body.source_photos, body.row_ix)
             if row["match_status"] != "merged":
                 raise HTTPException(400, "row is not merged")
@@ -332,6 +353,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
     @app.post("/api/dismiss")
     def api_dismiss(body: DismissBody) -> dict:
         with lock:
+            revision["n"] += 1
             dismissed.add(_sighting_key(body.photo, body.model_dump(exclude={"photo"})))
             return state()