Fix participant-file truncation: serialize + never-shrink saves, blocking final save

app.py

@@ -471,20 +471,57 @@ def _row_to_trial(row: pd.Series) -> dict:
     }
 
 
+# Per-participant save coordination. Uploads for a given participant file are
+# serialized through one lock, and we never overwrite a larger file with a
+# smaller (stale) snapshot. This prevents the out-of-order/last-writer-wins race
+# that previously truncated participant files when clicks were saved from
+# unsynchronized background threads.
+_SAVE_REGISTRY_LOCK = threading.Lock()
+_SAVE_ENTRIES: dict[str, dict] = {}
+
+
+def _save_entry(participant_file: str) -> dict:
+    with _SAVE_REGISTRY_LOCK:
+        entry = _SAVE_ENTRIES.get(participant_file)
+        if entry is None:
+            entry = {"lock": threading.Lock(), "saved_count": 0}
+            _SAVE_ENTRIES[participant_file] = entry
+        return entry
+
+
+def _reset_save_baseline(participant_file: str, count: int) -> None:
+    """Align the never-shrink guard with what's actually on HF at session start."""
+    entry = _save_entry(participant_file)
+    with entry["lock"]:
+        entry["saved_count"] = count
+
+
 def _save_results(participant_file: str, results: list[dict]) -> None:
     if not HF_TOKEN or not results:
         return
-    frame = pd.DataFrame(results, columns=RESULTS_COLUMNS)
-    buf = io.BytesIO()
-    frame.to_csv(buf, index=False)
-    buf.seek(0)
-    api.upload_file(
-        path_or_fileobj=buf,
-        path_in_repo=participant_file,
-        repo_id=RESULTS_REPO,
-        repo_type="dataset",
-        commit_message=f"Update {participant_file} (n={len(results)})",
-    )
+    snapshot = list(results)
+    entry = _save_entry(participant_file)
+    # Serialize all uploads for this participant so they can't race each other.
+    with entry["lock"]:
+        # Never replace a more-complete file with a stale/smaller snapshot.
+        if len(snapshot) <= entry["saved_count"]:
+            return
+        frame = pd.DataFrame(snapshot, columns=RESULTS_COLUMNS)
+        buf = io.BytesIO()
+        frame.to_csv(buf, index=False)
+        buf.seek(0)
+        try:
+            api.upload_file(
+                path_or_fileobj=buf,
+                path_in_repo=participant_file,
+                repo_id=RESULTS_REPO,
+                repo_type="dataset",
+                commit_message=f"Update {participant_file} (n={len(snapshot)})",
+            )
+        except Exception as exc:  # noqa: BLE001
+            print(f"[save] WARNING: upload failed for {participant_file} ({exc}).")
+            return
+        entry["saved_count"] = len(snapshot)
 
 
 # ---------------------------------------------------------------------------
@@ -548,6 +585,9 @@ def start_session(access_code: str):
 
     participant_file = _participant_filename(code)
     prior = _load_participant_results(participant_file)
+    # Baseline the never-shrink save guard to the file that's actually on HF,
+    # so a returning participant's saves grow from their real prior progress.
+    _reset_save_baseline(participant_file, len(prior))
 
     if _is_complete(prior):
         msg = DONE_ALREADY_HTML_TMPL.format(
@@ -649,14 +689,21 @@ def _make_choice(state: dict, side: str):
         }
     )
 
-    threading.Thread(
-        target=_save_results,
-        args=(state["participant_file"], list(state["results"])),
-        daemon=True,
-    ).start()
-
     state["current_idx"] += 1
-    if state["current_idx"] >= len(state["trials"]):
+    is_done = state["current_idx"] >= len(state["trials"])
+
+    if is_done:
+        # Final trial: save synchronously so completion is guaranteed persisted
+        # (all trials written) before we show the "done" panel.
+        _save_results(state["participant_file"], list(state["results"]))
+    else:
+        threading.Thread(
+            target=_save_results,
+            args=(state["participant_file"], list(state["results"])),
+            daemon=True,
+        ).start()
+
+    if is_done:
         total = state["total_trials"]
         return (
             state,