Conformance finding: worker-versioning no-compatible-worker behavior remains unproved (#200)

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -3202,20 +3202,33 @@ async def heartbeat_worker(
             body["heartbeat_interval_seconds"] = heartbeat_interval_seconds
         return await self._request("POST", "/worker/heartbeat", worker=True, json=body)
 
-    async def poll_workflow_task(
-        self, *, worker_id: str, task_queue: str, timeout: float = 35.0
-    ) -> Any:
+    async def poll_workflow_task_response(
+        self,
+        *,
+        worker_id: str,
+        task_queue: str,
+        timeout: float = 35.0,
+        build_id: str | None = None,
+        history_page_size: int | None = None,
+        poll_request_id: str | None = None,
+    ) -> dict[str, Any]:
         """Long-poll for the next workflow task on ``task_queue``.
 
-        Returns the task payload, or ``None`` on poll timeout. Worker-plane
+        Returns the full worker-protocol response envelope, including
+        ``poll_status`` when the server has no task to lease. Worker-plane
         endpoint — most applications use :class:`~durable_workflow.Worker`
         rather than calling this directly.
         """
         body: dict[str, Any] = {
             "worker_id": worker_id,
             "task_queue": task_queue,
-            "poll_request_id": f"wf-poll-{uuid.uuid4().hex}",
+            "poll_request_id": poll_request_id or f"wf-poll-{uuid.uuid4().hex}",
         }
+        if build_id:
+            body["build_id"] = build_id
+        if history_page_size is not None:
+            body["history_page_size"] = history_page_size
+
         for _ in range(2):
             try:
                 data = await self._request(
@@ -3224,9 +3237,26 @@ async def poll_workflow_task(
             except httpx.TimeoutException:
                 continue
 
-            return (data or {}).get("task")
+            return data if isinstance(data, dict) else {}
 
-        return None
+        return {"task": None, "poll_status": "timeout"}
+
+    async def poll_workflow_task(
+        self, *, worker_id: str, task_queue: str, timeout: float = 35.0
+    ) -> Any:
+        """Long-poll for the next workflow task on ``task_queue``.
+
+        Returns the task payload, or ``None`` on poll timeout. Worker-plane
+        endpoint — most applications use :class:`~durable_workflow.Worker`
+        rather than calling this directly.
+        """
+        data = await self.poll_workflow_task_response(
+            worker_id=worker_id,
+            task_queue=task_queue,
+            timeout=timeout,
+        )
+
+        return data.get("task")
 
     async def complete_workflow_task(
         self,

tests/test_client.py

@@ -2270,6 +2270,43 @@ async def test_poll_workflow_task_matches_polyglot_fixture(self, client: Client)
         assert isinstance(request_body["poll_request_id"], str)
         assert request_body["poll_request_id"] != ""
 
+    @pytest.mark.asyncio
+    async def test_poll_workflow_task_response_preserves_no_compatible_status(self, client: Client) -> None:
+        response = {
+            "task": None,
+            "poll_status": "no_compatible_worker",
+        }
+
+        with patch.object(client, "_request", new_callable=AsyncMock, return_value=response) as mock:
+            envelope = await client.poll_workflow_task_response(
+                worker_id="worker-v2",
+                task_queue="queue-1",
+                build_id="build-v2",
+                history_page_size=100,
+                timeout=3.0,
+            )
+
+        assert envelope == response
+        request_body = mock.await_args.kwargs["json"]
+        assert request_body["worker_id"] == "worker-v2"
+        assert request_body["task_queue"] == "queue-1"
+        assert request_body["build_id"] == "build-v2"
+        assert request_body["history_page_size"] == 100
+        assert isinstance(request_body["poll_request_id"], str)
+        assert mock.await_args.kwargs["timeout"] == 3.0
+
+    @pytest.mark.asyncio
+    async def test_poll_workflow_task_keeps_returning_none_for_no_compatible_status(self, client: Client) -> None:
+        response = {
+            "task": None,
+            "poll_status": "no_compatible_worker",
+        }
+
+        with patch.object(client, "poll_workflow_task_response", new_callable=AsyncMock, return_value=response):
+            task = await client.poll_workflow_task(worker_id="worker-v2", task_queue="queue-1")
+
+        assert task is None
+
     @pytest.mark.asyncio
     async def test_poll_workflow_task_retries_once_with_same_poll_request_id_after_timeout(
         self, client: Client