Let Python Worker report its build_id during registration

Accept a `build_id` keyword on `Worker(...)` and forward it through
`Client.register_worker` so the server's per-build-id rollout snapshot
for the task queue can attribute active, draining, and stale worker
counts to the right cohort. An empty or whitespace-only `build_id` is
rejected at construction rather than silently dropped.

Author: durable-workflow-ops

src/durable_workflow/worker.py

@@ -242,6 +242,7 @@ def __init__(
         workflows: Iterable[type] = (),
         activities: Iterable[Callable[..., Any]] = (),
         worker_id: str | None = None,
+        build_id: str | None = None,
         poll_timeout: float = 35.0,
         max_concurrent_workflow_tasks: int = 10,
         max_concurrent_activity_tasks: int = 10,
@@ -258,6 +259,12 @@ def __init__(
         }
         self.activities = {_activity_name(a): a for a in activities}
         self.worker_id = worker_id or f"py-worker-{uuid.uuid4().hex[:8]}"
+        if build_id is not None:
+            if not isinstance(build_id, str) or build_id.strip() == "":
+                raise ValueError("build_id must be a non-empty string when provided")
+            self.build_id: str | None = build_id
+        else:
+            self.build_id = None
         _guard_worker_workflow_fingerprints(self.worker_id, self.workflow_definition_fingerprints)
         if max_concurrent_workflow_tasks < 1:
             raise ValueError("max_concurrent_workflow_tasks must be at least 1")
@@ -346,6 +353,7 @@ async def _register(self) -> None:
             supported_activity_types=list(self.activities),
             max_concurrent_workflow_tasks=self.max_concurrent_workflow_tasks,
             max_concurrent_activity_tasks=self.max_concurrent_activity_tasks,
+            build_id=self.build_id,
         )
         log.info("worker %s registered on %s", self.worker_id, self.task_queue)
 

tests/test_worker.py

@@ -187,6 +187,50 @@ async def test_register_advertises_custom_concurrency_limits(self, mock_client:
         assert call_kwargs["max_concurrent_workflow_tasks"] == 3
         assert call_kwargs["max_concurrent_activity_tasks"] == 7
 
+    @pytest.mark.asyncio
+    async def test_register_forwards_build_id_when_configured(
+        self, mock_client: AsyncMock
+    ) -> None:
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[echo_activity],
+            worker_id="w-build",
+            build_id="release-2026.04.22-a1",
+        )
+        assert worker.build_id == "release-2026.04.22-a1"
+
+        await worker._register()
+
+        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_register_omits_build_id_when_not_configured(
+        self, mock_client: AsyncMock
+    ) -> None:
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[echo_activity],
+            worker_id="w-no-build",
+        )
+        assert worker.build_id is None
+
+        await worker._register()
+
+        call_kwargs = mock_client.register_worker.call_args.kwargs
+        assert call_kwargs["build_id"] is None
+
+    def test_constructor_rejects_empty_build_id(self, mock_client: AsyncMock) -> None:
+        with pytest.raises(ValueError, match="build_id"):
+            Worker(mock_client, task_queue="q1", build_id="")
+
+        with pytest.raises(ValueError, match="build_id"):
+            Worker(mock_client, task_queue="q1", build_id="   ")
+
     def test_constructor_rejects_non_positive_concurrency_limits(
         self, mock_client: AsyncMock
     ) -> None: