Re-audit round 2: 5 blind reviewers, 17 fixes, +12 tests

The re-run confirmed round 1 held and then caught second-order bugs in
its own fixes plus two long-standing ones everyone missed. TUI decisions
after a mid-session reload were counted but never written (rows are now
re-adopted into the fresh list on every save, preferring undecided slots
on duplicate keys); row_ix was computed by equality so duplicate rows
shared an ordinal (identity now, merges included, veto sends it); upload
job keys collided for two same-version copies (completions are counted
per key, so --limit or an interrupt can no longer strand the second
copy); diff consumes collids on exact-version matches (a vetoed
same-version second copy was silently swallowed) and splits mismatches:
report-only disagreement while an unclaimed copy exists, second-copy add
only when every copy is claimed.

Also: XML responses are validated and written atomically before caching
(a torn or truncated 200 body can never poison a re-run), JSON artifacts
write atomically, thing/search parsers refuse missing ids like the
collection parser, empty game names are refused by the upload queue, a
never-rendering version picker fails retryably instead of terminally,
the systemic-failure abort compares exception types, blocked same-title
entries defer as a group so positional pairing can't misalign,
truncation heads pick the earliest separator, diff messages tell the
truth when a token exists without a username, and the shared-constant
sweep now actually covers every module (statuses, search types, marker
names, client_for, ports). pydantic declared as a direct dependency.

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

pyproject.toml

@@ -15,6 +15,7 @@ dependencies = [
     "fastapi>=0.141.1",
     "uvicorn>=0.52.1",
     "playwright>=1.62.0",
+    "pydantic>=2.13.4",
 ]
 
 [project.scripts]

scripts/fixture_common.py

@@ -9,6 +9,8 @@ from __future__ import annotations
 
 from pathlib import Path
 
+from bggpipe.config import STUB_CACHE_MARKER_NAME, STUB_DATA_MARKER_NAME
+
 CACHE_MARKER_TEXT = (
     "This cache contains hand-written stub XML, not real BGG "
     "responses. Data resolved from it must not be uploaded.\n"
@@ -28,9 +30,9 @@ def esc(text: str) -> str:
 
 
 def write_cache_marker(target: Path) -> None:
-    (target / "STUB_FIXTURES.marker").write_text(CACHE_MARKER_TEXT)
+    (target / STUB_CACHE_MARKER_NAME).write_text(CACHE_MARKER_TEXT)
 
 
 def write_data_marker(data_dir: Path = Path("data")) -> None:
     data_dir.mkdir(parents=True, exist_ok=True)
-    (data_dir / "STUB_DATA.marker").write_text(DATA_MARKER_TEXT)
+    (data_dir / STUB_DATA_MARKER_NAME).write_text(DATA_MARKER_TEXT)

scripts/write_photo_fixtures.py

@@ -19,7 +19,7 @@ from pathlib import Path
 
 from fixture_common import esc, write_cache_marker, write_data_marker
 
-from bggpipe.bgg_client import cache_key
+from bggpipe.bgg_client import SEARCH_TYPES, cache_key
 
 TARGETS = (Path("tests/fixtures/bgg_cache"), Path("data/bgg_cache"))
 
@@ -27,7 +27,6 @@ TARGETS = (Path("tests/fixtures/bgg_cache"), Path("data/bgg_cache"))
 # gitignored, so this is what protects a fresh clone): upload refuses to
 # run while it exists.
 DATA_MARKER = Path("data/STUB_DATA.marker")
-SEARCH_TYPES = "boardgame,boardgameexpansion"
 
 BG, EXP = "boardgame", "boardgameexpansion"
 
@@ -345,7 +344,6 @@ def versions_xml(bgg_id: int, versions: list[tuple]) -> str:
     )
 
 
-
 def main() -> None:
     files: dict[str, str] = {}
     for query, results in SEARCHES.items():

scripts/write_stub_fixtures.py

@@ -16,10 +16,9 @@ from pathlib import Path
 
 from fixture_common import esc, write_cache_marker
 
-from bggpipe.bgg_client import cache_key
+from bggpipe.bgg_client import SEARCH_TYPES, cache_key
 
 FIXTURE_CACHE = Path("tests/fixtures/bgg_cache")
-SEARCH_TYPES = "boardgame,boardgameexpansion"
 
 
 def search_item(bgg_id: int, name: str, year: int | None, type_: str) -> str:

src/bggpipe/bgg_client.py

@@ -20,6 +20,7 @@ import httpx
 
 from bggpipe import __version__
 from bggpipe.config import Config
+from bggpipe.fsio import atomic_write_text
 from bggpipe.models import (
     BGGResponseError,
     CollectionItem,
@@ -29,9 +30,13 @@ from bggpipe.models import (
     parse_search,
     parse_things,
     parse_things_full,
+    validate_response,
 )
 
 BASE_URL = "https://boardgamegeek.com/xmlapi2"
+# One home for the search-type filter: the fixture generators must build
+# cache keys with the byte-identical string or every lookup silently misses.
+SEARCH_TYPES = "boardgame,boardgameexpansion"
 QUEUE_BACKOFF = (2.0, 5.0, 10.0, 30.0)  # sleeps between the 5 attempts (spec)
 MAX_ATTEMPTS = 5
 _UNSAFE = re.compile(r"[^A-Za-z0-9._=,-]+")
@@ -120,14 +125,13 @@ class BGGClient:
                 continue
 
             response.raise_for_status()
-            if "<errors" in response.text[:120]:
-                # BGG serves some errors as HTTP 200 <errors> XML (bad
-                # username etc.) — caching one would poison every re-run
-                raise BGGResponseError(
-                    f"BGG error document for /{endpoint}: {response.text[:200]}"
-                )
-            self.cache_dir.mkdir(parents=True, exist_ok=True)
-            cache_path.write_text(response.text)
+            try:
+                # error documents AND malformed/truncated bodies must never
+                # reach the cache — they would poison every future run
+                validate_response(response.text)
+            except BGGResponseError as err:
+                raise BGGResponseError(f"/{endpoint}: {err}") from err
+            atomic_write_text(cache_path, response.text)
             return response.text
 
         raise BGGQueueTimeout(
@@ -138,9 +142,7 @@ class BGGClient:
 
     # -- typed endpoint wrappers ------------------------------------------
 
-    def search(
-        self, query: str, types: str = "boardgame,boardgameexpansion"
-    ) -> list[SearchResult]:
+    def search(self, query: str, types: str = SEARCH_TYPES) -> list[SearchResult]:
         return parse_search(self.get_xml("search", {"query": query, "type": types}))
 
     def things(

src/bggpipe/config.py

@@ -16,6 +16,10 @@ from pathlib import Path
 
 DEFAULT_CONFIG_PATH = Path("config.toml")
 DEFAULT_REVIEW_PORT = 8377
+# Provenance marker filenames — the upload guard and both fixture
+# generators must agree on these exactly.
+STUB_CACHE_MARKER_NAME = "STUB_FIXTURES.marker"
+STUB_DATA_MARKER_NAME = "STUB_DATA.marker"
 
 
 @dataclass(frozen=True)
@@ -70,8 +74,8 @@ class Config:
     def stub_marker_paths(self) -> tuple[Path, Path]:
         # gitignored (travels with the stub XML) + committed (guards clones)
         return (
-            self.cache_dir / "STUB_FIXTURES.marker",
-            self.data_dir / "STUB_DATA.marker",
+            self.cache_dir / STUB_CACHE_MARKER_NAME,
+            self.data_dir / STUB_DATA_MARKER_NAME,
         )
 
 

src/bggpipe/diff.py

@@ -8,12 +8,12 @@ Two collection sources:
   logged-in-user exemption.
 
 Outputs both artifacts:
-- to_add.csv    — recognized games not in the collection, including
-  additional copies whose confident version matches no owned copy;
+- to_add.csv    — recognized games not in the collection, plus additional
+  copies once every owned copy is claimed by another match row;
 - to_update.csv — owned, VERSION-LESS entries where matching produced a
-  confident version (version_auto/version_approved). Strictly additive:
-  entries that already carry a version are never touched — a further copy
-  with a different version becomes a to_add row instead.
+  confident version. Strictly additive: entries that already carry a
+  version are never touched — a version mismatch against an unclaimed
+  copy is reported as a disagreement, nothing more.
 """
 
 from __future__ import annotations
@@ -25,10 +25,11 @@ from pathlib import Path
 
 import typer
 
-from bggpipe.bgg_client import BGGClient, client_for
+from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
 from bggpipe.config import Config
 from bggpipe.models import (
     CONFIDENT_VERSION_STATUSES,
+    RECOGNIZED_MATCH_STATUSES,
     CollectionItem,
     parse_collection,
 )
@@ -55,6 +56,7 @@ class DiffResult:
     to_update: list[dict] = field(default_factory=list)
     already_owned: list[str] = field(default_factory=list)  # title_raw
     second_copies: list[str] = field(default_factory=list)  # notes for adds
+    disagreements: list[str] = field(default_factory=list)  # report-only
     unseen: list[CollectionItem] = field(default_factory=list)
     pending: list[str] = field(default_factory=list)  # ambiguous/unmatched titles
     rejected: int = 0
@@ -99,30 +101,7 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
                 p for p in row["source_photos"].split(";") if p
             )
 
-    for row in rows:
-        status = row["match_status"]
-        if status == "rejected":
-            result.rejected += 1
-            continue
-        if status == "merged":
-            result.merged += 1  # represented by its survivor row
-            continue
-        if status not in ("auto", "approved") or not row["bgg_id"]:
-            result.pending.append(row["title_raw"])
-            continue
-
-        result.recognized += 1
-        bgg_id = int(row["bgg_id"])
-        copies = by_object.get(bgg_id, [])
-        if copies:
-            seen_object_ids.add(bgg_id)
-
-        confident = (
-            row["version_status"] in CONFIDENT_VERSION_STATUSES and row["version_id"]
-        )
-        version_id = int(row["version_id"]) if confident else None
-
-        def add_row(row: dict = row, confident: bool = confident) -> dict:
+    def add_row(row: dict, confident: bool) -> dict:
         photos = {p for p in row["source_photos"].split(";") if p}
         photos |= merged_photos.get(row["title_raw"], set())
         return {
@@ -136,24 +115,52 @@ def compute_diff(rows: list[dict], collection: list[CollectionItem]) -> DiffResu
             "source_photos": ";".join(sorted(photos)),
         }
 
-        if not copies:
-            result.to_add.append(add_row())
+    recognized: list[dict] = []
+    for row in rows:
+        status = row["match_status"]
+        if status == "rejected":
+            result.rejected += 1
             continue
-
-        if not confident:
-            # bare id with unknown version: owned if any copy exists
-            result.already_owned.append(row["title_raw"])
+        if status == "merged":
+            result.merged += 1  # represented by its survivor row
             continue
-
-        if any(c.version_id == version_id for c in copies):
-            result.already_owned.append(row["title_raw"])
+        if status not in RECOGNIZED_MATCH_STATUSES or not row["bgg_id"]:
+            result.pending.append(row["title_raw"])
             continue
+        result.recognized += 1
+        recognized.append(row)
+        if by_object.get(int(row["bgg_id"])):
+            seen_object_ids.add(int(row["bgg_id"]))
 
-        versionless = [
-            c
-            for c in copies
-            if c.version_id is None and c.coll_id not in consumed_collids
+    def unconsumed(bgg_id: int) -> list[CollectionItem]:
+        return [
+            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"]
+        )
+
+    # 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)):
+        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]
+        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
+            consumed_collids.add(matching[0].coll_id)
+            result.already_owned.append(row["title_raw"])
+            continue
+        versionless = [c for c in remaining if c.version_id is None]
         if versionless:
             target = versionless[0]
             consumed_collids.add(target.coll_id)
@@ -167,18 +174,49 @@ 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:
+            # most likely the same physical box mis-scored — report, never
+            # touch, never duplicate (spec: report the disagreement)
+            consumed_collids.add(remaining[0].coll_id)
+            result.already_owned.append(row["title_raw"])
+            result.disagreements.append(
+                f"{row['title_raw']}: photo suggests version "
+                f"{row['version_name']!r} ({row['version_id']}) but the "
+                "remaining collection entry carries a different version — "
+                "left untouched"
+            )
         else:
-            # no remaining copy can take this version: every copy either
-            # already carries a different version or was consumed by another
-            # match — per spec this is an additional physical copy to ADD
-            # (existing entries are never touched)
-            result.to_add.append(add_row())
+            # 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 "
                 f"{row['version_name']!r} ({row['version_id']}) — every "
                 "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)):
+        bgg_id = int(row["bgg_id"])
+        if not by_object.get(bgg_id):
+            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:
+            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"
+            )
+
     result.unseen = [
         item for item in collection if item.object_id not in seen_object_ids
     ]
@@ -201,10 +239,25 @@ def run_diff(cfg: Config, *, client: BGGClient | None = None) -> DiffResult:
         typer.echo(f"{cfg.matches_path} is empty — run `bggpipe resolve` first.")
         raise typer.Exit(code=1)
 
-    if os.environ.get("BGG_API_TOKEN") and cfg.bgg_username:
+    has_token = bool(os.environ.get("BGG_API_TOKEN"))
+    if has_token and cfg.bgg_username:
         typer.echo("Fetching live collection from BGG…")
         client = client or client_for(cfg)
+        try:
             collection = client.collection_full(cfg.bgg_username, refresh=True)
+        except BGGAuthError 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.")
+            collection = load_snapshot_collection(cfg.data_dir)
+    else:
+        if has_token:
+            # saying "No BGG_API_TOKEN" here would be false and misdirect
+            # the user's debugging — the missing half is the username
+            typer.echo(
+                "BGG_API_TOKEN is set but BGG_USERNAME is not (is .env "
+                "loaded?) — using snapshot files instead of live mode."
+            )
         else:
             typer.echo(
                 "No BGG_API_TOKEN — using collection snapshot files in "
@@ -228,6 +281,10 @@ def run_diff(cfg: Config, *, client: BGGClient | None = None) -> DiffResult:
         typer.echo("\nSecond copies (verify these on the dry run before upload):")
         for line in result.second_copies:
             typer.echo(f"  - {line}")
+    if result.disagreements:
+        typer.echo("\nVersion disagreements (reported only — nothing changed):")
+        for line in result.disagreements:
+            typer.echo(f"  - {line}")
     if result.pending:
         typer.echo("\nStill pending review: " + ", ".join(result.pending))
     if result.unseen:

src/bggpipe/enrich.py

@@ -19,16 +19,20 @@ import json
 
 import typer
 
-from bggpipe.bgg_client import BGGAuthError, BGGClient
+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.resolve import read_matches
 
 BATCH_SIZE = 20
-_CONFIDENT_VERSION = ("version_auto", "version_approved")
 
 
 def _version_info(row: dict) -> dict | None:
-    if row["version_status"] not in _CONFIDENT_VERSION or not row["version_id"]:
+    if row["version_status"] not in CONFIDENT_VERSION_STATUSES or not row["version_id"]:
         return None
     version_id = int(row["version_id"])
     for cand in json.loads(row["version_candidates_json"] or "[]"):
@@ -60,7 +64,7 @@ def run_enrich(
 
     targets: list[tuple[str, int, dict | None]] = []
     for row in rows:
-        if row["match_status"] not in ("auto", "approved") or not row["bgg_id"]:
+        if row["match_status"] not in RECOGNIZED_MATCH_STATUSES or not row["bgg_id"]:
             continue
         version = _version_info(row)
         key = f"{row['bgg_id']}:{version['version_id']}" if version else row["bgg_id"]
@@ -70,7 +74,7 @@ def run_enrich(
     games: dict = json.loads(games_path.read_text()) if games_path.exists() else {}
 
     need = sorted({bgg_id for key, bgg_id, _ in targets if refresh or key not in games})
-    client = client or BGGClient(cfg.cache_dir, cfg.rate_limit_seconds)
+    client = client or client_for(cfg)
 
     fetched: dict[int, dict] = {}
     blocked = False
@@ -90,7 +94,9 @@ def run_enrich(
             updated += 1
 
     games_path.parent.mkdir(parents=True, exist_ok=True)
-    games_path.write_text(json.dumps(games, indent=2, ensure_ascii=False) + "\n")
+    atomic_write_text(
+        games_path, json.dumps(games, indent=2, ensure_ascii=False) + "\n"
+    )
 
     typer.echo(
         f"games.json: {len(games)} entr{'y' if len(games) == 1 else 'ies'} "

src/bggpipe/extract.py

@@ -19,6 +19,7 @@ from pathlib import Path
 import typer
 
 from bggpipe.config import Config
+from bggpipe.fsio import atomic_write_text
 from bggpipe.normalize import normalize_title
 
 IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".heic"}
@@ -105,7 +106,8 @@ def prepare_image(path: Path) -> tuple[str, str]:
 
 def parse_vision_response(text: str) -> tuple[list[dict], list[dict], int]:
     """Parse the model's JSON defensively: strip code fences, locate the
-    payload amid any prose. Returns (title entries, unidentified sightings).
+    payload amid any prose. Returns (title entries, unidentified sightings,
+    dropped-malformed-entry count).
     A bare JSON array (the pre-unidentified response shape) still parses —
     it's all titles."""
     cleaned = _CODE_FENCE.sub("", text).strip()
@@ -268,9 +270,11 @@ def rebuild_artifacts(
             unidentified[photo] = data["unidentified"]
     deduped = dedupe_entries(entries)
     titles_path.parent.mkdir(parents=True, exist_ok=True)
-    titles_path.write_text(json.dumps(deduped, indent=2, ensure_ascii=False) + "\n")
-    unidentified_path.write_text(
-        json.dumps(unidentified, indent=2, ensure_ascii=False) + "\n"
+    atomic_write_text(
+        titles_path, json.dumps(deduped, indent=2, ensure_ascii=False) + "\n"
+    )
+    atomic_write_text(
+        unidentified_path, json.dumps(unidentified, indent=2, ensure_ascii=False) + "\n"
     )
     return deduped, unidentified
 
@@ -311,7 +315,9 @@ def run_extract(
             typer.echo(f"  {photo.name}: already extracted, skipping")
             continue
         result = extract_photo(photo, vision)
-        raw_path.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n")
+        atomic_write_text(
+            raw_path, json.dumps(result, indent=2, ensure_ascii=False) + "\n"
+        )
         note = (
             f" ({len(result['unidentified'])} unidentified)"
             if result["unidentified"]

src/bggpipe/fsio.py

@@ -0,0 +1,18 @@
+"""Atomic file writes for every accumulated artifact.
+
+A kill mid-write must never leave a torn file that poisons future runs —
+the same tmp + os.replace guarantee write_matches gives matches.csv,
+available to JSON artifacts and the XML response cache.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import 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.write_text(text)
+    os.replace(tmp, path)

src/bggpipe/models.py

@@ -12,10 +12,11 @@ class BGGResponseError(Exception):
     """The API returned a well-formed error document (e.g. bad username)."""
 
 
-# The one status predicate the whole pipeline shares: a version is trusted
+# The two status predicates the whole pipeline shares: a version is trusted
 # for diff/upload/enrich only when matching produced it confidently or a
-# human approved it.
+# 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")
 
 
 @dataclass(frozen=True)
@@ -78,9 +79,11 @@ def _attr_int(elem: ET.Element | None, attr: str = "value") -> int | None:
 
 def parse_search(xml_text: str) -> list[SearchResult]:
     results = []
+    skipped = 0
     for item in _root(xml_text).findall("item"):
         name_elem = item.find("name")
         if name_elem is None or item.get("id") is None:
+            skipped += 1  # tolerate stragglers; wholesale drift raises below
             continue
         results.append(
             SearchResult(
@@ -91,6 +94,11 @@ def parse_search(xml_text: str) -> list[SearchResult]:
                 type=item.get("type", "boardgame"),
             )
         )
+    if skipped and not results:
+        raise BGGResponseError(
+            f"search response had {skipped} item(s), none parseable — "
+            "schema drift? Bad data must not look like no results."
+        )
     return results
 
 
@@ -125,7 +133,7 @@ def parse_things(xml_text: str) -> list[ThingDetails]:
         ]
         things.append(
             ThingDetails(
-                bgg_id=int(item.get("id", 0)),
+                bgg_id=int(_required_attr(item, "id")),
                 name=name.get("value", "") if name is not None else "",
                 year=_attr_int(item.find("yearpublished")),
                 type=item.get("type", "boardgame"),
@@ -181,7 +189,7 @@ def parse_things_full(xml_text: str) -> list[dict]:
                     best_player_counts.append(results.get("numplayers"))
         games.append(
             {
-                "bgg_id": int(item.get("id", 0)),
+                "bgg_id": int(_required_attr(item, "id")),
                 "type": item.get("type", "boardgame"),
                 "name": name.get("value", "") if name is not None else "",
                 "year": _attr_int(item.find("yearpublished")),
@@ -215,16 +223,26 @@ def parse_things_full(xml_text: str) -> list[dict]:
     return games
 
 
-def _required_attr(item, name: str) -> str:
+def _required_attr(item: ET.Element, name: str) -> str:
     value = item.get(name)
     if not value:
         raise BGGResponseError(
-            f"collection item missing {name!r} — truncated or unexpected "
-            "response; refusing to feed it to the diff"
+            f"response item missing {name!r} — truncated or unexpected "
+            "response; refusing to coerce a missing id"
         )
     return value
 
 
+def validate_response(xml_text: str) -> None:
+    """Raise BGGResponseError for error documents AND malformed XML — the
+    client calls this before caching, so a torn or truncated 200 body can
+    never poison the cache."""
+    try:
+        _root(xml_text)
+    except ET.ParseError as err:
+        raise BGGResponseError(f"malformed XML: {err}") from err
+
+
 def parse_collection(xml_text: str) -> list[CollectionItem]:
     items = []
     for item in _root(xml_text).findall("item"):
@@ -240,7 +258,11 @@ def parse_collection(xml_text: str) -> list[CollectionItem]:
                 name=item.findtext("name", default=""),
                 subtype=item.get("subtype", "boardgame"),
                 own=status is not None and status.get("own") == "1",
-                year=int(year_text) if year_text and year_text.isdigit() else None,
+                year=(
+                    int(year_text)
+                    if year_text and year_text.lstrip("-").isdigit()
+                    else None
+                ),
                 version_id=(
                     int(version_item.get("id"))
                     if version_item is not None and version_item.get("id")

src/bggpipe/resolve.py

@@ -22,7 +22,11 @@ from rapidfuzz import fuzz
 from bggpipe.bgg_client import BGGAuthError, BGGClient, client_for
 from bggpipe.config import Config
 from bggpipe.extract import cues_conflict
-from bggpipe.models import CONFIDENT_VERSION_STATUSES, GameVersion
+from bggpipe.models import (
+    CONFIDENT_VERSION_STATUSES,
+    RECOGNIZED_MATCH_STATUSES,
+    GameVersion,
+)
 from bggpipe.normalize import normalize_title
 
 FUZZY_THRESHOLD = 90
@@ -149,8 +153,10 @@ def load_titles(path: Path) -> list[TitleEntry]:
         entries.append(
             TitleEntry(
                 title_raw=title_raw,
-                title_normalized=raw.get("title_normalized")
-                or normalize_title(title_raw),
+                # always recompute: a stale/hand-written stored value would
+                # silently break exact matching (both sides must normalize
+                # by the CURRENT rules)
+                title_normalized=normalize_title(title_raw),
                 confidence=raw.get("confidence", "high"),
                 publisher_hint=raw.get("publisher_hint") or "",
                 edition_hint=raw.get("edition_hint") or "",
@@ -173,9 +179,10 @@ def _truncation_heads(title_raw: str) -> list[str]:
     before the first subtitle separator, before a "(The) Game ..."
     descriptor, then the first two words as a last resort."""
     heads: list[str] = []
-    sep_head = next(
-        (title_raw.split(sep)[0] for sep in _SEPARATORS if sep in title_raw), None
-    )
+    present = [(title_raw.find(sep), sep) for sep in _SEPARATORS if sep in title_raw]
+    # earliest separator wins — priority order would let " - " late in the
+    # title beat an early ": ", yielding heads like "Blorvath: Quest"
+    sep_head = title_raw.split(min(present)[1])[0] if present else None
     if sep_head:
         heads.append(sep_head)
     match = _GAME_WORD.search(title_raw)
@@ -203,13 +210,14 @@ def _plausible_candidates(
     """Search BGG and keep plausible candidates, one per id: exact-normalized
     or fuzzy>=90 against the FULL title, or — on truncated retries — exact
     (only exact: truncation must stay conservative) against the head."""
+    # a fully non-Latin title normalizes to "" — empty-vs-empty is not a
+    # match (token_sort_ratio("", "") is 100), and searching would only
+    # spend rate-limited requests to prove nothing
+    if not entry.title_normalized:
+        return []
     by_id: dict[int, Candidate] = {}
     for result in client.search(query):
         norm = normalize_title(result.name)
-        # a fully non-Latin title normalizes to "" — empty-vs-empty is not a
-        # match (and token_sort_ratio("", "") is 100, so guard fuzzy too)
-        if not entry.title_normalized:
-            continue
         exact = norm == entry.title_normalized
         fuzzy = fuzz.token_sort_ratio(norm, entry.title_normalized)
         if not exact and fuzzy < FUZZY_THRESHOLD:
@@ -427,7 +435,7 @@ def dedupe_matches(rows: list[dict], titles: list[TitleEntry]) -> list[MergeEven
 
     groups: dict[tuple[str, str], list[dict]] = {}
     for row in rows:
-        if row["match_status"] not in ("auto", "approved") or not row["bgg_id"]:
+        if row["match_status"] not in RECOGNIZED_MATCH_STATUSES or not row["bgg_id"]:
             continue
         if row.get("dedupe_veto"):
             # a human already ruled "this is a genuinely separate copy" —
@@ -516,10 +524,17 @@ def run_resolve(
     skipped = 0
     photos_updated = False
     blocked: list[str] = []
+    blocked_titles: set[str] = set()
     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]
             photos = ";".join(entry.source_photos)
@@ -534,6 +549,7 @@ def run_resolve(
             # No API token yet: 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")
             continue
         new_rows.append(row)

src/bggpipe/review.py

@@ -110,18 +110,37 @@ class ReviewSession:
         self.warnings.append(message)
         self.console.print(f"[yellow]{message}[/yellow]")
 
-    def _save(self, row: dict | None = None) -> None:
-        """Atomic write of the in-memory rows. If another process rewrote the
-        file since we loaded (resolve in a second terminal), reload first and
-        re-apply `row` — the one decision being saved — by identity, so
-        neither side's work is silently lost."""
-        if self.reload_if_changed() and row is not None:
+    def _adopt(self, row: dict) -> bool:
+        """Swap `row` (possibly an orphaned reference from before a reload)
+        back into self.rows by identity or key. Key collisions (duplicate
+        two-edition rows) prefer a still-undecided slot. False = row is gone
+        from the file entirely."""
+        if any(r is row for r in self.rows):
+            return True
         key = (row["title_raw"], row["source_photos"])
-            for i, fresh in enumerate(self.rows):
-                if (fresh["title_raw"], fresh["source_photos"]) == key:
-                    self.rows[i] = row
-                    break
-            else:
+        candidates = [
+            i
+            for i, r in enumerate(self.rows)
+            if (r["title_raw"], r["source_photos"]) == key
+        ]
+        if not candidates:
+            return False
+        undecided = [
+            i
+            for i in candidates
+            if self.rows[i]["match_status"] in ("ambiguous", "unmatched", "merged")
+        ]
+        self.rows[(undecided or candidates)[0]] = row
+        return True
+
+    def _save(self, row: dict | None = None) -> None:
+        """Atomic write of the in-memory rows. A TUI loop iterates row
+        references snapshotted before any reload, and reload_if_changed can
+        swap self.rows at every save — so the decided row must always be
+        re-adopted into the CURRENT list, or the decision would be counted
+        but never written."""
+        self.reload_if_changed()
+        if row is not None and not self._adopt(row):
             self._warn(
                 f"{row['title_raw']!r} disappeared from matches.csv while "
                 "you decided — decision NOT saved"

src/bggpipe/templates/review.html

@@ -282,12 +282,17 @@ async function post(url, body) {
     return;
   }
   if (!res.ok) {
-    const detail = await res.json().then(d => d.detail).catch(() => res.statusText);
-    alert("That didn't save: " + detail);
+    const detail = await res.json().then(d => d.detail).catch(() => null);
+    alert("That didn't save: " + (detail ?? res.statusText));
     return;
   }
+  try {
     STATE = await res.json();
     render();
+  } catch (err) {
+    // the decision saved server-side; only the re-render failed
+    errorBanner(`saved, but the page failed to refresh (${err.message || err}) — reload the page`);
+  }
 }
 
 function cueChips(cues) {
@@ -432,6 +437,7 @@ function render() {
     html += `<h2>Merges <span class="count">— duplicate reads folded into one game; veto if wrong</span></h2>`;
     html += s.merges.map(mg => `
       <section class="card merge actionable" data-kind="merge"
+               data-rowix="${mg.row_ix}"
                data-title="${esc(mg.title_raw)}" data-photos="${esc(mg.source_photos)}">
         <div class="body">
           <span class="cname">${esc(mg.title_raw)}</span>
@@ -522,6 +528,7 @@ const dismiss = t => post("/api/dismiss", {
 });
 const vetoMerge = card => post("/api/veto-merge", {
   title_raw: card.dataset.title, source_photos: card.dataset.photos,
+  row_ix: rowIx(card),
 });
 
 document.addEventListener("keydown", e => {
@@ -566,7 +573,7 @@ setInterval(async () => {
     if (++pollMisses >= 3) errorBanner(`lost contact (${err.message || err})`);
     return;
   }
-  if (pollMisses >= 3) showBanner("");  // recovered: clear the lost-contact banner
+  if (pollMisses >= 3) render();  // recovered: rebuild banners from state
   pollMisses = 0;
   if (JSON.stringify(fresh) !== JSON.stringify(STATE)) {
     STATE = fresh;

src/bggpipe/upload.py

@@ -21,7 +21,6 @@ domcontentloaded plus explicit element waits.
 
 from __future__ import annotations
 
-import contextlib
 import csv
 import os
 import random
@@ -67,17 +66,24 @@ class UploadJob:
 
     @property
     def key(self) -> tuple[str, str, str]:
-        # A second copy of the same game (different version) is a distinct
-        # add; updates are keyed by the physical copy they amend.
-        if self.action == "update":
-            return ("update", self.collid, "")
-        return ("add", self.bgg_id, self.version_id)
+        return _key(self.action, self.bgg_id, self.collid, self.version_id)
+
+
+def _key(
+    action: str, bgg_id: str, collid: str, version_id: str
+) -> tuple[str, str, str]:
+    # A second copy of the same game (different version) is a distinct add;
+    # updates are keyed by the physical copy they amend. NOTE: two copies
+    # with the SAME (bgg_id, version) — vetoed duplicates — share a key, so
+    # build_queue counts completions per key instead of treating the key as
+    # unique.
+    if action == "update":
+        return ("update", collid, "")
+    return ("add", bgg_id, version_id)
 
 
 def _job_key(row: dict) -> tuple[str, str, str]:
-    if row["action"] == "update":
-        return ("update", row["collid"], "")
-    return ("add", row["bgg_id"], row["version_id"])
+    return _key(row["action"], row["bgg_id"], row["collid"], row["version_id"])
 
 
 def _read_csv(path: Path) -> list[dict]:
@@ -105,12 +111,21 @@ def build_queue(
     log_rows: list[dict],
     *,
     retry_failed: bool = False,
-) -> tuple[list[UploadJob], int, int]:
+) -> tuple[list[UploadJob], int, int, list[UploadJob]]:
     """Turn the diff outputs into pending jobs, minus work the log says is
-    done. Returns (jobs, skipped_done, skipped_failed)."""
-    latest: dict[tuple[str, str, str], str] = {}
+    done. Returns (jobs, skipped_done, skipped_failed, deferred) — deferred
+    being same-game updates held for a later run.
+
+    Completions are COUNTED per key, not looked up: two vetoed duplicate
+    copies share a key, and one logged success must complete exactly one
+    of them."""
+    done_count: Counter[tuple[str, str, str]] = Counter()
+    last_status: dict[tuple[str, str, str], str] = {}
     for row in log_rows:
-        latest[_job_key(row)] = row["status"]
+        k = _job_key(row)
+        if row["status"] in DONE_STATUSES:
+            done_count[k] += 1
+        last_status[k] = row["status"]
 
     candidates = [
         UploadJob(
@@ -137,11 +152,21 @@ def build_queue(
     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:
-        status = latest.get(job.key)
-        if status in DONE_STATUSES:
+        occurrence = seen[job.key]
+        seen[job.key] += 1
+        if not job.name:
+            # an empty name (manual id whose lookup failed) would make the
+            # name-driven selectors match ANY heading/row — refuse loudly
+            typer.echo(
+                f"  refusing to queue bgg_id {job.bgg_id}: empty game name "
+                "— re-review this match so the name resolves"
+            )
+            skipped_failed += 1
+        elif occurrence < done_count[job.key]:
             skipped_done += 1
-        elif status == "failed" and not retry_failed:
+        elif last_status.get(job.key) == "failed" and not retry_failed:
             skipped_failed += 1
         elif job.action == "update" and job.bgg_id in update_game_seen:
             # The row-edit flow finds rows by game name, not collid — a
@@ -281,8 +306,17 @@ class PlaywrightUploader:
         cancelled — when the name never shows up."""
         dialog.get_by_role("button", name="Set version/edition").click()
         pattern = re.compile(re.escape(version_name), re.I)
-        with contextlib.suppress(self._timeout_error):  # empty list is legal
-            dialog.get_by_role("listitem").first.wait_for(timeout=10_000)
+        try:
+            dialog.get_by_role("listitem").first.wait_for(timeout=15_000)
+        except self._timeout_error as err:
+            # A version resolve found on BGG cannot legitimately be missing
+            # from the picker — an unrendered list means a slow page or
+            # changed markup. Raising keeps the attempt retryable instead
+            # of a terminal (and false) added_no_version.
+            raise RuntimeError(
+                "version picker never rendered — site slow or markup "
+                "changed; attempt is retryable"
+            ) from err
         for _ in range(MAX_VERSION_PAGES):
             items = dialog.get_by_role("listitem").filter(has_text=pattern)
             if items.count():
@@ -396,7 +430,8 @@ def _process(
         suffix = f" — {note}" if note else ""
         typer.echo(f"  {job.name}: {status}{suffix}")
         if status == "failed":
-            consecutive = (note, consecutive[1] + 1 if note == consecutive[0] else 1)
+            kind = note.split(":", 1)[0]  # exception type from _scrub format
+            consecutive = (kind, consecutive[1] + 1 if kind == consecutive[0] else 1)
             if consecutive[1] >= 3:
                 typer.echo(
                     "  aborting — 3 identical consecutive failures look "
@@ -418,14 +453,33 @@ def verify_uploads(log_rows: list[dict], collection: list[CollectionItem]) -> li
         by_collid[item.coll_id] = item
 
     problems = []
+    added_copies: Counter[int] = Counter()
+    seen_add_keys: set[tuple[str, str, str]] = set()
     latest: dict[tuple[str, str, str], dict] = {}
     for row in log_rows:
-        latest[_job_key(row)] = row
+        k = _job_key(row)
+        latest[k] = row
+        if row["action"] == "add" and row["status"] in DONE_STATUSES:
+            added_copies[int(row["bgg_id"])] += 1
     for row in latest.values():
         if row["status"] in ("added", "added_no_version"):
             copies = by_object.get(int(row["bgg_id"]), [])
             if not copies:
                 problems.append(f"{row['name']}: logged added but not in collection")
+            elif (
+                _job_key(row) not in seen_add_keys
+                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))
+                problems.append(
+                    f"{row['name']}: {added_copies[int(row['bgg_id'])]} "
+                    f"add(s) logged but only {len(copies)} cop"
+                    f"{'y' if len(copies) == 1 else 'ies'} in the collection"
+                    " — a second-copy add may have edited an existing entry"
+                )
             elif (
                 row["status"] == "added"  # no_version: absence is expected
                 and row["version_id"]
@@ -494,6 +548,11 @@ def run_upload(
         typer.echo(f"{cfg.to_add_path} not found — run `bggpipe diff` first.")
         raise typer.Exit(code=1)
     to_add = _read_csv(cfg.to_add_path)
+    if not cfg.to_update_path.exists():
+        typer.echo(
+            f"note: {cfg.to_update_path} not found — no version updates "
+            "queued (re-run `bggpipe diff` if that's unexpected)"
+        )
     to_update = _read_csv(cfg.to_update_path)
     log_rows = _read_csv(log_path)
 

src/bggpipe/webreview.py

@@ -28,8 +28,11 @@ from pydantic import BaseModel
 from rich.console import Console
 
 from bggpipe.bgg_client import BGGClient
-from bggpipe.config import DEFAULT_REVIEW_PORT as DEFAULT_PORT  # single home
-from bggpipe.config import Config
+from bggpipe.config import DEFAULT_REVIEW_PORT, Config
+from bggpipe.models import (
+    CONFIDENT_VERSION_STATUSES,
+    RECOGNIZED_MATCH_STATUSES,
+)
 from bggpipe.review import ReviewSession
 
 
@@ -51,6 +54,13 @@ def load_thumbnails(cache_dir: Path) -> dict[int, str]:
     return thumbnails
 
 
+def _ix_of(rows: list[dict], row: dict) -> int:
+    """Index by IDENTITY: list.index compares by ==, which returns the
+    first of two equal duplicate rows for both — defeating the ordinal
+    disambiguation row_ix exists to provide."""
+    return next(i for i, r in enumerate(rows) if r is row)
+
+
 def _sighting_key(photo: str, sighting: dict) -> str:
     return "|".join(
         [
@@ -129,9 +139,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
         if session.reload_if_changed():
             thumbnails = load_thumbnails(cfg.cache_dir)
 
-    def find_row(
-        title_raw: str, source_photos: str, row_ix: int | None = None
-    ) -> dict:
+    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
@@ -139,9 +147,18 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
             row = session.rows[row_ix]
             if row["title_raw"] == title_raw and row["source_photos"] == source_photos:
                 return row
-        for row in session.rows:
-            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?")
 
     def photo_names() -> set[str]:
@@ -156,7 +173,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
         for c in candidates:
             c["thumbnail"] = thumbnails.get(c.get("bgg_id"))
         return {
-            "row_ix": session.rows.index(row),
+            "row_ix": _ix_of(session.rows, row),
             "title_raw": row["title_raw"],
             "source_photos": row["source_photos"],
             "match_status": row["match_status"],
@@ -173,7 +190,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
 
     def version_payload(row: dict) -> dict:
         return {
-            "row_ix": session.rows.index(row),
+            "row_ix": _ix_of(session.rows, row),
             "title_raw": row["title_raw"],
             "source_photos": row["source_photos"],
             "bgg_name": row["bgg_name"],
@@ -208,8 +225,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
         version_updates = sum(
             1
             for r in session.rows
-            if r["version_status"] in ("version_auto", "version_approved")
-            and r["version_id"]
+            if r["version_status"] in CONFIDENT_VERSION_STATUSES and r["version_id"]
         )
         available = photo_names()
         sightings = []
@@ -223,6 +239,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
                     )
         merges = [
             {
+                "row_ix": _ix_of(session.rows, r),
                 "title_raw": r["title_raw"],
                 "source_photos": r["source_photos"],
                 "merged_into": r.get("merged_into", ""),
@@ -240,7 +257,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
             "catalog": catalog,
             "decisions": session.decisions,
             "summary": {
-                "recognized": counts.get("auto", 0) + counts.get("approved", 0),
+                "recognized": sum(counts.get(s, 0) for s in RECOGNIZED_MATCH_STATUSES),
                 "ambiguous": counts.get("ambiguous", 0),
                 "unmatched": counts.get("unmatched", 0),
                 "rejected": counts.get("rejected", 0),
@@ -315,9 +332,7 @@ def create_app(cfg: Config, *, client: BGGClient | None = None) -> FastAPI:
     @app.post("/api/dismiss")
     def api_dismiss(body: DismissBody) -> dict:
         with lock:
-            dismissed.add(
-                _sighting_key(body.photo, body.model_dump(exclude={"photo"}))
-            )
+            dismissed.add(_sighting_key(body.photo, body.model_dump(exclude={"photo"})))
             return state()
 
     @app.get("/photos/{name}")
@@ -355,7 +370,7 @@ def _dev_app() -> FastAPI:
 def run_web_review(
     cfg: Config,
     *,
-    port: int = DEFAULT_PORT,
+    port: int = DEFAULT_REVIEW_PORT,
     dev: bool = False,
     config_path: Path | None = None,
 ) -> None:

tests/test_client.py

@@ -1,3 +1,5 @@
+"""BGG client tests: canned transports, fake clocks — never online."""
+
 from __future__ import annotations
 
 import random
@@ -182,3 +184,13 @@ def test_collection_item_missing_collid_refuses_to_parse(tmp_path):
     )
     with pytest.raises(BGGResponseError):
         parse_collection(bad_xml)
+
+
+def test_malformed_xml_raises_and_is_never_cached(tmp_path):
+    from bggpipe.models import BGGResponseError
+
+    torn = '<items total="1"><item type="boardgame" id="13"><na'
+    client, _, _ = make_client(tmp_path, [(200, torn)])
+    with pytest.raises(BGGResponseError):
+        client.get_xml("search", {"query": "catan", "type": "boardgame"})
+    assert list((tmp_path / "cache").glob("*.xml")) == []

tests/test_diff.py

@@ -10,6 +10,10 @@ from bggpipe.diff import compute_diff, load_snapshot_collection
 from bggpipe.models import CollectionItem
 
 FIXTURES = Path(__file__).parent / "fixtures"
+SNAPSHOT_NAMES = (
+    "collection_snapshot_base.xml",
+    "collection_snapshot_expansions.xml",
+)
 
 
 def _item(object_id, coll_id, name="Game", version_id=None, own=True):
@@ -115,10 +119,10 @@ def test_owned_with_matching_version_is_just_owned():
     assert not result.to_update and not result.to_add
 
 
-def test_confident_version_matching_no_copy_is_a_second_copy_to_add():
-    # Spec: a (bgg_id, version_id) pair is owned only if a collection item
-    # matches BOTH. All copies carry different versions -> this is an
-    # additional physical copy; existing entries are never edited.
+def test_version_mismatch_with_unclaimed_copy_is_report_only():
+    # ONE row, ONE copy with a different version: most likely the same
+    # physical box mis-scored. Spec: report the disagreement, touch nothing,
+    # and never risk uploading a duplicate entry.
     result = compute_diff(
         [
             _match(
@@ -131,10 +135,47 @@ def test_confident_version_matching_no_copy_is_a_second_copy_to_add():
         ],
         [_item(266192, 5, version_id=465063)],
     )
-    assert not result.to_update  # additive only: never edit a set version
-    assert [r["version_id"] for r in result.to_add] == ["521212"]
-    assert "fourth printing" in result.second_copies[0]
-    assert result.already_owned == []
+    assert not result.to_update and not result.to_add
+    assert result.already_owned == ["Wingspan"]
+    assert "fourth printing" in result.disagreements[0]
+
+
+def test_vetoed_duplicate_of_same_version_is_a_real_second_copy():
+    # Two rows, same confident version, ONE owned copy with that version:
+    # a human vetoed the merge ("these ARE two boxes"), so the exact-version
+    # match must consume the copy and the second row must become an add.
+    rows = [
+        _match("Catan", "13", vstatus="version_auto", vid="123", vname="3rd ed."),
+        _match("Catan", "13", vstatus="version_auto", vid="123", vname="3rd ed."),
+    ]
+    result = compute_diff(rows, [_item(13, 900, version_id=123)])
+    assert result.already_owned == ["Catan"]
+    assert [r["version_id"] for r in result.to_add] == ["123"]
+    assert len(result.second_copies) == 1
+
+
+def test_bare_duplicate_beyond_owned_count_is_added_versionless():
+    # Two vetoed version-unknown rows, one owned copy: the extra bare row
+    # is a version-less second copy, not silently "already owned".
+    rows = [_match("Catan", "13"), _match("Catan", "13")]
+    result = compute_diff(rows, [_item(13, 900)])
+    assert result.already_owned == ["Catan"]
+    (added,) = result.to_add
+    assert added["version_id"] == ""
+    assert len(result.second_copies) == 1
+
+
+def test_bare_row_does_not_steal_versionless_copy_from_confident_update():
+    # ordering independence: the confident row upgrades the versionless
+    # copy even when a bare row of the same game appears first in the file
+    rows = [
+        _match("Catan", "13"),
+        _match("Catan", "13", vstatus="version_auto", vid="55", vname="5th ed."),
+    ]
+    result = compute_diff(rows, [_item(13, 900), _item(13, 901)])
+    assert [u["version_id"] for u in result.to_update] == ["55"]
+    assert result.to_add == []
+    assert result.already_owned.count("Catan") == 2
 
 
 def test_version_unknown_owned_by_bare_id():
@@ -237,3 +278,22 @@ def test_run_diff_outputs_feed_upload_unchanged(tmp_path, monkeypatch):
     assert [(j.action, j.bgg_id, j.name) for j in fake.calls] == [
         ("add", "266192", "Wingspan")
     ]
+
+
+def test_token_without_username_says_so(tmp_path, monkeypatch, capsys):
+    import shutil
+
+    from bggpipe.diff import run_diff
+    from bggpipe.resolve import write_matches
+
+    monkeypatch.setenv("BGG_API_TOKEN", "tok")
+    monkeypatch.delenv("BGG_USERNAME", raising=False)
+    cfg = Config(data_dir=tmp_path)  # bgg_username defaults to ""
+    fixtures = Path(__file__).parent / "fixtures"
+    for name in SNAPSHOT_NAMES:
+        shutil.copy(fixtures / name, tmp_path / name)
+    write_matches(cfg.matches_path, [_match("Catan", "13")])
+    run_diff(cfg)
+    out = capsys.readouterr().out
+    assert "BGG_API_TOKEN is set but BGG_USERNAME is not" in out
+    assert "No BGG_API_TOKEN" not in out  # the old message was a lie here

tests/test_resolve.py

@@ -590,3 +590,44 @@ def test_empty_normalized_title_never_matches(client):
     entry = TitleEntry(title_raw="风声", title_normalized="")
     # any cached query works; candidates must be rejected regardless of name
     assert _plausible_candidates(client, entry, "Catan") == []
+
+
+def test_blocked_same_title_entry_defers_the_whole_group(tmp_path):
+    # entry1 of a two-edition title is blocked (no token); entry2 must NOT
+    # resolve, or its row would occupy entry1's pairing slot next run
+    import httpx as _httpx
+
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    (data_dir / "titles.json").write_text(
+        json.dumps(
+            [
+                {
+                    "title_raw": "Catan",
+                    "edition_hint": "3rd edition",
+                    "source_photos": ["a.jpg"],
+                },
+                {
+                    "title_raw": "Catan",
+                    "edition_hint": "5th edition",
+                    "source_photos": ["b.jpg"],
+                },
+            ]
+        )
+    )
+    cfg = Config(data_dir=data_dir)
+    blocked_client = BGGClient(
+        cache_dir=tmp_path / "empty_cache",
+        transport=_httpx.MockTransport(
+            lambda req: _httpx.Response(401, text="Unauthorized")
+        ),
+        sleep=lambda s: None,
+    )
+    run_resolve(cfg, client=blocked_client)
+    assert read_matches(cfg.matches_path) == []  # both deferred, none misplaced
+
+
+def test_truncation_separator_chosen_by_position():
+    heads = _truncation_heads("Blorvath: Quest of the Zzyzx - 2nd Edition")
+    assert heads[0] == "Blorvath"
+    assert "Blorvath: Quest" not in heads  # the comment's guarantee, now true

tests/test_review.py

@@ -382,3 +382,40 @@ def test_manual_id_unknown_to_bgg_warns_instead_of_crashing(tmp_path):
     assert session.rows[0]["match_status"] == "approved"
     assert session.rows[0]["bgg_id"] == "999999"
     assert any("no game with id 999999" in w for w in session.warnings)
+
+
+def test_every_tui_decision_after_external_rewrite_is_saved(tmp_path):
+    # THE round-2 catch: the TUI iterates row references snapshotted before
+    # any reload; after decision 1 triggers a reload, decisions 2..N used
+    # to be counted but never written.
+    from bggpipe.resolve import read_matches, write_matches
+
+    cfg = _setup(
+        tmp_path,
+        [
+            _row(title_raw="Alpha", match_status="unmatched"),
+            _row(title_raw="Beta", match_status="unmatched"),
+            _row(title_raw="Gamma", match_status="unmatched"),
+        ],
+    )
+    session = ReviewSession(
+        cfg,
+        console=quiet_console(),
+        input_fn=scripted(),
+        client=unauthorized_client(tmp_path),
+    )
+    stale_refs = list(session.pending_rows())  # what run() iterates
+
+    external = read_matches(cfg.matches_path)
+    external.append(_row(title_raw="Newcomer", match_status="auto", bgg_id="7"))
+    write_matches(cfg.matches_path, external)
+
+    for ref in stale_refs:  # decision 1 reloads; 2 and 3 are orphaned refs
+        session.decide_reject(ref)
+
+    saved = {r["title_raw"]: r for r in read_matches(cfg.matches_path)}
+    assert [saved[t]["match_status"] for t in ("Alpha", "Beta", "Gamma")] == (
+        ["rejected"] * 3
+    )
+    assert "Newcomer" in saved  # the external row survived too
+    assert session.decisions == 3

tests/test_upload.py

@@ -383,3 +383,45 @@ def test_real_run_without_credentials_exits_before_any_browser(tmp_path, monkeyp
     with pytest.raises(typer.Exit):
         run_upload(cfg, sleep=lambda s: None, now=NOW)  # uploader=None: real path
     assert not (tmp_path / "upload_log.csv").exists()
+
+
+def test_same_key_second_copy_survives_limit_and_interrupts(tmp_path):
+    # two vetoed duplicate copies share ("add", bgg_id, version): one logged
+    # success must complete exactly ONE of them, not both
+    cfg = _cfg(tmp_path)
+    twin = _add_row(bgg_id="13", name="Catan", version_id="123", version_name="3rd")
+    _seed_data(tmp_path, to_add=[dict(twin), dict(twin)])
+
+    first = FakeUploader()
+    run_upload(cfg, uploader=first, limit=1, sleep=lambda s: None, now=NOW)
+    assert len(first.calls) == 1
+
+    second = FakeUploader()
+    run_upload(cfg, uploader=second, sleep=lambda s: None, now=NOW)
+    assert len(second.calls) == 1  # the second copy, not zero, not two
+
+    third = FakeUploader()
+    assert run_upload(cfg, uploader=third, sleep=lambda s: None, now=NOW) == []
+
+
+def test_consecutive_failure_counter_resets_on_success(tmp_path):
+    class FlakyPairs(FakeUploader):
+        def add_game(self, job):
+            self.calls.append(job)
+            if job.bgg_id in ("1", "2", "4", "5"):
+                raise RuntimeError("dialog never appeared")
+            return "added", ""
+
+    cfg = _cfg(tmp_path)
+    _seed_data(tmp_path, to_add=[_add_row(bgg_id=str(i)) for i in range(1, 7)])
+    fake = FlakyPairs()
+    run_upload(cfg, uploader=fake, sleep=lambda s: None, now=NOW)
+    assert len(fake.calls) == 6  # fail,fail,ok,fail,fail,ok — never aborts
+
+
+def test_empty_game_name_is_refused_not_uploaded(tmp_path):
+    cfg = _cfg(tmp_path)
+    _seed_data(tmp_path, to_add=[_add_row(bgg_id="42", name="")])
+    fake = FakeUploader()
+    results = run_upload(cfg, uploader=fake, sleep=lambda s: None, now=NOW)
+    assert results == [] and fake.calls == []

tests/test_webreview.py

@@ -403,3 +403,32 @@ def test_session_warnings_surface_in_state(tmp_path):
     warnings = res.json()["warnings"]
     assert any("couldn't look up id 42" in w for w in warnings)
     assert warnings == web.get("/api/state").json()["warnings"]
+
+
+def test_duplicate_rows_are_individually_decidable_via_row_ix(tmp_path):
+    # two editions of one game in one photo: byte-identical rows. The
+    # ordinal must land each decision on its own row.
+    from bggpipe.resolve import read_matches as read_m
+    from bggpipe.resolve import write_matches as write_m
+
+    cfg = make_cfg(tmp_path)
+    dup = _row(title_raw="Twins", match_status="unmatched")
+    write_m(cfg.matches_path, [dict(dup), dict(dup)])
+    web = TestClient(create_app(cfg, client=unauthorized_client(tmp_path)))
+
+    pending = web.get("/api/state").json()["pending"]
+    assert [p["title_raw"] for p in pending] == ["Twins", "Twins"]
+    assert pending[0]["row_ix"] != pending[1]["row_ix"]  # identity, not ==
+
+    second = pending[1]
+    web.post(
+        "/api/decision",
+        json={
+            "title_raw": second["title_raw"],
+            "source_photos": second["source_photos"],
+            "row_ix": second["row_ix"],
+            "action": "reject",
+        },
+    )
+    rows = read_m(cfg.matches_path)
+    assert [r["match_status"] for r in rows] == ["unmatched", "rejected"]

uv.lock

@@ -64,6 +64,7 @@ dependencies = [
     { name = "pillow" },
     { name = "pillow-heif" },
     { name = "playwright" },
+    { name = "pydantic" },
     { name = "rapidfuzz" },
     { name = "rich" },
     { name = "typer" },
@@ -85,6 +86,7 @@ requires-dist = [
     { name = "pillow", specifier = ">=12.3.0" },
     { name = "pillow-heif", specifier = ">=1.5.0" },
     { name = "playwright", specifier = ">=1.62.0" },
+    { name = "pydantic", specifier = ">=2.13.4" },
     { name = "rapidfuzz", specifier = ">=3.9" },
     { name = "rich", specifier = ">=15.0.0" },
     { name = "typer", specifier = ">=0.12" },