Conformance blocker: cover full skew refusal matrix beyond smoke (#188)

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -19,6 +19,7 @@
 from __future__ import annotations
 
 import asyncio
+import os
 import time
 import uuid
 import warnings
@@ -59,6 +60,14 @@ def _default_sdk_version() -> str:
 DEFAULT_SDK_VERSION = _default_sdk_version()
 
 
+def _protocol_version_from_env(name: str, default: str) -> str:
+    value = os.environ.get(name)
+    if value is None or value.strip() == "":
+        return default
+
+    return value.strip()
+
+
 def _route_for_metrics(path: str) -> str:
     clean_path = path.split("?", 1)[0]
     parts = [part for part in clean_path.strip("/").split("/") if part]
@@ -1262,9 +1271,15 @@ def _headers(self, *, worker: bool = False) -> dict[str, str]:
             h["Authorization"] = f"Bearer {token}"
         h["X-Namespace"] = self.namespace
         if worker:
-            h["X-Durable-Workflow-Protocol-Version"] = PROTOCOL_VERSION
+            h["X-Durable-Workflow-Protocol-Version"] = _protocol_version_from_env(
+                "DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION",
+                PROTOCOL_VERSION,
+            )
         else:
-            h["X-Durable-Workflow-Control-Plane-Version"] = CONTROL_PLANE_VERSION
+            h["X-Durable-Workflow-Control-Plane-Version"] = _protocol_version_from_env(
+                "DURABLE_WORKFLOW_CONTROL_PLANE_VERSION",
+                CONTROL_PLANE_VERSION,
+            )
         return h
 
     def _auth_token(self, *, worker: bool = False) -> str | None:

tests/test_client.py

@@ -58,6 +58,16 @@ def test_worker_headers(self, client: Client) -> None:
         assert h["X-Durable-Workflow-Protocol-Version"] == PROTOCOL_VERSION
         assert "X-Durable-Workflow-Control-Plane-Version" not in h
 
+    def test_conformance_protocol_header_overrides(self, client: Client, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("DURABLE_WORKFLOW_CONTROL_PLANE_VERSION", "3")
+        monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "2.0")
+
+        h = client._headers(worker=False)
+        assert h["X-Durable-Workflow-Control-Plane-Version"] == "3"
+
+        worker_headers = client._headers(worker=True)
+        assert worker_headers["X-Durable-Workflow-Protocol-Version"] == "2.0"
+
     def test_no_token(self) -> None:
         c = Client("http://localhost:8080")
         h = c._headers()