Guard Python workflow re-registration

Issue: zorporation/durable-workflow#448

Loop-ID: build-02

Author: durable-workflow-ops

README.md

@@ -133,6 +133,11 @@ server. Query routing and synchronous pre-accept update validator execution are
 still server-side follow-ups; use those paths only with deployments that
 advertise support for the target workflow type.
 
+Workers fingerprint registered workflow class definitions and advertise those
+fingerprints during registration. Re-registering the same `worker_id` with a
+changed class body for an already advertised workflow type raises immediately;
+restart the worker process with a new id before serving changed workflow code.
+
 ## Features
 
 - **Async-first**: Built on `httpx` and `asyncio`
@@ -141,6 +146,7 @@ advertise support for the target workflow type.
 - **HTTP/JSON protocol**: No gRPC, no protobuf dependencies
 - **Codec envelopes**: Avro payloads by default, with JSON decode compatibility for existing history
 - **Payload-size warnings**: Structured warnings before oversized workflow, activity, signal, update, query, or search-attribute payloads reach the server
+- **Workflow definition guard**: Worker registration refuses same-id hot reloads when a workflow class definition changed
 - **Worker interceptors**: Typed hooks around workflow tasks, activity calls, and query tasks for tracing, logging, and custom metrics
 - **Metrics hooks**: Pluggable counters and histograms, with an optional Prometheus adapter
 

src/durable_workflow/client.py

@@ -1103,6 +1103,7 @@ async def register_worker(
         worker_id: str,
         task_queue: str,
         supported_workflow_types: list[str] | None = None,
+        workflow_definition_fingerprints: dict[str, str] | None = None,
         supported_activity_types: list[str] | None = None,
         runtime: str = "python",
         sdk_version: str | None = None,
@@ -1123,6 +1124,8 @@ async def register_worker(
             "supported_workflow_types": supported_workflow_types or [],
             "supported_activity_types": supported_activity_types or [],
         }
+        if workflow_definition_fingerprints is not None:
+            body["workflow_definition_fingerprints"] = workflow_definition_fingerprints
         return await self._request("POST", "/worker/register", worker=True, json=body)
 
     async def poll_workflow_task(

src/durable_workflow/worker.py

@@ -17,11 +17,14 @@
 
 import asyncio
 import contextlib
+import hashlib
+import inspect
 import logging
 import time
 import traceback
 import uuid
 from collections.abc import Awaitable, Callable, Iterable
+from types import FunctionType
 from typing import Any
 
 from . import serializer
@@ -54,6 +57,7 @@
 log = logging.getLogger("durable_workflow.worker")
 
 _TERMINAL_WORKFLOW_STATUSES = {"completed", "failed", "terminated", "canceled", "cancelled"}
+_WORKER_WORKFLOW_FINGERPRINTS: dict[tuple[str, str], str] = {}
 
 
 def _command_payload_codec(codec: object) -> str:
@@ -72,6 +76,70 @@ def _string_or_none(value: Any) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+def _callable_fingerprint_payload(value: object) -> str:
+    if isinstance(value, staticmethod | classmethod):
+        value = value.__func__
+    if not isinstance(value, FunctionType):
+        return repr(value)
+
+    try:
+        return inspect.getsource(value)
+    except (OSError, TypeError):
+        code = value.__code__
+        return repr(
+            (
+                code.co_argcount,
+                code.co_posonlyargcount,
+                code.co_kwonlyargcount,
+                code.co_code,
+                code.co_consts,
+                code.co_names,
+                code.co_varnames,
+            )
+        )
+
+
+def _workflow_definition_fingerprint(cls: type) -> str:
+    h = hashlib.sha256()
+    h.update(b"durable-workflow-python.workflow-definition.v1\0")
+    h.update(f"{cls.__module__}.{cls.__qualname__}\0".encode())
+    h.update(f"{_workflow_name(cls)}\0".encode())
+
+    for registry_name in (
+        "__workflow_signals__",
+        "__workflow_queries__",
+        "__workflow_updates__",
+        "__workflow_update_validators__",
+    ):
+        registry = getattr(cls, registry_name, {}) or {}
+        h.update(registry_name.encode())
+        h.update(repr(sorted(registry.items())).encode())
+
+    try:
+        h.update(inspect.getsource(cls).encode())
+    except (OSError, TypeError):
+        for attr, value in sorted(cls.__dict__.items()):
+            if attr.startswith("__") and attr.endswith("__"):
+                continue
+            h.update(attr.encode())
+            h.update(_callable_fingerprint_payload(value).encode())
+
+    return f"sha256:{h.hexdigest()}"
+
+
+def _guard_worker_workflow_fingerprints(worker_id: str, fingerprints: dict[str, str]) -> None:
+    for workflow_type, fingerprint in fingerprints.items():
+        key = (worker_id, workflow_type)
+        previous = _WORKER_WORKFLOW_FINGERPRINTS.get(key)
+        if previous is not None and previous != fingerprint:
+            raise RuntimeError(
+                "Workflow definition changed for worker "
+                f"{worker_id!r} and workflow type {workflow_type!r}; "
+                "restart the worker process before re-registering this workflow type."
+            )
+        _WORKER_WORKFLOW_FINGERPRINTS[key] = fingerprint
+
+
 def _manifest_version(manifest: Any) -> str:
     if isinstance(manifest, dict):
         value = manifest.get("version")
@@ -166,8 +234,13 @@ def __init__(
         self.client = client
         self.task_queue = task_queue
         self.workflows = {_workflow_name(w): w for w in workflows}
+        self.workflow_definition_fingerprints = {
+            workflow_type: _workflow_definition_fingerprint(workflow_cls)
+            for workflow_type, workflow_cls in self.workflows.items()
+        }
         self.activities = {_activity_name(a): a for a in activities}
         self.worker_id = worker_id or f"py-worker-{uuid.uuid4().hex[:8]}"
+        _guard_worker_workflow_fingerprints(self.worker_id, self.workflow_definition_fingerprints)
         self._poll_timeout = poll_timeout
         self._stop = asyncio.Event()
         self._wf_semaphore = asyncio.Semaphore(max_concurrent_workflow_tasks)
@@ -244,6 +317,7 @@ async def _register(self) -> None:
             worker_id=self.worker_id,
             task_queue=self.task_queue,
             supported_workflow_types=list(self.workflows),
+            workflow_definition_fingerprints=self.workflow_definition_fingerprints,
             supported_activity_types=list(self.activities),
         )
         log.info("worker %s registered on %s", self.worker_id, self.task_queue)

tests/test_client.py

@@ -629,11 +629,13 @@ async def test_register(self, client: Client) -> None:
                 worker_id="w1",
                 task_queue="q1",
                 supported_workflow_types=["greeter"],
+                workflow_definition_fingerprints={"greeter": "sha256:abc"},
                 supported_activity_types=["greet"],
             )
             assert result["registered"] is True
             body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json")
             assert body["runtime"] == "python"
+            assert body["workflow_definition_fingerprints"] == {"greeter": "sha256:abc"}
 
     @pytest.mark.asyncio
     async def test_register_advertises_installed_package_version(self, client: Client) -> None:

tests/test_worker.py

@@ -127,8 +127,57 @@ async def test_register(self, mock_client: AsyncMock) -> None:
         call_kwargs = mock_client.register_worker.call_args.kwargs
         assert call_kwargs["task_queue"] == "q1"
         assert "test-wf" in call_kwargs["supported_workflow_types"]
+        assert call_kwargs["workflow_definition_fingerprints"]["test-wf"].startswith("sha256:")
         assert "test-act" in call_kwargs["supported_activity_types"]
 
+    def test_constructor_rejects_changed_workflow_definition_for_same_worker_id(
+        self, mock_client: AsyncMock
+    ) -> None:
+        @workflow.defn(name="reloadable-wf")
+        class ReloadableWorkflowV1:
+            def run(self, ctx):  # type: ignore[no-untyped-def]
+                return "v1"
+
+        @workflow.defn(name="reloadable-wf")
+        class ReloadableWorkflowV2:
+            def run(self, ctx):  # type: ignore[no-untyped-def]
+                return "v2"
+
+        Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[ReloadableWorkflowV1],
+            activities=[],
+            worker_id="reload-worker",
+        )
+
+        with pytest.raises(RuntimeError, match="Workflow definition changed"):
+            Worker(
+                mock_client,
+                task_queue="q1",
+                workflows=[ReloadableWorkflowV2],
+                activities=[],
+                worker_id="reload-worker",
+            )
+
+    def test_constructor_allows_same_workflow_definition_for_same_worker_id(
+        self, mock_client: AsyncMock
+    ) -> None:
+        Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[],
+            worker_id="stable-worker",
+        )
+        Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[],
+            worker_id="stable-worker",
+        )
+
     @pytest.mark.asyncio
     async def test_register_calls_cluster_info(self, mock_client: AsyncMock) -> None:
         worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[])