Advertise Python worker admission capacity

Issue: zorporation/durable-workflow#488

Loop-ID: build-03

Author: durable-workflow-ops

README.md

@@ -156,6 +156,11 @@ 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.
 
+Workers also advertise their local workflow and activity concurrency limits
+during registration. Tune `max_concurrent_workflow_tasks` and
+`max_concurrent_activity_tasks` on `Worker(...)` to align local semaphores with
+the server's task-queue admission and operator visibility surfaces.
+
 ## Replay captured histories
 
 Use `Replayer` to debug a captured history without connecting to a live server:

src/durable_workflow/client.py

@@ -1134,6 +1134,8 @@ async def register_worker(
         supported_workflow_types: list[str] | None = None,
         workflow_definition_fingerprints: dict[str, str] | None = None,
         supported_activity_types: list[str] | None = None,
+        max_concurrent_workflow_tasks: int | None = None,
+        max_concurrent_activity_tasks: int | None = None,
         runtime: str = "python",
         sdk_version: str | None = None,
     ) -> Any:
@@ -1145,6 +1147,11 @@ async def register_worker(
         """
         if sdk_version is None:
             sdk_version = DEFAULT_SDK_VERSION
+        if max_concurrent_workflow_tasks is not None and max_concurrent_workflow_tasks < 1:
+            raise ValueError("max_concurrent_workflow_tasks must be at least 1")
+        if max_concurrent_activity_tasks is not None and max_concurrent_activity_tasks < 1:
+            raise ValueError("max_concurrent_activity_tasks must be at least 1")
+
         body: dict[str, Any] = {
             "worker_id": worker_id,
             "task_queue": task_queue,
@@ -1155,6 +1162,10 @@ async def register_worker(
         }
         if workflow_definition_fingerprints is not None:
             body["workflow_definition_fingerprints"] = workflow_definition_fingerprints
+        if max_concurrent_workflow_tasks is not None:
+            body["max_concurrent_workflow_tasks"] = max_concurrent_workflow_tasks
+        if max_concurrent_activity_tasks is not None:
+            body["max_concurrent_activity_tasks"] = max_concurrent_activity_tasks
         return await self._request("POST", "/worker/register", worker=True, json=body)
 
     async def poll_workflow_task(

src/durable_workflow/worker.py

@@ -241,7 +241,14 @@ def __init__(
         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)
+        if max_concurrent_workflow_tasks < 1:
+            raise ValueError("max_concurrent_workflow_tasks must be at least 1")
+        if max_concurrent_activity_tasks < 1:
+            raise ValueError("max_concurrent_activity_tasks must be at least 1")
+
         self._poll_timeout = poll_timeout
+        self.max_concurrent_workflow_tasks = max_concurrent_workflow_tasks
+        self.max_concurrent_activity_tasks = max_concurrent_activity_tasks
         self._stop = asyncio.Event()
         self._wf_semaphore = asyncio.Semaphore(max_concurrent_workflow_tasks)
         self._act_semaphore = asyncio.Semaphore(max_concurrent_activity_tasks)
@@ -319,6 +326,8 @@ async def _register(self) -> None:
             supported_workflow_types=list(self.workflows),
             workflow_definition_fingerprints=self.workflow_definition_fingerprints,
             supported_activity_types=list(self.activities),
+            max_concurrent_workflow_tasks=self.max_concurrent_workflow_tasks,
+            max_concurrent_activity_tasks=self.max_concurrent_activity_tasks,
         )
         log.info("worker %s registered on %s", self.worker_id, self.task_queue)
 

tests/test_client.py

@@ -665,6 +665,36 @@ async def test_register(self, client: Client) -> None:
             assert body["runtime"] == "python"
             assert body["workflow_definition_fingerprints"] == {"greeter": "sha256:abc"}
 
+    @pytest.mark.asyncio
+    async def test_register_sends_worker_capacity_when_configured(self, client: Client) -> None:
+        resp = _mock_response(201, {"worker_id": "w1", "registered": True})
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            await client.register_worker(
+                worker_id="w1",
+                task_queue="q1",
+                max_concurrent_workflow_tasks=3,
+                max_concurrent_activity_tasks=7,
+            )
+            body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json")
+            assert body["max_concurrent_workflow_tasks"] == 3
+            assert body["max_concurrent_activity_tasks"] == 7
+
+    @pytest.mark.asyncio
+    async def test_register_rejects_non_positive_worker_capacity(self, client: Client) -> None:
+        with pytest.raises(ValueError, match="max_concurrent_workflow_tasks"):
+            await client.register_worker(
+                worker_id="w1",
+                task_queue="q1",
+                max_concurrent_workflow_tasks=0,
+            )
+
+        with pytest.raises(ValueError, match="max_concurrent_activity_tasks"):
+            await client.register_worker(
+                worker_id="w1",
+                task_queue="q1",
+                max_concurrent_activity_tasks=0,
+            )
+
     @pytest.mark.asyncio
     async def test_register_advertises_installed_package_version(self, client: Client) -> None:
         from importlib.metadata import version as _pkg_version

tests/test_worker.py

@@ -138,6 +138,33 @@ async def test_register(self, mock_client: AsyncMock) -> None:
         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"]
+        assert call_kwargs["max_concurrent_workflow_tasks"] == 10
+        assert call_kwargs["max_concurrent_activity_tasks"] == 10
+
+    @pytest.mark.asyncio
+    async def test_register_advertises_custom_concurrency_limits(self, mock_client: AsyncMock) -> None:
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[echo_activity],
+            worker_id="w-capacity",
+            max_concurrent_workflow_tasks=3,
+            max_concurrent_activity_tasks=7,
+        )
+        await worker._register()
+        call_kwargs = mock_client.register_worker.call_args.kwargs
+        assert call_kwargs["max_concurrent_workflow_tasks"] == 3
+        assert call_kwargs["max_concurrent_activity_tasks"] == 7
+
+    def test_constructor_rejects_non_positive_concurrency_limits(
+        self, mock_client: AsyncMock
+    ) -> None:
+        with pytest.raises(ValueError, match="max_concurrent_workflow_tasks"):
+            Worker(mock_client, task_queue="q1", max_concurrent_workflow_tasks=0)
+
+        with pytest.raises(ValueError, match="max_concurrent_activity_tasks"):
+            Worker(mock_client, task_queue="q1", max_concurrent_activity_tasks=0)
 
     def test_constructor_rejects_changed_workflow_definition_for_same_worker_id(
         self, mock_client: AsyncMock