feat: add debug.list_tasks to expose GET /v1/tasks (#2059)

Adds a list_tasks() method to the sync and async debug namespace that
calls the server's distributed-tasks endpoint, so users can inspect
long-running background operations (e.g. reindexing) tracked across
cluster nodes without shelling out to curl.

Includes DistributedTask/DistributedTaskUnit pydantic models mirroring
the server's DistributedTasks OpenAPI schema, mock tests covering the
populated and empty-response cases, and integration test stubs
following the existing debug test conventions.

Author: muhammadshayan124

integration/test_client_debug.py

@@ -63,3 +63,20 @@ def test_get_object_multi_node(
         debug_obj = client.debug.get_object_over_rest(collection.name, uuid, node_name=node_name)
         assert debug_obj is not None
         assert str(debug_obj.uuid) == str(uuid)
+
+
+def test_list_tasks(client_factory: ClientFactory) -> None:
+    client = client_factory()
+
+    tasks = client.debug.list_tasks()
+
+    assert isinstance(tasks, dict)
+
+
+@pytest.mark.asyncio
+async def test_list_tasks_async(async_client_factory: AsyncClientFactory) -> None:
+    client = await async_client_factory()
+
+    tasks = await client.debug.list_tasks()
+
+    assert isinstance(tasks, dict)

mock_tests/test_debug.py

@@ -0,0 +1,56 @@
+import weaviate
+from weaviate.classes.debug import DistributedTask
+from pytest_httpserver import HTTPServer
+
+
+def test_list_tasks(weaviate_client: weaviate.WeaviateClient, weaviate_mock: HTTPServer) -> None:
+    weaviate_mock.expect_request("/v1/tasks").respond_with_json(
+        {
+            "reindex": [
+                {
+                    "id": "task-1",
+                    "version": 1,
+                    "status": "running",
+                    "startedAt": "2026-01-01T00:00:00Z",
+                    "finishedNodes": ["node1"],
+                    "payload": {"collection": "MyCollection"},
+                },
+                {
+                    "id": "task-2",
+                    "version": 1,
+                    "status": "finished",
+                    "startedAt": "2026-01-01T00:00:00Z",
+                    "finishedAt": "2026-01-01T00:05:00Z",
+                    "finishedNodes": ["node1", "node2"],
+                },
+            ]
+        }
+    )
+
+    tasks = weaviate_client.debug.list_tasks()
+
+    assert list(tasks.keys()) == ["reindex"]
+    assert len(tasks["reindex"]) == 2
+
+    first = tasks["reindex"][0]
+    assert isinstance(first, DistributedTask)
+    assert first.id == "task-1"
+    assert first.status == "running"
+    assert first.finished_at is None
+    assert first.finished_nodes == ["node1"]
+    assert first.payload == {"collection": "MyCollection"}
+
+    second = tasks["reindex"][1]
+    assert second.id == "task-2"
+    assert second.status == "finished"
+    assert second.finished_nodes == ["node1", "node2"]
+
+
+def test_list_tasks_empty(
+    weaviate_client: weaviate.WeaviateClient, weaviate_mock: HTTPServer
+) -> None:
+    weaviate_mock.expect_request("/v1/tasks").respond_with_json({})
+
+    tasks = weaviate_client.debug.list_tasks()
+
+    assert tasks == {}

weaviate/classes/debug.py

@@ -1,5 +1,7 @@
-from weaviate.debug.types import DebugRESTObject
+from weaviate.debug.types import DebugRESTObject, DistributedTask, DistributedTaskUnit
 
 __all__ = [
     "DebugRESTObject",
+    "DistributedTask",
+    "DistributedTaskUnit",
 ]

weaviate/debug/async_.pyi

@@ -1,8 +1,8 @@
-from typing import Optional
+from typing import Dict, List, Optional
 
 from weaviate.classes.config import ConsistencyLevel
 from weaviate.connect.v4 import ConnectionAsync
-from weaviate.debug.types import DebugRESTObject
+from weaviate.debug.types import DebugRESTObject, DistributedTask
 from weaviate.types import UUID
 
 from .executor import _DebugExecutor
@@ -17,3 +17,4 @@ class _DebugAsync(_DebugExecutor[ConnectionAsync]):
         node_name: Optional[str] = None,
         tenant: Optional[str] = None,
     ) -> Optional[DebugRESTObject]: ...
+    async def list_tasks(self) -> Dict[str, List[DistributedTask]]: ...

weaviate/debug/executor.py

@@ -1,11 +1,11 @@
-from typing import Dict, Generic, Optional
+from typing import Dict, Generic, List, Optional
 
 from httpx import Response
 
 from weaviate.classes.config import ConsistencyLevel
 from weaviate.connect import executor
 from weaviate.connect.v4 import ConnectionType, _ExpectedStatusCodes
-from weaviate.debug.types import DebugRESTObject
+from weaviate.debug.types import DebugRESTObject, DistributedTask
 from weaviate.types import UUID
 
 
@@ -50,3 +50,24 @@ def resp(response: Response) -> Optional[DebugRESTObject]:
             error_msg="Object was not retrieved",
             status_codes=_ExpectedStatusCodes(ok_in=[200, 404], error="get object"),
         )
+
+    def list_tasks(self) -> executor.Result[Dict[str, List[DistributedTask]]]:
+        """Use the REST API endpoint /tasks to list all distributed tasks currently active or available in the cluster.
+
+        Distributed tasks are long-running background operations, such as reindexing, that are tracked
+        across the cluster's nodes. The returned mapping is keyed by task namespace.
+        """
+
+        def resp(response: Response) -> Dict[str, List[DistributedTask]]:
+            return {
+                namespace: [DistributedTask(**task) for task in tasks]
+                for namespace, tasks in response.json().items()
+            }
+
+        return executor.execute(
+            response_callback=resp,
+            method=self._connection.get,
+            path="/tasks",
+            error_msg="Distributed tasks were not retrieved",
+            status_codes=_ExpectedStatusCodes(ok_in=200, error="list distributed tasks"),
+        )

weaviate/debug/sync.pyi

@@ -1,8 +1,8 @@
-from typing import Optional
+from typing import Dict, List, Optional
 
 from weaviate.classes.config import ConsistencyLevel
 from weaviate.connect.v4 import ConnectionSync
-from weaviate.debug.types import DebugRESTObject
+from weaviate.debug.types import DebugRESTObject, DistributedTask
 from weaviate.types import UUID
 
 from .executor import _DebugExecutor
@@ -17,3 +17,4 @@ class _Debug(_DebugExecutor[ConnectionSync]):
         node_name: Optional[str] = None,
         tenant: Optional[str] = None,
     ) -> Optional[DebugRESTObject]: ...
+    def list_tasks(self) -> Dict[str, List[DistributedTask]]: ...

weaviate/debug/types.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
 from pydantic import BaseModel, Field
 
@@ -15,3 +15,29 @@ class DebugRESTObject(BaseModel):
     uuid: uuid_package.UUID = Field(..., alias="id")
     vector: Optional[list[float]] = Field(None)
     vectors: Optional[Dict[str, list[float]]] = Field(None)
+
+
+class DistributedTaskUnit(BaseModel):
+    """A unit of a distributed task."""
+
+    id: str = Field(...)  # noqa: A003
+    node_id: str = Field(..., alias="nodeId")
+    status: str = Field(...)
+    progress: Optional[float] = Field(None)
+    error: Optional[str] = Field(None)
+    updated_at: Optional[datetime] = Field(None, alias="updatedAt")
+    finished_at: Optional[datetime] = Field(None, alias="finishedAt")
+
+
+class DistributedTask(BaseModel):
+    """Metadata about a distributed task running in the cluster."""
+
+    id: str = Field(...)  # noqa: A003
+    version: int = Field(...)
+    status: str = Field(...)
+    started_at: datetime = Field(..., alias="startedAt")
+    finished_at: Optional[datetime] = Field(None, alias="finishedAt")
+    finished_nodes: List[str] = Field(default_factory=list, alias="finishedNodes")
+    error: Optional[str] = Field(None)
+    payload: Optional[Dict[str, Any]] = Field(None)
+    units: Optional[List[DistributedTaskUnit]] = Field(None)