feat: merge draft step maps

This commit is contained in:
lda
2026-06-25 18:42:40 +07:00 Verified
parent 2d2477fb12
commit bcbdb81228
20 changed files with 359 additions and 14 deletions
+3
View File
@@ -57,6 +57,9 @@ clear operator feedback before adding more architecture.
- Completed: focused draft edit helpers are exposed through RPC/CLI, and
`wf deploy create` is accepted as an alias for `wf deploy save`. Docs now
distinguish draft shape from raw plan shape for agent authoring.
- Completed: `wf draft set-input` and `wf draft set-output` now accept
`--merge`, preserving existing bindings when agents split map edits across
multiple revisions.
- Completed: `wf schema` now lists workflow document/component models, emits
compact JSON outlines for agent discovery, and emits valid self-contained
JSON Schema with `--verbose`.
+6
View File
@@ -296,6 +296,7 @@ wf draft set-name concat_ws --revision 1 --name concat_ws_v2
wf draft set-route concat_ws --revision 2 --step call --outcome ok --to __end__
wf draft set-input concat_ws --revision 3 --step call --map input.items=items --map input.separator=separator
wf draft set-output concat_ws --revision 4 --step call --map value=state.value
wf draft set-input concat_ws --revision 5 --step call --merge --map input.limit=limit
```
`set-input` maps graph source paths to node-local input fields:
@@ -304,6 +305,11 @@ wf draft set-output concat_ws --revision 4 --step call --map value=state.value
`set-output` maps node-local output fields to workflow state paths:
`text=state.text` means `local.text -> state.text`.
By default, `set-input` and `set-output` replace the whole map for that step.
Use repeated `--map` flags in one command when you know the complete map. Use
`--merge` when adding or updating one entry across a later revision while
preserving existing bindings.
Validate:
```bash
+6
View File
@@ -41,7 +41,9 @@ wf draft patch <workspace_id> --revision <n> --input-file patch.json
wf draft set-name <workspace_id> --revision <n> --name <name>
wf draft set-route <workspace_id> --revision <n> --step <step_id> --outcome <outcome> --to <target>
wf draft set-input <workspace_id> --revision <n> --step <step_id> --map input.text=text
wf draft set-input <workspace_id> --revision <n> --step <step_id> --merge --map input.other=other
wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=state.text
wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map other=state.other
wf draft validate <workspace_id>
wf draft save <workspace_id> --artifact <artifact_id> --version <n> --title <title>
@@ -67,6 +69,10 @@ Use public CLI surfaces before broader documentation or implementation search:
Use `wf schema <name> --verbose` only when the complete JSON Schema is required;
the default compact outline is preferred for agent context.
For `draft set-input` and `draft set-output`, repeated `--map` flags in one
command define the complete replacement map. If you split map edits across
multiple commands, pass `--merge` or the later command replaces the earlier map.
## Rules
- Use explicit `--config <path>` for examples, challenge workspaces, and
@@ -80,7 +80,9 @@ CLI equivalents:
wf draft set-name <workspace_id> --revision <n> --name <name>
wf draft set-route <workspace_id> --revision <n> --step <step_id> --outcome ok --to <target>
wf draft set-input <workspace_id> --revision <n> --step <step_id> --map input.text=text
wf draft set-input <workspace_id> --revision <n> --step <step_id> --merge --map input.other=other
wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=state.text
wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map other=state.other
```
`set-input` direction: `input.text=text` means graph source `input.text` maps to
@@ -89,6 +91,10 @@ node-local target `local.text`.
`set-output` direction: `text=state.text` means node-local source `local.text`
maps to graph target `state.text`.
Without `--merge`, `set-input` and `set-output` replace the whole map for that
step. Use repeated `--map` flags in one command for a complete replacement. Use
`--merge` only when adding/updating entries over multiple revisions.
Use JSON Patch for structural edits the helpers do not cover.
For larger patches, write a JSON Patch array to a file and pass it with
@@ -16,6 +16,8 @@ validated, runnable deployment.
5. Inspect/patch/validate the workspace until valid.
- Use focused CLI commands (`set-name`, `set-route`, `set-input`, `set-output`)
for common edits.
- `set-input` and `set-output` replace full maps by default; pass `--merge`
only when adding or updating one entry across a later revision.
- Use JSON Patch only for general structural edits.
6. Save an artifact.
- Draft artifact:
+97 -2
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from typing import Any
from wf_artifacts import (
@@ -207,7 +207,15 @@ class WorkflowDraftApi:
revision: int,
step_id: str,
input_map: dict[str, str],
merge: bool = False,
) -> dict[str, Any]:
input_values: dict[str, Any] = {}
if merge:
existing_map, input_values = self._step_input_maps(
workspace_id=workspace_id,
step_id=step_id,
)
input_map = {**existing_map, **input_map}
return await self.patch_draft_workspace(
workspace_id=workspace_id,
revision=revision,
@@ -215,7 +223,7 @@ class WorkflowDraftApi:
{
"op": "replace",
"path": f"/steps/{_escape_json_pointer(step_id)}/input",
"value": _draft_input_bindings_payload(input_map, {}),
"value": _draft_input_bindings_payload(input_map, input_values),
}
],
)
@@ -227,7 +235,13 @@ class WorkflowDraftApi:
revision: int,
step_id: str,
output_map: dict[str, str],
merge: bool = False,
) -> dict[str, Any]:
if merge:
output_map = {
**self._step_output_map(workspace_id=workspace_id, step_id=step_id),
**output_map,
}
return await self.patch_draft_workspace(
workspace_id=workspace_id,
revision=revision,
@@ -240,6 +254,21 @@ class WorkflowDraftApi:
],
)
def _step_input_maps(
self,
*,
workspace_id: str,
step_id: str,
) -> tuple[dict[str, str], dict[str, Any]]:
workspace = self._draft_store().get_workspace(workspace_id)
step = _draft_step(workspace.draft, step_id)
return _input_maps_from_payload(step.get("input", []))
def _step_output_map(self, *, workspace_id: str, step_id: str) -> dict[str, str]:
workspace = self._draft_store().get_workspace(workspace_id)
step = _draft_step(workspace.draft, step_id)
return _output_map_from_payload(step.get("output", []))
async def create_minimal_draft_workspace(
self,
*,
@@ -374,6 +403,72 @@ def _draft_output_bindings_payload(output_map: dict[str, str]) -> list[dict[str,
]
def _draft_step(draft: Mapping[str, Any], step_id: str) -> Mapping[str, Any]:
steps = draft.get("steps", {})
if not isinstance(steps, Mapping):
raise KeyError("draft steps are not available")
step = steps[step_id]
if not isinstance(step, Mapping):
raise KeyError(f"draft step {step_id!r} is not an object")
return step
def _input_maps_from_payload(
payload: Any,
) -> tuple[dict[str, str], dict[str, Any]]:
"""Read stored canonical input bindings back into focused draft maps."""
input_map: dict[str, str] = {}
input_values: dict[str, Any] = {}
if not isinstance(payload, list):
return input_map, input_values
for item in payload:
if not isinstance(item, Mapping) or "target" not in item:
continue
target = _path_text(item["target"], expected_root="local")
if "path" in item:
input_map[_path_text(item["path"])] = target
elif "value" in item:
input_values[target] = item["value"]
return input_map, input_values
def _output_map_from_payload(payload: Any) -> dict[str, str]:
"""Read stored canonical output bindings back into the focused output map."""
output_map: dict[str, str] = {}
if not isinstance(payload, list):
return output_map
for item in payload:
if not isinstance(item, Mapping):
continue
if "source" in item and "target" in item:
output_map[_path_text(item["source"], expected_root="local")] = _path_text(
item["target"],
expected_root="state",
)
return output_map
def _path_text(value: Any, *, expected_root: str | None = None) -> str:
"""Return compact dotted text for stored structural path JSON."""
if isinstance(value, str):
return value
if not isinstance(value, Mapping):
raise ValueError(f"expected path object, got {value!r}")
root = value.get("root")
if expected_root is not None and root != expected_root:
raise ValueError(f"expected {expected_root} path root")
if not isinstance(root, str):
raise ValueError("path root must be a string")
raw_parts = value.get("parts", [])
if not isinstance(raw_parts, list) or not all(
isinstance(part, str) for part in raw_parts
):
raise ValueError("path parts must be strings")
if root == "local":
return "." if not raw_parts else ".".join(raw_parts)
return root if not raw_parts else f"{root}.{'.'.join(raw_parts)}"
def _graph_path_payload(value: str | GraphSourcePath) -> dict[str, str | list[str]]:
path = value if isinstance(value, GraphSourcePath) else GraphSourcePath.parse(value)
return GraphSourcePath._serialize(path)
+4
View File
@@ -335,12 +335,14 @@ class WorkflowApi:
revision: int,
step_id: str,
input_map: dict[str, str],
merge: bool = False,
) -> dict[str, Any]:
return await self.drafts.set_step_input_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
input_map=input_map,
merge=merge,
)
async def set_step_output_map(
@@ -350,12 +352,14 @@ class WorkflowApi:
revision: int,
step_id: str,
output_map: dict[str, str],
merge: bool = False,
) -> dict[str, Any]:
return await self.drafts.set_step_output_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
output_map=output_map,
merge=merge,
)
async def create_minimal_draft_workspace(
+2
View File
@@ -101,6 +101,7 @@ class WorkflowDraftSurface(Protocol):
revision: int,
step_id: str,
input_map: dict[str, str],
merge: bool = False,
) -> dict[str, Any]: ...
async def set_step_output_map(
@@ -110,6 +111,7 @@ class WorkflowDraftSurface(Protocol):
revision: int,
step_id: str,
output_map: dict[str, str],
merge: bool = False,
) -> dict[str, Any]: ...
async def validate_draft_workspace(
+36 -4
View File
@@ -193,11 +193,26 @@ def set_step_input_map(
list[str] | None,
typer.Option(
"--map",
help="Input binding SOURCE=LOCAL_TARGET. Repeatable. Example: input.text=text",
help="One input binding SOURCE=LOCAL_TARGET. Repeat in one command.",
),
] = None,
merge: Annotated[
bool,
typer.Option(
"--merge",
help="Preserve existing input bindings and add/update the passed --map entries.",
),
] = False,
) -> None:
"""Replace one step's input map without writing JSON Patch manually."""
"""Set one step's input map without writing JSON Patch manually.
Default behavior replaces the full input map for this step. Pass all desired
--map entries in one command for a complete replacement. Use --merge only
when adding or updating entries across a later revision.
Run `wf draft validate <workspace_id>` after map edits; validation reports
unresolved paths and conflicting writes.
"""
input_map = _parse_map_flags(mapping)
context = load_cli_context(ctx)
emit_json(
@@ -208,6 +223,7 @@ def set_step_input_map(
revision=revision,
step_id=step_id,
input_map=input_map,
merge=merge,
),
)
)
@@ -225,11 +241,26 @@ def set_step_output_map(
list[str] | None,
typer.Option(
"--map",
help="Output binding LOCAL_SOURCE=STATE_TARGET. Repeatable. Example: text=state.text",
help="One output binding LOCAL_SOURCE=STATE_TARGET. Repeat in one command.",
),
] = None,
merge: Annotated[
bool,
typer.Option(
"--merge",
help="Preserve existing output bindings and add/update the passed --map entries.",
),
] = False,
) -> None:
"""Replace one step's output map without writing JSON Patch manually."""
"""Set one step's output map without writing JSON Patch manually.
Default behavior replaces the full output map for this step. Pass all
desired --map entries in one command for a complete replacement. Use
--merge only when adding or updating entries across a later revision.
Run `wf draft validate <workspace_id>` after map edits; validation reports
unresolved paths and conflicting writes.
"""
output_map = _parse_map_flags(mapping)
context = load_cli_context(ctx)
emit_json(
@@ -240,6 +271,7 @@ def set_step_output_map(
revision=revision,
step_id=step_id,
output_map=output_map,
merge=merge,
),
)
)
+16 -2
View File
@@ -206,21 +206,35 @@ class SetDraftRouteRequest(BaseModel):
class SetStepInputMapRequest(BaseModel):
"""Typed MCP request for replacing one step input map."""
"""Typed MCP request for replacing or merging one step input map."""
workspace_id: WorkspaceId
revision: int = Field(ge=1, description="Expected current workspace revision.")
step_id: str = Field(description="Draft step id whose input map should change.")
input_map: DraftPathMap
merge: bool = Field(
default=False,
description=(
"When false, replace the full input map. When true, preserve existing "
"bindings and add/update only input_map entries."
),
)
class SetStepOutputMapRequest(BaseModel):
"""Typed MCP request for replacing one step output map."""
"""Typed MCP request for replacing or merging one step output map."""
workspace_id: WorkspaceId
revision: int = Field(ge=1, description="Expected current workspace revision.")
step_id: str = Field(description="Draft step id whose output map should change.")
output_map: DraftPathMap
merge: bool = Field(
default=False,
description=(
"When false, replace the full output map. When true, preserve existing "
"bindings and add/update only output_map entries."
),
)
class DeleteDraftWorkspaceRequest(BaseModel):
+6 -2
View File
@@ -399,7 +399,8 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
name="wf.workflow.set_step_input_map",
title="Set Step Input Map",
description=(
"Replace one compatibility step input map in a draft workspace. "
"Replace or merge one compatibility step input map in a draft workspace. "
"Set merge=true to preserve existing bindings while adding/updating entries. "
"New one-capability bootstraps should prefer canonical input bindings."
),
)
@@ -412,6 +413,7 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
revision=request.revision,
step_id=request.step_id,
input_map=request.input_map,
merge=request.merge,
)
)
@@ -419,7 +421,8 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
name="wf.workflow.set_step_output_map",
title="Set Step Output Map",
description=(
"Replace one compatibility step output map in a draft workspace. "
"Replace or merge one compatibility step output map in a draft workspace. "
"Set merge=true to preserve existing bindings while adding/updating entries. "
"New one-capability bootstraps should prefer canonical output bindings."
),
)
@@ -432,6 +435,7 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
revision=request.revision,
step_id=request.step_id,
output_map=request.output_map,
merge=request.merge,
)
)
@@ -108,6 +108,7 @@ class RpcDraftClientMixin:
revision: int,
step_id: str,
input_map: dict[str, str],
merge: bool = False,
) -> dict[str, Any]:
return await self._call(
"workflow.draft_workspaces.set_step_input_map",
@@ -116,6 +117,7 @@ class RpcDraftClientMixin:
"revision": revision,
"step_id": step_id,
"input_map": input_map,
"merge": merge,
},
)
@@ -126,6 +128,7 @@ class RpcDraftClientMixin:
revision: int,
step_id: str,
output_map: dict[str, str],
merge: bool = False,
) -> dict[str, Any]:
return await self._call(
"workflow.draft_workspaces.set_step_output_map",
@@ -134,6 +137,7 @@ class RpcDraftClientMixin:
"revision": revision,
"step_id": step_id,
"output_map": output_map,
"merge": merge,
},
)
@@ -154,6 +154,7 @@ def register_methods(
revision=params.revision,
step_id=params.step_id,
input_map=params.input_map,
merge=params.merge,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@@ -171,6 +172,7 @@ def register_methods(
revision=params.revision,
step_id=params.step_id,
output_map=params.output_map,
merge=params.merge,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
+2
View File
@@ -130,6 +130,7 @@ class SetStepInputMapParams(RpcParamsModel):
revision: int = Field(ge=1)
step_id: str = Field(min_length=1)
input_map: dict[str, str]
merge: bool = False
class SetStepOutputMapParams(RpcParamsModel):
@@ -137,6 +138,7 @@ class SetStepOutputMapParams(RpcParamsModel):
revision: int = Field(ge=1)
step_id: str = Field(min_length=1)
output_map: dict[str, str]
merge: bool = False
class ValidateDraftWorkspaceParams(RpcParamsModel):
+52
View File
@@ -241,6 +241,58 @@ async def test_draft_workspace_patch_helpers_update_revision_and_bindings(
]
@pytest.mark.asyncio
async def test_step_map_helpers_merge_with_existing_bindings(tmp_path: Path) -> None:
artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch_helper_merge")
api, _service = _draft_api(artifact_store)
await api.create_draft_workspace(
workspace_id="echo_ws",
draft=_echo_draft(),
)
input_mapped = await api.set_step_input_map(
workspace_id="echo_ws",
revision=1,
step_id="echo",
input_map={"input.extra": "extra"},
merge=True,
)
output_mapped = await api.set_step_output_map(
workspace_id="echo_ws",
revision=2,
step_id="echo",
output_map={"extra": "state.extra"},
merge=True,
)
replaced = await api.set_step_input_map(
workspace_id="echo_ws",
revision=3,
step_id="echo",
input_map={"input.final": "final"},
)
fetched = await api.get_draft_workspace(workspace_id="echo_ws", include_draft=True)
assert input_mapped["revision"] == 2
assert output_mapped["revision"] == 3
assert replaced["revision"] == 4
assert fetched["draft"]["steps"]["echo"]["input"] == [
{
"target": {"root": "local", "parts": ["final"]},
"path": {"root": "input", "parts": ["final"]},
}
]
assert fetched["draft"]["steps"]["echo"]["output"] == [
{
"source": {"root": "local", "parts": ["echoed"]},
"target": {"root": "state", "parts": ["echoed"]},
},
{
"source": {"root": "local", "parts": ["extra"]},
"target": {"root": "state", "parts": ["extra"]},
},
]
@pytest.mark.asyncio
async def test_validate_draft_workspace_refreshes_status(tmp_path: Path) -> None:
artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_validate_workspace")
+16
View File
@@ -126,3 +126,19 @@ def test_wf_draft_create_from_capability_help_exists() -> None:
assert result.exit_code == 0
assert "--title" in result.output
def test_wf_draft_map_help_explains_replace_merge_and_validate() -> None:
input_result = runner.invoke(app, ["draft", "set-input", "--help"])
output_result = runner.invoke(app, ["draft", "set-output", "--help"])
assert input_result.exit_code == 0
assert output_result.exit_code == 0
input_help = " ".join(input_result.output.split())
output_help = " ".join(output_result.output.split())
assert "replaces the full input map" in input_help
assert "Use --merge only" in input_help
assert "draft validate" in input_help
assert "replaces the full output map" in output_help
assert "Use --merge only" in output_help
assert "draft validate" in output_help
+44 -2
View File
@@ -1027,6 +1027,38 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
"value=state.value",
],
)
input_merged = runner.invoke(
app,
[
*base_args,
"draft",
"set-input",
"focused_ws",
"--revision",
"5",
"--step",
"call",
"--map",
"input.extra=extra",
"--merge",
],
)
output_merged = runner.invoke(
app,
[
*base_args,
"draft",
"set-output",
"focused_ws",
"--revision",
"6",
"--step",
"call",
"--map",
"extra=state.extra",
"--merge",
],
)
inspected = runner.invoke(
app,
[*base_args, "draft", "inspect", "focused_ws", "--include-draft"],
@@ -1036,6 +1068,8 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
assert routed.exit_code == 0, routed.output
assert input_mapped.exit_code == 0, input_mapped.output
assert output_mapped.exit_code == 0, output_mapped.output
assert input_merged.exit_code == 0, input_merged.output
assert output_merged.exit_code == 0, output_merged.output
assert inspected.exit_code == 0, inspected.output
payload = json.loads(inspected.output)
draft = payload["draft"]
@@ -1045,13 +1079,21 @@ def test_wf_draft_focused_edit_commands_use_rpc_target(monkeypatch, tmp_path) ->
{
"target": {"root": "local", "parts": ["value"]},
"path": {"root": "input", "parts": ["value"]},
}
},
{
"target": {"root": "local", "parts": ["extra"]},
"path": {"root": "input", "parts": ["extra"]},
},
]
assert draft["steps"]["call"]["output"] == [
{
"source": {"root": "local", "parts": ["value"]},
"target": {"root": "state", "parts": ["value"]},
}
},
{
"source": {"root": "local", "parts": ["extra"]},
"target": {"root": "state", "parts": ["extra"]},
},
]
+5
View File
@@ -107,6 +107,11 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
assert "input" in from_capability_request["properties"]
assert "output" in from_capability_request["properties"]
assert "output_map" in from_capability_request["properties"]
set_input_schema = tools_by_name[
"wf.workflow.set_step_input_map"
].inputSchema
set_input_request = set_input_schema["properties"]["request"]
assert "merge" in set_input_request["properties"]
from_capability_output = tools_by_name[
"wf.workflow.create_draft_workspace_from_capability"
].outputSchema
+34 -2
View File
@@ -700,6 +700,28 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
"output_map": {"value": "state.value"},
},
)
input_merged = await _rpc(
client,
"workflow.draft_workspaces.set_step_input_map",
{
"workspace_id": "focused_ws",
"revision": 5,
"step_id": "call",
"input_map": {"input.extra": "extra"},
"merge": True,
},
)
output_merged = await _rpc(
client,
"workflow.draft_workspaces.set_step_output_map",
{
"workspace_id": "focused_ws",
"revision": 6,
"step_id": "call",
"output_map": {"extra": "state.extra"},
"merge": True,
},
)
fetched = await _rpc(
client,
"workflow.draft_workspaces.get",
@@ -710,6 +732,8 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
assert routed["result"]["revision"] == 3
assert input_mapped["result"]["revision"] == 4
assert output_mapped["result"]["revision"] == 5
assert input_merged["result"]["revision"] == 6
assert output_merged["result"]["revision"] == 7
draft = fetched["result"]["draft"]
assert draft["name"] == "focused_renamed"
assert draft["routes"]["call"]["ok"] == "__end__"
@@ -717,13 +741,21 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
{
"target": {"root": "local", "parts": ["value"]},
"path": {"root": "input", "parts": ["value"]},
}
},
{
"target": {"root": "local", "parts": ["extra"]},
"path": {"root": "input", "parts": ["extra"]},
},
]
assert draft["steps"]["call"]["output"] == [
{
"source": {"root": "local", "parts": ["value"]},
"target": {"root": "state", "parts": ["value"]},
}
},
{
"source": {"root": "local", "parts": ["extra"]},
"target": {"root": "state", "parts": ["extra"]},
},
]
@@ -487,11 +487,27 @@ async def test_rpc_client_draft_workspace_focused_edit_methods(tmp_path) -> None
step_id="call",
output_map={"value": "state.value"},
)
input_merged = await client.set_step_input_map(
workspace_id="client_focused_ws",
revision=5,
step_id="call",
input_map={"input.extra": "extra"},
merge=True,
)
output_merged = await client.set_step_output_map(
workspace_id="client_focused_ws",
revision=6,
step_id="call",
output_map={"extra": "state.extra"},
merge=True,
)
assert named["revision"] == 2
assert routed["revision"] == 3
assert input_mapped["revision"] == 4
assert output_mapped["revision"] == 5
assert input_merged["revision"] == 6
assert output_merged["revision"] == 7
async def test_rpc_client_diagnoses_source(tmp_path) -> None: