[cross-repo from server#280] Conformance blocker: make worker-versioning measurable against current artifacts (#174)

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -597,12 +597,29 @@ class TaskQueueBuildIdCohort:
     sdk_versions: list[str]
     last_heartbeat_at: str | None = None
     first_seen_at: str | None = None
+    drain_intent: str | None = None
+    drained_at: str | None = None
+    promoted_at: str | None = None
+    rolled_back_at: str | None = None
+    new_start_selected: bool = False
+    workflow_definition_fingerprint_count: int = 0
+    workflow_definition_fingerprint_conflicts: list[dict[str, Any]] | 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")
+        fingerprint_conflicts_raw = data.get("workflow_definition_fingerprint_conflicts")
+        fingerprint_conflicts: list[dict[str, Any]] | None = None
+        if isinstance(fingerprint_conflicts_raw, list):
+            fingerprint_conflicts = []
+            for item in fingerprint_conflicts_raw:
+                if not isinstance(item, dict):
+                    continue
+                fingerprint_conflicts.append(
+                    {str(key): value for key, value in item.items() if isinstance(key, str)}
+                )
         return cls(
             build_id=data.get("build_id"),
             rollout_status=str(data.get("rollout_status") or ""),
@@ -614,6 +631,15 @@ def from_dict(cls, data: dict[str, Any]) -> TaskQueueBuildIdCohort:
             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"),
+            drain_intent=data.get("drain_intent") if isinstance(data.get("drain_intent"), str) else None,
+            drained_at=data.get("drained_at") if isinstance(data.get("drained_at"), str) else None,
+            promoted_at=data.get("promoted_at") if isinstance(data.get("promoted_at"), str) else None,
+            rolled_back_at=data.get("rolled_back_at") if isinstance(data.get("rolled_back_at"), str) else None,
+            new_start_selected=bool(data.get("new_start_selected")),
+            workflow_definition_fingerprint_count=int(
+                data.get("workflow_definition_fingerprint_count") or 0
+            ),
+            workflow_definition_fingerprint_conflicts=fingerprint_conflicts,
             raw=data,
         )
 
@@ -650,28 +676,41 @@ def from_dict(cls, data: dict[str, Any]) -> TaskQueueBuildIdRollout:
 class TaskQueueBuildIdRolloutState:
     """Operator-recorded drain intent for one ``(task_queue, build_id)`` cohort.
 
-    Returned by ``drain_task_queue_build_id`` and ``resume_task_queue_build_id``.
+    Returned by ``drain_task_queue_build_id``,
+    ``promote_task_queue_build_id``, and ``resume_task_queue_build_id``.
     ``build_id`` is ``None`` for the unversioned cohort (workers registered
     without a build identifier). ``drain_intent`` is ``"active"`` or
     ``"draining"``. ``drained_at`` is set only when ``drain_intent`` is
     ``"draining"``; repeated drains do not shift the timestamp.
+    ``promoted_at`` and ``new_start_selected`` identify the cohort currently
+    selected for fresh workflow starts.
     """
 
     namespace: str | None
     task_queue: str
     build_id: str | None
     drain_intent: str
     drained_at: str | None
+    promoted_at: str | None = None
+    rolled_back_at: str | None = None
+    new_start_selected: bool = False
+    deployment: dict[str, Any] | None = None
     raw: dict[str, Any] | None = None
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> TaskQueueBuildIdRolloutState:
+        deployment_raw = data.get("deployment")
+        deployment = dict(deployment_raw) if isinstance(deployment_raw, dict) else None
         return cls(
             namespace=data.get("namespace"),
             task_queue=str(data.get("task_queue") or ""),
             build_id=data.get("build_id") if isinstance(data.get("build_id"), str) else None,
             drain_intent=str(data.get("drain_intent") or ""),
             drained_at=data.get("drained_at") if isinstance(data.get("drained_at"), str) else None,
+            promoted_at=data.get("promoted_at") if isinstance(data.get("promoted_at"), str) else None,
+            rolled_back_at=data.get("rolled_back_at") if isinstance(data.get("rolled_back_at"), str) else None,
+            new_start_selected=bool(data.get("new_start_selected")),
+            deployment=deployment,
             raw=data,
         )
 
@@ -1895,6 +1934,24 @@ async def drain_task_queue_build_id(
             action="drain",
         )
 
+    async def promote_task_queue_build_id(
+        self,
+        task_queue: str,
+        build_id: str | None,
+    ) -> TaskQueueBuildIdRolloutState:
+        """Select a build-id cohort for fresh workflow starts on a task queue.
+
+        New workflow starts pin to ``build_id`` after promotion. Existing
+        workflow runs keep their stamped compatibility marker and continue
+        routing only to compatible workers. Pass ``None`` to promote the
+        unversioned cohort.
+        """
+        return await self._mutate_task_queue_build_id_rollout(
+            task_queue,
+            build_id,
+            action="promote",
+        )
+
     async def resume_task_queue_build_id(
         self,
         task_queue: str,

src/durable_workflow/sync.py

@@ -340,6 +340,16 @@ def drain_task_queue_build_id(
         )
         return result
 
+    def promote_task_queue_build_id(
+        self,
+        task_queue: str,
+        build_id: str | None,
+    ) -> TaskQueueBuildIdRolloutState:
+        result: TaskQueueBuildIdRolloutState = _run(
+            self._async.promote_task_queue_build_id(task_queue, build_id)
+        )
+        return result
+
     def resume_task_queue_build_id(
         self,
         task_queue: str,

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

@@ -0,0 +1,44 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "task_queue.build_id.promote",
+  "request": {
+    "method": "POST",
+    "path": "/task-queues/orders-critical/build-ids/promote",
+    "body": {
+      "build_id": "build-2026.04.21-z9"
+    }
+  },
+  "semantic_body": {
+    "namespace": "orders-prod",
+    "task_queue": "orders-critical",
+    "build_id": "build-2026.04.21-z9",
+    "drain_intent": "active",
+    "drained_at": null,
+    "promoted_at": "2026-04-22T10:00:00Z",
+    "new_start_selected": true
+  },
+  "response_body": {
+    "namespace": "orders-prod",
+    "task_queue": "orders-critical",
+    "build_id": "build-2026.04.21-z9",
+    "drain_intent": "active",
+    "drained_at": null,
+    "promoted_at": "2026-04-22T10:00:00Z",
+    "new_start_selected": true
+  },
+  "cli": {
+    "argv": {
+      "task-queue": "orders-critical",
+      "--build-id": "build-2026.04.21-z9",
+      "--json": true
+    }
+  },
+  "sdk_python": {
+    "method": "promote_task_queue_build_id",
+    "args": {
+      "task_queue": "orders-critical",
+      "build_id": "build-2026.04.21-z9"
+    }
+  }
+}

tests/test_client.py

@@ -1533,6 +1533,17 @@ async def test_list_task_queue_build_ids_surfaces_cohort_worker_counts(
                         "total_worker_count": 3,
                         "runtimes": ["worker-runtime"],
                         "sdk_versions": ["polyglot-sdk/2.0.0"],
+                        "drain_intent": "active",
+                        "drained_at": None,
+                        "promoted_at": "2026-04-22T10:00:00Z",
+                        "new_start_selected": True,
+                        "workflow_definition_fingerprint_count": 2,
+                        "workflow_definition_fingerprint_conflicts": [
+                            {
+                                "workflow_type": "orders.Checkout",
+                                "fingerprint_count": 2,
+                            }
+                        ],
                         "last_heartbeat_at": "2026-04-22T09:30:00Z",
                         "first_seen_at": "2026-04-22T08:00:00Z",
                     },
@@ -1571,6 +1582,16 @@ async def test_list_task_queue_build_ids_surfaces_cohort_worker_counts(
         assert alpha.total_worker_count == 3
         assert alpha.runtimes == ["worker-runtime"]
         assert alpha.sdk_versions == ["polyglot-sdk/2.0.0"]
+        assert alpha.drain_intent == "active"
+        assert alpha.promoted_at == "2026-04-22T10:00:00Z"
+        assert alpha.new_start_selected is True
+        assert alpha.workflow_definition_fingerprint_count == 2
+        assert alpha.workflow_definition_fingerprint_conflicts == [
+            {
+                "workflow_type": "orders.Checkout",
+                "fingerprint_count": 2,
+            }
+        ]
         assert unversioned.build_id is None
         assert unversioned.rollout_status == "stale_only"
         assert unversioned.stale_worker_count == 1
@@ -1607,6 +1628,38 @@ async def test_drain_task_queue_build_id_matches_polyglot_fixture(self, client:
         assert result.drain_intent == semantic["drain_intent"]
         assert result.drained_at == semantic["drained_at"]
 
+    @pytest.mark.asyncio
+    async def test_promote_task_queue_build_id_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = (
+            Path(__file__).parent
+            / "fixtures"
+            / "control-plane"
+            / "task-queue-build-id-promote-parity.json"
+        )
+        fixture = json.loads(fixture_path.read_text())
+        assert fixture["operation"] == "task_queue.build_id.promote"
+        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.promote_task_queue_build_id(**sdk["args"])
+
+        assert mock.call_args.args[0] == fixture["request"]["method"]
+        assert mock.call_args.args[1] == f"/api{fixture['request']['path']}"
+        body = mock.call_args.kwargs.get("json")
+        assert body == fixture["request"]["body"]
+
+        semantic = fixture["semantic_body"]
+        assert result.namespace == semantic["namespace"]
+        assert result.task_queue == semantic["task_queue"]
+        assert result.build_id == semantic["build_id"]
+        assert result.drain_intent == semantic["drain_intent"]
+        assert result.drained_at is None
+        assert result.promoted_at == semantic["promoted_at"]
+        assert result.new_start_selected == semantic["new_start_selected"]
+
     @pytest.mark.asyncio
     async def test_resume_task_queue_build_id_matches_polyglot_fixture(self, client: Client) -> None:
         fixture_path = (

tests/test_sync.py

@@ -294,6 +294,34 @@ def test_drain_task_queue_build_id(self) -> None:
             )
             assert mock.call_args.kwargs.get("json") == {"build_id": "build-alpha"}
 
+    def test_promote_task_queue_build_id(self) -> None:
+        client = Client("http://localhost:8080")
+        resp = _mock_response(
+            200,
+            {
+                "namespace": "default",
+                "task_queue": "orders",
+                "build_id": "build-alpha",
+                "drain_intent": "active",
+                "drained_at": None,
+                "promoted_at": "2026-04-22T10:00:00Z",
+                "new_start_selected": True,
+            },
+        )
+        with patch.object(
+            client._async._http, "request", new_callable=AsyncMock, return_value=resp
+        ) as mock:
+            result = client.promote_task_queue_build_id("orders", "build-alpha")
+            assert result.drain_intent == "active"
+            assert result.promoted_at == "2026-04-22T10:00:00Z"
+            assert result.new_start_selected is True
+            assert result.build_id == "build-alpha"
+            assert mock.call_args.args[:2] == (
+                "POST",
+                "/api/task-queues/orders/build-ids/promote",
+            )
+            assert mock.call_args.kwargs.get("json") == {"build_id": "build-alpha"}
+
     def test_resume_task_queue_build_id_for_unversioned_cohort(self) -> None:
         client = Client("http://localhost:8080")
         resp = _mock_response(