Consume recovery job state in the tracking UI

docs/api_reference/source/rest-api.rst

@@ -6966,6 +6966,28 @@ Reference to an issue associated with this trace
 | issue_name | ``STRING`` | The name of the issue this assessment references This field is required. |
 +------------+------------+--------------------------------------------------------------------------+
 
+.. _mlflowJobProgress:
+
+JobProgress
+-----------
+
+
+
+Structured best-effort progress payload for a running job.
+
+
++------------+------------+-------------+
+| Field Name |    Type    | Description |
++============+============+=============+
+| phase      | ``STRING`` |             |
++------------+------------+-------------+
+| completed  | ``INT64``  |             |
++------------+------------+-------------+
+| total      | ``INT64``  |             |
++------------+------------+-------------+
+| unit       | ``STRING`` |             |
++------------+------------+-------------+
+
 .. _mlflowJobState:
 
 JobState
@@ -6977,15 +6999,21 @@ Generic job state message combining status with metadata.
 Provides a unified way to represent job state across different job types.
 
 
-+---------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
-|  Field Name   |                      Type                      |                                         Description                                          |
-+===============+================================================+==============================================================================================+
-| status        | :ref:`mlflowjobstatus`                         | Current status of the job.                                                                   |
-+---------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
-| error_message | ``STRING``                                     | Error message if the job failed. Only set when status is JOB_STATUS_FAILED.                  |
-+---------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
-| metadata      | An array of :ref:`mlflowjobstatemetadataentry` | Additional metadata as key-value pairs. Can be used to store job-specific state information. |
-+---------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
++---------------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
+|     Field Name      |                      Type                      |                                         Description                                          |
++=====================+================================================+==============================================================================================+
+| status              | :ref:`mlflowjobstatus`                         | Current status of the job.                                                                   |
++---------------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
+| error_message       | ``STRING``                                     | Error message if the job failed. Only set when status is JOB_STATUS_FAILED.                  |
++---------------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
+| metadata            | An array of :ref:`mlflowjobstatemetadataentry` | Additional metadata as key-value pairs. Can be used to store job-specific state information. |
++---------------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
+| status_message      | ``STRING``                                     | Latest best-effort in-flight status message.                                                 |
++---------------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
+| progress_payload    | :ref:`mlflowjobprogress`                       | Latest best-effort structured progress payload.                                              |
++---------------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
+| progress_updated_at | ``INT64``                                      | Timestamp of the latest progress update in milliseconds since epoch.                         |
++---------------------+------------------------------------------------+----------------------------------------------------------------------------------------------+
 
 .. _mlflowLinkPromptsToTrace:
 
@@ -9600,21 +9628,23 @@ JobStatus
 Generic status enum for MLflow jobs.
 Can be used across different job types (optimization, scorer, etc.).
 
-+------------------------+----------------------------------+
-|          Name          |           Description            |
-+========================+==================================+
-| JOB_STATUS_UNSPECIFIED |                                  |
-+------------------------+----------------------------------+
-| JOB_STATUS_PENDING     | Job is queued, waiting to start. |
-+------------------------+----------------------------------+
-| JOB_STATUS_IN_PROGRESS | Job is currently running.        |
-+------------------------+----------------------------------+
-| JOB_STATUS_COMPLETED   | Job completed successfully.      |
-+------------------------+----------------------------------+
-| JOB_STATUS_FAILED      | Job failed with an error.        |
-+------------------------+----------------------------------+
-| JOB_STATUS_CANCELED    | Job was canceled by user.        |
-+------------------------+----------------------------------+
++---------------------------+----------------------------------------------------------------------------+
+|           Name            |                                Description                                 |
++===========================+============================================================================+
+| JOB_STATUS_UNSPECIFIED    |                                                                            |
++---------------------------+----------------------------------------------------------------------------+
+| JOB_STATUS_PENDING        | Job is queued, waiting to start.                                           |
++---------------------------+----------------------------------------------------------------------------+
+| JOB_STATUS_IN_PROGRESS    | Job is currently running.                                                  |
++---------------------------+----------------------------------------------------------------------------+
+| JOB_STATUS_COMPLETED      | Job completed successfully.                                                |
++---------------------------+----------------------------------------------------------------------------+
+| JOB_STATUS_FAILED         | Job failed with an error.                                                  |
++---------------------------+----------------------------------------------------------------------------+
+| JOB_STATUS_CANCELED       | Job was canceled by user.                                                  |
++---------------------------+----------------------------------------------------------------------------+
+| JOB_STATUS_NEEDS_RECOVERY | Job backend work may still exist, but the current watcher is unresponsive. |
++---------------------------+----------------------------------------------------------------------------+
 
 .. _mlflowLoggedModelStatus:
 

mlflow/entities/_job.py

@@ -1,11 +1,131 @@
 import json
-from typing import Any
+from dataclasses import dataclass
+from typing import Any, Literal
 
 from mlflow.entities._job_status import JobStatus
 from mlflow.entities._mlflow_object import _MlflowObject
+from mlflow.protos.jobs_pb2 import JobProgress as ProtoJobProgress
 from mlflow.utils.workspace_utils import resolve_entity_workspace_name
 
 
+@dataclass
+class JobProgress:
+    """
+    Structured best-effort progress payload for an in-flight job.
+
+    This keeps progress machine-readable while still allowing it to be
+    stored as JSON in the backing row and sent through the job APIs.
+
+    Attributes:
+        phase: Short label for the current stage of work, such as
+            ``"scoring traces"`` or ``"uploading artifacts"``.
+        completed: Number of work units finished so far.
+        total: Total number of work units, when known.
+        unit: Human-readable name for the work unit, such as ``"trace"``
+            or ``"file"``.
+
+    Example:
+        A trace-scoring job that has processed 42 out of 100 traces could
+        report ``JobProgress(phase="scoring traces", completed=42,
+        total=100, unit="trace")``.
+    """
+
+    phase: str | None = None
+    completed: int | None = None
+    total: int | None = None
+    unit: str | None = None
+
+    @classmethod
+    def from_dict(cls, payload: dict[str, Any]) -> "JobProgress":
+        return cls(
+            phase=payload.get("phase"),
+            completed=payload.get("completed"),
+            total=payload.get("total"),
+            unit=payload.get("unit"),
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        payload = {}
+        if self.phase is not None:
+            payload["phase"] = self.phase
+        if self.completed is not None:
+            payload["completed"] = self.completed
+        if self.total is not None:
+            payload["total"] = self.total
+        if self.unit is not None:
+            payload["unit"] = self.unit
+        return payload
+
+    @classmethod
+    def from_proto(cls, proto: ProtoJobProgress) -> "JobProgress":
+        return cls(
+            phase=proto.phase or None,
+            completed=proto.completed if proto.HasField("completed") else None,
+            total=proto.total if proto.HasField("total") else None,
+            unit=proto.unit or None,
+        )
+
+    def to_proto(self) -> ProtoJobProgress:
+        progress = ProtoJobProgress()
+        if self.phase is not None:
+            progress.phase = self.phase
+        if self.completed is not None:
+            progress.completed = self.completed
+        if self.total is not None:
+            progress.total = self.total
+        if self.unit is not None:
+            progress.unit = self.unit
+        return progress
+
+
+ScopedPermissionResourceType = Literal["experiment", "gateway_endpoint", "prompt"]
+ScopedPermissionName = Literal["READ", "USE", "EDIT"]
+
+
+@dataclass(frozen=True)
+class JobScopedPermission:
+    """
+    A single resource permission granted to a job token.
+
+    Attributes:
+        resource_type: Type of protected MLflow resource.
+        resource_identifier: Stable identifier for the protected resource.
+        workspace: Workspace that owns the resource, when workspace scoping is
+            relevant to authorization.
+        permission: Permission granted for the resource.
+
+    Example:
+        A job allowed to post assessments for experiment ``123`` in workspace
+        ``team-a`` could include ``JobScopedPermission(
+        resource_type="experiment", resource_identifier="123",
+        workspace="team-a", permission="EDIT")``.
+    """
+
+    resource_type: ScopedPermissionResourceType
+    resource_identifier: str
+    workspace: str | None = None
+    permission: ScopedPermissionName = "READ"
+
+    @classmethod
+    def from_dict(cls, payload: dict[str, Any]) -> "JobScopedPermission":
+        return cls(
+            resource_type=payload["resource_type"],
+            resource_identifier=payload["resource_identifier"],
+            workspace=payload.get("workspace"),
+            permission=payload["permission"],
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        payload = {
+            "resource_type": self.resource_type,
+            "resource_identifier": self.resource_identifier,
+            "permission": self.permission,
+        }
+        if self.workspace is not None:
+            payload["workspace"] = self.workspace
+        return payload
+
+
 class Job(_MlflowObject):
     """
     MLflow entity representing a Job.
@@ -24,6 +144,14 @@ def __init__(
         last_update_time: int,
         workspace: str | None = None,
         status_details: dict[str, Any] | None = None,
+        error_message: str | None = None,
+        executor_backend: str | None = None,
+        lease_expires_at: int | None = None,
+        status_message: str | None = None,
+        progress_payload: JobProgress | dict[str, Any] | None = None,
+        progress_updated_at: int | None = None,
+        token_hash: str | None = None,
+        scoped_permissions: list[JobScopedPermission] | list[dict[str, Any]] | None = None,
     ):
         super().__init__()
         self._job_id = job_id
@@ -37,6 +165,23 @@ def __init__(
         self._last_update_time = last_update_time
         self._workspace = resolve_entity_workspace_name(workspace)
         self._status_details = status_details
+        self._error_message = error_message
+        self._executor_backend = executor_backend
+        self._lease_expires_at = lease_expires_at
+        self._status_message = status_message
+        self._progress_payload = (
+            JobProgress.from_dict(progress_payload)
+            if isinstance(progress_payload, dict)
+            else progress_payload
+        )
+        self._progress_updated_at = progress_updated_at
+        self._token_hash = token_hash
+        self._scoped_permissions = (
+            [JobScopedPermission.from_dict(permission) for permission in scoped_permissions]
+            if scoped_permissions is not None
+            and all(isinstance(permission, dict) for permission in scoped_permissions)
+            else scoped_permissions
+        )
 
     @property
     def job_id(self) -> str:
@@ -93,6 +238,8 @@ def parsed_result(self) -> Any:
         If job status is FAILED, the parsed result is the error string.
         Otherwise, the parsed result is None.
         """
+        if self.result is None:
+            return None
         if self.status == JobStatus.SUCCEEDED:
             return json.loads(self.result)
         return self.result
@@ -117,5 +264,93 @@ def status_details(self) -> dict[str, Any] | None:
         """Job status details containing other runtime information."""
         return self._status_details
 
+    @property
+    def error_message(self) -> str | None:
+        """
+        Human-readable terminal error for operators and UI surfaces.
+
+        This is set for failed or timed-out jobs when a terminal error message
+        is available. It may be absent for successful, canceled, pending, or
+        in-flight jobs.
+        """
+        if self._error_message is not None:
+            return self._error_message
+        if self.status in {JobStatus.FAILED, JobStatus.TIMEOUT} and isinstance(self.result, str):
+            return self.result
+        return None
+
+    @property
+    def executor_backend(self) -> str | None:
+        """
+        Persisted executor backend selected for the job.
+
+        This is framework coordination state used to keep retries, cancellation,
+        and recovery pinned to the same backend choice for the lifetime of the
+        job row.
+        """
+        return self._executor_backend
+
+    @property
+    def lease_expires_at(self) -> int | None:
+        """
+        Expiration timestamp for the job's short-lived execution lease.
+
+        This is distinct from terminal outcome fields and exists so recovery
+        logic can tell whether a `RUNNING` job still appears healthy.
+        """
+        return self._lease_expires_at
+
+    @property
+    def status_message(self) -> str | None:
+        """
+        Latest best-effort in-flight status message.
+
+        This is the lightweight plain-text progress channel for operators and
+        simple UI surfaces. It exists so jobs can report useful progress text
+        even when they do not emit a structured `progress_payload`.
+        """
+        return self._status_message
+
+    @property
+    def progress_payload(self) -> JobProgress | None:
+        """
+        Latest best-effort structured progress payload.
+
+        This is intentionally distinct from `status_message`: the string message
+        is the plain-text progress channel, while `progress_payload` is the
+        machine-readable form that can carry fields such as phase, completed,
+        total, and unit for richer shared progress UIs.
+        """
+        return self._progress_payload
+
+    @property
+    def progress_updated_at(self) -> int | None:
+        """
+        Timestamp of the latest structured or message-based progress update, in
+        milliseconds since the UNIX epoch.
+        """
+        return self._progress_updated_at
+
+    @property
+    def token_hash(self) -> str | None:
+        """
+        Persisted hash of the remote-execution job token.
+
+        This is internal framework auth state. The plaintext token is never
+        stored on the entity or in the database row.
+        """
+        return self._token_hash
+
+    @property
+    def scoped_permissions(self) -> list[JobScopedPermission] | None:
+        """
+        Persisted permissions scoped to this job's remote-execution contract.
+
+        Each entry describes one protected resource the job token may access,
+        including its resource type, stable identifier, workspace, and granted
+        permission.
+        """
+        return self._scoped_permissions
+
     def __repr__(self) -> str:
         return f"<Job(job_id={self.job_id}, job_name={self.job_name}, workspace={self.workspace})>"