Merge pull request #1993 from transformerlab/fix/caching-issue-lab-finish

Fix caching issue when saving score during lab finish

Author: deep1401

api/localprovider_pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
   "soundfile==0.13.1",
   "tensorboardX==2.6.2.2",
   "timm==1.0.15",
-  "transformerlab==0.1.31",
+  "transformerlab==0.1.32",
   "transformerlab-inference==0.2.52",
   "transformers==4.57.1",
   "wandb==0.23.1",

api/pyproject.toml

@@ -25,7 +25,7 @@ dependencies = [
   "python-dotenv==1.1.0",
   "python-multipart==0.0.20",
   "sqlalchemy[asyncio]==2.0.40",
-  "transformerlab==0.1.31",
+  "transformerlab==0.1.32",
   "transformerlab-inference==0.2.52",
   "uvicorn==0.35.0",
   "watchfiles==1.0.5",

lab-sdk/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "transformerlab"
-version = "0.1.31"
+version = "0.1.32"
 description = "Python SDK for Transformer Lab"
 readme = "README.md"
 requires-python = ">=3.10"

lab-sdk/src/lab/lab_facade.py

@@ -630,15 +630,28 @@ def finish(
         except Exception:
             pass
         _run_async(self._job.update_progress(100))  # type: ignore[union-attr]
-        _run_async(
-            self._job.update_job_data_fields(  # type: ignore[union-attr]
-                {
-                    "completion_status": "success",
-                    "completion_details": message,
-                    "end_time": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()),
-                }
-            )
-        )
+
+        # Resolve score before the first write so it's included atomically with the
+        # other completion fields. This prevents a window where completion_status is
+        # "success" but score is still null (observable by caches or concurrent readers).
+        resolved_score: Dict[str, Any] = {}
+        if isinstance(score, dict):
+            resolved_score.update(score)
+        resolved_score.setdefault("discard", False)
+
+        completion_fields: Dict[str, Any] = {
+            "completion_status": "success",
+            "completion_details": message,
+            "end_time": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()),
+            "score": resolved_score,
+        }
+        if additional_output_path is not None and additional_output_path.strip() != "":
+            completion_fields["additional_output_path"] = additional_output_path
+        if plot_data_path is not None and plot_data_path.strip() != "":
+            completion_fields["plot_data_path"] = plot_data_path
+
+        _run_async(self._job.update_job_data_fields(completion_fields))  # type: ignore[union-attr]
+
         # Best-effort Trackio integration: finish our own run if we created one,
         # then capture the active Trackio DB (managed or user-created) into job artifacts.
         try:
@@ -662,15 +675,6 @@ def finish(
         except Exception:
             # Never let optional Trackio integration break finish()
             logger.debug("Trackio integration failed during finish()", exc_info=True)
-        resolved_score: Dict[str, Any] = {}
-        if isinstance(score, dict):
-            resolved_score.update(score)
-        resolved_score.setdefault("discard", False)
-        _run_async(self._job.update_job_data_field("score", resolved_score))  # type: ignore[union-attr]
-        if additional_output_path is not None and additional_output_path.strip() != "":
-            _run_async(self._job.update_job_data_field("additional_output_path", additional_output_path))  # type: ignore[union-attr]
-        if plot_data_path is not None and plot_data_path.strip() != "":
-            _run_async(self._job.update_job_data_field("plot_data_path", plot_data_path))  # type: ignore[union-attr]
 
         # Important: update cached_jobs only when all completion fields are already written.
         _run_async(self._job.update_status(JobStatus.COMPLETE))  # type: ignore[union-attr]