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

feat: add draft control step commands

Browse Source
This commit is contained in:
lda
2026-07-21 00:46:52 +07:00 Verified
parent 63e7a17f9e
commit 18e8fb90e3
4 changed files with 909 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_cli/commands/draft_add.py
+327 -8
View File
@@ -1,14 +1,37 @@
from __future__ import annotations
from typing import Annotated
from pathlib import Path
from typing import Annotated, Literal
import typer
from pydantic import ValidationError
from wf_artifacts.drafts.models import (
DraftEndPayload,
DraftEndStep,
DraftForeachPayload,
DraftForeachStep,
DraftInterruptPayload,
DraftInterruptStep,
DraftJoinStep,
DraftStep,
)
from wf_cli.context import load_cli_context_from_typer as load_cli_context
from wf_cli.io import emit_json
from wf_cli.remote_errors import run_cli_operation
from wf_core.models.schemas import SchemaRef
from wf_core.models.steps import (
ForeachConcurrentPolicy,
)
from .draft_options import _parse_map_flags, _parse_output_map_flags, _parse_route_flags
from .draft_options import (
_parse_map_flags,
_parse_output_map_flags,
_parse_route_flags,
_parse_step_input_map_flags,
parse_json_file,
route_source,
)
app = typer.Typer(
name="add",
@@ -17,6 +40,39 @@ app = typer.Typer(
)
def _submit_step(
ctx: typer.Context,
*,
workspace_id: str,
revision: int,
step_id: str,
step: DraftStep,
from_step: str | None,
from_outcome: str | None,
routes: dict[str, str] | None,
) -> None:
"""Send every typed step through the same local-or-remote API boundary."""
context = load_cli_context(ctx)
emit_json(
run_cli_operation(
context,
context.handlers.add_step(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
step=step,
incoming=route_source(from_step, from_outcome),
routes=routes,
),
)
)
def _as_bad_parameter(exc: ValidationError) -> typer.BadParameter:
"""Keep model validation failures on Click's concise input-error surface."""
return typer.BadParameter(str(exc))
@app.command("capability")
def add_step_from_capability(
ctx: typer.Context,
@@ -99,13 +155,276 @@ def add_step_from_capability(
)
# These subgroups establish the public command boundary for later task slices.
# They intentionally have no command bodies until their typed options are ready.
@app.command("interrupt")
def add_interrupt_step(
ctx: typer.Context,
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
revision: Annotated[
int, typer.Option("--revision", min=1, help="Expected workspace revision.")
],
step_id: Annotated[str, typer.Option("--step", help="New draft step id.")],
kind: Annotated[str, typer.Option("--kind", help="Interrupt request kind.")],
from_step: Annotated[
str | None, typer.Option("--from-step", help="Incoming step id.")
] = None,
from_outcome: Annotated[
str | None,
typer.Option("--from-outcome", help="Outcome on --from-step (default: ok)."),
] = None,
request_schema_file: Annotated[
Path | None,
typer.Option(
"--request-schema-file", help="JSON Schema for the interrupt request."
),
] = None,
resume_schema_file: Annotated[
Path | None,
typer.Option(
"--resume-schema-file", help="JSON Schema for the resume payload."
),
] = None,
request: Annotated[
list[str] | None,
typer.Option(
"--request",
help="Request binding GRAPH_SOURCE=LOCAL_TARGET. Repeat as needed.",
),
] = None,
resume: Annotated[
list[str] | None,
typer.Option(
"--resume",
help="Resume binding LOCAL_SOURCE=STATE_TARGET. Repeat as needed.",
),
] = None,
outcome: Annotated[
list[str] | None,
typer.Option("--outcome", help="Declared resume outcome. Repeat as needed."),
] = None,
route: Annotated[
list[str] | None,
typer.Option("--route", help="Route mapping OUTCOME=TARGET. Repeat as needed."),
] = None,
) -> None:
"""Add a typed interrupt and its request/resume contract."""
request_map = _parse_step_input_map_flags(request, option_name="--request")
resume_map = _parse_output_map_flags(resume, option_name="--resume")
routes = _parse_route_flags(route)
try:
request_schema = (
SchemaRef.model_validate(
parse_json_file(
request_schema_file, option_name="--request-schema-file"
)
)
if request_schema_file is not None
else None
)
resume_schema = (
SchemaRef.model_validate(
parse_json_file(resume_schema_file, option_name="--resume-schema-file")
)
if resume_schema_file is not None
else None
)
step = DraftInterruptStep(
interrupt=DraftInterruptPayload.model_validate(
{
"kind": kind,
"request_schema": request_schema,
"resume_schema": resume_schema,
"request": [
{"path": source, "target": target}
for source, target in request_map.items()
],
"resume": [
{"source": source, "target": target}
for source, target in resume_map.items()
],
"outcomes": outcome or ["submitted"],
}
)
)
except ValidationError as exc:
raise _as_bad_parameter(exc) from exc
_submit_step(
ctx,
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
step=step,
from_step=from_step,
from_outcome=from_outcome,
routes=routes or None,
)
@app.command("foreach")
def add_foreach_step(
ctx: typer.Context,
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
revision: Annotated[
int, typer.Option("--revision", min=1, help="Expected workspace revision.")
],
step_id: Annotated[str, typer.Option("--step", help="New draft step id.")],
over: Annotated[
str, typer.Option("--over", help="Graph path containing the item list.")
],
as_: Annotated[
str, typer.Option("--as", help="Context key for the current item.")
],
mode: Annotated[
Literal["serial", "concurrent"],
typer.Option("--mode", help="Item admission mode."),
] = "serial",
item_error: Annotated[
Literal["fail", "skip", "collect"],
typer.Option("--item-error", help="Per-item failure policy."),
] = "fail",
collect_to: Annotated[
str | None,
typer.Option("--collect-to", help="State path for collected item errors."),
] = None,
max_active: Annotated[
int | None,
typer.Option("--max-active", min=1, help="Concurrent active-item limit."),
] = None,
max_outstanding: Annotated[
int | None,
typer.Option(
"--max-outstanding", min=1, help="Concurrent outstanding-item limit."
),
] = None,
from_step: Annotated[
str | None, typer.Option("--from-step", help="Incoming step id.")
] = None,
from_outcome: Annotated[
str | None,
typer.Option("--from-outcome", help="Outcome on --from-step (default: ok)."),
] = None,
route: Annotated[
list[str] | None,
typer.Option("--route", help="Route mapping OUTCOME=TARGET. Repeat as needed."),
] = None,
) -> None:
"""Add a foreach loop with explicit item and concurrency policies."""
if mode == "serial" and (max_active is not None or max_outstanding is not None):
raise typer.BadParameter(
"--max-active and --max-outstanding require --mode concurrent"
)
try:
concurrent_options: dict[str, int] = {}
if max_active is not None:
concurrent_options["max_active"] = max_active
if max_outstanding is not None:
concurrent_options["max_outstanding"] = max_outstanding
concurrent = (
ForeachConcurrentPolicy.model_validate(concurrent_options)
if mode == "concurrent"
else None
)
step = DraftForeachStep(
foreach=DraftForeachPayload.model_validate(
{
"over": over,
"as": as_,
"mode": mode,
"item_error": {
"action": item_error,
"collect_to": collect_to,
},
"concurrent": concurrent,
}
)
)
except ValidationError as exc:
raise _as_bad_parameter(exc) from exc
_submit_step(
ctx,
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
step=step,
from_step=from_step,
from_outcome=from_outcome,
routes=_parse_route_flags(route) or None,
)
@app.command("join")
def add_join_step(
ctx: typer.Context,
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
revision: Annotated[
int, typer.Option("--revision", min=1, help="Expected workspace revision.")
],
step_id: Annotated[str, typer.Option("--step", help="New draft step id.")],
from_step: Annotated[
str | None, typer.Option("--from-step", help="Incoming step id.")
] = None,
from_outcome: Annotated[
str | None,
typer.Option("--from-outcome", help="Outcome on --from-step (default: ok)."),
] = None,
route: Annotated[
list[str] | None,
typer.Option("--route", help="Route mapping OUTCOME=TARGET. Repeat as needed."),
] = None,
) -> None:
"""Add a join step."""
_submit_step(
ctx,
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
step=DraftJoinStep(join={}),
from_step=from_step,
from_outcome=from_outcome,
routes=_parse_route_flags(route) or None,
)
@app.command("end")
def add_end_step(
ctx: typer.Context,
workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
revision: Annotated[
int, typer.Option("--revision", min=1, help="Expected workspace revision.")
],
step_id: Annotated[str, typer.Option("--step", help="New draft step id.")],
outcome: Annotated[
str, typer.Option("--outcome", help="Public workflow outcome.")
] = "ok",
from_step: Annotated[
str | None, typer.Option("--from-step", help="Incoming step id.")
] = None,
from_outcome: Annotated[
str | None,
typer.Option("--from-outcome", help="Outcome on --from-step (default: ok)."),
] = None,
) -> None:
"""Add an explicit terminal outcome step."""
try:
step = DraftEndStep(end=DraftEndPayload(outcome=outcome))
except ValidationError as exc:
raise _as_bad_parameter(exc) from exc
_submit_step(
ctx,
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
step=step,
from_step=from_step,
from_outcome=from_outcome,
routes=None,
)
# These subgroups establish the public boundary for the next command slice.
# They intentionally have no bodies until their typed decision options are ready.
for _name in (
"interrupt",
"foreach",
"join",
"end",
"when",
"choose",
"match",
src/wf_cli/commands/draft_options.py
+15 -7
View File
@@ -35,10 +35,12 @@ def _parse_map_flags(values: list[str] | None) -> dict[str, str]:
)
def _parse_output_map_flags(values: list[str] | None) -> dict[str, str]:
def _parse_output_map_flags(
values: list[str] | None, *, option_name: str = "--bind-output"
) -> dict[str, str]:
parsed = _parse_assignment_flags(
values,
option_name="--bind-output",
option_name=option_name,
expected="LOCAL_OUTPUT=STATE_TARGET",
)
for local_output, state_target in parsed.items():
@@ -46,27 +48,33 @@ def _parse_output_map_flags(values: list[str] | None) -> dict[str, str]:
LocalPath.parse(local_output)
except PathResolutionError as exc:
raise typer.BadParameter(
f"--bind-output source {local_output!r} must be a node-local "
f"{option_name} source {local_output!r} must be a node-local "
"output path such as value or ."
) from exc
try:
StatePath.parse(state_target)
except PathResolutionError as exc:
raise typer.BadParameter(
f"--bind-output target {state_target!r} must be a state path "
f"{option_name} target {state_target!r} must be a state path "
"such as state.value"
) from exc
return parsed
def _parse_step_input_map_flags(values: list[str] | None) -> dict[str, str]:
def _parse_step_input_map_flags(
values: list[str] | None, *, option_name: str = "--map"
) -> dict[str, str]:
"""Parse graph-source to bare-local input mappings for one draft step."""
parsed = _parse_map_flags(values)
parsed = _parse_assignment_flags(
values,
option_name=option_name,
expected="GRAPH_SOURCE=LOCAL_TARGET",
)
for source, target in parsed.items():
if target.startswith("local."):
bare_target = target.removeprefix("local.")
raise typer.BadParameter(
"--map target must be a bare local field; "
f"{option_name} target must be a bare local field; "
f"use {source}={bare_target}, not {source}={target}"
)
return parsed