Expose task queue admission in Python SDK

Issue: zorporation/durable-workflow#488

Loop-ID: build-01

Author: durable-workflow-ops

README.md

@@ -159,7 +159,17 @@ 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.
+the server's task-queue admission and operator visibility surfaces. Use
+`Client.list_task_queues()` or `Client.describe_task_queue("orders")` to read
+the server-side workflow, activity, and query-task admission status before
+tuning those local limits:
+
+```python
+queues = await client.list_task_queues()
+for queue in queues.task_queues:
+    workflow_admission = queue.admission.workflow_tasks if queue.admission else None
+    print(queue.name, workflow_admission.status if workflow_admission else "unknown")
+```
 
 ## Replay captured histories
 

src/durable_workflow/__init__.py

@@ -17,6 +17,11 @@
     ScheduleList,
     ScheduleSpec,
     ScheduleTriggerResult,
+    TaskQueueAdmission,
+    TaskQueueDescription,
+    TaskQueueList,
+    TaskQueueQueryAdmission,
+    TaskQueueTaskAdmission,
     WorkflowExecution,
     WorkflowHandle,
     WorkflowList,
@@ -100,6 +105,11 @@
     "ScheduleSpec",
     "ScheduleTriggerResult",
     "StartChildWorkflow",
+    "TaskQueueAdmission",
+    "TaskQueueDescription",
+    "TaskQueueList",
+    "TaskQueueQueryAdmission",
+    "TaskQueueTaskAdmission",
     "Worker",
     "WorkerInterceptor",
     "WorkflowExecution",

src/durable_workflow/client.py

@@ -25,6 +25,7 @@
 from importlib.metadata import PackageNotFoundError
 from importlib.metadata import version as _pkg_version
 from typing import Any
+from urllib.parse import quote
 
 import httpx
 
@@ -102,6 +103,127 @@ class WorkflowList:
     next_page_token: str | None = None
 
 
+@dataclass
+class TaskQueueTaskAdmission:
+    """Workflow/activity admission state for one task queue."""
+
+    status: str | None = None
+    budget_source: str | None = None
+    server_budget_source: str | None = None
+    active_worker_count: int | None = None
+    configured_slot_count: int | None = None
+    leased_count: int | None = None
+    ready_count: int | None = None
+    available_slot_count: int | None = None
+    server_max_active_leases_per_queue: int | None = None
+    server_active_lease_count: int | None = None
+    server_remaining_active_lease_capacity: int | None = None
+    server_lock_required: bool | None = None
+    server_lock_supported: bool | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> TaskQueueTaskAdmission | None:
+        if data is None:
+            return None
+        return cls(
+            status=data.get("status"),
+            budget_source=data.get("budget_source"),
+            server_budget_source=data.get("server_budget_source"),
+            active_worker_count=data.get("active_worker_count"),
+            configured_slot_count=data.get("configured_slot_count"),
+            leased_count=data.get("leased_count"),
+            ready_count=data.get("ready_count"),
+            available_slot_count=data.get("available_slot_count"),
+            server_max_active_leases_per_queue=data.get("server_max_active_leases_per_queue"),
+            server_active_lease_count=data.get("server_active_lease_count"),
+            server_remaining_active_lease_capacity=data.get("server_remaining_active_lease_capacity"),
+            server_lock_required=data.get("server_lock_required"),
+            server_lock_supported=data.get("server_lock_supported"),
+        )
+
+
+@dataclass
+class TaskQueueQueryAdmission:
+    """Worker-routed query-task admission state for one task queue."""
+
+    status: str | None = None
+    budget_source: str | None = None
+    max_pending_per_queue: int | None = None
+    approximate_pending_count: int | None = None
+    remaining_pending_capacity: int | None = None
+    lock_required: bool | None = None
+    lock_supported: bool | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> TaskQueueQueryAdmission | None:
+        if data is None:
+            return None
+        return cls(
+            status=data.get("status"),
+            budget_source=data.get("budget_source"),
+            max_pending_per_queue=data.get("max_pending_per_queue"),
+            approximate_pending_count=data.get("approximate_pending_count"),
+            remaining_pending_capacity=data.get("remaining_pending_capacity"),
+            lock_required=data.get("lock_required"),
+            lock_supported=data.get("lock_supported"),
+        )
+
+
+@dataclass
+class TaskQueueAdmission:
+    """Server-side admission budgets for workflow, activity, and query tasks."""
+
+    workflow_tasks: TaskQueueTaskAdmission | None = None
+    activity_tasks: TaskQueueTaskAdmission | None = None
+    query_tasks: TaskQueueQueryAdmission | None = None
+    raw: dict[str, Any] | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any] | None) -> TaskQueueAdmission:
+        payload = data or {}
+        return cls(
+            workflow_tasks=TaskQueueTaskAdmission.from_dict(payload.get("workflow_tasks")),
+            activity_tasks=TaskQueueTaskAdmission.from_dict(payload.get("activity_tasks")),
+            query_tasks=TaskQueueQueryAdmission.from_dict(payload.get("query_tasks")),
+            raw=payload,
+        )
+
+
+@dataclass
+class TaskQueueDescription:
+    """Current server visibility and admission state for one task queue."""
+
+    name: str
+    namespace: str | None = None
+    stats: dict[str, Any] | None = None
+    admission: TaskQueueAdmission | None = None
+    pollers: list[dict[str, Any]] | None = None
+    current_leases: list[dict[str, Any]] | None = None
+    raw: dict[str, Any] | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> TaskQueueDescription:
+        pollers = data.get("pollers")
+        current_leases = data.get("current_leases")
+        return cls(
+            name=data.get("name", ""),
+            namespace=data.get("namespace"),
+            stats=data.get("stats"),
+            admission=TaskQueueAdmission.from_dict(data.get("admission")),
+            pollers=pollers if isinstance(pollers, list) else None,
+            current_leases=current_leases if isinstance(current_leases, list) else None,
+            raw=data,
+        )
+
+
+@dataclass
+class TaskQueueList:
+    """One task-queue visibility page returned by the server."""
+
+    namespace: str | None
+    task_queues: list[TaskQueueDescription]
+
+
 @dataclass
 class ScheduleSpec:
     """Calendar or interval rules for a scheduled workflow."""
@@ -571,6 +693,38 @@ async def health(self) -> dict[str, Any]:
             )
         return result
 
+    # ── Task queues ────────────────────────────────────────────────────
+    async def list_task_queues(self) -> TaskQueueList:
+        """List task queues with server-side admission status.
+
+        Admission data describes server budgets and observed backlog. Worker
+        constructor limits remain local semaphores that are advertised during
+        registration.
+        """
+        data = await self._request("GET", "/task-queues")
+        items = data.get("task_queues", []) if isinstance(data, dict) else []
+        return TaskQueueList(
+            namespace=data.get("namespace") if isinstance(data, dict) else None,
+            task_queues=[
+                TaskQueueDescription.from_dict(item)
+                for item in items
+                if isinstance(item, dict)
+            ],
+        )
+
+    async def describe_task_queue(self, name: str) -> TaskQueueDescription:
+        """Return backlog, poller, lease, and admission detail for ``name``."""
+        data = await self._request("GET", f"/task-queues/{quote(name, safe='')}", context=name)
+        if not isinstance(data, dict):
+            raise ServerError(
+                200,
+                {
+                    "reason": "invalid_task_queue_response",
+                    "message": f"expected JSON object, got {type(data).__name__}",
+                },
+            )
+        return TaskQueueDescription.from_dict(data)
+
     # ── Workflows ──────────────────────────────────────────────────────
     async def start_workflow(
         self,

tests/test_client.py

@@ -361,6 +361,103 @@ async def test_list(self, client: Client) -> None:
             assert result.executions[0].workflow_id == "wf-1"
 
 
+class TestTaskQueues:
+    @pytest.mark.asyncio
+    async def test_list_task_queues_parses_admission(self, client: Client) -> None:
+        resp = _mock_response(200, {
+            "namespace": "ns1",
+            "task_queues": [
+                {
+                    "name": "orders",
+                    "stats": {"approximate_backlog_count": 2},
+                    "admission": {
+                        "workflow_tasks": {
+                            "status": "throttled",
+                            "budget_source": "worker_registration.max_concurrent_workflow_tasks",
+                            "active_worker_count": 2,
+                            "configured_slot_count": 10,
+                            "leased_count": 1,
+                            "ready_count": 2,
+                            "available_slot_count": 9,
+                            "server_budget_source": "server.admission.workflow_tasks.max_active_leases_per_queue",
+                            "server_max_active_leases_per_queue": 1,
+                            "server_active_lease_count": 1,
+                            "server_remaining_active_lease_capacity": 0,
+                            "server_lock_required": True,
+                            "server_lock_supported": True,
+                        },
+                        "activity_tasks": {"status": "accepting", "configured_slot_count": 5},
+                        "query_tasks": {
+                            "status": "full",
+                            "budget_source": "server.query_tasks.max_pending_per_queue",
+                            "max_pending_per_queue": 10,
+                            "approximate_pending_count": 10,
+                            "remaining_pending_capacity": 0,
+                            "lock_required": True,
+                            "lock_supported": True,
+                        },
+                    },
+                }
+            ],
+        })
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.list_task_queues()
+
+            assert result.namespace == "ns1"
+            assert len(result.task_queues) == 1
+            queue = result.task_queues[0]
+            assert queue.name == "orders"
+            assert queue.stats == {"approximate_backlog_count": 2}
+            assert queue.admission is not None
+            assert queue.admission.workflow_tasks is not None
+            assert queue.admission.workflow_tasks.status == "throttled"
+            assert queue.admission.workflow_tasks.server_remaining_active_lease_capacity == 0
+            assert queue.admission.activity_tasks is not None
+            assert queue.admission.activity_tasks.configured_slot_count == 5
+            assert queue.admission.query_tasks is not None
+            assert queue.admission.query_tasks.status == "full"
+            assert queue.admission.query_tasks.lock_supported is True
+            assert queue.admission.raw is not None
+            assert queue.admission.raw["query_tasks"]["max_pending_per_queue"] == 10
+            assert mock.call_args.args[:2] == ("GET", "/api/task-queues")
+
+    @pytest.mark.asyncio
+    async def test_describe_task_queue_parses_details_and_escapes_name(self, client: Client) -> None:
+        resp = _mock_response(200, {
+            "name": "orders/high priority",
+            "namespace": "ns1",
+            "stats": {"pollers": {"active_count": 1}},
+            "pollers": [{"worker_id": "w1", "status": "active"}],
+            "current_leases": [{"task_id": "t1", "task_type": "workflow"}],
+            "admission": {"workflow_tasks": {"status": "accepting"}},
+        })
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.describe_task_queue("orders/high priority")
+
+            assert result.name == "orders/high priority"
+            assert result.namespace == "ns1"
+            assert result.pollers == [{"worker_id": "w1", "status": "active"}]
+            assert result.current_leases == [{"task_id": "t1", "task_type": "workflow"}]
+            assert result.admission is not None
+            assert result.admission.workflow_tasks is not None
+            assert result.admission.workflow_tasks.status == "accepting"
+            assert mock.call_args.args[:2] == ("GET", "/api/task-queues/orders%2Fhigh%20priority")
+
+    @pytest.mark.asyncio
+    async def test_describe_task_queue_rejects_non_object_response(self, client: Client) -> None:
+        resp = httpx.Response(
+            status_code=200,
+            content=b"[]",
+            headers={"content-type": "application/json"},
+            request=httpx.Request("GET", "http://test"),
+        )
+        with (
+            patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp),
+            pytest.raises(ServerError, match="invalid_task_queue_response"),
+        ):
+            await client.describe_task_queue("orders")
+
+
 class TestErrorMapping:
     @pytest.mark.asyncio
     async def test_401_unauthorized(self, client: Client) -> None: