Expose worker build-id rollout on the Python SDK

Add `Client.list_task_queue_build_ids(task_queue)` and its sync mirror so
operators using the Python SDK can verify which build cohorts can claim
work on a queue before draining or removing an older build. The method
calls the server's task-queue build-ids endpoint and returns a
`TaskQueueBuildIdRollout` with per-cohort rollout status, active,
draining, and stale worker counts, runtime and SDK-version mix,
first-seen timestamp, and most-recent heartbeat. Unversioned workers are
grouped under a cohort whose `build_id` is `None`.

Consume the `task-queue-build-ids-parity.json` fixture byte-identically
with the CLI so CI catches any divergence before it hits operators.

Issue: zorporation/durable-workflow#491
Loop-ID: build-01

Author: durable-workflow-ops

src/durable_workflow/__init__.py

@@ -32,6 +32,8 @@
     StoragePayloadTestResult,
     StorageTestResult,
     TaskQueueAdmission,
+    TaskQueueBuildIdCohort,
+    TaskQueueBuildIdRollout,
     TaskQueueDescription,
     TaskQueueList,
     TaskQueueQueryAdmission,
@@ -177,6 +179,8 @@
     "StorageTestResult",
     "StartChildWorkflow",
     "TaskQueueAdmission",
+    "TaskQueueBuildIdCohort",
+    "TaskQueueBuildIdRollout",
     "TaskQueueDescription",
     "TaskQueueList",
     "TaskQueueQueryAdmission",

src/durable_workflow/client.py

@@ -467,6 +467,73 @@ class TaskQueueList:
     task_queues: list[TaskQueueDescription]
 
 
+@dataclass
+class TaskQueueBuildIdCohort:
+    """Per-build-id rollout state for one task queue.
+
+    ``build_id`` is ``None`` for the cohort of workers that registered
+    without a build identifier (the legacy unversioned default).
+    """
+
+    build_id: str | None
+    rollout_status: str
+    active_worker_count: int
+    draining_worker_count: int
+    stale_worker_count: int
+    total_worker_count: int
+    runtimes: list[str]
+    sdk_versions: list[str]
+    last_heartbeat_at: str | None = None
+    first_seen_at: str | None = None
+    raw: dict[str, Any] | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> TaskQueueBuildIdCohort:
+        runtimes = data.get("runtimes")
+        sdk_versions = data.get("sdk_versions")
+        return cls(
+            build_id=data.get("build_id"),
+            rollout_status=str(data.get("rollout_status") or ""),
+            active_worker_count=int(data.get("active_worker_count") or 0),
+            draining_worker_count=int(data.get("draining_worker_count") or 0),
+            stale_worker_count=int(data.get("stale_worker_count") or 0),
+            total_worker_count=int(data.get("total_worker_count") or 0),
+            runtimes=[r for r in runtimes if isinstance(r, str)] if isinstance(runtimes, list) else [],
+            sdk_versions=[v for v in sdk_versions if isinstance(v, str)] if isinstance(sdk_versions, list) else [],
+            last_heartbeat_at=data.get("last_heartbeat_at"),
+            first_seen_at=data.get("first_seen_at"),
+            raw=data,
+        )
+
+
+@dataclass
+class TaskQueueBuildIdRollout:
+    """Build-id rollout snapshot returned by the server for one task queue."""
+
+    namespace: str | None
+    task_queue: str
+    stale_after_seconds: int | None
+    build_ids: list[TaskQueueBuildIdCohort]
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> TaskQueueBuildIdRollout:
+        items = data.get("build_ids")
+        return cls(
+            namespace=data.get("namespace"),
+            task_queue=str(data.get("task_queue") or ""),
+            stale_after_seconds=(
+                int(data["stale_after_seconds"])
+                if isinstance(data.get("stale_after_seconds"), int)
+                else None
+            ),
+            build_ids=[
+                TaskQueueBuildIdCohort.from_dict(item)
+                for item in (items if isinstance(items, list) else [])
+                if isinstance(item, dict)
+            ],
+        )
+
+
 @dataclass
 class WorkerDescription:
     """Current server view of one registered worker."""
@@ -1303,6 +1370,28 @@ async def describe_task_queue(self, name: str) -> TaskQueueDescription:
             )
         return TaskQueueDescription.from_dict(data)
 
+    async def list_task_queue_build_ids(self, task_queue: str) -> TaskQueueBuildIdRollout:
+        """Return the build-id rollout snapshot for ``task_queue``.
+
+        Use this before draining or removing an older build to confirm which
+        build cohorts can still claim work on the queue. Unversioned workers
+        are grouped under a cohort whose ``build_id`` is ``None``.
+        """
+        data = await self._request(
+            "GET",
+            f"/task-queues/{quote(task_queue, safe='')}/build-ids",
+            context=task_queue,
+        )
+        if not isinstance(data, dict):
+            raise ServerError(
+                200,
+                {
+                    "reason": "invalid_task_queue_build_ids_response",
+                    "message": f"expected JSON object, got {type(data).__name__}",
+                },
+            )
+        return TaskQueueBuildIdRollout.from_dict(data)
+
     # ── Search attributes ─────────────────────────────────────────────
     async def list_search_attributes(self) -> SearchAttributeList:
         """List system and custom search attribute definitions for this namespace."""

src/durable_workflow/sync.py

@@ -17,6 +17,7 @@
     ScheduleSpec,
     ScheduleTriggerResult,
     StorageTestResult,
+    TaskQueueBuildIdRollout,
     TaskQueueDescription,
     TaskQueueList,
     WorkflowCommandResult,
@@ -284,6 +285,12 @@ def describe_task_queue(self, name: str) -> TaskQueueDescription:
         result: TaskQueueDescription = _run(self._async.describe_task_queue(name))
         return result
 
+    def list_task_queue_build_ids(self, task_queue: str) -> TaskQueueBuildIdRollout:
+        result: TaskQueueBuildIdRollout = _run(
+            self._async.list_task_queue_build_ids(task_queue)
+        )
+        return result
+
     def list_namespaces(self) -> NamespaceList:
         result: NamespaceList = _run(self._async.list_namespaces())
         return result

tests/fixtures/control-plane/task-queue-build-ids-parity.json

@@ -0,0 +1,90 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "task_queue.build_ids",
+  "request": {
+    "method": "GET",
+    "path": "/task-queues/orders-critical/build-ids"
+  },
+  "semantic_body": {
+    "namespace": "orders-prod",
+    "task_queue": "orders-critical",
+    "build_ids": [
+      "build-2026.04.22-a1",
+      "build-2026.04.21-z9",
+      null
+    ],
+    "rollout_statuses": {
+      "build-2026.04.22-a1": "active",
+      "build-2026.04.21-z9": "draining",
+      "unversioned": "stale_only"
+    }
+  },
+  "response_body": {
+    "namespace": "orders-prod",
+    "task_queue": "orders-critical",
+    "stale_after_seconds": 60,
+    "build_ids": [
+      {
+        "build_id": "build-2026.04.22-a1",
+        "rollout_status": "active",
+        "active_worker_count": 2,
+        "draining_worker_count": 0,
+        "stale_worker_count": 0,
+        "total_worker_count": 2,
+        "runtimes": [
+          "worker-runtime"
+        ],
+        "sdk_versions": [
+          "polyglot-sdk/2.0.0"
+        ],
+        "last_heartbeat_at": "2026-04-22T09:30:00Z",
+        "first_seen_at": "2026-04-22T08:00:00Z"
+      },
+      {
+        "build_id": "build-2026.04.21-z9",
+        "rollout_status": "draining",
+        "active_worker_count": 0,
+        "draining_worker_count": 1,
+        "stale_worker_count": 0,
+        "total_worker_count": 1,
+        "runtimes": [
+          "worker-runtime"
+        ],
+        "sdk_versions": [
+          "polyglot-sdk/1.9.8"
+        ],
+        "last_heartbeat_at": "2026-04-22T09:29:00Z",
+        "first_seen_at": "2026-04-21T10:00:00Z"
+      },
+      {
+        "build_id": null,
+        "rollout_status": "stale_only",
+        "active_worker_count": 0,
+        "draining_worker_count": 0,
+        "stale_worker_count": 1,
+        "total_worker_count": 1,
+        "runtimes": [
+          "worker-runtime"
+        ],
+        "sdk_versions": [
+          "polyglot-sdk/1.0.0"
+        ],
+        "last_heartbeat_at": "2026-04-20T12:00:00Z",
+        "first_seen_at": "2026-04-20T11:00:00Z"
+      }
+    ]
+  },
+  "cli": {
+    "argv": {
+      "task-queue": "orders-critical",
+      "--json": true
+    }
+  },
+  "sdk_python": {
+    "method": "list_task_queue_build_ids",
+    "args": {
+      "task_queue": "orders-critical"
+    }
+  }
+}

tests/test_client.py

@@ -1014,6 +1014,103 @@ async def test_describe_task_queue_matches_polyglot_fixture(self, client: Client
         assert result.admission.query_tasks.status == semantic["query_admission_status"]
         assert [lease["task_id"] for lease in result.current_leases or []] == semantic["current_lease_ids"]
 
+    @pytest.mark.asyncio
+    async def test_list_task_queue_build_ids_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = (
+            Path(__file__).parent
+            / "fixtures"
+            / "control-plane"
+            / "task-queue-build-ids-parity.json"
+        )
+        fixture = json.loads(fixture_path.read_text())
+        assert fixture["operation"] == "task_queue.build_ids"
+        sdk = fixture["sdk_python"]
+        resp = _mock_response(200, fixture["response_body"])
+
+        with patch.object(
+            client._http, "request", new_callable=AsyncMock, return_value=resp
+        ) as mock:
+            result = await client.list_task_queue_build_ids(**sdk["args"])
+
+        assert mock.call_args.args[0] == fixture["request"]["method"]
+        assert mock.call_args.args[1] == f"/api{fixture['request']['path']}"
+
+        semantic = fixture["semantic_body"]
+        assert result.namespace == semantic["namespace"]
+        assert result.task_queue == semantic["task_queue"]
+        assert result.stale_after_seconds == fixture["response_body"]["stale_after_seconds"]
+        assert [cohort.build_id for cohort in result.build_ids] == semantic["build_ids"]
+
+        rollout_statuses = {
+            (cohort.build_id if cohort.build_id is not None else "unversioned"): cohort.rollout_status
+            for cohort in result.build_ids
+        }
+        assert rollout_statuses == semantic["rollout_statuses"]
+
+    @pytest.mark.asyncio
+    async def test_list_task_queue_build_ids_surfaces_cohort_worker_counts(
+        self, client: Client
+    ) -> None:
+        resp = _mock_response(
+            200,
+            {
+                "namespace": "default",
+                "task_queue": "orders",
+                "stale_after_seconds": 90,
+                "build_ids": [
+                    {
+                        "build_id": "build-alpha",
+                        "rollout_status": "active_with_draining",
+                        "active_worker_count": 2,
+                        "draining_worker_count": 1,
+                        "stale_worker_count": 0,
+                        "total_worker_count": 3,
+                        "runtimes": ["worker-runtime"],
+                        "sdk_versions": ["polyglot-sdk/2.0.0"],
+                        "last_heartbeat_at": "2026-04-22T09:30:00Z",
+                        "first_seen_at": "2026-04-22T08:00:00Z",
+                    },
+                    {
+                        "build_id": None,
+                        "rollout_status": "stale_only",
+                        "active_worker_count": 0,
+                        "draining_worker_count": 0,
+                        "stale_worker_count": 1,
+                        "total_worker_count": 1,
+                        "runtimes": [],
+                        "sdk_versions": [],
+                        "last_heartbeat_at": None,
+                        "first_seen_at": None,
+                    },
+                ],
+            },
+        )
+
+        with patch.object(
+            client._http, "request", new_callable=AsyncMock, return_value=resp
+        ) as mock:
+            result = await client.list_task_queue_build_ids("orders")
+
+        assert mock.call_args.args[0] == "GET"
+        assert mock.call_args.args[1] == "/api/task-queues/orders/build-ids"
+        assert result.task_queue == "orders"
+        assert result.stale_after_seconds == 90
+        assert len(result.build_ids) == 2
+
+        alpha, unversioned = result.build_ids
+        assert alpha.build_id == "build-alpha"
+        assert alpha.rollout_status == "active_with_draining"
+        assert alpha.active_worker_count == 2
+        assert alpha.draining_worker_count == 1
+        assert alpha.total_worker_count == 3
+        assert alpha.runtimes == ["worker-runtime"]
+        assert alpha.sdk_versions == ["polyglot-sdk/2.0.0"]
+        assert unversioned.build_id is None
+        assert unversioned.rollout_status == "stale_only"
+        assert unversioned.stale_worker_count == 1
+        assert unversioned.last_heartbeat_at is None
+        assert unversioned.first_seen_at is None
+
     @pytest.mark.asyncio
     async def test_list_task_queues_parses_admission(self, client: Client) -> None:
         resp = _mock_response(200, {

tests/test_sync.py

@@ -161,6 +161,43 @@ def test_describe_task_queue(self) -> None:
             assert result.admission.activity_tasks.status == "no_slots"
             assert mock.call_args.args[:2] == ("GET", "/api/task-queues/orders%2Fhigh%20priority")
 
+    def test_list_task_queue_build_ids(self) -> None:
+        client = Client("http://localhost:8080")
+        resp = _mock_response(
+            200,
+            {
+                "namespace": "default",
+                "task_queue": "orders",
+                "stale_after_seconds": 60,
+                "build_ids": [
+                    {
+                        "build_id": "build-alpha",
+                        "rollout_status": "active",
+                        "active_worker_count": 2,
+                        "draining_worker_count": 0,
+                        "stale_worker_count": 0,
+                        "total_worker_count": 2,
+                        "runtimes": ["worker-runtime"],
+                        "sdk_versions": ["polyglot-sdk/2.0.0"],
+                        "last_heartbeat_at": "2026-04-22T09:30:00Z",
+                        "first_seen_at": "2026-04-22T08:00:00Z",
+                    }
+                ],
+            },
+        )
+        with patch.object(
+            client._async._http, "request", new_callable=AsyncMock, return_value=resp
+        ) as mock:
+            result = client.list_task_queue_build_ids("orders")
+            assert result.task_queue == "orders"
+            assert result.stale_after_seconds == 60
+            assert len(result.build_ids) == 1
+            cohort = result.build_ids[0]
+            assert cohort.build_id == "build-alpha"
+            assert cohort.rollout_status == "active"
+            assert cohort.total_worker_count == 2
+            assert mock.call_args.args[:2] == ("GET", "/api/task-queues/orders/build-ids")
+
 
 class TestSyncClientRunVisibility:
     def test_list_workflow_runs(self) -> None: