feat: let a job see that its worker is draining (#11612)

* feat: let a job see that its worker is draining

* fix: gate worker draining check on job visibility and export sdk helper

* fix: keep draining across re-registration and ignore parked jobs

* fix: reset draining only on worker startup registration

* chore: update ee-repo-ref to 55feb2e84332179faa4e17aeb419f7f480df79a6

This commit updates the EE repository reference after PR #857 was merged in windmill-ee-private.

Previous ee-repo-ref: e5010fa046a3aba93595d1bf44eb6a0cef2b5273

New ee-repo-ref: 55feb2e84332179faa4e17aeb419f7f480df79a6

Automated by sync-ee-ref workflow.

---------

Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>

python-client/wmill/wmill/client.py

@@ -823,6 +823,22 @@ class Windmill:
         else:
             return r.json()
 
+    def worker_is_draining(self, job_id: Optional[str] = None) -> bool:
+        """Whether the worker running the job has received its shutdown signal.
+
+        A draining worker waits for its current job to finish before exiting. A long-running
+        script can poll this and return early so the worker can be replaced.
+
+        Args:
+            job_id: Job ID (defaults to current WM_JOB_ID)
+
+        Returns:
+            True if the worker running the job is draining
+        """
+        job_id = job_id or os.environ.get("WM_JOB_ID")
+
+        return self.get(f"/w/{self.workspace}/jobs/worker_is_draining/{job_id}").json()
+
     def set_flow_user_state(self, key: str, value: Any) -> None:
         """Set the user state of a flow at a given key"""
         flow_id = self.get_root_job_id()
@@ -2027,6 +2043,17 @@ def get_progress(job_id: Optional[str] = None) -> Any:
     return _client.get_progress(job_id)
 
 
+@init_global_client
+def worker_is_draining(job_id: Optional[str] = None) -> bool:
+    """
+    Whether the worker running this job has received its shutdown signal.
+
+    A draining worker waits for its current job to finish before exiting. A long-running
+    script can poll this and return early so the worker can be replaced.
+    """
+    return _client.worker_is_draining(job_id)
+
+
 def set_shared_state_pickle(value: Any, path="state.pickle") -> None:
     """
     Set the state in the shared folder using pickle