Fail closed on SDK protocol compatibility checks

durable-workflow-ops · durable-workflow-ops · commit 86c4eb3e913d · 2026-04-17T23:55:50.000Z
diff --git a/README.md b/README.md
@@ -77,9 +77,16 @@ Full documentation is available at [durable-workflow.github.io/docs/2.0/sdks/pyt
 
 ## Compatibility
 
-SDK version 0.2.x is compatible with Durable Workflow server 0.x prerelease images and the Server 2.x protocol line.
+SDK version 0.2.x is compatible with servers that advertise these protocol
+manifests from `GET /api/cluster/info`:
 
-The worker automatically checks server version at startup and raises a clear error if incompatible.
+- `control_plane.version: "2"`
+- `control_plane.request_contract.schema: durable-workflow.v2.control-plane-request.contract` version `1`
+- `worker_protocol.version: "1.0"`
+
+The top-level server `version` is build identity only. The worker checks these
+protocol manifests at startup and fails closed when compatibility is missing,
+unknown, or undiscoverable.
 
 ## Development
 
diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py
@@ -20,6 +20,8 @@
 
 PROTOCOL_VERSION = "1.0"
 CONTROL_PLANE_VERSION = "2"
+CONTROL_PLANE_REQUEST_CONTRACT_SCHEMA = "durable-workflow.v2.control-plane-request.contract"
+CONTROL_PLANE_REQUEST_CONTRACT_VERSION = 1
 
 
 def _default_sdk_version() -> str:
@@ -313,7 +315,7 @@ async def _do_request() -> httpx.Response:
         return resp.json()
 
     async def get_cluster_info(self) -> dict[str, Any]:
-        """Fetch server version and capabilities from /api/cluster/info."""
+        """Fetch server build identity, capabilities, and protocol manifests."""
         result = await self._request("GET", "/cluster/info", worker=False, context="get_cluster_info")
         if not isinstance(result, dict):
             raise ServerError(
diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py
@@ -10,7 +10,13 @@
 
 from . import serializer
 from .activity import ActivityContext, ActivityInfo, _set_context
-from .client import Client
+from .client import (
+    CONTROL_PLANE_REQUEST_CONTRACT_SCHEMA,
+    CONTROL_PLANE_REQUEST_CONTRACT_VERSION,
+    CONTROL_PLANE_VERSION,
+    PROTOCOL_VERSION,
+    Client,
+)
 from .errors import ActivityCancelled, AvroNotInstalledError, NonRetryableError
 from .workflow import replay
 
@@ -25,6 +31,71 @@ def _workflow_name(cls: type) -> str:
     return getattr(cls, "__workflow_name__", cls.__name__)
 
 
+def _manifest_version(manifest: Any) -> str:
+    if isinstance(manifest, dict):
+        value = manifest.get("version")
+        if isinstance(value, str | int):
+            return str(value)
+    return "missing"
+
+
+def _validate_server_compatibility(info: dict[str, Any]) -> None:
+    control_plane = info.get("control_plane")
+    if not isinstance(control_plane, dict):
+        raise RuntimeError(
+            "Server compatibility error: missing control_plane manifest; "
+            f"sdk-python 0.2.x requires control_plane.version {CONTROL_PLANE_VERSION}."
+        )
+
+    control_plane_version = _manifest_version(control_plane)
+    if control_plane_version != CONTROL_PLANE_VERSION:
+        raise RuntimeError(
+            "Server compatibility error: unsupported control_plane.version "
+            f"{control_plane_version!r}; sdk-python 0.2.x requires {CONTROL_PLANE_VERSION!r}."
+        )
+
+    request_contract = control_plane.get("request_contract")
+    if not isinstance(request_contract, dict):
+        raise RuntimeError(
+            "Server compatibility error: missing control_plane.request_contract; "
+            f"expected {CONTROL_PLANE_REQUEST_CONTRACT_SCHEMA} v{CONTROL_PLANE_REQUEST_CONTRACT_VERSION}."
+        )
+
+    request_schema = request_contract.get("schema")
+    request_version = request_contract.get("version")
+    if (
+        request_schema != CONTROL_PLANE_REQUEST_CONTRACT_SCHEMA
+        or not _contract_version_matches(request_version, CONTROL_PLANE_REQUEST_CONTRACT_VERSION)
+    ):
+        raise RuntimeError(
+            "Server compatibility error: unsupported control_plane.request_contract "
+            f"{request_schema!r} v{request_version!r}; expected "
+            f"{CONTROL_PLANE_REQUEST_CONTRACT_SCHEMA} v{CONTROL_PLANE_REQUEST_CONTRACT_VERSION}."
+        )
+
+    worker_protocol = info.get("worker_protocol")
+    if not isinstance(worker_protocol, dict):
+        raise RuntimeError(
+            "Server compatibility error: missing worker_protocol manifest; "
+            f"sdk-python 0.2.x requires worker_protocol.version {PROTOCOL_VERSION}."
+        )
+
+    worker_protocol_version = _manifest_version(worker_protocol)
+    if worker_protocol_version != PROTOCOL_VERSION:
+        raise RuntimeError(
+            "Server compatibility error: unsupported worker_protocol.version "
+            f"{worker_protocol_version!r}; sdk-python 0.2.x requires {PROTOCOL_VERSION!r}."
+        )
+
+
+def _contract_version_matches(value: Any, expected: int) -> bool:
+    if isinstance(value, int):
+        return value == expected
+    if isinstance(value, str) and value.isdigit():
+        return int(value) == expected
+    return False
+
+
 class Worker:
     def __init__(
         self,
@@ -52,32 +123,18 @@ def __init__(
         self._in_flight: set[asyncio.Task[Any]] = set()
 
     async def _register(self) -> None:
-        # Check server version compatibility
         try:
             info = await self.client.get_cluster_info()
-            server_version = info.get("version", "unknown")
-
-            # Parse major version (e.g., "0.1.9" -> 0, "2.0.0" -> 2)
-            try:
-                major = int(server_version.split(".")[0])
-            except (ValueError, IndexError):
-                log.warning("unable to parse server version %r; skipping compatibility check", server_version)
-                major = None
-
-            # Require server major version 0 or 2 (0.x is pre-release, 2.x is stable).
-            # SDK 0.2.x is compatible with server 0.x prereleases and 2.x protocol releases.
-            if major is not None and major not in (0, 2):
-                raise RuntimeError(
-                    f"Server version {server_version} is incompatible with sdk-python 0.2.x "
-                    f"(requires server 0.x or 2.x). "
-                    f"Upgrade the server or use a compatible SDK version."
-                )
-
-            log.debug("server version %s is compatible", server_version)
         except Exception as e:
-            if isinstance(e, RuntimeError) and "incompatible" in str(e):
-                raise
-            log.warning("failed to check server version compatibility: %s", e)
+            raise RuntimeError(f"Server compatibility error: unable to read /api/cluster/info: {e}") from e
+
+        _validate_server_compatibility(info)
+        log.debug(
+            "server compatibility accepted: app_version=%s control_plane=%s worker_protocol=%s",
+            info.get("version", "unknown"),
+            _manifest_version(info.get("control_plane")),
+            _manifest_version(info.get("worker_protocol")),
+        )
 
         await self.client.register_worker(
             worker_id=self.worker_id,
diff --git a/tests/test_worker.py b/tests/test_worker.py
@@ -7,7 +7,13 @@
 import pytest
 
 from durable_workflow import activity, workflow
-from durable_workflow.client import Client
+from durable_workflow.client import (
+    CONTROL_PLANE_REQUEST_CONTRACT_SCHEMA,
+    CONTROL_PLANE_REQUEST_CONTRACT_VERSION,
+    CONTROL_PLANE_VERSION,
+    PROTOCOL_VERSION,
+    Client,
+)
 from durable_workflow.worker import Worker
 
 
@@ -38,10 +44,29 @@ def mock_client() -> AsyncMock:
     client.complete_activity_task = AsyncMock(return_value={"outcome": "completed"})
     client.fail_workflow_task = AsyncMock(return_value={"outcome": "failed"})
     client.fail_activity_task = AsyncMock(return_value={"outcome": "failed"})
-    client.get_cluster_info = AsyncMock(return_value={"version": "2.0.0"})
+    client.get_cluster_info = AsyncMock(return_value=compatible_cluster_info())
     return client
 
 
+def compatible_cluster_info(**overrides: object) -> dict[str, object]:
+    info: dict[str, object] = {
+        "version": "not-authoritative",
+        "control_plane": {
+            "version": CONTROL_PLANE_VERSION,
+            "request_contract": {
+                "schema": CONTROL_PLANE_REQUEST_CONTRACT_SCHEMA,
+                "version": CONTROL_PLANE_REQUEST_CONTRACT_VERSION,
+                "operations": {},
+            },
+        },
+        "worker_protocol": {
+            "version": PROTOCOL_VERSION,
+        },
+    }
+    info.update(overrides)
+    return info
+
+
 class TestWorkerRegistration:
     @pytest.mark.asyncio
     async def test_register(self, mock_client: AsyncMock) -> None:
@@ -66,48 +91,59 @@ async def test_register_calls_cluster_info(self, mock_client: AsyncMock) -> None
         mock_client.get_cluster_info.assert_awaited_once()
 
     @pytest.mark.asyncio
-    async def test_register_accepts_server_major_0(self, mock_client: AsyncMock) -> None:
-        mock_client.get_cluster_info = AsyncMock(return_value={"version": "0.1.9"})
+    async def test_register_uses_protocol_manifests_not_top_level_app_version(
+        self, mock_client: AsyncMock
+    ) -> None:
+        mock_client.get_cluster_info = AsyncMock(return_value=compatible_cluster_info(version="3.0.0"))
         worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
         await worker._register()
         mock_client.register_worker.assert_awaited_once()
 
     @pytest.mark.asyncio
-    async def test_register_accepts_server_major_2(self, mock_client: AsyncMock) -> None:
-        mock_client.get_cluster_info = AsyncMock(return_value={"version": "2.3.4"})
+    async def test_register_rejects_missing_control_plane_manifest(self, mock_client: AsyncMock) -> None:
+        mock_client.get_cluster_info = AsyncMock(return_value=compatible_cluster_info(control_plane=None))
         worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
-        await worker._register()
-        mock_client.register_worker.assert_awaited_once()
+        with pytest.raises(RuntimeError, match="missing control_plane manifest"):
+            await worker._register()
+        mock_client.register_worker.assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_register_rejects_incompatible_major_1(self, mock_client: AsyncMock) -> None:
-        mock_client.get_cluster_info = AsyncMock(return_value={"version": "1.4.0"})
+    async def test_register_rejects_unsupported_control_plane_version(self, mock_client: AsyncMock) -> None:
+        mock_client.get_cluster_info = AsyncMock(
+            return_value=compatible_cluster_info(control_plane={"version": "3", "request_contract": {}})
+        )
         worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
-        with pytest.raises(RuntimeError, match="incompatible"):
+        with pytest.raises(RuntimeError, match="unsupported control_plane.version"):
             await worker._register()
         mock_client.register_worker.assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_register_rejects_incompatible_major_3(self, mock_client: AsyncMock) -> None:
-        mock_client.get_cluster_info = AsyncMock(return_value={"version": "3.0.0"})
+    async def test_register_rejects_missing_request_contract(self, mock_client: AsyncMock) -> None:
+        mock_client.get_cluster_info = AsyncMock(
+            return_value=compatible_cluster_info(control_plane={"version": CONTROL_PLANE_VERSION})
+        )
         worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
-        with pytest.raises(RuntimeError, match="incompatible"):
+        with pytest.raises(RuntimeError, match="missing control_plane.request_contract"):
             await worker._register()
         mock_client.register_worker.assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_register_skips_check_on_unparseable_version(self, mock_client: AsyncMock) -> None:
-        mock_client.get_cluster_info = AsyncMock(return_value={"version": "unknown"})
+    async def test_register_rejects_unsupported_worker_protocol_version(self, mock_client: AsyncMock) -> None:
+        mock_client.get_cluster_info = AsyncMock(
+            return_value=compatible_cluster_info(worker_protocol={"version": "2.0"})
+        )
         worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
-        await worker._register()
-        mock_client.register_worker.assert_awaited_once()
+        with pytest.raises(RuntimeError, match="unsupported worker_protocol.version"):
+            await worker._register()
+        mock_client.register_worker.assert_not_called()
 
     @pytest.mark.asyncio
-    async def test_register_continues_when_cluster_info_fails(self, mock_client: AsyncMock) -> None:
+    async def test_register_fails_closed_when_cluster_info_fails(self, mock_client: AsyncMock) -> None:
         mock_client.get_cluster_info = AsyncMock(side_effect=RuntimeError("network down"))
         worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])
-        await worker._register()
-        mock_client.register_worker.assert_awaited_once()
+        with pytest.raises(RuntimeError, match="unable to read /api/cluster/info"):
+            await worker._register()
+        mock_client.register_worker.assert_not_called()
 
 
 class TestWorkflowTaskExecution:
