PR Feedback

Author: andystaples

.vscode/mcp.json

@@ -57,6 +57,7 @@ class ActivityWorkItem:
     task_id: int
     input: Optional[str]
     completion_token: int
+    version: Optional[str] = None
 
 
 @dataclass
@@ -439,14 +440,50 @@ def RestartInstance(self, request: pb.RestartInstanceRequest, context):
 
     @staticmethod
     def _parse_work_item_filters(request: pb.GetWorkItemsRequest):
-        """Extract filter name sets from the request, or None if unfiltered."""
+        """Extract filters from the request.
+
+        Returns a tuple of three values, one per work-item category.  Each
+        value is either ``None`` (no filtering -- dispatch everything) or a
+        ``dict`` mapping a task name to a ``frozenset`` of accepted versions
+        (empty frozenset means *any* version of that name is accepted).
+        An empty ``dict`` means the worker opted into filtering for that
+        category but listed no names, so *nothing* should match.
+        """
         if not request.HasField("workItemFilters"):
             return None, None, None
         wf = request.workItemFilters
-        orch_names = {f.name for f in wf.orchestrations} if wf.orchestrations else None
-        activity_names = {f.name for f in wf.activities} if wf.activities else None
-        entity_names = {f.name for f in wf.entities} if wf.entities else None
-        return orch_names, activity_names, entity_names
+
+        def _build_filter(filters):
+            result: dict[str, frozenset[str]] = {}
+            for f in filters:
+                versions = frozenset(f.versions) if f.versions else frozenset()
+                existing = result.get(f.name, frozenset())
+                result[f.name] = existing | versions
+            return result
+
+        orch_filter = _build_filter(wf.orchestrations)
+        activity_filter = _build_filter(wf.activities)
+        entity_filter = {f.name: frozenset() for f in wf.entities}
+        return orch_filter, activity_filter, entity_filter
+
+    @staticmethod
+    def _matches_filter(name: str, version: Optional[str],
+                        filt: Optional[dict[str, frozenset[str]]]) -> bool:
+        """Check whether a work item matches the parsed filter.
+
+        *filt* is ``None`` when the worker did not opt into filtering
+        (everything matches).  Otherwise it is a dict mapping accepted
+        names to a frozenset of accepted versions.  An empty frozenset
+        means any version of that name is accepted.
+        """
+        if filt is None:
+            return True
+        accepted_versions = filt.get(name)
+        if accepted_versions is None:
+            return False
+        if not accepted_versions:
+            return True  # empty set -- any version
+        return (version or "") in accepted_versions
 
     def GetWorkItems(self, request: pb.GetWorkItemsRequest, context):
         """Streams work items to the worker (orchestration and activity work items)."""
@@ -468,8 +505,9 @@ def GetWorkItems(self, request: pb.GetWorkItemsRequest, context):
                         if not instance or not instance.pending_events:
                             continue
 
-                        # Skip if orchestration name doesn't match filters
-                        if orch_filter is not None and instance.name not in orch_filter:
+                        # Skip if orchestration doesn't match filters
+                        if not self._matches_filter(
+                                instance.name, instance.version, orch_filter):
                             skipped_orchs.append(instance_id)
                             continue
 
@@ -515,7 +553,9 @@ def GetWorkItems(self, request: pb.GetWorkItemsRequest, context):
                         matched_activity = None
                         while self._activity_queue:
                             candidate = self._activity_queue.popleft()
-                            if activity_filter is not None and candidate.name not in activity_filter:
+                            if not self._matches_filter(
+                                    candidate.name, candidate.version,
+                                    activity_filter):
                                 skipped.append(candidate)
                                 continue
                             matched_activity = candidate
@@ -548,7 +588,9 @@ def GetWorkItems(self, request: pb.GetWorkItemsRequest, context):
                                 if entity_filter is not None:
                                     try:
                                         parsed = EntityInstanceId.parse(entity_id)
-                                        if parsed.entity not in entity_filter:
+                                        if not self._matches_filter(
+                                                parsed.entity, None,
+                                                entity_filter):
                                             skipped_entities.append(entity_id)
                                             continue
                                     except ValueError:
@@ -1313,12 +1355,15 @@ def _process_schedule_task_action(self, instance: OrchestrationInstance,
             instance.status = pb.ORCHESTRATION_STATUS_RUNNING
 
         # Queue activity for execution
+        task_version = schedule_task.version.value \
+            if schedule_task.HasField("version") else None
         self._activity_queue.append(ActivityWorkItem(
             instance_id=instance.instance_id,
             name=task_name,
             task_id=task_id,
             input=input_value,
-            completion_token=instance.completion_token
+            completion_token=instance.completion_token,
+            version=task_version,
         ))
         self._work_available.set()
 

durabletask/testing/in_memory_backend.py

@@ -13,7 +13,7 @@
 from threading import Event, Thread
 from types import GeneratorType
 from enum import Enum
-from typing import Any, Generator, Optional, Sequence, Tuple, TypeVar, Union
+from typing import Any, Generator, Optional, Sequence, Tuple, TypeVar, Union, overload
 import uuid
 from packaging.version import InvalidVersion, parse
 
@@ -181,8 +181,11 @@ def __init__(self, name: str):
 
         Args:
             name: The name of the entity to filter.
+                  The name is normalized to lowercase to match
+                  entity registration and instance ID conventions.
         """
-        self.name = name
+        EntityInstanceId.validate_entity_name(name)
+        self.name = name.lower()
 
 
 class WorkItemFilters:
@@ -516,6 +519,15 @@ def use_versioning(self, version: VersioningOptions) -> None:
             raise RuntimeError("Cannot set default version while the worker is running.")
         self._registry.versioning = version
 
+    @overload
+    def use_work_item_filters(self) -> None: ...
+
+    @overload
+    def use_work_item_filters(self, filters: WorkItemFilters) -> None: ...
+
+    @overload
+    def use_work_item_filters(self, filters: None) -> None: ...
+
     def use_work_item_filters(
         self,
         filters: Union[WorkItemFilters, None, object] = _AUTO_GENERATE_FILTERS,

durabletask/worker.py

@@ -47,6 +47,18 @@ def test_defaults(self):
         f = EntityWorkItemFilter(name="myentity")
         assert f.name == "myentity"
 
+    def test_name_normalized_to_lowercase(self):
+        f = EntityWorkItemFilter(name="Counter")
+        assert f.name == "counter"
+
+    def test_invalid_name_raises(self):
+        with pytest.raises(ValueError):
+            EntityWorkItemFilter(name="bad@name")
+
+    def test_empty_name_raises(self):
+        with pytest.raises(ValueError):
+            EntityWorkItemFilter(name="")
+
 
 # ---------------------------------------------------------------------------
 # WorkItemFilters construction

tests/durabletask/test_work_item_filters.py

@@ -12,6 +12,8 @@
     ActivityWorkItemFilter,
     EntityWorkItemFilter,
     OrchestrationWorkItemFilter,
+    VersioningOptions,
+    VersionMatchStrategy,
     WorkItemFilters,
 )
 from durabletask.testing import create_test_backend
@@ -256,3 +258,131 @@ def ping(self, _):
 
     assert matched_invoked
     assert not unmatched_invoked
+
+
+# ------------------------------------------------------------------
+# Tests: version-aware filtering with strict versioning
+# ------------------------------------------------------------------
+
+def _simple_v2_orchestrator(ctx: task.OrchestrationContext, input: int):
+    """Orchestrator that returns immediately (no activities) for version tests."""
+    return input + 1
+
+
+def test_strict_version_matching_orchestration_completes():
+    """Orchestration scheduled with the matching version is processed."""
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
+        w.add_orchestrator(_simple_v2_orchestrator)
+        w.use_versioning(VersioningOptions(
+            version="2.0",
+            match_strategy=VersionMatchStrategy.STRICT,
+        ))
+        w.use_work_item_filters()  # auto-generate with version
+        w.start()
+
+        c = client.TaskHubGrpcClient(host_address=HOST)
+        id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=10, version="2.0")
+        state = c.wait_for_orchestration_completion(id, timeout=30)
+
+    assert state is not None
+    assert state.runtime_status == client.OrchestrationStatus.COMPLETED
+    assert state.serialized_output == "11"
+
+
+def test_strict_version_incompatible_orchestration_stays_pending():
+    """Orchestration with an incompatible version is not dispatched and stays pending."""
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
+        w.add_orchestrator(_simple_v2_orchestrator)
+        w.use_versioning(VersioningOptions(
+            version="2.0",
+            match_strategy=VersionMatchStrategy.STRICT,
+        ))
+        w.use_work_item_filters()
+        w.start()
+
+        c = client.TaskHubGrpcClient(host_address=HOST)
+
+        # Schedule with version "1.0" — incompatible with the worker's "2.0"
+        bad_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=5, version="1.0")
+
+        # Schedule a compatible one so we can confirm the worker is active
+        good_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=5, version="2.0")
+        good_state = c.wait_for_orchestration_completion(good_id, timeout=30)
+
+        assert good_state is not None
+        assert good_state.runtime_status == client.OrchestrationStatus.COMPLETED
+
+        # The incompatible orchestration must remain pending (not failed)
+        bad_state = c.get_orchestration_state(bad_id)
+        assert bad_state is not None
+        assert bad_state.runtime_status == client.OrchestrationStatus.PENDING
+
+
+def test_strict_version_no_version_orchestration_stays_pending():
+    """Orchestration scheduled without a version is not dispatched by a strict worker."""
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
+        w.add_orchestrator(_simple_v2_orchestrator)
+        w.use_versioning(VersioningOptions(
+            version="2.0",
+            match_strategy=VersionMatchStrategy.STRICT,
+        ))
+        w.use_work_item_filters()
+        w.start()
+
+        c = client.TaskHubGrpcClient(host_address=HOST)
+
+        # Schedule without any version
+        no_ver_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=1)
+
+        # Schedule a compatible one to prove the worker is running
+        good_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=1, version="2.0")
+        good_state = c.wait_for_orchestration_completion(good_id, timeout=30)
+        assert good_state is not None
+        assert good_state.runtime_status == client.OrchestrationStatus.COMPLETED
+
+        # The unversioned orchestration must remain pending
+        no_ver_state = c.get_orchestration_state(no_ver_id)
+        assert no_ver_state is not None
+        assert no_ver_state.runtime_status == client.OrchestrationStatus.PENDING
+
+
+def test_strict_version_explicit_filters_with_versions():
+    """Explicit filters with version constraints enforce strict matching."""
+    custom_filters = WorkItemFilters(
+        orchestrations=[
+            OrchestrationWorkItemFilter(
+                name=task.get_name(_simple_v2_orchestrator),
+                versions=["3.0"],
+            ),
+        ],
+    )
+
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
+        w.add_orchestrator(_simple_v2_orchestrator)
+        w.use_work_item_filters(custom_filters)
+        w.start()
+
+        c = client.TaskHubGrpcClient(host_address=HOST)
+
+        # Version "2.0" does not match the filter's "3.0"
+        bad_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=1, version="2.0")
+
+        # Version "3.0" should match
+        good_id = c.schedule_new_orchestration(
+            _simple_v2_orchestrator, input=1, version="3.0")
+        good_state = c.wait_for_orchestration_completion(good_id, timeout=30)
+
+        assert good_state is not None
+        assert good_state.runtime_status == client.OrchestrationStatus.COMPLETED
+        assert good_state.serialized_output == "2"
+
+        # Mismatched version must remain pending
+        bad_state = c.get_orchestration_state(bad_id)
+        assert bad_state is not None
+        assert bad_state.runtime_status == client.OrchestrationStatus.PENDING