Expose workflow controls in Python sync client

Add sync facade and workflow-handle helpers for run visibility, history export, repair, and archive operations so Python callers can use the same control-plane methods through async handles and the blocking client.

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -523,6 +523,29 @@ async def describe(self) -> WorkflowExecution:
         """Return the server's current view of this workflow. See :meth:`Client.describe_workflow`."""
         return await self._client.describe_workflow(self.workflow_id)
 
+    async def get_history(self) -> Any:
+        """Fetch this run's durable history. See :meth:`Client.get_history`."""
+        if self.run_id is None:
+            raise ValueError("run_id is required to fetch workflow history from a handle")
+        return await self._client.get_history(self.workflow_id, self.run_id)
+
+    async def export_history(self) -> Any:
+        """Export this run's history as a replay bundle. See :meth:`Client.export_history`."""
+        if self.run_id is None:
+            raise ValueError("run_id is required to export workflow history from a handle")
+        return await self._client.export_history(self.workflow_id, self.run_id)
+
+    async def list_runs(self) -> WorkflowRunList:
+        """List all runs in this workflow execution chain. See :meth:`Client.list_workflow_runs`."""
+        return await self._client.list_workflow_runs(self.workflow_id)
+
+    async def describe_run(self, run_id: str | None = None) -> WorkflowRun:
+        """Return one run's detailed status. See :meth:`Client.describe_workflow_run`."""
+        selected_run_id = run_id or self.run_id
+        if selected_run_id is None:
+            raise ValueError("run_id is required to describe a workflow run from a handle")
+        return await self._client.describe_workflow_run(self.workflow_id, selected_run_id)
+
     async def signal(self, signal_name: str, args: list[Any] | None = None) -> None:
         """Deliver an external signal to this workflow. See :meth:`Client.signal_workflow`."""
         await self._client.signal_workflow(self.workflow_id, signal_name, args=args)
@@ -539,6 +562,14 @@ async def terminate(self, *, reason: str | None = None) -> None:
         """Forcefully stop this workflow. See :meth:`Client.terminate_workflow`."""
         await self._client.terminate_workflow(self.workflow_id, reason=reason)
 
+    async def repair(self) -> WorkflowCommandResult:
+        """Ask the server to repair this workflow. See :meth:`Client.repair_workflow`."""
+        return await self._client.repair_workflow(self.workflow_id)
+
+    async def archive(self, *, reason: str | None = None) -> WorkflowCommandResult:
+        """Move this terminal workflow into the archive tier. See :meth:`Client.archive_workflow`."""
+        return await self._client.archive_workflow(self.workflow_id, reason=reason)
+
     async def update(
         self,
         update_name: str,

src/durable_workflow/sync.py

@@ -16,9 +16,12 @@
     ScheduleTriggerResult,
     TaskQueueDescription,
     TaskQueueList,
+    WorkflowCommandResult,
     WorkflowExecution,
     WorkflowHandle,
     WorkflowList,
+    WorkflowRun,
+    WorkflowRunList,
 )
 from .metrics import MetricsRecorder
 from .retry_policy import TransportRetryPolicy
@@ -54,6 +57,20 @@ def describe(self) -> WorkflowExecution:
         result: WorkflowExecution = _run(self._handle.describe())
         return result
 
+    def get_history(self) -> Any:
+        return _run(self._handle.get_history())
+
+    def export_history(self) -> Any:
+        return _run(self._handle.export_history())
+
+    def list_runs(self) -> WorkflowRunList:
+        result: WorkflowRunList = _run(self._handle.list_runs())
+        return result
+
+    def describe_run(self, run_id: str | None = None) -> WorkflowRun:
+        result: WorkflowRun = _run(self._handle.describe_run(run_id))
+        return result
+
     def signal(self, signal_name: str, args: list[Any] | None = None) -> None:
         _run(self._handle.signal(signal_name, args=args))
 
@@ -66,6 +83,14 @@ def cancel(self, *, reason: str | None = None) -> None:
     def terminate(self, *, reason: str | None = None) -> None:
         _run(self._handle.terminate(reason=reason))
 
+    def repair(self) -> WorkflowCommandResult:
+        result: WorkflowCommandResult = _run(self._handle.repair())
+        return result
+
+    def archive(self, *, reason: str | None = None) -> WorkflowCommandResult:
+        result: WorkflowCommandResult = _run(self._handle.archive(reason=reason))
+        return result
+
     def update(
         self,
         update_name: str,
@@ -262,6 +287,14 @@ def get_history(self, workflow_id: str, run_id: str) -> Any:
     def export_history(self, workflow_id: str, run_id: str) -> Any:
         return _run(self._async.export_history(workflow_id, run_id))
 
+    def list_workflow_runs(self, workflow_id: str) -> WorkflowRunList:
+        result: WorkflowRunList = _run(self._async.list_workflow_runs(workflow_id))
+        return result
+
+    def describe_workflow_run(self, workflow_id: str, run_id: str) -> WorkflowRun:
+        result: WorkflowRun = _run(self._async.describe_workflow_run(workflow_id, run_id))
+        return result
+
     def signal_workflow(self, workflow_id: str, signal_name: str, *, args: list[Any] | None = None) -> None:
         _run(self._async.signal_workflow(workflow_id, signal_name, args=args))
 
@@ -274,6 +307,14 @@ def cancel_workflow(self, workflow_id: str, *, reason: str | None = None) -> Non
     def terminate_workflow(self, workflow_id: str, *, reason: str | None = None) -> None:
         _run(self._async.terminate_workflow(workflow_id, reason=reason))
 
+    def repair_workflow(self, workflow_id: str) -> WorkflowCommandResult:
+        result: WorkflowCommandResult = _run(self._async.repair_workflow(workflow_id))
+        return result
+
+    def archive_workflow(self, workflow_id: str, *, reason: str | None = None) -> WorkflowCommandResult:
+        result: WorkflowCommandResult = _run(self._async.archive_workflow(workflow_id, reason=reason))
+        return result
+
     def update_workflow(
         self,
         workflow_id: str,

tests/test_client.py

@@ -426,6 +426,57 @@ async def test_describe_run_request_matches_polyglot_fixture(self, client: Clien
         assert run.actions == response["actions"]
 
 
+class TestWorkflowHandleControlPlane:
+    @pytest.mark.asyncio
+    async def test_run_visibility_delegates_to_client(self, client: Client) -> None:
+        handle = WorkflowHandle(client, workflow_id="wf-1", run_id="r1", workflow_type="greeter")
+        client.get_history = AsyncMock(return_value={"events": []})
+        client.export_history = AsyncMock(return_value={"schema": "durable.workflow.history.v2"})
+        client.list_workflow_runs = AsyncMock(return_value="runs")
+        client.describe_workflow_run = AsyncMock(return_value="run")
+
+        assert await handle.get_history() == {"events": []}
+        assert await handle.export_history() == {"schema": "durable.workflow.history.v2"}
+        assert await handle.list_runs() == "runs"
+        assert await handle.describe_run() == "run"
+
+        client.get_history.assert_awaited_once_with("wf-1", "r1")
+        client.export_history.assert_awaited_once_with("wf-1", "r1")
+        client.list_workflow_runs.assert_awaited_once_with("wf-1")
+        client.describe_workflow_run.assert_awaited_once_with("wf-1", "r1")
+
+    @pytest.mark.asyncio
+    async def test_run_specific_methods_require_run_id(self, client: Client) -> None:
+        handle = WorkflowHandle(client, workflow_id="wf-1")
+
+        with pytest.raises(ValueError, match="run_id is required"):
+            await handle.get_history()
+        with pytest.raises(ValueError, match="run_id is required"):
+            await handle.export_history()
+        with pytest.raises(ValueError, match="run_id is required"):
+            await handle.describe_run()
+
+    @pytest.mark.asyncio
+    async def test_describe_run_accepts_explicit_run_id(self, client: Client) -> None:
+        handle = WorkflowHandle(client, workflow_id="wf-1")
+        client.describe_workflow_run = AsyncMock(return_value="run")
+
+        assert await handle.describe_run("r2") == "run"
+        client.describe_workflow_run.assert_awaited_once_with("wf-1", "r2")
+
+    @pytest.mark.asyncio
+    async def test_maintenance_delegates_to_client(self, client: Client) -> None:
+        handle = WorkflowHandle(client, workflow_id="wf-1", run_id="r1", workflow_type="greeter")
+        client.repair_workflow = AsyncMock(return_value="repair")
+        client.archive_workflow = AsyncMock(return_value="archive")
+
+        assert await handle.repair() == "repair"
+        assert await handle.archive(reason="retention") == "archive"
+
+        client.repair_workflow.assert_awaited_once_with("wf-1")
+        client.archive_workflow.assert_awaited_once_with("wf-1", reason="retention")
+
+
 class TestSignalWorkflow:
     @pytest.mark.asyncio
     async def test_signal(self, client: Client) -> None:

tests/test_sync.py

@@ -7,6 +7,7 @@
 import httpx
 import pytest
 
+from durable_workflow.client import WorkflowHandle
 from durable_workflow.sync import Client, SyncWorkflowHandle
 
 
@@ -161,6 +162,101 @@ def test_describe_task_queue(self) -> None:
             assert mock.call_args.args[:2] == ("GET", "/api/task-queues/orders%2Fhigh%20priority")
 
 
+class TestSyncClientRunVisibility:
+    def test_list_workflow_runs(self) -> None:
+        client = Client("http://localhost:8080")
+        resp = _mock_response(
+            200,
+            {
+                "workflow_id": "wf-1",
+                "run_count": 2,
+                "runs": [
+                    {"workflow_id": "wf-1", "run_id": "r1", "workflow_type": "greeter", "status": "completed"},
+                    {
+                        "workflow_id": "wf-1",
+                        "run_id": "r2",
+                        "workflow_type": "greeter",
+                        "status": "running",
+                        "is_current_run": True,
+                    },
+                ],
+            },
+        )
+        with patch.object(client._async._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = client.list_workflow_runs("wf-1")
+
+        assert result.workflow_id == "wf-1"
+        assert result.run_count == 2
+        assert [run.run_id for run in result.runs] == ["r1", "r2"]
+        assert result.runs[1].is_current_run is True
+        assert mock.call_args.args[:2] == ("GET", "/api/workflows/wf-1/runs")
+
+    def test_describe_workflow_run(self) -> None:
+        client = Client("http://localhost:8080")
+        resp = _mock_response(
+            200,
+            {
+                "workflow_id": "wf-1",
+                "run_id": "r1",
+                "workflow_type": "greeter",
+                "status": "completed",
+                "status_bucket": "terminal",
+                "actions": {"archive": {"enabled": True}},
+            },
+        )
+        with patch.object(client._async._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            run = client.describe_workflow_run("wf-1", "r1")
+
+        assert run.workflow_id == "wf-1"
+        assert run.run_id == "r1"
+        assert run.status_bucket == "terminal"
+        assert run.actions == {"archive": {"enabled": True}}
+        assert mock.call_args.args[:2] == ("GET", "/api/workflows/wf-1/runs/r1")
+
+
+class TestSyncClientMaintenance:
+    def test_repair_workflow(self) -> None:
+        client = Client("http://localhost:8080")
+        resp = _mock_response(
+            200,
+            {
+                "workflow_id": "wf-1",
+                "outcome": "accepted",
+                "command_status": "queued",
+                "command_id": "cmd-1",
+            },
+        )
+        with patch.object(client._async._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = client.repair_workflow("wf-1")
+
+        assert result.workflow_id == "wf-1"
+        assert result.outcome == "accepted"
+        assert result.command_status == "queued"
+        assert result.command_id == "cmd-1"
+        assert mock.call_args.args[:2] == ("POST", "/api/workflows/wf-1/repair")
+
+    def test_archive_workflow(self) -> None:
+        client = Client("http://localhost:8080")
+        resp = _mock_response(
+            200,
+            {
+                "workflow_id": "wf-1",
+                "outcome": "accepted",
+                "command_status": "completed",
+                "command_id": "cmd-2",
+            },
+        )
+        with patch.object(client._async._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            result = client.archive_workflow("wf-1", reason="retention policy")
+
+        assert result.workflow_id == "wf-1"
+        assert result.outcome == "accepted"
+        assert result.command_status == "completed"
+        assert result.command_id == "cmd-2"
+        assert mock.call_args.args[:2] == ("POST", "/api/workflows/wf-1/archive")
+        assert mock.call_args.kwargs["json"] == {"reason": "retention policy"}
+
+
 class TestSyncClientUpdate:
     def test_update(self) -> None:
         client = Client("http://localhost:8080")
@@ -187,6 +283,26 @@ def test_handle_update(self) -> None:
             assert result["outcome"] == "completed"
 
 
+class TestSyncWorkflowHandleControlPlane:
+    def test_run_visibility_and_maintenance_delegate_to_async_handle(self) -> None:
+        async_handle = WorkflowHandle(AsyncMock(), workflow_id="wf-1", run_id="r1", workflow_type="greeter")
+        async_handle.get_history = AsyncMock(return_value={"events": []})  # type: ignore[method-assign]
+        async_handle.export_history = AsyncMock(return_value={"schema": "durable.workflow.history.v2"})  # type: ignore[method-assign]
+        async_handle.list_runs = AsyncMock(return_value=[])  # type: ignore[method-assign]
+        async_handle.describe_run = AsyncMock(return_value={"run_id": "r1"})  # type: ignore[method-assign]
+        async_handle.repair = AsyncMock(return_value={"outcome": "accepted"})  # type: ignore[method-assign]
+        async_handle.archive = AsyncMock(return_value={"outcome": "completed"})  # type: ignore[method-assign]
+        handle = SyncWorkflowHandle(async_handle)
+
+        assert handle.get_history() == {"events": []}
+        assert handle.export_history() == {"schema": "durable.workflow.history.v2"}
+        assert handle.list_runs() == []
+        assert handle.describe_run() == {"run_id": "r1"}
+        assert handle.repair() == {"outcome": "accepted"}
+        assert handle.archive(reason="retention") == {"outcome": "completed"}
+        async_handle.archive.assert_awaited_once_with(reason="retention")
+
+
 class TestSyncClientContextManager:
     def test_context_manager(self) -> None:
         with Client("http://localhost:8080") as client: