Expose workflow maintenance commands in Python SDK

Author: durable-workflow-ops

src/durable_workflow/__init__.py

@@ -31,6 +31,7 @@
     TaskQueueList,
     TaskQueueQueryAdmission,
     TaskQueueTaskAdmission,
+    WorkflowCommandResult,
     WorkflowExecution,
     WorkflowHandle,
     WorkflowList,
@@ -167,6 +168,7 @@
     "TaskQueueTaskAdmission",
     "Worker",
     "WorkerInterceptor",
+    "WorkflowCommandResult",
     "WorkflowExecution",
     "WorkflowTaskHandler",
     "WorkflowTaskInterceptorContext",

src/durable_workflow/client.py

@@ -176,6 +176,27 @@ def from_dict(
         )
 
 
+@dataclass
+class WorkflowCommandResult:
+    """Machine-readable outcome returned by workflow control commands."""
+
+    workflow_id: str
+    outcome: str
+    command_status: str | None = None
+    command_id: str | None = None
+    raw: dict[str, Any] | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any], *, workflow_id: str | None = None) -> WorkflowCommandResult:
+        return cls(
+            workflow_id=data.get("workflow_id", workflow_id or ""),
+            outcome=data.get("outcome", ""),
+            command_status=data.get("command_status"),
+            command_id=data.get("command_id"),
+            raw=data,
+        )
+
+
 @dataclass
 class WorkflowRunList:
     """All known durable runs for one workflow execution, oldest first."""
@@ -1213,6 +1234,19 @@ async def terminate_workflow(self, workflow_id: str, *, reason: str | None = Non
             body["reason"] = reason
         await self._request("POST", f"/workflows/{workflow_id}/terminate", json=body, context=workflow_id)
 
+    async def repair_workflow(self, workflow_id: str) -> WorkflowCommandResult:
+        """Ask the server to repair a stalled workflow, returning the command outcome."""
+        data = await self._request("POST", f"/workflows/{workflow_id}/repair", json={}, context=workflow_id)
+        return WorkflowCommandResult.from_dict(data, workflow_id=workflow_id)
+
+    async def archive_workflow(self, workflow_id: str, *, reason: str | None = None) -> WorkflowCommandResult:
+        """Move a terminal workflow into the archive tier, returning the command outcome."""
+        body: dict[str, Any] = {}
+        if reason is not None:
+            body["reason"] = reason
+        data = await self._request("POST", f"/workflows/{workflow_id}/archive", json=body, context=workflow_id)
+        return WorkflowCommandResult.from_dict(data, workflow_id=workflow_id)
+
     async def update_workflow(
         self,
         workflow_id: str,

tests/fixtures/control-plane/workflow-archive-parity.json

@@ -0,0 +1,39 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "workflow.archive",
+  "request": {
+    "method": "POST",
+    "path": "/workflows/wf-polyglot-231/archive"
+  },
+  "semantic_body": {
+    "workflow_id": "wf-polyglot-231",
+    "reason": "retention policy cleanup"
+  },
+  "response_body": {
+    "workflow_id": "wf-polyglot-231",
+    "outcome": "archived",
+    "command_status": "accepted",
+    "command_id": "cmd-polyglot-archive-231"
+  },
+  "cli": {
+    "argv": {
+      "workflow-id": "wf-polyglot-231",
+      "--reason": "retention policy cleanup",
+      "--json": true
+    },
+    "expected_body": {
+      "reason": "retention policy cleanup"
+    }
+  },
+  "sdk_python": {
+    "method": "archive_workflow",
+    "args": {
+      "workflow_id": "wf-polyglot-231",
+      "reason": "retention policy cleanup"
+    },
+    "expected_body": {
+      "reason": "retention policy cleanup"
+    }
+  }
+}

tests/fixtures/control-plane/workflow-repair-parity.json

@@ -0,0 +1,32 @@
+{
+  "schema": "durable-workflow.polyglot.control-plane-request-fixture",
+  "version": 1,
+  "operation": "workflow.repair",
+  "request": {
+    "method": "POST",
+    "path": "/workflows/wf-polyglot-231/repair"
+  },
+  "semantic_body": {
+    "workflow_id": "wf-polyglot-231"
+  },
+  "response_body": {
+    "workflow_id": "wf-polyglot-231",
+    "outcome": "repair_requested",
+    "command_status": "accepted",
+    "command_id": "cmd-polyglot-repair-231"
+  },
+  "cli": {
+    "argv": {
+      "workflow-id": "wf-polyglot-231",
+      "--json": true
+    },
+    "expected_body": {}
+  },
+  "sdk_python": {
+    "method": "repair_workflow",
+    "args": {
+      "workflow_id": "wf-polyglot-231"
+    },
+    "expected_body": {}
+  }
+}

tests/test_client.py

@@ -642,6 +642,52 @@ async def test_terminate_request_matches_polyglot_fixture(self, client: Client)
         assert sdk["args"]["workflow_id"] == fixture["semantic_body"]["workflow_id"]
 
 
+class TestWorkflowMaintenanceCommands:
+    @pytest.mark.asyncio
+    async def test_repair_request_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "workflow-repair-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.repair_workflow(**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") or call_args[1].get("json")) == sdk["expected_body"]
+
+        assert result.workflow_id == fixture["semantic_body"]["workflow_id"]
+        assert result.outcome == fixture["response_body"]["outcome"]
+        assert result.command_status == fixture["response_body"]["command_status"]
+        assert result.command_id == fixture["response_body"]["command_id"]
+
+    @pytest.mark.asyncio
+    async def test_archive_request_matches_polyglot_fixture(self, client: Client) -> None:
+        fixture_path = Path(__file__).parent / "fixtures" / "control-plane" / "workflow-archive-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.archive_workflow(**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']}"
+        body = call_args.kwargs.get("json") or call_args[1].get("json")
+
+        assert body == sdk["expected_body"]
+        assert body["reason"] == fixture["semantic_body"]["reason"]
+        assert result.workflow_id == fixture["semantic_body"]["workflow_id"]
+        assert result.outcome == fixture["response_body"]["outcome"]
+        assert result.command_status == fixture["response_body"]["command_status"]
+        assert result.command_id == fixture["response_body"]["command_id"]
+
+
 class TestQueryWorkflow:
     @pytest.mark.asyncio
     async def test_query(self, client: Client) -> None: