feat: replace draft output bindings from cli

src/wf_cli/commands/draft_options.py

@@ -8,10 +8,16 @@ import typer
 from pydantic import TypeAdapter, ValidationError
 
 from wf_api.surface import RouteSource
-from wf_core.models.steps import InputBinding, InputPathBinding, InputValueBinding
+from wf_core.models.steps import (
+    InputBinding,
+    InputPathBinding,
+    InputValueBinding,
+    OutputBinding,
+)
 from wf_core.paths import GraphSourcePath, LocalPath, PathResolutionError, StatePath
 
 _INPUT_BINDINGS_ADAPTER = TypeAdapter(list[InputBinding])
+_OUTPUT_BINDINGS_ADAPTER = TypeAdapter(list[OutputBinding])
 
 
 def _parse_assignment_flags(
@@ -170,6 +176,37 @@ def parse_step_input_bindings_file(path: Path) -> list[InputBinding]:
         raise validation_error_as_bad_parameter(exc) from exc
 
 
+def parse_step_output_binding_flags(
+    values: list[str] | None,
+) -> list[OutputBinding]:
+    """Parse ordered local-to-state outputs without collapsing fan-out."""
+    bindings: list[OutputBinding] = []
+    for item in values or []:
+        source, separator, target = item.partition("=")
+        if separator != "=" or not source or not target:
+            raise typer.BadParameter("--map must use LOCAL_SOURCE=STATE_TARGET")
+        try:
+            bindings.append(
+                OutputBinding(
+                    source=LocalPath.parse(source),
+                    target=StatePath.parse(target),
+                )
+            )
+        except (PathResolutionError, ValidationError) as exc:
+            raise typer.BadParameter(str(exc)) from exc
+    return bindings
+
+
+def parse_step_output_bindings_file(path: Path) -> list[OutputBinding]:
+    """Read and validate an ordered canonical output-binding list."""
+    try:
+        return _OUTPUT_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_route_flags(values: list[str] | None) -> dict[str, str]:
     return _parse_assignment_flags(
         values,

src/wf_cli/commands/drafts.py

@@ -9,12 +9,15 @@ import typer
 from wf_cli.commands import draft_add
 from wf_cli.commands.draft_options import (
     _parse_map_flags,
+    _parse_output_map_flags,
     _parse_route_flags,
     _parse_step_input_map_flags,
     parse_json_object_file,
     parse_step_input_binding_flags,
     parse_step_input_bindings_file,
     parse_step_input_value_flags,
+    parse_step_output_binding_flags,
+    parse_step_output_bindings_file,
 )
 from wf_cli.context import load_cli_context_from_typer as load_cli_context
 from wf_cli.formats import ListOutputFormat, emit_list_payload
@@ -500,38 +503,87 @@ def set_step_output_map(
         list[str] | None,
         typer.Option(
             "--map",
-            help="One output binding LOCAL_SOURCE=STATE_TARGET. Repeat in one command.",
+            help=(
+                "LOCAL_SOURCE=STATE_TARGET canonical output binding. Repeat to "
+                "replace the ordered list."
+            ),
         ),
     ] = None,
+    bindings_file: Annotated[
+        Path | None,
+        typer.Option(
+            "--bindings-file",
+            help="Replace with an ordered canonical JSON array of output bindings.",
+        ),
+    ] = None,
+    clear: Annotated[
+        bool,
+        typer.Option("--clear", help="Replace with no bindings."),
+    ] = False,
     merge: Annotated[
         bool,
         typer.Option(
             "--merge",
-            help="Preserve existing output bindings and add/update the passed --map entries.",
+            help=(
+                "Compatibility-only and potentially lossy: preserve existing "
+                "bindings and add/update --map entries."
+            ),
         ),
     ] = False,
 ) -> None:
-    """Set one step's output map without writing JSON Patch manually.
+    """Replace one step's ordered output bindings without writing JSON Patch manually.
 
-    Default behavior replaces the full output map for this step. Pass all
+    By default, ``--map LOCAL_SOURCE=STATE_TARGET`` replaces the complete
-    desired --map entries in one command for a complete replacement. Use
+    ordered canonical binding list. ``--bindings-file`` accepts an ordered
-    --merge only when adding or updating entries across a later revision.
+    canonical JSON array, and ``--clear`` replaces with no bindings. Use
+    ``--merge`` only with ``--map`` for compatibility-only and potentially
+    lossy map edits.
 
     Run `wf draft validate <workspace_id>` after map edits; validation reports
     unresolved paths and conflicting writes.
     """
-    output_map = _parse_map_flags(mapping)
+    has_maps = bool(mapping)
+    has_file = bindings_file is not None
+    selected_modes = sum((has_maps, has_file, clear))
+    if selected_modes == 0:
+        raise typer.BadParameter("provide --map, --bindings-file, or --clear")
+    if selected_modes > 1:
+        raise typer.BadParameter(
+            "--bindings-file and --clear cannot be combined with --map"
+        )
+    if merge and (has_file or clear):
+        raise typer.BadParameter(
+            "--merge is supported only for compatibility map-only edits"
+        )
+
     context = load_cli_context(ctx)
+    if merge:
+        output_map = _parse_output_map_flags(mapping)
+        operation = context.handlers.set_step_output_map(
+            workspace_id=workspace_id,
+            revision=revision,
+            step_id=step_id,
+            output_map=output_map,
+            merge=True,
+        )
+    else:
+        bindings = (
+            parse_step_output_bindings_file(bindings_file)
+            if bindings_file is not None
+            else []
+            if clear
+            else parse_step_output_binding_flags(mapping)
+        )
+        operation = context.handlers.set_step_output_bindings(
+            workspace_id=workspace_id,
+            revision=revision,
+            step_id=step_id,
+            bindings=bindings,
+        )
     emit_json(
         run_cli_operation(
             context,
-            context.handlers.set_step_output_map(
+            operation,
-                workspace_id=workspace_id,
-                revision=revision,
-                step_id=step_id,
-                output_map=output_map,
-                merge=merge,
-            ),
         )
     )
 

tests/wf_cli/test_app.py

@@ -9,6 +9,7 @@ import typer
 from typer.testing import CliRunner
 
 from wf_cli.app import app
+from wf_cli.commands import draft_options
 from wf_cli.commands.draft_options import (
     parse_json_file,
     parse_step_input_binding_flags,
@@ -20,6 +21,80 @@ from wf_cli.commands.draft_options import (
 runner = CliRunner()
 
 
+def _parse_step_output_binding_flags(values: list[str] | None):
+    parser = getattr(draft_options, "parse_step_output_binding_flags", None)
+    assert callable(parser), "step output binding flag parser is not available"
+    return parser(values)
+
+
+def _parse_step_output_bindings_file(path):
+    parser = getattr(draft_options, "parse_step_output_bindings_file", None)
+    assert callable(parser), "step output bindings file parser is not available"
+    return parser(path)
+
+
+def test_draft_options_parse_step_output_bindings_preserves_source_fan_out() -> None:
+    bindings = _parse_step_output_binding_flags(
+        [
+            "report.title=state.report.title",
+            "report.title=state.audit.title",
+        ]
+    )
+
+    assert [binding.model_dump(mode="json") for binding in bindings] == [
+        {"source": "report.title", "target": "state.report.title"},
+        {"source": "report.title", "target": "state.audit.title"},
+    ]
+
+
+def test_draft_options_parse_step_output_bindings_file_preserves_order(
+    tmp_path,
+) -> None:
+    path = tmp_path / "output-bindings.json"
+    path.write_text(
+        json.dumps(
+            [
+                {"source": "report.title", "target": "state.report.title"},
+                {"source": "report.title", "target": "state.audit.title"},
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    bindings = _parse_step_output_bindings_file(path)
+
+    assert [binding.model_dump(mode="json") for binding in bindings] == [
+        {"source": "report.title", "target": "state.report.title"},
+        {"source": "report.title", "target": "state.audit.title"},
+    ]
+
+
+@pytest.mark.parametrize(
+    ("payload", "expected_text"),
+    [
+        ({"source": "report.title", "target": "state.title"}, "list"),
+        ([{"target": "state.title"}], "source"),
+        (
+            [{"source": "report.title", "target": "state.title", "extra": 1}],
+            "extra",
+        ),
+        ([{"source": "report..title", "target": "state.title"}], "path"),
+        ([{"source": "report.title", "target": "title"}], "unknown path root"),
+    ],
+)
+def test_draft_options_parse_step_output_bindings_file_rejects_invalid_payload(
+    tmp_path, payload, expected_text: str
+) -> None:
+    path = tmp_path / "invalid-output-bindings.json"
+    path.write_text(json.dumps(payload), encoding="utf-8")
+
+    with pytest.raises(typer.BadParameter) as exc_info:
+        _parse_step_output_bindings_file(path)
+
+    assert expected_text in str(exc_info.value).lower()
+    assert "traceback" not in str(exc_info.value).lower()
+
+
 def test_draft_options_parse_json_file_reports_invalid_input(tmp_path) -> None:
     path = tmp_path / "invalid.json"
     path.write_text("{", encoding="utf-8")
@@ -576,8 +651,10 @@ def test_wf_draft_map_help_explains_replace_merge_and_validate() -> None:
     assert "input.text=text" in input_help
     assert "input.text=local.text" in input_help
     assert "input.title=report.title" in input_help
-    assert "replaces the full output map" in output_help
+    assert "LOCAL_SOURCE=STATE_TARGET" in output_help
-    assert "Use --merge only" in output_help
+    assert "ordered canonical JSON array" in output_help
+    assert "replace with no bindings" in output_help
+    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 "Use --merge only" in workflow_output_help
@@ -1815,3 +1892,214 @@ def test_wf_draft_set_input_rejects_malformed_local_path(monkeypatch) -> None:
     assert result.exit_code == 2
     assert "rootless node-local path" in result.output
     assert "context loaded" not in result.output
+
+
+def test_wf_draft_set_output_replaces_repeated_maps_with_canonical_bindings(
+    monkeypatch,
+) -> None:
+    binding_calls: list[dict[str, Any]] = []
+    map_calls: list[dict[str, Any]] = []
+
+    class FakeHandlers:
+        async def set_step_output_bindings(self, **kwargs: Any) -> dict[str, Any]:
+            binding_calls.append(kwargs)
+            return {"revision": 2, "status": "valid"}
+
+        async def set_step_output_map(self, **kwargs: Any) -> dict[str, Any]:
+            map_calls.append(kwargs)
+            return {"revision": 2, "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-output",
+            "report_ws",
+            "--revision",
+            "1",
+            "--step",
+            "render",
+            "--map",
+            "report.title=state.report.title",
+            "--map",
+            "report.title=state.audit.title",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert map_calls == []
+    assert [
+        binding.model_dump(mode="json") for binding in binding_calls[0]["bindings"]
+    ] == [
+        {"source": "report.title", "target": "state.report.title"},
+        {"source": "report.title", "target": "state.audit.title"},
+    ]
+
+
+def test_wf_draft_set_output_replaces_from_bindings_file(monkeypatch, tmp_path) -> None:
+    calls: list[dict[str, Any]] = []
+
+    class FakeHandlers:
+        async def set_step_output_bindings(self, **kwargs: Any) -> dict[str, Any]:
+            calls.append(kwargs)
+            return {"revision": 2, "status": "valid"}
+
+    bindings_path = tmp_path / "output-bindings.json"
+    bindings_path.write_text(
+        json.dumps(
+            [
+                {"source": "report.title", "target": "state.report.title"},
+                {"source": "report.title", "target": "state.audit.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-output",
+            "report_ws",
+            "--revision",
+            "1",
+            "--step",
+            "render",
+            "--bindings-file",
+            str(bindings_path),
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert [binding.model_dump(mode="json") for binding in calls[0]["bindings"]] == [
+        {"source": "report.title", "target": "state.report.title"},
+        {"source": "report.title", "target": "state.audit.title"},
+    ]
+
+
+def test_wf_draft_set_output_clear_sends_empty_binding_list(monkeypatch) -> None:
+    calls: list[dict[str, Any]] = []
+
+    class FakeHandlers:
+        async def set_step_output_bindings(self, **kwargs: Any) -> dict[str, Any]:
+            calls.append(kwargs)
+            return {"revision": 2, "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-output",
+            "report_ws",
+            "--revision",
+            "1",
+            "--step",
+            "render",
+            "--clear",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert calls[0]["bindings"] == []
+
+
+@pytest.mark.parametrize(
+    ("extra_args", "expected_error"),
+    [
+        ([], "provide --map, --bindings-file, or --clear"),
+        (
+            ["--bindings-file", "bindings.json", "--map", "value=state.value"],
+            "cannot be combined with --map",
+        ),
+        (
+            ["--clear", "--map", "value=state.value"],
+            "cannot be combined with --map",
+        ),
+        (
+            ["--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_output_rejects_invalid_mode_combinations(
+    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-output",
+            "report_ws",
+            "--revision",
+            "1",
+            "--step",
+            "render",
+            *extra_args,
+        ],
+    )
+
+    assert result.exit_code == 2
+    assert "context loaded" not in result.output
+    assert expected_error in " ".join(result.output.split())
+
+
+def test_wf_draft_set_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_step_output_bindings(self, **kwargs: Any) -> dict[str, Any]:
+            binding_calls.append(kwargs)
+            return {"revision": 2, "status": "valid"}
+
+        async def set_step_output_map(self, **kwargs: Any) -> dict[str, Any]:
+            map_calls.append(kwargs)
+            return {"revision": 2, "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-output",
+            "report_ws",
+            "--revision",
+            "1",
+            "--step",
+            "render",
+            "--map",
+            "value=state.value",
+            "--merge",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    assert binding_calls == []
+    assert map_calls == [
+        {
+            "workspace_id": "report_ws",
+            "revision": 1,
+            "step_id": "render",
+            "output_map": {"value": "state.value"},
+            "merge": True,
+        }
+    ]

tests/wf_cli/test_remote_target.py

@@ -1530,6 +1530,100 @@ def test_wf_draft_set_input_replaces_canonical_bindings_over_rpc(
     assert rpc_methods.count("workflow.draft_workspaces.set_step_input_bindings") == 1
 
 
+def test_wf_draft_set_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")
+    bindings_path = tmp_path / "output-bindings.json"
+    bindings_path.write_text(
+        json.dumps(
+            [
+                {"source": "value", "target": "state.report"},
+                {"source": "value", "target": "state.audit"},
+            ]
+        ),
+        encoding="utf-8",
+    )
+    runner = CliRunner()
+    base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
+    created = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "create",
+            "output_bindings_ws",
+            "--capability",
+            "wf.std.constant",
+            "--name",
+            "output_bindings",
+        ],
+    )
+    assert created.exit_code == 0, created.output
+    rpc_calls.clear()
+
+    replaced = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "set-output",
+            "output_bindings_ws",
+            "--revision",
+            "1",
+            "--step",
+            "call",
+            "--bindings-file",
+            str(bindings_path),
+        ],
+    )
+
+    assert replaced.exit_code == 0, replaced.output
+    assert [method for method, _params in rpc_calls] == [
+        "workflow.draft_workspaces.set_step_output_bindings"
+    ]
+    assert rpc_calls[0][1]["bindings"] == [
+        {"source": "value", "target": "state.report"},
+        {"source": "value", "target": "state.audit"},
+    ]
+
+    rpc_calls.clear()
+    merged = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "set-output",
+            "output_bindings_ws",
+            "--revision",
+            "2",
+            "--step",
+            "call",
+            "--map",
+            "value=state.compat",
+            "--merge",
+        ],
+    )
+
+    assert merged.exit_code == 0, merged.output
+    assert [method for method, _params in rpc_calls] == [
+        "workflow.draft_workspaces.set_step_output_map"
+    ]
+
+
 def test_wf_draft_add_capability_uses_rpc_target(monkeypatch, tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     _patch_rpc_client_to_server(monkeypatch, server)