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

feat: replace workflow output bindings from cli

Browse Source
This commit is contained in:
lda
2026-07-26 21:26:06 +07:00 Verified
parent 54a32c7f41
commit b0a01a63b0
4 changed files with 582 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_cli/commands/draft_options.py
+48 -6
View File
@@ -107,7 +107,16 @@ def validation_error_as_bad_parameter(
def parse_step_input_binding_flags(
values: list[str] | None,
) -> list[InputPathBinding]:
"""Parse ordered canonical path bindings without collapsing source fan-out."""
"""Parse ordered step-input path bindings without collapsing source fan-out."""
return _parse_input_path_binding_flags(values, target_label="node-local")
def _parse_input_path_binding_flags(
values: list[str] | None,
*,
target_label: str,
) -> list[InputPathBinding]:
"""Parse ordered GRAPH_SOURCE=LOCAL_TARGET bindings for one CLI audience."""
bindings: list[InputPathBinding] = []
for item in values or []:
source, separator, target = item.partition("=")
@@ -116,7 +125,7 @@ def parse_step_input_binding_flags(
if target.startswith("local."):
bare_target = target.removeprefix("local.")
raise typer.BadParameter(
"--map target must be a rootless node-local path; "
f"--map target must be a rootless {target_label} path; "
f"use {source}={bare_target}, not {source}={target}"
)
try:
@@ -129,7 +138,7 @@ def parse_step_input_binding_flags(
target_path = LocalPath.parse(target)
except PathResolutionError as exc:
raise typer.BadParameter(
f"--map target must be a rootless node-local path: {exc}"
f"--map target must be a rootless {target_label} path: {exc}"
) from exc
bindings.append(InputPathBinding(path=source_path, target=target_path))
return bindings
@@ -138,7 +147,16 @@ def parse_step_input_binding_flags(
def parse_step_input_value_flags(
values: list[str] | None,
) -> list[InputValueBinding]:
"""Parse ordered canonical literal bindings from LOCAL_TARGET=JSON flags."""
"""Parse ordered step-input literal bindings from LOCAL_TARGET=JSON flags."""
return _parse_input_value_binding_flags(values, target_label="node-local")
def _parse_input_value_binding_flags(
values: list[str] | None,
*,
target_label: str,
) -> list[InputValueBinding]:
"""Parse ordered LOCAL_TARGET=JSON bindings for one CLI audience."""
bindings: list[InputValueBinding] = []
for item in values or []:
target, separator, raw_value = item.partition("=")
@@ -146,7 +164,7 @@ def parse_step_input_value_flags(
raise typer.BadParameter("--value must use LOCAL_TARGET=JSON")
if target.startswith("local."):
raise typer.BadParameter(
"--value target must be a rootless node-local path"
f"--value target must be a rootless {target_label} path"
)
try:
value = json.loads(raw_value)
@@ -161,7 +179,7 @@ def parse_step_input_value_flags(
raise validation_error_as_bad_parameter(exc) from exc
except PathResolutionError as exc:
raise typer.BadParameter(
f"--value target must be a valid rootless local path: {exc}"
f"--value target must be a valid rootless {target_label} path: {exc}"
) from exc
return bindings
@@ -176,6 +194,30 @@ def parse_step_input_bindings_file(path: Path) -> list[InputBinding]:
raise validation_error_as_bad_parameter(exc) from exc
def parse_workflow_output_binding_flags(
values: list[str] | None,
) -> list[InputPathBinding]:
"""Parse ordered canonical workflow-output path bindings."""
return _parse_input_path_binding_flags(values, target_label="workflow-output")
def parse_workflow_output_value_flags(
values: list[str] | None,
) -> list[InputValueBinding]:
"""Parse ordered canonical workflow-output literal bindings."""
return _parse_input_value_binding_flags(values, target_label="workflow-output")
def parse_workflow_output_bindings_file(path: Path) -> list[InputBinding]:
"""Read an ordered canonical workflow-output binding list."""
try:
return _INPUT_BINDINGS_ADAPTER.validate_python(
parse_json_file(path, option_name="--bindings-file")
)
except ValidationError as exc:
raise validation_error_as_bad_parameter(exc) from exc
def parse_step_output_binding_flags(
values: list[str] | None,
) -> list[OutputBinding]:
src/wf_cli/commands/drafts.py
+98 -19
View File
@@ -18,6 +18,9 @@ from wf_cli.commands.draft_options import (
parse_step_input_value_flags,
parse_step_output_binding_flags,
parse_step_output_bindings_file,
parse_workflow_output_binding_flags,
parse_workflow_output_bindings_file,
parse_workflow_output_value_flags,
)
from wf_cli.context import load_cli_context_from_typer as load_cli_context
from wf_cli.formats import ListOutputFormat, emit_list_payload
@@ -607,47 +610,123 @@ def set_workflow_output(
typer.Option(
"--map",
help=(
"One output binding GRAPH_SOURCE=OUTPUT_FIELD, for example "
"state.markdown=markdown. Repeat in one command."
"GRAPH_SOURCE=OUTPUT_TARGET canonical path binding. Repeat to "
"replace the complete ordered workflow output binding list."
),
),
] = None,
literal_values: Annotated[
list[str] | None,
typer.Option(
"--value",
help=(
"OUTPUT_TARGET=JSON canonical literal binding. Repeat after "
"--map bindings in the replacement list."
),
),
] = None,
bindings_file: Annotated[
Path | None,
typer.Option(
"--bindings-file",
help=(
"Replace with an ordered canonical JSON array of path/value "
"workflow output bindings."
),
),
] = None,
clear: Annotated[
bool,
typer.Option(
"--clear",
help=(
"Remove explicit bindings and restore implicit same-name state "
"fallback from output_schema."
),
),
] = False,
merge: Annotated[
bool,
typer.Option(
"--merge",
help="Preserve existing workflow output bindings and add/update the passed --map entries.",
help=(
"Compatibility-only and potentially lossy: preserve existing "
"bindings and add/update map-only entries."
),
),
] = False,
) -> None:
"""Set the top-level workflow output projection without writing JSON Patch manually.
"""Replace the workflow's public output projection with canonical bindings.
Default behavior replaces the full workflow output map. Pass all desired
--map entries in one command for a complete replacement. Use --merge only
when adding or updating entries across a later revision.
By default, ``--map GRAPH_SOURCE=OUTPUT_TARGET`` and
``--value OUTPUT_TARGET=JSON`` replace the complete ordered workflow output
binding list. ``--bindings-file`` accepts an ordered canonical JSON array
and preserves exact path/value interleaving. ``--clear`` restores the
implicit same-name state fallback declared by ``output_schema``. Use
--merge only with --map for compatibility-only and potentially lossy edits.
This edits WorkflowDraft.output (top-level workflow output). Use
wf draft set-output for step-level output bindings.
For single-field input/state sources, missing output_schema fields are
projected automatically from the source schema.
Repeat --map for multiple mappings:
--map state.markdown=markdown --map state.title=title
To round-trip exact ordering, run wf draft inspect --include-draft and
export draft.output for a later --bindings-file replacement.
Run `wf draft validate <workspace_id>` after editing the projection.
"""
output_map = _parse_map_flags(mapping)
has_maps = bool(mapping)
has_values = bool(literal_values)
has_file = bindings_file is not None
has_convenience = has_maps or has_values
if not has_convenience and not has_file and not clear:
raise typer.BadParameter("provide --map, --value, --bindings-file, or --clear")
if merge and (has_values or has_file or clear):
raise typer.BadParameter(
"--merge is supported only for compatibility map-only edits"
)
if has_file and (has_convenience or clear):
raise typer.BadParameter(
"--bindings-file is mutually exclusive with --map, --value, and --clear"
)
if clear and has_convenience:
raise typer.BadParameter("--clear is mutually exclusive with --map and --value")
if merge:
output_map = _parse_map_flags(mapping)
bindings = None
else:
output_map = None
bindings = (
parse_workflow_output_bindings_file(bindings_file)
if bindings_file is not None
else []
if clear
else [
*parse_workflow_output_binding_flags(mapping),
*parse_workflow_output_value_flags(literal_values),
]
)
context = load_cli_context(ctx)
if merge:
assert output_map is not None
operation = context.handlers.set_workflow_output_map(
workspace_id=workspace_id,
revision=revision,
output_map=output_map,
merge=True,
)
else:
assert bindings is not None
operation = context.handlers.set_workflow_output_bindings(
workspace_id=workspace_id,
revision=revision,
bindings=bindings,
)
emit_json(
run_cli_operation(
context,
context.handlers.set_workflow_output_map(
workspace_id=workspace_id,
revision=revision,
output_map=output_map,
merge=merge,
),
operation,
)
)
tests/wf_cli/test_app.py
+329 -3
View File
@@ -16,6 +16,9 @@ from wf_cli.commands.draft_options import (
parse_step_input_value_flags,
parse_step_output_binding_flags,
parse_step_output_bindings_file,
parse_workflow_output_binding_flags,
parse_workflow_output_bindings_file,
parse_workflow_output_value_flags,
route_source,
)
@@ -155,6 +158,107 @@ def test_draft_options_parse_step_input_bindings_file_validates_union(
parse_step_input_bindings_file(path)
def test_parse_workflow_output_flags_preserves_path_then_literal_order() -> None:
bindings = [
*parse_workflow_output_binding_flags(
[
"state.title=report.title",
"state.title=audit.title",
]
),
*parse_workflow_output_value_flags(['format="markdown"', "optional=null"]),
]
assert [binding.model_dump(mode="json") for binding in bindings] == [
{"path": "state.title", "target": "report.title"},
{"path": "state.title", "target": "audit.title"},
{"value": "markdown", "target": "format"},
{"value": None, "target": "optional"},
]
def test_parse_workflow_output_bindings_file_preserves_mixed_order(
tmp_path,
) -> None:
path = tmp_path / "workflow-output-bindings.json"
path.write_text(
json.dumps(
[
{"path": "state.title", "target": "report.title"},
{"value": "markdown", "target": "format"},
{"path": "input.audit", "target": "audit"},
]
),
encoding="utf-8",
)
bindings = parse_workflow_output_bindings_file(path)
assert [binding.model_dump(mode="json") for binding in bindings] == [
{"path": "state.title", "target": "report.title"},
{"value": "markdown", "target": "format"},
{"path": "input.audit", "target": "audit"},
]
@pytest.mark.parametrize(
("parser", "values", "expected_text"),
[
(
parse_workflow_output_binding_flags,
["unknown.title=report.title"],
"graph source path",
),
(
parse_workflow_output_binding_flags,
["state.title=local.report.title"],
"rootless workflow-output path",
),
(
parse_workflow_output_value_flags,
['local.format="markdown"'],
"rootless workflow-output path",
),
(
parse_workflow_output_value_flags,
["format=not-json"],
"invalid JSON",
),
],
)
def test_parse_workflow_output_flags_report_compact_errors(
parser, values: list[str], expected_text: str
) -> None:
with pytest.raises(typer.BadParameter) as exc_info:
parser(values)
message = str(exc_info.value)
assert expected_text in message
assert "Traceback" not in message
@pytest.mark.parametrize(
("payload", "expected_text"),
[
({"path": "state.title", "target": "title"}, "list"),
(
[{"path": "state.title", "value": "x", "target": "title"}],
"validation errors",
),
],
)
def test_parse_workflow_output_bindings_file_rejects_invalid_payload(
tmp_path, payload, expected_text: str
) -> None:
path = tmp_path / "invalid-workflow-output-bindings.json"
path.write_text(json.dumps(payload), encoding="utf-8")
with pytest.raises(typer.BadParameter) as exc_info:
parse_workflow_output_bindings_file(path)
assert expected_text in str(exc_info.value)
def test_wf_help_lists_lifecycle_groups() -> None:
result = runner.invoke(app, ["--help"])
@@ -645,10 +749,15 @@ def test_wf_draft_map_help_explains_replace_merge_and_validate() -> None:
assert "replace with no bindings" in output_help.lower()
assert "compatibility-only and potentially lossy" in output_help
assert "draft validate" in output_help
assert "replaces the full workflow output map" in workflow_output_help
assert "complete ordered workflow output binding list" in workflow_output_help
assert "GRAPH_SOURCE=OUTPUT_TARGET" in workflow_output_help
assert "OUTPUT_TARGET=JSON" in workflow_output_help
assert "ordered canonical JSON array" in workflow_output_help
assert "same-name state fallback" in workflow_output_help
assert "inspect --include-draft" in workflow_output_help
assert "draft.output" in workflow_output_help
assert "Use --merge only" in workflow_output_help
assert "GRAPH_SOURCE=OUTPUT_FIELD" in workflow_output_help
assert "output_schema fields are projected" in workflow_output_help
assert "compatibility-only and potentially lossy" in workflow_output_help
assert "draft validate" in workflow_output_help
@@ -2130,3 +2239,220 @@ def test_wf_draft_set_output_merge_keeps_compatibility_map_handler(monkeypatch)
"merge": True,
}
]
def test_wf_draft_set_workflow_output_replaces_with_canonical_bindings(
monkeypatch,
) -> None:
binding_calls: list[dict[str, Any]] = []
map_calls: list[dict[str, Any]] = []
class FakeHandlers:
async def set_workflow_output_bindings(self, **kwargs: Any) -> dict[str, Any]:
binding_calls.append(kwargs)
return {"revision": 5, "status": "valid"}
async def set_workflow_output_map(self, **kwargs: Any) -> dict[str, Any]:
map_calls.append(kwargs)
return {"revision": 5, "status": "valid"}
context = SimpleNamespace(handlers=FakeHandlers(), verbose=False)
monkeypatch.setattr("wf_cli.commands.drafts.load_cli_context", lambda _ctx: context)
result = runner.invoke(
app,
[
"draft",
"set-workflow-output",
"report_ws",
"--revision",
"4",
"--map",
"state.title=report.title",
"--map",
"state.title=audit.title",
"--value",
'format="markdown"',
],
)
assert result.exit_code == 0, result.output
assert map_calls == []
assert [
binding.model_dump(mode="json") for binding in binding_calls[0]["bindings"]
] == [
{"path": "state.title", "target": "report.title"},
{"path": "state.title", "target": "audit.title"},
{"value": "markdown", "target": "format"},
]
def test_wf_draft_set_workflow_output_replaces_from_bindings_file(
monkeypatch, tmp_path
) -> None:
calls: list[dict[str, Any]] = []
class FakeHandlers:
async def set_workflow_output_bindings(self, **kwargs: Any) -> dict[str, Any]:
calls.append(kwargs)
return {"revision": 5, "status": "valid"}
path = tmp_path / "workflow-output-bindings.json"
path.write_text(
json.dumps(
[
{"value": "markdown", "target": "format"},
{"path": "state.title", "target": "report.title"},
]
),
encoding="utf-8",
)
context = SimpleNamespace(handlers=FakeHandlers(), verbose=False)
monkeypatch.setattr("wf_cli.commands.drafts.load_cli_context", lambda _ctx: context)
result = runner.invoke(
app,
[
"draft",
"set-workflow-output",
"report_ws",
"--revision",
"4",
"--bindings-file",
str(path),
],
)
assert result.exit_code == 0, result.output
assert [binding.model_dump(mode="json") for binding in calls[0]["bindings"]] == [
{"value": "markdown", "target": "format"},
{"path": "state.title", "target": "report.title"},
]
def test_wf_draft_set_workflow_output_clear_restores_fallback(monkeypatch) -> None:
calls: list[dict[str, Any]] = []
class FakeHandlers:
async def set_workflow_output_bindings(self, **kwargs: Any) -> dict[str, Any]:
calls.append(kwargs)
return {"revision": 5, "status": "valid"}
context = SimpleNamespace(handlers=FakeHandlers(), verbose=False)
monkeypatch.setattr("wf_cli.commands.drafts.load_cli_context", lambda _ctx: context)
result = runner.invoke(
app,
[
"draft",
"set-workflow-output",
"report_ws",
"--revision",
"4",
"--clear",
],
)
assert result.exit_code == 0, result.output
assert calls[0]["bindings"] == []
@pytest.mark.parametrize(
("extra_args", "expected_error"),
[
([], "provide --map, --value, --bindings-file, or --clear"),
(
["--bindings-file", "bindings.json", "--map", "state.x=x"],
"--bindings-file is mutually exclusive",
),
(
["--bindings-file", "bindings.json", "--value", "x=1"],
"--bindings-file is mutually exclusive",
),
(
["--bindings-file", "bindings.json", "--clear"],
"--bindings-file is mutually exclusive",
),
(["--clear", "--map", "state.x=x"], "--clear is mutually exclusive"),
(["--clear", "--value", "x=1"], "--clear is mutually exclusive"),
(
["--merge", "--value", "x=1"],
"--merge is supported only for compatibility map-only edits",
),
(
["--merge", "--bindings-file", "bindings.json"],
"--merge is supported only for compatibility map-only edits",
),
(
["--merge", "--clear"],
"--merge is supported only for compatibility map-only edits",
),
],
)
def test_wf_draft_set_workflow_output_rejects_invalid_modes_before_context(
monkeypatch, extra_args: list[str], expected_error: str
) -> None:
monkeypatch.setattr(
"wf_cli.commands.drafts.load_cli_context",
lambda _ctx: (_ for _ in ()).throw(AssertionError("context loaded")),
)
result = runner.invoke(
app,
[
"draft",
"set-workflow-output",
"report_ws",
"--revision",
"4",
*extra_args,
],
)
assert result.exit_code == 2
assert expected_error in " ".join(result.output.split())
assert "context loaded" not in result.output
def test_wf_draft_set_workflow_output_merge_keeps_compatibility_map_handler(
monkeypatch,
) -> None:
binding_calls: list[dict[str, Any]] = []
map_calls: list[dict[str, Any]] = []
class FakeHandlers:
async def set_workflow_output_bindings(self, **kwargs: Any) -> dict[str, Any]:
binding_calls.append(kwargs)
return {"revision": 5, "status": "valid"}
async def set_workflow_output_map(self, **kwargs: Any) -> dict[str, Any]:
map_calls.append(kwargs)
return {"revision": 5, "status": "valid"}
context = SimpleNamespace(handlers=FakeHandlers(), verbose=False)
monkeypatch.setattr("wf_cli.commands.drafts.load_cli_context", lambda _ctx: context)
result = runner.invoke(
app,
[
"draft",
"set-workflow-output",
"report_ws",
"--revision",
"4",
"--map",
"state.title=report.title",
"--merge",
],
)
assert result.exit_code == 0, result.output
assert binding_calls == []
assert map_calls == [
{
"workspace_id": "report_ws",
"revision": 4,
"output_map": {"state.title": "report.title"},
"merge": True,
}
]
tests/wf_cli/test_remote_target.py
+107 -1
View File
@@ -1248,9 +1248,110 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
]
def test_wf_draft_set_workflow_output_uses_rpc_target(monkeypatch, tmp_path) -> None:
def test_wf_draft_set_workflow_output_replaces_canonical_bindings_over_rpc(
monkeypatch, tmp_path
) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
_patch_rpc_client_to_server(monkeypatch, server)
rpc_calls: list[tuple[str, dict[str, Any]]] = []
original_call = RpcClientTransport._call
async def recording_call(
self: RpcClientTransport, method: str, params: dict[str, Any]
) -> dict[str, Any]:
rpc_calls.append((method, params))
return await original_call(self, method, params)
monkeypatch.setattr(RpcClientTransport, "_call", recording_call)
config_path = tmp_path / "wf.json"
config_path.write_text('{"version": 1}', encoding="utf-8")
state_schema_path = tmp_path / "state-schema.json"
state_schema_path.write_text(
json.dumps(
{
"type": "object",
"properties": {"title": {"type": "string"}},
}
),
encoding="utf-8",
)
output_schema_path = tmp_path / "output-schema.json"
output_schema_path.write_text(
json.dumps(
{
"type": "object",
"properties": {"format": {"type": "string"}},
}
),
encoding="utf-8",
)
runner = CliRunner()
base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
created = runner.invoke(
app,
[
*base_args,
"draft",
"create",
"report",
"--name",
"report",
"--state-schema-file",
str(state_schema_path),
"--output-schema-file",
str(output_schema_path),
],
)
assert created.exit_code == 0, created.output
rpc_calls.clear()
result = runner.invoke(
app,
[
*base_args,
"draft",
"set-workflow-output",
"report",
"--revision",
"1",
"--map",
"state.title=report.title",
"--value",
'format="markdown"',
],
)
assert result.exit_code == 0, result.output
assert rpc_calls == [
(
"workflow.draft_workspaces.set_workflow_output_bindings",
{
"workspace_id": "report",
"revision": 1,
"bindings": [
{"path": "state.title", "target": "report.title"},
{"value": "markdown", "target": "format"},
],
},
)
]
def test_wf_draft_set_workflow_output_merge_uses_compatibility_rpc_target(
monkeypatch, tmp_path
) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
_patch_rpc_client_to_server(monkeypatch, server)
rpc_methods: list[str] = []
original_call = RpcClientTransport._call
async def recording_call(
self: RpcClientTransport, method: str, params: dict[str, Any]
) -> dict[str, Any]:
rpc_methods.append(method)
return await original_call(self, method, params)
monkeypatch.setattr(RpcClientTransport, "_call", recording_call)
config_path = tmp_path / "wf.json"
config_path.write_text('{"version": 1}', encoding="utf-8")
runner = CliRunner()
@@ -1277,6 +1378,7 @@ def test_wf_draft_set_workflow_output_uses_rpc_target(monkeypatch, tmp_path) ->
"1",
"--map",
"state.markdown=markdown",
"--merge",
],
)
inspected = runner.invoke(
@@ -1289,6 +1391,10 @@ def test_wf_draft_set_workflow_output_uses_rpc_target(monkeypatch, tmp_path) ->
assert inspected.exit_code == 0, inspected.output
draft = json.loads(inspected.output)["draft"]
assert draft["output"] == [{"path": "state.markdown", "target": "markdown"}]
assert rpc_methods[-2:] == [
"workflow.draft_workspaces.set_workflow_output_map",
"workflow.draft_workspaces.get",
]
def test_wf_draft_remove_route_uses_rpc_target(monkeypatch, tmp_path) -> None: