fix(localprocess): ensure __get_job_status returns Complete correctly

docs/source/train/options.rst

@@ -60,3 +60,37 @@ Options Reference
 .. autoclass:: kubeflow.trainer.options.ContainerPatch
    :members:
    :show-inheritance:
+
+Using options with TrainerClient
+===============================
+
+The ``options`` parameter in ``TrainerClient`` allows users to customize runtime behavior
+and backend-specific configurations for training jobs.
+
+It provides flexibility to control how training jobs are executed depending on the
+selected backend (e.g., Kubernetes, local, container).
+
+Example
+-------
+
+.. code-block:: python
+
+    from kubeflow.trainer import TrainerClient, CustomTrainer
+
+    def train_fn():
+        print("Training...")
+
+    client = TrainerClient()
+
+    job_id = client.train(
+        trainer=CustomTrainer(func=train_fn),
+        options={
+            "epochs": 10,
+            "batch_size": 32
+        }
+    )
+
+    client.wait_for_job_status(job_id)
+
+The ``options`` dictionary can include different parameters depending on the backend
+and runtime configuration.

kubeflow/trainer/api/trainer_client.py

@@ -265,6 +265,36 @@ def wait_for_job_status(
             polling_interval=polling_interval,
             callbacks=callbacks,
         )
+    def get_job_progress(self, name: str) -> dict:
+        """Get progress of a TrainJob.
+
+        Args:
+            name: Name of the TrainJob.
+
+        Returns:
+            Dictionary containing job status and progress.
+        """
+
+        # Get job details
+        job = self.get_job(name=name)
+
+        status = job.status if hasattr(job, "status") else "Unknown"
+
+        if status == "Running":
+            progress = "In Progress"
+        elif status in ["Complete", "Succeeded"]:
+            progress = "100%"
+        elif status == "Failed":
+            progress = "Error"
+        else:
+            progress = "Unknown"
+
+        return {
+            "job_id": name,
+            "status": status,
+            "progress": progress,
+        }
+
 
     def delete_job(self, name: str):
         """Delete the TrainJob.

kubeflow/trainer/backends/localprocess/backend.py

@@ -264,17 +264,22 @@ def __get_job_status(self, job: LocalBackendJobs) -> str:
         if not job.steps:
             return constants.TRAINJOB_CREATED
         statuses = [_step.job.status for _step in job.steps]
-        # if status is running or failed will take precedence over completed
+
+    # Priority: Failed > Running > Created > Complete
         if constants.TRAINJOB_FAILED in statuses:
-            status = constants.TRAINJOB_FAILED
-        elif constants.TRAINJOB_RUNNING in statuses:
-            status = constants.TRAINJOB_RUNNING
-        elif constants.TRAINJOB_CREATED in statuses:
-            status = constants.TRAINJOB_CREATED
-        else:
-            status = constants.TRAINJOB_COMPLETE
+            return constants.TRAINJOB_FAILED
+
+        if constants.TRAINJOB_RUNNING in statuses:
+            return constants.TRAINJOB_RUNNING
+        if constants.TRAINJOB_CREATED in statuses:
+            return constants.TRAINJOB_CREATED
+
+    # ✅ NEW FIX: Ensure all steps are actually complete
+        if all(status == constants.TRAINJOB_COMPLETE for status in statuses):
+            return constants.TRAINJOB_COMPLETE
 
-        return status
+    # fallback (safety)
+        return constants.TRAINJOB_RUNNING
 
     def __register_job(
         self,