Expose workflow run visibility in Python SDK

Author: durable-workflow-ops

src/durable_workflow/__init__.py

@@ -34,6 +34,8 @@
     WorkflowExecution,
     WorkflowHandle,
     WorkflowList,
+    WorkflowRun,
+    WorkflowRunList,
 )
 from .errors import (
     ActivityCancelled,
@@ -171,6 +173,8 @@
     "WorkflowHandle",
     "WorkflowList",
     "WorkflowPayloadDecodeError",
+    "WorkflowRun",
+    "WorkflowRunList",
     "activity",
     "parse_auth_composition_contract",
     "sync",

src/durable_workflow/client.py

@@ -105,6 +105,86 @@ class WorkflowList:
     next_page_token: str | None = None
 
 
+@dataclass
+class WorkflowRun:
+    """Current server view of one durable run in a workflow execution chain."""
+
+    workflow_id: str
+    run_id: str
+    workflow_type: str
+    status: str | None = None
+    namespace: str | None = None
+    task_queue: str | None = None
+    run_number: int | None = None
+    run_count: int | None = None
+    is_current_run: bool | None = None
+    status_bucket: str | None = None
+    business_key: str | None = None
+    compatibility: str | None = None
+    payload_codec: str | None = None
+    input: Any = None
+    output: Any = None
+    memo: dict[str, Any] | None = None
+    search_attributes: dict[str, Any] | None = None
+    actions: dict[str, Any] | None = None
+    started_at: str | None = None
+    closed_at: str | None = None
+    last_progress_at: str | None = None
+    closed_reason: str | None = None
+    wait_kind: str | None = None
+    wait_reason: str | None = None
+    execution_timeout_seconds: int | None = None
+    run_timeout_seconds: int | None = None
+    execution_deadline_at: str | None = None
+    run_deadline_at: str | None = None
+
+    @classmethod
+    def from_dict(
+        cls, data: dict[str, Any], *, workflow_id: str | None = None, run_id: str | None = None
+    ) -> WorkflowRun:
+        return cls(
+            workflow_id=data.get("workflow_id", workflow_id or ""),
+            run_id=data.get("run_id", run_id or ""),
+            workflow_type=data.get("workflow_type", ""),
+            status=data.get("status"),
+            namespace=data.get("namespace"),
+            task_queue=data.get("task_queue"),
+            run_number=data.get("run_number"),
+            run_count=data.get("run_count"),
+            is_current_run=data.get("is_current_run"),
+            status_bucket=data.get("status_bucket"),
+            business_key=data.get("business_key"),
+            compatibility=data.get("compatibility"),
+            payload_codec=data.get("payload_codec"),
+            input=data.get("input"),
+            output=data.get("output"),
+            memo=data.get("memo") if isinstance(data.get("memo"), dict) else None,
+            search_attributes=(
+                data.get("search_attributes") if isinstance(data.get("search_attributes"), dict) else None
+            ),
+            actions=data.get("actions") if isinstance(data.get("actions"), dict) else None,
+            started_at=data.get("started_at"),
+            closed_at=data.get("closed_at"),
+            last_progress_at=data.get("last_progress_at"),
+            closed_reason=data.get("closed_reason"),
+            wait_kind=data.get("wait_kind"),
+            wait_reason=data.get("wait_reason"),
+            execution_timeout_seconds=data.get("execution_timeout_seconds"),
+            run_timeout_seconds=data.get("run_timeout_seconds"),
+            execution_deadline_at=data.get("execution_deadline_at"),
+            run_deadline_at=data.get("run_deadline_at"),
+        )
+
+
+@dataclass
+class WorkflowRunList:
+    """All known durable runs for one workflow execution, oldest first."""
+
+    workflow_id: str
+    run_count: int
+    runs: list[WorkflowRun]
+
+
 @dataclass
 class TaskQueueTaskAdmission:
     """Workflow/activity admission state for one task queue."""
@@ -1016,6 +1096,24 @@ async def get_history(self, workflow_id: str, run_id: str) -> Any:
             "GET", f"/workflows/{workflow_id}/runs/{run_id}/history", context=workflow_id
         )
 
+    async def list_workflow_runs(self, workflow_id: str) -> WorkflowRunList:
+        """List all durable runs in one workflow execution chain, oldest first."""
+        data = await self._request("GET", f"/workflows/{workflow_id}/runs", context=workflow_id)
+        runs = [
+            WorkflowRun.from_dict(item, workflow_id=data.get("workflow_id", workflow_id))
+            for item in data.get("runs", [])
+        ]
+        return WorkflowRunList(
+            workflow_id=data.get("workflow_id", workflow_id),
+            run_count=data.get("run_count", len(runs)),
+            runs=runs,
+        )
+
+    async def describe_workflow_run(self, workflow_id: str, run_id: str) -> WorkflowRun:
+        """Return detailed status, payload, and actionability for one specific workflow run."""
+        data = await self._request("GET", f"/workflows/{workflow_id}/runs/{run_id}", context=workflow_id)
+        return WorkflowRun.from_dict(data, workflow_id=workflow_id, run_id=run_id)
+
     async def send_webhook_bridge_event(
         self,
         adapter: str,

tests/fixtures/control-plane/workflow-list-runs-parity.json

@@ -0,0 +1,59 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "workflow.list_runs",
+  "request": {
+    "method": "GET",
+    "path": "/workflows/wf-polyglot-231/runs"
+  },
+  "semantic_body": {
+    "workflow_id": "wf-polyglot-231",
+    "run_count": 2,
+    "run_ids": [
+      "run-polyglot-231-001",
+      "run-polyglot-231-002"
+    ]
+  },
+  "response_body": {
+    "workflow_id": "wf-polyglot-231",
+    "run_count": 2,
+    "runs": [
+      {
+        "workflow_id": "wf-polyglot-231",
+        "run_id": "run-polyglot-231-001",
+        "run_number": 1,
+        "workflow_type": "orders.process",
+        "status": "completed",
+        "status_bucket": "completed",
+        "task_queue": "orders",
+        "started_at": "2026-04-22T05:00:00Z",
+        "closed_at": "2026-04-22T05:04:00Z",
+        "is_current_run": false
+      },
+      {
+        "workflow_id": "wf-polyglot-231",
+        "run_id": "run-polyglot-231-002",
+        "run_number": 2,
+        "workflow_type": "orders.process",
+        "status": "running",
+        "status_bucket": "active",
+        "task_queue": "orders",
+        "started_at": "2026-04-22T05:05:00Z",
+        "closed_at": null,
+        "is_current_run": true
+      }
+    ]
+  },
+  "cli": {
+    "argv": {
+      "workflow-id": "wf-polyglot-231",
+      "--json": true
+    }
+  },
+  "sdk_python": {
+    "method": "list_workflow_runs",
+    "args": {
+      "workflow_id": "wf-polyglot-231"
+    }
+  }
+}

tests/fixtures/control-plane/workflow-show-run-parity.json

@@ -0,0 +1,73 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "workflow.show_run",
+  "request": {
+    "method": "GET",
+    "path": "/workflows/wf-polyglot-231/runs/run-polyglot-231-002"
+  },
+  "semantic_body": {
+    "workflow_id": "wf-polyglot-231",
+    "run_id": "run-polyglot-231-002"
+  },
+  "response_body": {
+    "workflow_id": "wf-polyglot-231",
+    "run_id": "run-polyglot-231-002",
+    "workflow_type": "orders.process",
+    "namespace": "orders-prod",
+    "business_key": "order-42",
+    "status": "running",
+    "status_bucket": "active",
+    "run_number": 2,
+    "run_count": 2,
+    "is_current_run": true,
+    "task_queue": "orders",
+    "compatibility": "build-2026-04-22",
+    "payload_codec": "avro",
+    "execution_timeout_seconds": 86400,
+    "run_timeout_seconds": 3600,
+    "execution_deadline_at": "2026-04-23T05:00:00Z",
+    "run_deadline_at": "2026-04-22T06:05:00Z",
+    "started_at": "2026-04-22T05:05:00Z",
+    "closed_at": null,
+    "last_progress_at": "2026-04-22T05:06:00Z",
+    "closed_reason": null,
+    "wait_kind": "timer",
+    "wait_reason": "shipment deadline",
+    "input": [
+      {
+        "order_id": 42,
+        "priority": "gold"
+      }
+    ],
+    "memo": {
+      "source": "polyglot-fixture"
+    },
+    "search_attributes": {
+      "CustomerId": "cust-42",
+      "Tier": "gold"
+    },
+    "actions": {
+      "can_query": true,
+      "can_signal": true,
+      "can_cancel": true,
+      "can_terminate": true,
+      "can_repair": false,
+      "can_archive": false
+    }
+  },
+  "cli": {
+    "argv": {
+      "workflow-id": "wf-polyglot-231",
+      "run-id": "run-polyglot-231-002",
+      "--json": true
+    }
+  },
+  "sdk_python": {
+    "method": "describe_workflow_run",
+    "args": {
+      "workflow_id": "wf-polyglot-231",
+      "run_id": "run-polyglot-231-002"
+    }
+  }
+}

tests/test_client.py

@@ -345,6 +345,62 @@ async def test_history_request_matches_polyglot_fixture(self, client: Client) ->
         assert history == fixture["response_body"]
 
 
+class TestWorkflowRunVisibility:
+    @pytest.mark.asyncio
+    async def test_list_runs_request_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "workflow-list-runs-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+        sdk = fixture["sdk_python"]
+
+        resp = _mock_response(200, fixture["response_body"])
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = await client.list_workflow_runs(**sdk["args"])
+
+        call_args = mock.call_args
+        assert call_args.args[0] == fixture["request"]["method"]
+        assert call_args.args[1] == f"/api{fixture['request']['path']}"
+        assert call_args.kwargs.get("json") is None
+
+        semantic = fixture["semantic_body"]
+        assert sdk["args"]["workflow_id"] == semantic["workflow_id"]
+        assert result.workflow_id == semantic["workflow_id"]
+        assert result.run_count == semantic["run_count"]
+        assert [run.run_id for run in result.runs] == semantic["run_ids"]
+        assert result.runs[0].workflow_type == fixture["response_body"]["runs"][0]["workflow_type"]
+        assert result.runs[1].is_current_run is True
+
+    @pytest.mark.asyncio
+    async def test_describe_run_request_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "workflow-show-run-parity.json"
+        fixture = json.loads(fixture_path.read_text())
+        sdk = fixture["sdk_python"]
+
+        resp = _mock_response(200, fixture["response_body"])
+
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            run = await client.describe_workflow_run(**sdk["args"])
+
+        call_args = mock.call_args
+        assert call_args.args[0] == fixture["request"]["method"]
+        assert call_args.args[1] == f"/api{fixture['request']['path']}"
+        assert call_args.kwargs.get("json") is None
+
+        semantic = fixture["semantic_body"]
+        response = fixture["response_body"]
+        assert sdk["args"]["workflow_id"] == semantic["workflow_id"]
+        assert sdk["args"]["run_id"] == semantic["run_id"]
+        assert run.workflow_id == semantic["workflow_id"]
+        assert run.run_id == semantic["run_id"]
+        assert run.workflow_type == response["workflow_type"]
+        assert run.status == response["status"]
+        assert run.status_bucket == response["status_bucket"]
+        assert run.business_key == response["business_key"]
+        assert run.memo == response["memo"]
+        assert run.search_attributes == response["search_attributes"]
+        assert run.actions == response["actions"]
+
+
 class TestSignalWorkflow:
     @pytest.mark.asyncio
     async def test_signal(self, client: Client) -> None: