[Release] Rename save_last_job_result to save_last_job_status

Rename method and parameter to better reflect the new JobStatus type.
Add a warning log when the returned job ID doesn't match the expected
one. Update FakeJobResult to FakeJobStatus in tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: aslonnie

release/ray_release/job_manager/anyscale_job_manager.py

@@ -96,8 +96,12 @@ def _run_job(
         logger.info(f"Link to job: " f"{format_link(self.job_url())}")
         return
 
-    def save_last_job_result(self, value):
-        self._last_job_result = value
+    def save_last_job_status(self, status):
+        if status and hasattr(status, "id") and status.id != self._job_id:
+            logger.warning(
+                f"Job ID mismatch: expected {self._job_id}, got {status.id}"
+            )
+        self._last_job_result = status
 
     def job_id(self) -> Optional[str]:
         return self._job_id
@@ -198,7 +202,7 @@ def _wait_job(self, timeout: int):
                     next_status += 30
 
                 result = self._get_job_status_with_retry()
-                self.save_last_job_result(result)
+                self.save_last_job_status(result)
                 status = self._last_job_status()
 
                 if not job_running and status in {
@@ -219,7 +223,7 @@ def _wait_job(self, timeout: int):
                 time.sleep(1)
 
         result = self._get_job_status_with_retry()
-        self.save_last_job_result(result)
+        self.save_last_job_status(result)
         status = self._last_job_status()
         assert status in terminal_state
         if status == JobState.FAILED and not job_running:

release/ray_release/tests/test_anyscale_job_manager.py

@@ -2,15 +2,17 @@
 
 import pytest
 
+from anyscale.job.models import JobState
+
 from ray_release.anyscale_util import Anyscale
 from ray_release.cluster_manager.cluster_manager import ClusterManager
 from ray_release.job_manager.anyscale_job_manager import AnyscaleJobManager
 from ray_release.test import Test
 
 
-class FakeJobResult:
-    def __init__(self, _id: str):
-        self.id = _id
+class FakeJobStatus:
+    def __init__(self, state: JobState):
+        self.state = state
 
 
 class FakeSDK(Anyscale):
@@ -32,6 +34,7 @@ def test_get_last_logs_long_running_job():
     anyscale_job_manager = AnyscaleJobManager(cluster_manager=cluster_manager)
     anyscale_job_manager._duration = 4 * 3_600 + 1
     anyscale_job_manager._job_id = "foo"
+    anyscale_job_manager.save_last_job_status(FakeJobStatus(state=JobState.SUCCEEDED))
     assert anyscale_job_manager.get_last_logs() is None