feat: replace draft step input bindings

src/wf_api/draft_authoring.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
+from copy import deepcopy
 from dataclasses import dataclass
 from typing import Any
 
@@ -22,8 +23,11 @@ from wf_artifacts.drafts.models import (
     DraftUseStep,
     DraftWhenStep,
 )
+from wf_core.local_paths import has_overlapping_paths, paths_overlap
 from wf_core.models.steps import (
     InputBinding,
+    InputPathBinding,
+    InputValueBinding,
     OutputBinding,
 )
 from wf_core.paths import (
@@ -58,7 +62,9 @@ from .operation_context import WorkflowOperationContext
 from .schema_projection import (
     project_output_property_to_state_schema,
     project_schema_path_to_schema_path,
+    schema_fragment_at_path,
     schema_path_exists,
+    validate_json_value_at_schema_path,
 )
 
 
@@ -72,6 +78,56 @@ def _local_parts(path: str) -> tuple[str, ...]:
     return LocalPath.parse(path.removeprefix("local.")).parts
 
 
+def _draft_schema(draft: Mapping[str, Any], key: str) -> dict[str, Any]:
+    """Return an isolated mutable copy of one draft schema document."""
+    value = draft.get(key, {})
+    if not isinstance(value, dict):
+        raise ValueError(f"draft {key} must be an object")
+    return deepcopy(value)
+
+
+def _overlapping_input_targets_error(
+    bindings: Sequence[InputBinding],
+) -> ValueError:
+    """Describe the first overlapping binding pair with stable input indexes."""
+    for left_index, left in enumerate(bindings):
+        for right_index in range(left_index + 1, len(bindings)):
+            right = bindings[right_index]
+            if paths_overlap(left.target, right.target):
+                return ValueError(
+                    f"bindings[{left_index}].target {str(left.target)!r} "
+                    f"overlaps bindings[{right_index}].target "
+                    f"{str(right.target)!r}"
+                )
+    raise AssertionError("overlap error requested without overlapping targets")
+
+
+def _step_input_bindings_patch(
+    *,
+    workspace: WorkflowDraftWorkspace,
+    step_id: str,
+    bindings: list[dict[str, Any]],
+    input_schema: dict[str, Any],
+    state_schema: dict[str, Any],
+) -> list[dict[str, Any]]:
+    """Build one atomic patch for schemas and canonical step input bindings."""
+    patch: list[dict[str, Any]] = []
+    for key, value in (
+        ("input_schema", input_schema),
+        ("state_schema", state_schema),
+    ):
+        if workspace.draft.get(key, {}) != value:
+            patch.append({"op": "replace", "path": f"/{key}", "value": value})
+    patch.append(
+        {
+            "op": "replace",
+            "path": f"/steps/{escape_json_pointer(step_id)}/input",
+            "value": bindings,
+        }
+    )
+    return patch
+
+
 class WorkflowDraftAuthoringApi:
     """Capability-aware semantic edits over revisioned workflow drafts."""
 
@@ -302,6 +358,107 @@ class WorkflowDraftAuthoringApi:
             draft=draft,
         )
 
+    async def set_step_input_bindings(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        bindings: Sequence[InputBinding],
+    ) -> dict[str, Any]:
+        """Replace one capability step's canonical input bindings atomically."""
+        checked = self._workspace_if_revision_matches(
+            workspace_id=workspace_id,
+            revision=revision,
+        )
+        if isinstance(checked, dict):
+            return checked
+        workspace = checked
+        step = draft_step(workspace.draft, step_id)
+        capability_name = step.get("use")
+        if not isinstance(capability_name, str) or not capability_name:
+            raise ValueError(
+                f"draft step {step_id!r} does not declare a capability use"
+            )
+        spec = self.context.specs.get_qualified_spec(capability_name)
+        capability_schema = (
+            spec.input_schema_contract or spec.input_model.model_json_schema()
+        )
+
+        targets = [binding.target for binding in bindings]
+        if has_overlapping_paths(targets):
+            raise _overlapping_input_targets_error(bindings)
+
+        projected_input = _draft_schema(workspace.draft, "input_schema")
+        projected_state = _draft_schema(workspace.draft, "state_schema")
+        for index, binding in enumerate(bindings):
+            target_parts = binding.target.parts
+            try:
+                schema_fragment_at_path(
+                    capability_schema,
+                    target_parts,
+                    label="capability input schema",
+                )
+            except ValueError as exc:
+                raise ValueError(
+                    f"bindings[{index}].target {str(binding.target)!r} "
+                    f"is not declared by capability {capability_name!r}: {exc}"
+                ) from exc
+
+            if isinstance(binding, InputValueBinding):
+                if not target_parts and not isinstance(binding.value, Mapping):
+                    raise ValueError(
+                        f"bindings[{index}].value for target '.' must be a JSON object"
+                    )
+                validate_json_value_at_schema_path(
+                    schema=capability_schema,
+                    parts=target_parts,
+                    value=binding.value,
+                    label=f"bindings[{index}].value",
+                )
+                continue
+
+            if isinstance(binding, InputPathBinding):
+                source = binding.path
+                if source.root == "context":
+                    continue
+                target_schema = (
+                    projected_input if source.root == "input" else projected_state
+                )
+                if not schema_path_exists(target_schema, source.parts):
+                    target_schema = project_schema_path_to_schema_path(
+                        target_schema=target_schema,
+                        source_schema=capability_schema,
+                        source_parts=target_parts,
+                        target_parts=source.parts,
+                        allow_existing_equivalent=True,
+                    )
+                if source.root == "input":
+                    projected_input = target_schema
+                else:
+                    projected_state = target_schema
+
+        payload = [binding.model_dump(mode="json") for binding in bindings]
+        if (
+            step.get("input", []) == payload
+            and workspace.draft.get("input_schema", {}) == projected_input
+            and workspace.draft.get("state_schema", {}) == projected_state
+        ):
+            return summarize_draft_workspace(workspace)
+
+        patch = _step_input_bindings_patch(
+            workspace=workspace,
+            step_id=step_id,
+            bindings=payload,
+            input_schema=projected_input,
+            state_schema=projected_state,
+        )
+        return await self.drafts.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
+        )
+
     async def bind_draft(
         self,
         *,

src/wf_api/service.py

@@ -5,6 +5,7 @@ from typing import Any
 
 from wf_artifacts import ArtifactKind
 from wf_artifacts.drafts.models import DraftStep
+from wf_core.models.steps import InputBinding
 
 from .artifacts import WorkflowArtifactApi
 from .capabilities import WorkflowCapabilityApi
@@ -408,6 +409,21 @@ class WorkflowApi:
             merge=merge,
         )
 
+    async def set_step_input_bindings(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        bindings: Sequence[InputBinding],
+    ) -> dict[str, Any]:
+        return await self.draft_authoring.set_step_input_bindings(
+            workspace_id=workspace_id,
+            revision=revision,
+            step_id=step_id,
+            bindings=bindings,
+        )
+
     async def set_step_output_map(
         self,
         *,

src/wf_api/surface.py

@@ -5,6 +5,7 @@ from typing import Any, Protocol
 
 from wf_artifacts import ArtifactKind
 from wf_artifacts.drafts.models import DraftStep
+from wf_core.models.steps import InputBinding
 
 from .draft_authoring import RouteSource
 from .runs import TraceRangeLike
@@ -137,6 +138,15 @@ class WorkflowDraftSurface(Protocol):
         merge: bool = False,
     ) -> dict[str, Any]: ...
 
+    async def set_step_input_bindings(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        bindings: Sequence[InputBinding],
+    ) -> dict[str, Any]: ...
+
     async def set_step_output_map(
         self,
         *,

tests/wf_api/test_drafts_service.py

@@ -1,18 +1,22 @@
 from __future__ import annotations
 
+from dataclasses import replace
 from pathlib import Path
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 import pytest
-from pydantic import BaseModel, TypeAdapter
+from pydantic import BaseModel, Field, TypeAdapter
 
 from tests.wf_mcp.test_support import echo_tool
 from wf_api.draft_authoring import RouteSource, WorkflowDraftAuthoringApi
 from wf_api.drafts import WorkflowDraftApi
+from wf_api.models import RawWorkflowPlan
 from wf_api.service import WorkflowApi
 from wf_artifacts import FileDraftWorkspaceStore, FileWorkflowArtifactStore
 from wf_artifacts.drafts.models import DraftStep
 from wf_authoring import node
+from wf_core.models.steps import InputPathBinding, InputValueBinding
+from wf_core.paths import GraphSourcePath, LocalPath
 from wf_mcp.broker import WfMcpService
 from wf_mcp.broker.service.workflow_operation_context import context_from_service
 from wf_mcp.models import ConnectionConfig
@@ -114,6 +118,55 @@ def _nested_report_draft() -> dict[str, Any]:
     }
 
 
+class _StructuredRequest(BaseModel):
+    title: str
+    body: str
+    format: Literal["markdown", "json"]
+    note: str | None = None
+
+
+class _StructuredAudit(BaseModel):
+    title: str = ""
+
+
+class _StructuredReportInput(BaseModel):
+    request: _StructuredRequest
+    audit: _StructuredAudit = Field(default_factory=_StructuredAudit)
+
+
+class _StructuredReportOutput(BaseModel):
+    rendered: str
+
+
+@node(name="structured_report", outcomes=("ok",))
+def _structured_report(payload: _StructuredReportInput) -> _StructuredReportOutput:
+    return _StructuredReportOutput(
+        rendered=(
+            f"{payload.request.title}|{payload.request.body}|{payload.request.format}"
+        )
+    )
+
+
+def _structured_report_draft(
+    capability_name: str = "demo.personal.structured_report",
+) -> dict[str, Any]:
+    return {
+        "name": "structured_report",
+        "input_schema": {"type": "object", "properties": {}},
+        "state_schema": {"type": "object", "properties": {}},
+        "output_schema": {"type": "object", "properties": {}},
+        "start": "report",
+        "steps": {
+            "report": {
+                "use": capability_name,
+                "input": [],
+                "output": [],
+            }
+        },
+        "routes": {"report": {"ok": "__end__"}},
+    }
+
+
 def _draft_api(
     artifact_store: FileWorkflowArtifactStore,
     *,
@@ -138,6 +191,22 @@ def _draft_api(
     )
 
 
+async def _create_structured_binding_api(
+    tmp_path: Path,
+    workspace_id: str,
+) -> tuple[WorkflowDraftApi, WfMcpService, WorkflowApi]:
+    draft_api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / workspace_id),
+        register_echo=True,
+    )
+    service.register_specs("demo.personal", _structured_report)
+    await draft_api.create_draft_workspace(
+        workspace_id=workspace_id,
+        draft=_structured_report_draft(),
+    )
+    return draft_api, service, WorkflowApi(authoring.context)
+
+
 @pytest.mark.asyncio
 async def test_patch_draft_applies_json_patch(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_patch")
@@ -1363,6 +1432,443 @@ async def test_bind_draft_stale_revision_precedes_nested_local_path_error(
     assert result["diagnostics"][0]["code"] == "revision_conflict"
 
 
+@pytest.mark.asyncio
+async def test_set_step_input_bindings_replaces_structured_inputs_in_order(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_structured_binding_api(
+        tmp_path,
+        "structured_input",
+    )
+
+    result = await api.set_step_input_bindings(
+        workspace_id="structured_input",
+        revision=1,
+        step_id="report",
+        bindings=[
+            InputPathBinding(
+                path=GraphSourcePath.state("report", "title"),
+                target=LocalPath.of("request", "title"),
+            ),
+            InputPathBinding(
+                path=GraphSourcePath.state("report", "markdown"),
+                target=LocalPath.of("request", "body"),
+            ),
+            InputValueBinding(
+                target=LocalPath.of("request", "format"),
+                value="markdown",
+            ),
+        ],
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="structured_input",
+        include_draft=True,
+    )
+
+    assert result["revision"] == 2
+    assert inspected["draft"]["steps"]["report"]["input"] == [
+        {"target": "request.title", "path": "state.report.title"},
+        {"target": "request.body", "path": "state.report.markdown"},
+        {"target": "request.format", "value": "markdown"},
+    ]
+
+
+@pytest.mark.asyncio
+async def test_set_step_input_bindings_preserves_source_fan_out(tmp_path: Path) -> None:
+    draft_api, _service, api = await _create_structured_binding_api(
+        tmp_path,
+        "input_fan_out",
+    )
+
+    await api.set_step_input_bindings(
+        workspace_id="input_fan_out",
+        revision=1,
+        step_id="report",
+        bindings=[
+            InputPathBinding(
+                path=GraphSourcePath.state("report", "title"),
+                target=LocalPath.of("request", "title"),
+            ),
+            InputPathBinding(
+                path=GraphSourcePath.state("report", "title"),
+                target=LocalPath.of("audit", "title"),
+            ),
+        ],
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="input_fan_out",
+        include_draft=True,
+    )
+
+    assert inspected["draft"]["steps"]["report"]["input"] == [
+        {"target": "request.title", "path": "state.report.title"},
+        {"target": "audit.title", "path": "state.report.title"},
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("bindings", "message"),
+    [
+        (
+            [
+                InputValueBinding(
+                    target=LocalPath.of("request", "missing"),
+                    value="value",
+                )
+            ],
+            r"bindings\[0\]\.target 'request.missing' is not declared",
+        ),
+        (
+            [
+                InputValueBinding(
+                    target=LocalPath.of("request", "title"),
+                    value="first",
+                ),
+                InputValueBinding(
+                    target=LocalPath.of("request", "title"),
+                    value="second",
+                ),
+            ],
+            r"bindings\[0\]\.target 'request.title' overlaps bindings\[1\]",
+        ),
+        (
+            [
+                InputValueBinding(
+                    target=LocalPath.of("request"),
+                    value={
+                        "title": "Report",
+                        "body": "Body",
+                        "format": "markdown",
+                    },
+                ),
+                InputValueBinding(
+                    target=LocalPath.of("request", "title"),
+                    value="Report",
+                ),
+            ],
+            r"bindings\[0\]\.target 'request' overlaps bindings\[1\]",
+        ),
+        (
+            [
+                InputValueBinding(
+                    target=LocalPath.of("request", "format"),
+                    value="html",
+                )
+            ],
+            r"bindings\[0\]\.value does not satisfy schema at 'request.format'",
+        ),
+        (
+            [InputValueBinding(target=LocalPath.root(), value="not-an-object")],
+            r"bindings\[0\]\.value for target '\.' must be a JSON object",
+        ),
+    ],
+)
+async def test_set_step_input_bindings_rejects_semantic_errors_without_mutation(
+    tmp_path: Path,
+    bindings: list[InputPathBinding | InputValueBinding],
+    message: str,
+) -> None:
+    workspace_id = f"invalid_bindings_{len(message)}"
+    draft_api, _service, api = await _create_structured_binding_api(
+        tmp_path,
+        workspace_id,
+    )
+    before = await draft_api.get_draft_workspace(
+        workspace_id=workspace_id,
+        include_draft=True,
+    )
+
+    with pytest.raises(ValueError, match=message):
+        await api.set_step_input_bindings(
+            workspace_id=workspace_id,
+            revision=1,
+            step_id="report",
+            bindings=bindings,
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id=workspace_id,
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_set_step_input_bindings_rejects_remote_target_reference_without_mutation(
+    tmp_path: Path,
+) -> None:
+    draft_api, service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "remote_binding_target"),
+        register_echo=True,
+    )
+    remote_spec = replace(
+        _structured_report,
+        name="remote_structured_report",
+        input_schema_contract={
+            "type": "object",
+            "properties": {"request": {"$ref": "https://example.com/request.json"}},
+        },
+    )
+    service.register_specs("demo.personal", remote_spec)
+    await draft_api.create_draft_workspace(
+        workspace_id="remote_target",
+        draft=_structured_report_draft("demo.personal.remote_structured_report"),
+    )
+    api = WorkflowApi(authoring.context)
+    before = await draft_api.get_draft_workspace(
+        workspace_id="remote_target",
+        include_draft=True,
+    )
+
+    with pytest.raises(ValueError, match="unsupported reference"):
+        await api.set_step_input_bindings(
+            workspace_id="remote_target",
+            revision=1,
+            step_id="report",
+            bindings=[
+                InputValueBinding(
+                    target=LocalPath.of("request"),
+                    value={},
+                )
+            ],
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id="remote_target",
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+async def test_set_step_input_bindings_rejects_non_capability_step_without_mutation(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "non_capability_binding"),
+        register_echo=True,
+    )
+    draft = _structured_report_draft()
+    draft["steps"]["report"] = {"join": {}}
+    await draft_api.create_draft_workspace(workspace_id="non_capability", draft=draft)
+    api = WorkflowApi(authoring.context)
+    before = await draft_api.get_draft_workspace(
+        workspace_id="non_capability",
+        include_draft=True,
+    )
+
+    with pytest.raises(ValueError, match="does not declare a capability use"):
+        await api.set_step_input_bindings(
+            workspace_id="non_capability",
+            revision=1,
+            step_id="report",
+            bindings=[],
+        )
+
+    after = await draft_api.get_draft_workspace(
+        workspace_id="non_capability",
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "binding",
+    [
+        InputValueBinding(target=LocalPath.of("missing"), value="value"),
+        InputValueBinding(
+            target=LocalPath.of("request", "format"),
+            value="html",
+        ),
+    ],
+)
+async def test_set_step_input_bindings_stale_revision_wins_over_semantic_errors(
+    tmp_path: Path,
+    binding: InputValueBinding,
+) -> None:
+    draft_api, _service, api = await _create_structured_binding_api(
+        tmp_path,
+        f"stale_binding_{len(str(binding.target))}",
+    )
+    workspace_id = f"stale_binding_{len(str(binding.target))}"
+
+    result = await api.set_step_input_bindings(
+        workspace_id=workspace_id,
+        revision=2,
+        step_id="report",
+        bindings=[binding],
+    )
+
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id=workspace_id,
+        include_draft=True,
+    )
+    assert inspected["revision"] == 1
+
+
+@pytest.mark.asyncio
+async def test_set_step_input_bindings_projects_whole_capability_payload(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_structured_binding_api(
+        tmp_path,
+        "whole_payload_binding",
+    )
+
+    await api.set_step_input_bindings(
+        workspace_id="whole_payload_binding",
+        revision=1,
+        step_id="report",
+        bindings=[
+            InputPathBinding(
+                path=GraphSourcePath.input("payload"),
+                target=LocalPath.root(),
+            )
+        ],
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="whole_payload_binding",
+        include_draft=True,
+    )
+
+    payload_schema = inspected["draft"]["input_schema"]["properties"]["payload"]
+    assert payload_schema["properties"]["request"]["$ref"].endswith(
+        "/_StructuredRequest"
+    )
+    assert inspected["draft"]["steps"]["report"]["input"] == [
+        {"target": ".", "path": "input.payload"}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_set_step_input_bindings_projects_input_and_state_but_not_context(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_structured_binding_api(
+        tmp_path,
+        "multi_schema_binding",
+    )
+
+    await api.set_step_input_bindings(
+        workspace_id="multi_schema_binding",
+        revision=1,
+        step_id="report",
+        bindings=[
+            InputPathBinding(
+                path=GraphSourcePath.input("title"),
+                target=LocalPath.of("request", "title"),
+            ),
+            InputPathBinding(
+                path=GraphSourcePath.state("body"),
+                target=LocalPath.of("request", "body"),
+            ),
+            InputValueBinding(
+                target=LocalPath.of("request", "format"),
+                value="markdown",
+            ),
+            InputPathBinding(
+                path=GraphSourcePath.context("prior_outcome"),
+                target=LocalPath.of("request", "note"),
+            ),
+        ],
+    )
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="multi_schema_binding",
+        include_draft=True,
+    )
+    draft = inspected["draft"]
+
+    assert draft["input_schema"]["properties"]["title"]["type"] == "string"
+    assert draft["state_schema"]["properties"]["body"]["type"] == "string"
+    assert set(draft["input_schema"]["properties"]) == {"title"}
+    assert set(draft["state_schema"]["properties"]) == {"body"}
+
+
+@pytest.mark.asyncio
+async def test_set_step_input_bindings_accepts_explicit_null_and_exact_noop(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, api = await _create_structured_binding_api(
+        tmp_path,
+        "null_and_noop_binding",
+    )
+    bindings = [
+        InputValueBinding(
+            target=LocalPath.of("request", "note"),
+            value=None,
+        )
+    ]
+
+    first = await api.set_step_input_bindings(
+        workspace_id="null_and_noop_binding",
+        revision=1,
+        step_id="report",
+        bindings=bindings,
+    )
+    second = await api.set_step_input_bindings(
+        workspace_id="null_and_noop_binding",
+        revision=first["revision"],
+        step_id="report",
+        bindings=bindings,
+    )
+
+    assert first["revision"] == 2
+    assert second["revision"] == 2
+    inspected = await draft_api.get_draft_workspace(
+        workspace_id="null_and_noop_binding",
+        include_draft=True,
+    )
+    assert inspected["draft"]["steps"]["report"]["input"] == [
+        {"target": "request.note", "value": None}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_set_step_input_bindings_compiles_and_assembles_nested_payload(
+    tmp_path: Path,
+) -> None:
+    draft_api, service, api = await _create_structured_binding_api(
+        tmp_path,
+        "execute_structured_binding",
+    )
+    bindings = [
+        InputPathBinding(
+            path=GraphSourcePath.input("title"),
+            target=LocalPath.of("request", "title"),
+        ),
+        InputPathBinding(
+            path=GraphSourcePath.input("body"),
+            target=LocalPath.of("request", "body"),
+        ),
+        InputValueBinding(
+            target=LocalPath.of("request", "format"),
+            value="markdown",
+        ),
+    ]
+    await api.set_step_input_bindings(
+        workspace_id="execute_structured_binding",
+        revision=1,
+        step_id="report",
+        bindings=bindings,
+    )
+
+    compiled = await draft_api.compile_draft_workspace(
+        workspace_id="execute_structured_binding"
+    )
+    plan = RawWorkflowPlan.model_validate(compiled["compiled_plan"])
+    run = await service.run_workflow_from_plan(
+        plan,
+        {"title": "Thesis", "body": "Evidence"},
+    )
+
+    assert run.error is None
+    assert run.trace[0].output == {"rendered": "Thesis|Evidence|markdown"}
+
+
 @pytest.mark.asyncio
 async def test_bind_draft_rejects_unsupported_direction(tmp_path: Path) -> None:
     artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_bind_bad_direction")