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

feat: add run watch command

Browse Source
This commit is contained in:
lda
2026-06-05 17:40:10 +07:00 Verified
parent 04491b2550
commit d61b183793
4 changed files with 228 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/current_roadmap.md
+6 -7
View File
@@ -223,13 +223,12 @@ implementation state.
interrupted run, and resume it to completion through `RpcWorkflowApiClient`.
- Auth/source secrets boundary: keep registry desired state separate from
upstream credentials, and surface missing auth as validation diagnostics.
- Run watch/progress: start with polling over existing inspect/trace APIs;
defer SSE/WebSocket/MCP progress until the polling UX is proven insufficient.
- CLI remote error formatting: remote JSON-RPC workflow/admin errors currently
can surface as full Python tracebacks in CLI commands, for example
`wf source inspect missing.source`. Add a compact, user-facing error path for
expected remote `RuntimeError`/workflow errors while preserving tracebacks for
developer/debug mode.
- Completed: `wf run watch` starts run progress UX with polling over existing
`inspect_run` and optional bounded `read_run_trace`. SSE/WebSocket/MCP
progress remains deferred until polling UX proves insufficient.
- Completed: CLI remote error formatting now routes expected operation and
HTTP transport failures through compact Typer/Click errors by default.
`wf --verbose ...` preserves raw exception behavior for debugging.
- MCP package split direction: keep separating "MCP as a client transport"
from "MCP as an upstream workflow source provider." The future shape is
likely `wf_transport_mcp` for exposing workflow/admin surfaces to MCP
docs/wf_cli.md
+7
View File
@@ -255,6 +255,13 @@ Inspect a run without trace detail:
wf run inspect run_123
```
Poll a run until it stops:
```bash
wf run watch run_123 --interval 1
wf run watch run_123 --trace --trace-limit 25
```
Read a bounded trace slice:
```bash
src/wf_cli/commands/runs.py
+49
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
from pathlib import Path
import time
from typing import Annotated
import typer
@@ -15,6 +16,8 @@ app = typer.Typer(
no_args_is_help=True,
)
_STOPPED_RUN_STATUSES = frozenset({"completed", "failed", "interrupted", "blocked"})
@app.command("start")
def start_run(
@@ -67,6 +70,52 @@ def inspect_run(
emit_json(run_cli_operation(context, context.handlers.inspect_run(run_id=run_id)))
@app.command("watch")
def watch_run(
ctx: typer.Context,
run_id: Annotated[str, typer.Argument(help="Durable run id to watch.")],
interval: Annotated[
float,
typer.Option("--interval", min=0.0, help="Polling interval in seconds."),
] = 1.0,
timeout: Annotated[
float | None,
typer.Option("--timeout", min=0.1, help="Maximum seconds to watch."),
] = None,
include_trace: Annotated[
bool,
typer.Option("--trace", help="Include a bounded trace slice when stopped."),
] = False,
trace_from: Annotated[
int,
typer.Option("--trace-from", min=0, help="Trace slice start offset."),
] = 0,
trace_limit: Annotated[
int,
typer.Option("--trace-limit", min=1, max=100, help="Trace slice limit."),
] = 25,
) -> None:
"""Poll a durable run until it reaches a stopped status."""
context = load_cli_context_from_typer(ctx)
started_at = time.monotonic()
while True:
payload = run_cli_operation(context, context.handlers.inspect_run(run_id=run_id))
if payload.get("status") in _STOPPED_RUN_STATUSES:
if include_trace:
payload = run_cli_operation(
context,
context.handlers.read_run_trace(
run_id=run_id,
trace_range=TraceRange(start=trace_from, limit=trace_limit),
),
)
emit_json(payload)
return
if timeout is not None and time.monotonic() - started_at >= timeout:
raise typer.BadParameter(f"run {run_id!r} did not stop before timeout")
time.sleep(interval)
@app.command("resume")
def resume_run(
ctx: typer.Context,
tests/wf_cli/test_run_deploy.py
+166
View File
@@ -2,6 +2,7 @@ from __future__ import annotations
import json
from pathlib import Path
from typing import Any, cast
from unittest.mock import patch
from typer.testing import CliRunner
@@ -19,6 +20,11 @@ from tests.wf_mcp.workflow_surface.conftest import echo_artifact
runner = CliRunner()
class PendingRunHandlers:
async def inspect_run(self, *, run_id: str) -> dict[str, Any]:
return {"run_id": run_id, "status": "running"}
def _load_cli_context_with_specs(ctx: TyperContext | str | Path) -> CliContext:
"""Test-only hook: seed executable demo specs for CLI integration tests.
@@ -225,6 +231,166 @@ def test_wf_run_inspect_and_trace_existing_run() -> None:
assert traced_payload["trace"][0]["node_id"] == "echo"
def test_wf_run_watch_outputs_terminal_run_summary() -> None:
root = local_temp_root() / "wf_cli_run_watch_completed"
root.mkdir(parents=True, exist_ok=True)
config_path = _seed_echo_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",
"echo.personal",
"--input",
'{"text": "hello"}',
],
)
run_id = json.loads(start.output)["run_id"]
watched = runner.invoke(
app,
[
"--config",
str(config_path),
"run",
"watch",
run_id,
"--interval",
"0",
],
)
assert watched.exit_code == 0, watched.output
payload = json.loads(watched.output)
assert payload["run_id"] == run_id
assert payload["status"] == "completed"
assert payload["outcome"] == "ok"
def test_wf_run_watch_stops_on_interrupted_run() -> None:
root = local_temp_root() / "wf_cli_run_watch_interrupted"
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"]
watched = runner.invoke(
app,
[
"--config",
str(config_path),
"run",
"watch",
run_id,
"--interval",
"0",
],
)
assert watched.exit_code == 0, watched.output
payload = json.loads(watched.output)
assert payload["run_id"] == run_id
assert payload["status"] == "interrupted"
assert payload["resume_readiness"] == "ready"
def test_wf_run_watch_can_include_trace_slice() -> None:
root = local_temp_root() / "wf_cli_run_watch_trace"
root.mkdir(parents=True, exist_ok=True)
config_path = _seed_echo_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",
"echo.personal",
"--input",
'{"text": "hello"}',
],
)
run_id = json.loads(start.output)["run_id"]
watched = runner.invoke(
app,
[
"--config",
str(config_path),
"run",
"watch",
run_id,
"--interval",
"0",
"--trace",
"--trace-limit",
"1",
],
)
assert watched.exit_code == 0, watched.output
payload = json.loads(watched.output)
assert payload["run_id"] == run_id
assert payload["status"] == "completed"
assert payload["trace_limit"] == 1
assert payload["trace"][0]["node_id"] == "echo"
def test_wf_run_watch_times_out_for_unstopped_run() -> None:
fake_context = CliContext(
config_path=Path("dummy"),
service=None,
handlers=cast(Any, PendingRunHandlers()),
source_admin=cast(Any, None),
admin=cast(Any, None),
)
with patch(
"wf_cli.commands.runs.load_cli_context_from_typer",
lambda _ctx: fake_context,
):
result = runner.invoke(
app,
[
"run",
"watch",
"run_pending",
"--interval",
"0",
"--timeout",
"0.1",
],
)
assert result.exit_code != 0
assert "did not stop before timeout" in result.output
def test_wf_run_resume_interrupted_run() -> None:
root = local_temp_root() / "wf_cli_run_resume"
root.mkdir(parents=True, exist_ok=True)