Audit round 7, upload cluster: evidence over inference at every exit

Five blind reviewers swept the real-data-era surface; this lands the
upload findings, all verified against the code and the documented site
behavior before fixing.

The two HIGHs shared a root: logging outcomes the browser never proved.
add_game waited for an "Add To" button that an owned game's page does
not have — so a second-copy add could never succeed, and worse, an add
that LANDED but missed the log became an unretryable failure loop
(every retry: 30s timeout, logged failed, nothing ever settles).
add_game now polls for either button state: "In Collections" without
second_copy returns the previously-dead already_present status (the
landed-but-unlogged case heals itself on retry); with second_copy it
refuses loudly (that flow is unverified — add by hand). A save whose
dialog is slow to hide reloads the page and asks for ownership evidence
instead of guessing "failed". update_entry no longer trusts the editor
merely closing: the cell must settle on text matching the CHOSEN
version, else the AJAX save failed server-side and "updated" would
mark a job done forever that never touched the site.

Per-copy bookkeeping: stale_jobs endorsed per game, so rejecting one
of two queued editions let the rejected copy upload on the survivor's
endorsement — it now counts endorsements per (bgg_id, version) and
retires the game with "re-run diff" when a copy loses its backing.
annotate_queue stamped every row sharing a job key with the same log
status, so one success marked both vetoed duplicates done; completions
are now claimed one row per done log line.

Smaller findings: the version-drift note queued a doomed re-add after
warning about it (now skips — the entry exists on BGG; re-adding only
duplicates); the one-update-per-game deferral rested on a claim the
collid-exact editor disproves (removed — same-game updates run
together); the 3-identical-failures abort compared exception class
only, so three unrelated problems aborted a healthy run (now compares
whole messages).

Also from the test seat: run_upload's stale filtering finally executes
against a real matches.csv in tests; rejected credentials pin that no
anonymous storage state is saved; update_entry's three guarded exits
each have a test; _scrub's newline flattening is pinned.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016jXZFSTZQKzAC8fqpWSz9g

src/bggpipe/upload.py

@@ -58,6 +58,16 @@ DONE_STATUSES = {"added", "added_no_version", "updated", "already_present"}
 MAX_VERSION_PAGES = 40
 
 
+_YEARISH = re.compile(r"[\s(]*\d{4}(?:-\d+)?[\s)]*")
+
+
+def _loose_version_text(text: str) -> str:
+    """Version names as displayed differ from the API's by punctuation,
+    parens and year qualifiers — compare only the words that name it."""
+    text = _YEARISH.sub(" ", text or "")
+    return " ".join(re.sub(r"[^\w\s]", " ", text.casefold()).split())
+
+
 @dataclass(frozen=True)
 class UploadJob:
     action: str  # "add" | "update"
@@ -99,8 +109,14 @@ def annotate_queue(
     job succeeds — so a reader without the log sees finished work as
     outstanding forever."""
     last: dict[tuple[str, str, str], str] = {}
+    done_count: Counter[tuple[str, str, str]] = Counter()
     for row in log_rows:
         last[_job_key(row)] = row["status"]
+        if row["status"] in DONE_STATUSES:
+            done_count[_job_key(row)] += 1
+    # vetoed duplicate copies share a key: one success must mark ONE row
+    # done, not both, or the queue reports a copy uploaded that never was
+    claimed: Counter[tuple[str, str, str]] = Counter()
     out = []
     for row in queue_rows:
         key = _key(
@@ -110,6 +126,10 @@ def annotate_queue(
             row.get("version_id", ""),
         )
         status = last.get(key, "")
+        if status in DONE_STATUSES and claimed[key] >= done_count[key]:
+            status = ""  # completions exhausted: this copy is still pending
+        if status in DONE_STATUSES:
+            claimed[key] += 1
         out.append(
             {
                 **row,
@@ -134,20 +154,34 @@ def stale_jobs(queue_rows: list[dict], match_rows: list[dict]) -> dict[str, str]
         # driving the queue directly): absence is not a verdict
         return {}
     live: dict[str, list[dict]] = {}
+    endorsed: Counter[tuple[str, str]] = Counter()
     for row in match_rows:
         if row.get("bgg_id"):
             live.setdefault(row["bgg_id"], []).append(row)
+            if is_recognized(row):
+                endorsed[(row["bgg_id"], row.get("version_id", ""))] += 1
+    queued: Counter[tuple[str, str]] = Counter()
+    for row in queue_rows:
+        queued[(row.get("bgg_id") or "", row.get("version_id", ""))] += 1
     stale = {}
     for row in queue_rows:
         bgg_id = row.get("bgg_id") or ""
         rows = live.get(bgg_id, [])
-        if any(is_recognized(r) for r in rows):
-            continue
         if not rows:
             stale[bgg_id] = "no longer matched to this game in matches.csv"
-        else:
+            continue
+        if not any(is_recognized(r) for r in rows):
             statuses = sorted({r["match_status"] for r in rows})
             stale[bgg_id] = f"now {', '.join(statuses)} in matches.csv"
+            continue
+        # the game survives, but does THIS copy? Rejecting one of two
+        # editions must not ride along on the other's endorsement.
+        pair = (bgg_id, row.get("version_id", ""))
+        if queued[pair] > endorsed[pair]:
+            stale[bgg_id] = (
+                f"queued with version {row.get('version_id') or '(none)'} "
+                "but matches.csv no longer endorses that copy — re-run diff"
+            )
     return stale
 
 
@@ -248,8 +282,7 @@ def build_queue(
 
     jobs: list[UploadJob] = []
     skipped_done = skipped_failed = 0
-    deferred: list[UploadJob] = []
-    update_game_seen: set[str] = set()
+    deferred: list[UploadJob] = []  # kept for callers; nothing defers now
     seen: Counter[tuple[str, str, str]] = Counter()
     queued_versions: dict[str, set[str]] = {}
     for job in candidates:
@@ -266,11 +299,13 @@ def build_queue(
             prior = set()
         if prior and job.version_id not in prior:
             typer.echo(
-                f"  note: {job.name} was previously {job.action}ed with a "
+                f"  skipping {job.name}: 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)"
+                "the entry exists on BGG, so re-adding can only duplicate "
+                "it; correct the version by hand (additive-only rule)"
             )
+            skipped_done += 1
+            continue
         occurrence = seen[job.key]
         seen[job.key] += 1
         if not job.name:
@@ -285,15 +320,10 @@ def build_queue(
             skipped_done += 1
         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
-            # second same-game update this run could reopen the copy the
-            # first one just versioned and overwrite it. One per run; the
-            # next run (after --verify) picks up the rest.
-            deferred.append(job)
         else:
-            if job.action == "update":
-                update_game_seen.add(job.bgg_id)
+            # same-game updates coexist in one run: update_entry addresses
+            # the copy by collid and the edition by radio value, so a
+            # second update cannot reopen what the first just saved
             jobs.append(job)
     return jobs, skipped_done, skipped_failed, deferred
 
@@ -586,13 +616,36 @@ class PlaywrightUploader:
             self._page.wait_for_timeout(400)
         return False
 
+    def _owned_button(self):
+        """The owned-game page replaces "Add To" with "In Collections" —
+        positive evidence the collection already holds this game."""
+        return self._page.get_by_role("button", name="In Collections")
+
     def add_game(self, job: UploadJob) -> tuple[str, str]:
         self._ensure_logged_in()
         page = self._page
         # /boardgame/<id> redirects to the canonical slug for any subtype.
         self._goto(f"{BGG}/boardgame/{job.bgg_id}/")
-        add_btn = page.get_by_role("button", name="Add To").first
-        dialog = self._open_dialog(add_btn)
+        add_btn = page.get_by_role("button", name="Add To")
+        for _ in range(60):  # the header hydrates late: poll for EITHER state
+            if add_btn.count() and add_btn.first.is_visible():
+                break
+            owned = self._owned_button()
+            if owned.count() and owned.first.is_visible():
+                if job.second_copy:
+                    # the second-copy path goes through the In Collections
+                    # dialog, which this code has never driven live
+                    raise RuntimeError(
+                        "second copy of an owned game — the In Collections "
+                        "add-a-copy flow is unverified; add this copy by hand"
+                    )
+                return (
+                    "already_present",
+                    "the site already lists this game as owned — a previous "
+                    "attempt likely landed without reaching the log",
+                )
+            page.wait_for_timeout(500)
+        dialog = self._open_dialog(add_btn.first)
         # Wait for the form itself, NOT for a heading matching our stored
         # name: a match made through an ALTERNATE name (BGG 140509 is
         # "Dungeons & Dragons" to search, "Dragones Y Mazmorras" on the
@@ -612,7 +665,21 @@ class PlaywrightUploader:
                 note = f"version {job.version_name!r}: {why}; added without version"
         dialog.get_by_role("button", name="Save").click()
         # The dialog is hidden after save, not removed from the DOM.
-        dialog.wait_for(state="hidden", timeout=15_000)
+        try:
+            dialog.wait_for(state="hidden", timeout=15_000)
+        except self._timeout_error:
+            # a slow hide is not a failed save: reload and ask the page.
+            # Logging "failed" for a landed add would double-add on retry.
+            self._goto(f"{BGG}/boardgame/{job.bgg_id}/")
+            owned = self._owned_button()
+            try:
+                owned.first.wait_for(timeout=15_000)
+            except self._timeout_error as err:
+                raise RuntimeError(
+                    "the save dialog never closed and the page does not "
+                    "show the game as owned — the add may not have landed"
+                ) from err
+            note = (note + "; " if note else "") + "save confirmed via page reload"
         return status, note
 
     def update_entry(self, job: UploadJob) -> tuple[str, str]:
@@ -650,9 +717,20 @@ class PlaywrightUploader:
                 f"{job.collid} — entry left untouched"
             ) from err
         radio.first.click()  # fires CE_SaveData: no separate Save button
+        want = _loose_version_text(job.version_name)
         for _ in range(40):
             settled = " ".join((cell.first.text_content() or "").split())
             if settled and "editing" not in settled.casefold():
+                got = _loose_version_text(settled)
+                if want and want not in got and got not in want:
+                    # the editor closed but re-rendered its OLD content:
+                    # the AJAX save failed server-side. "updated" here
+                    # would mark the job done forever without evidence.
+                    raise RuntimeError(
+                        f"the version cell settled on {settled!r}, not the "
+                        f"chosen {job.version_name!r} — the save did not "
+                        "land; safe to retry (same radio, same result)"
+                    )
                 return "updated", ""
             page.wait_for_timeout(500)
         raise RuntimeError(
@@ -706,7 +784,7 @@ def _process(
         suffix = f" — {note}" if note else ""
         typer.echo(f"  {job.name}: {status}{suffix}")
         if status == "failed":
-            kind = note.split(":", 1)[0]  # exception type from _scrub format
+            kind = note  # IDENTICAL means the whole message, not the class
             consecutive = (kind, consecutive[1] + 1 if kind == consecutive[0] else 1)
             if consecutive[1] >= 3:
                 typer.echo(