lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add workflow run resume cli command

Browse Source
This commit is contained in:
lda
2026-06-03 18:23:45 +07:00 Verified
parent c5a206d783
commit 76d673e0c4
5 changed files with 285 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_api/surface.py
+9
View File
@@ -165,6 +165,15 @@ class WorkflowRunSurface(Protocol):
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]: ...
async def resume_run(
self,
*,
run_id: str,
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]: ...
async def inspect_run(
self,
*,
src/wf_cli/commands/runs.py
+52
View File
@@ -67,6 +67,58 @@ def inspect_run(
emit_json(asyncio.run(context.handlers.inspect_run(run_id=run_id)))
@app.command("resume")
def resume_run(
ctx: typer.Context,
run_id: Annotated[str, typer.Argument(help="Interrupted durable run id.")],
payload_json: Annotated[
str | None,
typer.Option("--payload", help="Resume payload JSON object."),
] = None,
payload_file: Annotated[
Path | None,
typer.Option("--payload-file", help="Path to resume payload JSON object."),
] = None,
outcome: Annotated[
str,
typer.Option("--outcome", help="Interrupt resume outcome."),
] = "submitted",
trace_from: Annotated[
int | None,
typer.Option("--trace-from", min=0, help="Optional trace slice start."),
] = None,
trace_limit: Annotated[
int | None,
typer.Option(
"--trace-limit", min=1, max=100, help="Optional trace slice limit."
),
] = None,
) -> None:
"""Resume an interrupted durable run.
The target store owns the paused run. With `--local`, this is the local file
store; with `--url`, this is the long-lived JSON-RPC server's store.
"""
try:
resume_payload = parse_json_input(
input_json=payload_json,
input_file=payload_file,
)
except CliInputError as exc:
raise typer.BadParameter(str(exc)) from exc
context = load_cli_context_from_typer(ctx)
trace_range = _optional_trace_range(start=trace_from, limit=trace_limit)
payload = asyncio.run(
context.handlers.resume_run(
run_id=run_id,
resume_payload=resume_payload,
resume_outcome=outcome,
trace_range=trace_range,
)
)
emit_json(payload)
@app.command("trace")
def trace_run(
ctx: typer.Context,
src/wf_transport_rpc_http/client_runs.py
+18
View File
@@ -26,6 +26,24 @@ class RpcRunClientMixin:
},
)
async def resume_run(
self,
*,
run_id: str,
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
return await self._call(
"workflow.runs.resume",
{
"run_id": run_id,
"resume_payload": resume_payload,
"resume_outcome": resume_outcome,
"trace_range": _trace_range_payload(trace_range),
},
)
async def inspect_run(self, *, run_id: str) -> dict[str, Any]:
return await self._call("workflow.runs.inspect", {"run_id": run_id})
tests/wf_cli/test_remote_target.py
+104
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import asyncio
import json
import httpx
@@ -219,6 +220,42 @@ def _constant_plan() -> RawWorkflowPlan:
)
def _interrupt_plan() -> RawWorkflowPlan:
return RawWorkflowPlan.model_validate(
{
"name": "remote_approval",
"input_schema": {
"type": "object",
"properties": {"message": {"type": "string"}},
"required": ["message"],
},
"state_schema": {"fields": {}},
"output_schema": {"type": "object", "properties": {}},
"outcomes": ["submitted"],
"start": "approval",
"nodes": [
{
"id": "approval",
"type": "interrupt",
"kind": "approval",
"request": [
{
"path": {"root": "input", "parts": ["message"]},
"target": {"root": "local", "parts": ["message"]},
}
],
"resume": [],
"outcomes": ["submitted"],
},
{"id": "end_submitted", "type": "end", "outcome": "submitted"},
],
"edges": [
{"from": "approval", "outcome": "submitted", "to": "end_submitted"}
],
}
)
def test_wf_cap_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
rpc_app = create_rpc_app(server)
@@ -341,3 +378,70 @@ def test_wf_remote_draft_artifact_deploy_lifecycle(monkeypatch, tmp_path) -> Non
)
assert validated_deployment.exit_code == 0, validated_deployment.output
assert '"status": "runnable"' in validated_deployment.output
def test_wf_remote_run_resume_interrupted_deployment(monkeypatch, tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
asyncio.run(
server.api.create_artifact_from_plan(
artifact_id="remote_approval",
version=1,
title="Remote Approval",
plan=_interrupt_plan(),
outcomes=("submitted",),
)
)
asyncio.run(
server.api.save_deployment(
{
"id": "remote_approval.default",
"artifact_id": "remote_approval",
"artifact_version": 1,
"bindings": [],
}
)
)
original_client = httpx.AsyncClient
monkeypatch.setattr(
"wf_transport_rpc_http.client.httpx.AsyncClient",
lambda *args, **kwargs: original_client(
transport=httpx.ASGITransport(app=create_rpc_app(server)),
base_url="http://test",
),
)
config_path = tmp_path / "wf.json"
config_path.write_text('{"version": 1}', encoding="utf-8")
runner = CliRunner()
base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
started = runner.invoke(
app,
[
*base_args,
"run",
"start",
"remote_approval.default",
"--input",
'{"message": "approve?"}',
],
)
assert started.exit_code == 0, started.output
started_payload = json.loads(started.output)
assert started_payload["status"] == "interrupted"
resumed = runner.invoke(
app,
[
*base_args,
"run",
"resume",
started_payload["run_id"],
"--payload",
"{}",
],
)
assert resumed.exit_code == 0, resumed.output
resumed_payload = json.loads(resumed.output)
assert resumed_payload["run_id"] == started_payload["run_id"]
assert resumed_payload["status"] == "completed"
assert resumed_payload["outcome"] == "submitted"
tests/wf_cli/test_run_deploy.py
+102 -2
View File
@@ -6,13 +6,13 @@ from unittest.mock import patch
from typer.testing import CliRunner
from wf_artifacts import FileWorkflowArtifactStore, WorkflowDeployment
from wf_artifacts import FileWorkflowArtifactStore, WorkflowArtifact, WorkflowDeployment
from wf_cli.app import app
from typer import Context as TyperContext
from wf_cli.context import CliContext, config_path_from_context, load_cli_context
from tests.wf_mcp.test_support import echo_tool, local_temp_root
from tests.wf_mcp.test_support import echo_tool, input_binding, local_temp_root
from tests.wf_mcp.workflow_surface.conftest import echo_artifact
@@ -74,6 +74,22 @@ def _seed_echo_deployment(root: Path) -> Path:
return config_path
def _seed_interrupt_deployment(root: Path) -> Path:
config_path = _write_config(root)
store_root = root / ".wf_mcp_store"
artifact_store = FileWorkflowArtifactStore(store_root)
artifact_store.save_artifact(_interrupt_artifact())
artifact_store.save_deployment(
WorkflowDeployment(
id="approval.personal",
artifact_id="approval",
artifact_version=1,
bindings=[],
)
)
return config_path
def test_wf_deploy_validate_outputs_json() -> None:
root = local_temp_root() / "wf_cli_deploy_validate"
root.mkdir(parents=True, exist_ok=True)
@@ -209,6 +225,49 @@ def test_wf_run_inspect_and_trace_existing_run() -> None:
assert traced_payload["trace"][0]["node_id"] == "echo"
def test_wf_run_resume_interrupted_run() -> None:
root = local_temp_root() / "wf_cli_run_resume"
root.mkdir(parents=True, exist_ok=True)
config_path = _seed_interrupt_deployment(root)
with patch(
"wf_cli.commands.runs.load_cli_context_from_typer", _load_cli_context_with_specs
):
start = runner.invoke(
app,
[
"--config",
str(config_path),
"run",
"start",
"approval.personal",
"--input",
'{"message": "send?"}',
],
)
run_id = json.loads(start.output)["run_id"]
resumed = runner.invoke(
app,
[
"--config",
str(config_path),
"run",
"resume",
run_id,
"--payload",
"{}",
],
)
assert resumed.exit_code == 0, resumed.output
payload = json.loads(resumed.output)
assert payload["run_id"] == run_id
assert payload["status"] == "completed"
assert payload["outcome"] == "submitted"
assert payload["resume_readiness"] == "not_applicable"
def test_wf_run_start_reports_bad_json() -> None:
root = local_temp_root() / "wf_cli_run_bad_json"
root.mkdir(parents=True, exist_ok=True)
@@ -232,3 +291,44 @@ def test_wf_run_start_reports_bad_json() -> None:
assert result.exit_code != 0
assert "invalid JSON" in result.stderr
def _interrupt_artifact() -> WorkflowArtifact:
return WorkflowArtifact(
id="approval",
version=1,
title="Approval",
input_schema={
"type": "object",
"properties": {"message": {"type": "string"}},
"required": ["message"],
},
output_schema={"type": "object", "properties": {}},
outcomes=("submitted",),
plan={
"name": "approval",
"input_schema": {
"type": "object",
"properties": {"message": {"type": "string"}},
"required": ["message"],
},
"state_schema": {"fields": {}},
"output_schema": {"type": "object", "properties": {}},
"outcomes": ["submitted"],
"start": "approval",
"nodes": [
{
"id": "approval",
"type": "interrupt",
"kind": "approval",
"request": [input_binding("input.message", "message")],
"resume": [],
"outcomes": ["submitted"],
},
{"id": "end_submitted", "type": "end", "outcome": "submitted"},
],
"edges": [
{"from": "approval", "outcome": "submitted", "to": "end_submitted"}
],
},
)