fix: address semantic draft review nits

src/wf_api/draft_authoring.py

@@ -4,6 +4,10 @@ from collections.abc import Sequence
 from dataclasses import dataclass
 from typing import Any
 
+from wf_artifacts.draft_workspaces.models import (
+    WorkflowDraftWorkspace,
+    summarize_draft_workspace,
+)
 from wf_core.models.steps import (
     InputBinding,
     OutputBinding,
@@ -45,6 +49,31 @@ class WorkflowDraftAuthoringApi:
         self.context = context
         self.drafts = drafts
 
+    def _workspace_if_revision_matches(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+    ) -> WorkflowDraftWorkspace | dict[str, Any]:
+        """Load a workspace for no-op edits while still enforcing optimistic locks."""
+        workspace = self.drafts._draft_store().get_workspace(workspace_id)
+        if workspace.revision == revision:
+            return workspace
+        return {
+            **summarize_draft_workspace(workspace),
+            "status": "conflict",
+            "diagnostics": [
+                {
+                    "code": "revision_conflict",
+                    "path": "revision",
+                    "message": (
+                        f"workspace {workspace.id!r} is at revision "
+                        f"{workspace.revision}, not {revision}"
+                    ),
+                }
+            ],
+        }
+
     def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
         try:
             spec = self.context.specs.get_qualified_spec(qualified_name)
@@ -317,9 +346,13 @@ class WorkflowDraftAuthoringApi:
             raise ValueError(f"routes for step {step_id!r} must be an object")
         merged = {**existing, **routes}
         if merged == existing:
-            return await self.drafts.get_draft_workspace(
+            checked = self._workspace_if_revision_matches(
                 workspace_id=workspace_id,
+                revision=revision,
             )
+            if isinstance(checked, dict):
+                return checked
+            return summarize_draft_workspace(checked)
         return await self.drafts.patch_draft_workspace(
             workspace_id=workspace_id,
             revision=revision,
@@ -342,9 +375,13 @@ class WorkflowDraftAuthoringApi:
     ) -> dict[str, Any]:
         """Update the target for multiple (step, outcome) pairs atomically."""
         if not branches:
-            return await self.drafts.get_draft_workspace(
+            checked = self._workspace_if_revision_matches(
                 workspace_id=workspace_id,
+                revision=revision,
             )
+            if isinstance(checked, dict):
+                return checked
+            return summarize_draft_workspace(checked)
         workspace = self.drafts._draft_store().get_workspace(workspace_id)
         draft_routes = workspace.draft.get("routes", {})
         if not isinstance(draft_routes, dict):
@@ -374,9 +411,13 @@ class WorkflowDraftAuthoringApi:
                 }
             )
         if not patch:
-            return await self.drafts.get_draft_workspace(
+            checked = self._workspace_if_revision_matches(
                 workspace_id=workspace_id,
+                revision=revision,
             )
+            if isinstance(checked, dict):
+                return checked
+            return summarize_draft_workspace(checked)
         return await self.drafts.patch_draft_workspace(
             workspace_id=workspace_id,
             revision=revision,

src/wf_api/draft_payloads.py

@@ -53,13 +53,13 @@ def state_root_field(value: str) -> str:
 
 
 def _local_path_payload(value: str) -> str:
-    return LocalPath._serialize(LocalPath.parse(value))
+    return str(LocalPath.parse(value))
 
 
 def _graph_path_payload(value: str | GraphSourcePath) -> str:
     path = value if isinstance(value, GraphSourcePath) else GraphSourcePath.parse(value)
-    return GraphSourcePath._serialize(path)
+    return str(path)
 
 
 def _state_path_payload(value: str) -> str:
-    return StatePath._serialize(StatePath.parse(value))
+    return str(StatePath.parse(value))

src/wf_api/schema_projection.py

@@ -32,10 +32,15 @@ def project_output_property_to_state_schema(
         raise ValueError(f"output field {output_field!r} is not a JSON Schema object")
 
     projected = deepcopy(state_schema)
+    state_type = projected.get("type")
+    if state_type is not None and state_type != "object":
+        raise ValueError("state_schema must be an object schema")
     projected.setdefault("type", "object")
     properties = projected.setdefault("properties", {})
     if not isinstance(properties, dict):
         raise ValueError("state_schema.properties must be an object")
+    if state_field in properties:
+        raise ValueError(f"state field {state_field!r} already exists")
     properties[state_field] = deepcopy(output_property)
 
     _merge_definition_block(projected, output_schema, "$defs")

src/wf_api/service.py

@@ -8,7 +8,7 @@ from wf_artifacts import ArtifactKind
 from .artifacts import WorkflowArtifactApi
 from .capabilities import WorkflowCapabilityApi
 from .deployments import WorkflowDeploymentApi
-from .draft_authoring import WorkflowDraftAuthoringApi
+from .draft_authoring import DraftOutcomeRef, WorkflowDraftAuthoringApi
 from .drafts import WorkflowDraftApi
 from .models import RawWorkflowPlan
 from .operation_context import WorkflowOperationContext
@@ -436,8 +436,6 @@ class WorkflowApi:
         branches: list[dict[str, str]],
         target: str,
     ) -> dict[str, Any]:
-        from .draft_authoring import DraftOutcomeRef
-
         refs = [
             DraftOutcomeRef(step_id=b["step_id"], outcome=b["outcome"])
             for b in branches

src/wf_cli/commands/drafts.py

@@ -12,18 +12,39 @@ from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_value
 from wf_cli.remote_errors import run_cli_operation
 
 
-def _parse_map_flags(values: list[str] | None) -> dict[str, str]:
+def _parse_assignment_flags(
+    values: list[str] | None,
+    *,
+    option_name: str,
+    expected: str,
+) -> dict[str, str]:
     parsed: dict[str, str] = {}
     for item in values or []:
         source, separator, target = item.partition("=")
         if separator != "=" or not source or not target:
-            raise typer.BadParameter("--map must use source=target")
+            raise typer.BadParameter(f"{option_name} must use {expected}")
         if source in parsed:
-            raise typer.BadParameter(f"duplicate --map for {source!r}")
+            raise typer.BadParameter(f"duplicate {option_name} for {source!r}")
         parsed[source] = target
     return parsed
 
 
+def _parse_map_flags(values: list[str] | None) -> dict[str, str]:
+    return _parse_assignment_flags(
+        values,
+        option_name="--map",
+        expected="source=target",
+    )
+
+
+def _parse_route_flags(values: list[str] | None) -> dict[str, str]:
+    return _parse_assignment_flags(
+        values,
+        option_name="--route",
+        expected="OUTCOME=TARGET",
+    )
+
+
 app = typer.Typer(
     name="draft",
     help="Create, inspect, patch, validate, and save draft workflows.",
@@ -373,15 +394,7 @@ def add_step_from_capability(
     """
     input_map = _parse_map_flags(input_mapping)
     bind_outputs = _parse_map_flags(output_mapping)
-    routes: dict[str, str] = {}
-    if route:
-        for r in route:
-            key, _, value = r.partition("=")
-            if not key or not value:
-                raise typer.BadParameter(
-                    f"invalid route: {r!r} (expected OUTCOME=TARGET)"
-                )
-            routes[key] = value
+    routes = _parse_route_flags(route)
     context = load_cli_context(ctx)
     emit_json(
         run_cli_operation(
@@ -418,15 +431,7 @@ def branch_draft(
     ] = None,
 ) -> None:
     """Branch multiple outcome routes on a single step atomically."""
-    routes: dict[str, str] = {}
-    if route:
-        for r in route:
-            key, _, value = r.partition("=")
-            if not key or not value:
-                raise typer.BadParameter(
-                    f"invalid route: {r!r} (expected OUTCOME=TARGET)"
-                )
-            routes[key] = value
+    routes = _parse_route_flags(route)
     context = load_cli_context(ctx)
     emit_json(
         run_cli_operation(

src/wf_mcp/workflow_surface/models.py

@@ -280,11 +280,11 @@ class AddStepFromCapabilityRequest(BaseModel):
             "require explicit routes."
         ),
     )
-    input_map: dict[str, str] = Field(
+    input_map: DraftPathMap = Field(
         default_factory=dict,
         description="Graph source path to node-local target field.",
     )
-    bind_outputs: dict[str, str] = Field(
+    bind_outputs: DraftPathMap = Field(
         default_factory=dict,
         description="Node-local output field to state path with schema projection.",
     )