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

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -3242,7 +3242,14 @@ async def poll_workflow_task_response(
         return {"task": None, "poll_status": "timeout"}
 
     async def poll_workflow_task(
-        self, *, worker_id: str, task_queue: str, timeout: float = 35.0
+        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,
     ) -> Any:
         """Long-poll for the next workflow task on ``task_queue``.
 
@@ -3254,6 +3261,9 @@ async def poll_workflow_task(
             worker_id=worker_id,
             task_queue=task_queue,
             timeout=timeout,
+            build_id=build_id,
+            history_page_size=history_page_size,
+            poll_request_id=poll_request_id,
         )
 
         return data.get("task")
@@ -3341,7 +3351,13 @@ async def workflow_task_history(
         )
 
     async def poll_query_task(
-        self, *, worker_id: str, task_queue: str, timeout: float = 35.0
+        self,
+        *,
+        worker_id: str,
+        task_queue: str,
+        timeout: float = 35.0,
+        build_id: str | None = None,
+        poll_request_id: str | None = None,
     ) -> Any:
         """Long-poll for the next workflow query task on ``task_queue``.
 
@@ -3351,8 +3367,11 @@ async def poll_query_task(
         body: dict[str, Any] = {
             "worker_id": worker_id,
             "task_queue": task_queue,
-            "poll_request_id": f"query-poll-{uuid.uuid4().hex}",
+            "poll_request_id": poll_request_id or f"query-poll-{uuid.uuid4().hex}",
         }
+        if build_id:
+            body["build_id"] = build_id
+
         for _ in range(2):
             try:
                 data = await self._request(
@@ -3426,7 +3445,13 @@ async def fail_query_task(
         )
 
     async def poll_activity_task(
-        self, *, worker_id: str, task_queue: str, timeout: float = 35.0
+        self,
+        *,
+        worker_id: str,
+        task_queue: str,
+        timeout: float = 35.0,
+        build_id: str | None = None,
+        poll_request_id: str | None = None,
     ) -> Any:
         """Long-poll for the next activity task on ``task_queue``.
 
@@ -3436,8 +3461,11 @@ async def poll_activity_task(
         body: dict[str, Any] = {
             "worker_id": worker_id,
             "task_queue": task_queue,
-            "poll_request_id": f"activity-poll-{uuid.uuid4().hex}",
+            "poll_request_id": poll_request_id or f"activity-poll-{uuid.uuid4().hex}",
         }
+        if build_id:
+            body["build_id"] = build_id
+
         for _ in range(2):
             try:
                 data = await self._request(

src/durable_workflow/worker.py

@@ -1588,6 +1588,7 @@ async def _poll_workflow_tasks(self) -> None:
                     worker_id=self.worker_id,
                     task_queue=self.task_queue,
                     timeout=self._poll_http_timeout,
+                    build_id=self.build_id,
                 )
             except Exception as e:
                 self._wf_semaphore.release()
@@ -1665,6 +1666,7 @@ async def _poll_activity_tasks(self) -> None:
                     worker_id=self.worker_id,
                     task_queue=self.task_queue,
                     timeout=self._poll_http_timeout,
+                    build_id=self.build_id,
                 )
             except Exception as e:
                 self._act_semaphore.release()
@@ -1702,6 +1704,7 @@ async def _poll_query_tasks(self, *, client: Client | None = None, track_tasks:
                     worker_id=self.worker_id,
                     task_queue=self.task_queue,
                     timeout=self._poll_http_timeout,
+                    build_id=self.build_id,
                 )
             except Exception as e:
                 self._record_poll_metrics("query", "error", time.perf_counter() - poll_start)
@@ -1960,6 +1963,7 @@ async def run_until(
                         worker_id=self.worker_id,
                         task_queue=self.task_queue,
                         timeout=self._poll_http_timeout,
+                        build_id=self.build_id,
                     )
                     self._record_poll_metrics(
                         "workflow",
@@ -1990,6 +1994,7 @@ async def run_until(
                     worker_id=self.worker_id,
                     task_queue=self.task_queue,
                     timeout=self._poll_http_timeout,
+                    build_id=self.build_id,
                 )
                 self._record_poll_metrics(
                     "activity",

tests/test_client.py

@@ -2302,10 +2302,15 @@ async def test_poll_workflow_task_keeps_returning_none_for_no_compatible_status(
             "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")
+        with patch.object(client, "poll_workflow_task_response", new_callable=AsyncMock, return_value=response) as mock:
+            task = await client.poll_workflow_task(
+                worker_id="worker-v2",
+                task_queue="queue-1",
+                build_id="build-v2",
+            )
 
         assert task is None
+        assert mock.await_args.kwargs["build_id"] == "build-v2"
 
     @pytest.mark.asyncio
     async def test_poll_workflow_task_retries_once_with_same_poll_request_id_after_timeout(
@@ -2335,12 +2340,36 @@ async def test_poll_activity_task_sends_poll_request_id(self, client: Client) ->
         response_task = {"task": {"task_id": "activity-task-123"}}
 
         with patch.object(client, "_request", new_callable=AsyncMock, return_value=response_task) as mock:
-            task = await client.poll_activity_task(worker_id="worker-1", task_queue="queue-1")
+            task = await client.poll_activity_task(
+                worker_id="worker-1",
+                task_queue="queue-1",
+                build_id="build-1",
+            )
+
+        assert task == response_task["task"]
+        request_body = mock.await_args.kwargs["json"]
+        assert request_body["worker_id"] == "worker-1"
+        assert request_body["task_queue"] == "queue-1"
+        assert request_body["build_id"] == "build-1"
+        assert isinstance(request_body["poll_request_id"], str)
+        assert request_body["poll_request_id"] != ""
+
+    @pytest.mark.asyncio
+    async def test_poll_query_task_sends_build_id(self, client: Client) -> None:
+        response_task = {"task": {"query_task_id": "query-task-123"}}
+
+        with patch.object(client, "_request", new_callable=AsyncMock, return_value=response_task) as mock:
+            task = await client.poll_query_task(
+                worker_id="worker-1",
+                task_queue="queue-1",
+                build_id="build-1",
+            )
 
         assert task == response_task["task"]
         request_body = mock.await_args.kwargs["json"]
         assert request_body["worker_id"] == "worker-1"
         assert request_body["task_queue"] == "queue-1"
+        assert request_body["build_id"] == "build-1"
         assert isinstance(request_body["poll_request_id"], str)
         assert request_body["poll_request_id"] != ""
 

tests/test_worker.py

@@ -403,6 +403,59 @@ async def test_register_forwards_build_id_when_configured(
         call_kwargs = mock_client.register_worker.call_args.kwargs
         assert call_kwargs["build_id"] == "release-2026.04.22-a1"
 
+    @pytest.mark.asyncio
+    async def test_poll_loops_forward_build_id_when_configured(
+        self, mock_client: AsyncMock
+    ) -> None:
+        workflow_called = asyncio.Event()
+        activity_called = asyncio.Event()
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[echo_activity],
+            worker_id="w-build-poll",
+            build_id="release-2026.04.22-a1",
+            poll_timeout=0.01,
+        )
+
+        async def workflow_poll_once(**_: object) -> None:
+            workflow_called.set()
+            await worker.stop()
+            return None
+
+        async def activity_poll_once(**_: object) -> None:
+            activity_called.set()
+            await worker.stop()
+            return None
+
+        mock_client.poll_workflow_task.side_effect = workflow_poll_once
+        await worker._poll_workflow_tasks()
+        workflow_kwargs = mock_client.poll_workflow_task.call_args.kwargs
+        assert workflow_called.is_set()
+        assert workflow_kwargs["build_id"] == "release-2026.04.22-a1"
+
+        worker._stop.clear()
+        mock_client.poll_activity_task.side_effect = activity_poll_once
+        await worker._poll_activity_tasks()
+        activity_kwargs = mock_client.poll_activity_task.call_args.kwargs
+        assert activity_called.is_set()
+        assert activity_kwargs["build_id"] == "release-2026.04.22-a1"
+
+        worker._stop.clear()
+        query_called = asyncio.Event()
+
+        async def query_poll_once(**_: object) -> None:
+            query_called.set()
+            await worker.stop()
+            return None
+
+        mock_client.poll_query_task.side_effect = query_poll_once
+        await worker._poll_query_tasks()
+        query_kwargs = mock_client.poll_query_task.call_args.kwargs
+        assert query_called.is_set()
+        assert query_kwargs["build_id"] == "release-2026.04.22-a1"
+
     @pytest.mark.asyncio
     async def test_register_omits_build_id_when_not_configured(
         self, mock_client: AsyncMock