wf_artifacts to end this "with" usage

src/wf_artifacts/drafts/adapter.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from typing import Any
-
 from wf_authoring import WorkflowBuilder
 from wf_authoring.dsl import PathExpr
 from wf_core import JoinNode, Workflow
@@ -44,8 +42,8 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
         return builder.use_ref(
             step.use,
             id=step_id,
-            input=_draft_input_bindings(step),
-            output=_draft_output_bindings(step),
+            input=step.input,
+            output=step.output,
             desc=step.desc,
         )
     if isinstance(step, DraftForeachStep):
@@ -89,24 +87,3 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
             default=step.match.default,
         ).entry
     raise TypeError(f"unsupported draft step {type(step)!r}")
-
-
-def _draft_input_bindings(step: DraftUseStep) -> list[dict[str, Any]]:
-    """Translate draft input maps into canonical core input binding structs.
-
-    Draft JSON keeps `in` and `with` because they are compact patch targets for
-    LLM clients. The compiled workflow should not re-emit deprecated builder map
-    sugar, so this adapter boundary converts them to `NodeUse.input`.
-    """
-    literal_bindings = [
-        {"target": target, "value": value} for target, value in step.with_.items()
-    ]
-    path_bindings = [
-        {"target": target, "path": source} for source, target in step.in_.items()
-    ]
-    return [*literal_bindings, *path_bindings]
-
-
-def _draft_output_bindings(step: DraftUseStep) -> list[dict[str, str]]:
-    """Translate draft output maps into canonical core output binding structs."""
-    return [{"source": source, "target": target} for source, target in step.out.items()]

src/wf_artifacts/drafts/api.py

@@ -73,6 +73,8 @@ def patch_workflow_draft(draft: JsonObject, patch: JsonPatch) -> JsonObject:
             )
         )
     result = validate_workflow_draft(patched)
+    if result["status"] == "valid":
+        patched = WorkflowDraft.model_validate(patched).model_dump(mode="json")
     return {"draft": patched, **result}
 
 

src/wf_artifacts/drafts/models.py

@@ -5,17 +5,20 @@ from typing import Any, Literal
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from wf_core.models.conditions import Condition
+from wf_core.models.steps import InputBinding, OutputBinding
 
 JsonObject = dict[str, Any]
-STEP_KIND_KEYS = frozenset({
-    "use",
-    "foreach",
-    "interrupt",
-    "join",
-    "when",
-    "choose",
-    "match",
-})
+STEP_KIND_KEYS = frozenset(
+    {
+        "use",
+        "foreach",
+        "interrupt",
+        "join",
+        "when",
+        "choose",
+        "match",
+    }
+)
 
 
 class DraftUseStep(BaseModel):
@@ -24,34 +27,67 @@ class DraftUseStep(BaseModel):
     model_config = ConfigDict(extra="forbid", populate_by_name=True)
 
     use: str
-    in_: dict[str, str] = Field(
-        default_factory=dict,
-        alias="in",
+    input: list[InputBinding] = Field(
+        default_factory=list,
         description=(
-            "Source-to-destination map from graph paths to node-local input "
-            "paths. Example: {'input.text': 'message'}. Values must be strings; "
-            "use 'with' for literals."
+            "Canonical input bindings for this capability. Use path bindings "
+            "for graph-to-local input and value bindings for literals."
         ),
     )
-    with_: dict[str, Any] = Field(
-        default_factory=dict,
-        alias="with",
+    output: list[OutputBinding] = Field(
+        default_factory=list,
         description=(
-            "Static node-local input values keyed by destination input field/path. "
-            "Example: {'value': 'CLICKED'}."
-        ),
-    )
-    out: dict[str, str] = Field(
-        default_factory=dict,
-        description=(
-            "Source-to-destination map from node-local output paths to workflow "
-            "state destinations. Example: {'echoed': 'state.echoed'}."
+            "Canonical output bindings from node-local output paths to workflow "
+            "state destinations."
         ),
     )
     desc: str | None = None
     retry: int | None = Field(default=None, ge=0)
     timeout_seconds: int | None = Field(default=None, gt=0)
 
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_legacy_maps(cls, value: object) -> object:
+        """Accept draft `in`/`with`/`out` maps as parse-only compatibility."""
+        if not isinstance(value, dict):
+            return value
+
+        data = dict(value)
+        legacy_input = data.pop("in", None)
+        legacy_with = data.pop("with", None)
+        legacy_output = data.pop("out", None)
+        if "input" in data and (legacy_input is not None or legacy_with is not None):
+            raise ValueError("cannot mix canonical input with legacy in/with maps")
+        if "output" in data and legacy_output is not None:
+            raise ValueError("cannot mix canonical output with legacy out map")
+
+        input_bindings = list(data.get("input", []))
+        output_bindings = list(data.get("output", []))
+        if legacy_with is not None:
+            if not isinstance(legacy_with, dict):
+                raise ValueError("draft use 'with' must be a mapping")
+            input_bindings.extend(
+                {"target": target, "value": literal}
+                for target, literal in legacy_with.items()
+            )
+        if legacy_input is not None:
+            if not isinstance(legacy_input, dict):
+                raise ValueError("draft use 'in' must be a mapping")
+            input_bindings.extend(
+                {"target": target, "path": source}
+                for source, target in legacy_input.items()
+            )
+        if legacy_output is not None:
+            if not isinstance(legacy_output, dict):
+                raise ValueError("draft use 'out' must be a mapping")
+            output_bindings.extend(
+                {"source": source, "target": target}
+                for source, target in legacy_output.items()
+            )
+        data["input"] = input_bindings
+        data["output"] = output_bindings
+        return data
+
 
 class DraftForeachPayload(BaseModel):
     """Payload for one draft foreach step."""