fix: lower draft output bindings through state

src/wf_api/draft_authoring.py

@@ -12,7 +12,12 @@ from wf_core.models.steps import (
     InputBinding,
     OutputBinding,
 )
-from wf_core.paths import GraphSourcePath, LocalPath
+from wf_core.paths import (
+    GraphSourcePath,
+    LocalPath,
+    format_toml_path_segments,
+    parse_toml_path_segments,
+)
 
 from .constants import (
     DEFAULT_CALL_STEP_ID,
@@ -190,11 +195,19 @@ class WorkflowDraftAuthoringApi:
             if not source_path.startswith("local.")
             else ("local", LocalPath.parse(source_path).parts)
         )
-        target_root, target_parts = (
-            _graph_parts(target_path)
-            if not target_path.startswith("local.")
-            else ("local", LocalPath.parse(target_path).parts)
-        )
+        if target_path.startswith("output."):
+            # GraphSourcePath excludes output targets, but output fields still
+            # use the same canonical TOML-key grammar as other workflow paths.
+            output_path_parts = parse_toml_path_segments(target_path)
+            target_root = output_path_parts[0]
+            target_parts = output_path_parts[1:]
+            if target_root != "output" or not target_parts:
+                raise ValueError("output path must name a field, such as output.result")
+        elif target_path.startswith("local."):
+            target_root = "local"
+            target_parts = LocalPath.parse(target_path).parts
+        else:
+            target_root, target_parts = _graph_parts(target_path)
 
         if target_root == "local" and source_root in {"input", "state"}:
             local_field = _local_field(target_path)
@@ -228,15 +241,89 @@ class WorkflowDraftAuthoringApi:
                 ],
             )
 
-        if source_root == "local" and target_root in {"state", "output"}:
+        if source_root == "local" and target_root == "output":
+            local_field = _local_field(source_path)
+            output_schema_source = (
+                spec.output_schema_contract or spec.output_model.model_json_schema()
+            )
+            state_path_str = format_toml_path_segments(("state", *target_parts))
+            output_target_str = format_toml_path_segments(target_parts)
+
+            state_schema = workspace.draft.get("state_schema", {})
+            if not isinstance(state_schema, dict):
+                raise ValueError("draft state_schema must be an object")
+            projected_state = project_property_to_schema_path(
+                target_schema=state_schema,
+                source_schema=output_schema_source,
+                source_field=local_field,
+                target_parts=target_parts,
+                allow_existing_equivalent=True,
+            )
+
+            output_schema = workspace.draft.get("output_schema", {})
+            if not isinstance(output_schema, dict):
+                raise ValueError("draft output_schema must be an object")
+            projected_output = project_property_to_schema_path(
+                target_schema=output_schema,
+                source_schema=output_schema_source,
+                source_field=local_field,
+                target_parts=target_parts,
+                allow_existing_equivalent=True,
+            )
+
+            output_map = {
+                **self.drafts._step_output_map(
+                    workspace_id=workspace_id, step_id=step_id
+                ),
+                local_field: state_path_str,
+            }
+
+            existing_output = workspace.draft.get("output")
+            if isinstance(existing_output, list):
+                output_bindings = [
+                    b
+                    for b in existing_output
+                    if not (
+                        isinstance(b, dict) and b.get("target") == output_target_str
+                    )
+                ]
+            else:
+                output_bindings = []
+            output_bindings.append(
+                {"path": state_path_str, "target": output_target_str}
+            )
+
+            return await self.drafts.patch_draft_workspace(
+                workspace_id=workspace_id,
+                revision=revision,
+                patch=[
+                    {
+                        "op": "replace",
+                        "path": "/state_schema",
+                        "value": projected_state,
+                    },
+                    {
+                        "op": "replace",
+                        "path": "/output_schema",
+                        "value": projected_output,
+                    },
+                    {
+                        "op": "replace",
+                        "path": f"/steps/{escape_json_pointer(step_id)}/output",
+                        "value": output_bindings_payload(output_map),
+                    },
+                    {"op": "replace", "path": "/output", "value": output_bindings},
+                ],
+            )
+
+        if source_root == "local" and target_root == "state":
             local_field = _local_field(source_path)
             output_schema = (
                 spec.output_schema_contract or spec.output_model.model_json_schema()
             )
-            schema_key = "state_schema" if target_root == "state" else "output_schema"
-            target_schema = workspace.draft.get(schema_key, {})
+            target_schema = workspace.draft.get("state_schema", {})
             if not isinstance(target_schema, dict):
-                raise ValueError(f"draft {schema_key} must be an object")
+                raise ValueError("draft state_schema must be an object")
             projected = project_property_to_schema_path(
                 target_schema=target_schema,
                 source_schema=output_schema,
@@ -253,7 +340,7 @@ class WorkflowDraftAuthoringApi:
                 workspace_id=workspace_id,
                 revision=revision,
                 patch=[
-                    {"op": "replace", "path": f"/{schema_key}", "value": projected},
+                    {"op": "replace", "path": "/state_schema", "value": projected},
                     {
                         "op": "replace",
                         "path": f"/steps/{escape_json_pointer(step_id)}/output",

src/wf_api/drafts.py

@@ -514,17 +514,31 @@ def _draft_repair_hint(
     workspace_id: str,
     revision: int,
 ) -> str | None:
-    if diagnostic.get("code") != "invalid_destination_path":
-        return None
+    code = diagnostic.get("code")
     step_id = diagnostic.get("step_id")
     details = diagnostic.get("details")
     if not isinstance(step_id, str) or not isinstance(details, dict):
         return None
-    output_field = details.get("output_field")
-    state_path = details.get("state_path")
-    if not isinstance(output_field, str) or not isinstance(state_path, str):
-        return None
-    return (
-        f"wf draft bind {workspace_id} --revision {revision} "
-        f"--step {step_id} --from local.{output_field} --to {state_path}"
-    )
+
+    if code == "invalid_destination_path":
+        output_field = details.get("output_field")
+        state_path = details.get("state_path")
+        if not isinstance(output_field, str) or not isinstance(state_path, str):
+            return None
+        return (
+            f"wf draft bind {workspace_id} --revision {revision} "
+            f"--step {step_id} --from local.{output_field} --to {state_path}"
+        )
+
+    if code == "invalid_source_path":
+        source_path = details.get("source_path")
+        target_field = details.get("target_field")
+        if not isinstance(source_path, str) or not isinstance(target_field, str):
+            return None
+        if source_path.startswith("input."):
+            return (
+                f"wf draft bind {workspace_id} --revision {revision} "
+                f"--step {step_id} --from {source_path} --to local.{target_field}"
+            )
+
+    return None

src/wf_api/schema_projection.py

@@ -14,8 +14,13 @@ def project_property_to_schema_path(
     source_schema: JsonObject,
     source_field: str,
     target_parts: tuple[str, ...],
+    allow_existing_equivalent: bool = False,
 ) -> JsonObject:
-    """Copy one source property schema into a target JSON Schema object path."""
+    """Copy one source property schema into a target JSON Schema object path.
+
+    ``allow_existing_equivalent`` accepts exact schema equality only. It does
+    not attempt semantic JSON Schema compatibility analysis.
+    """
     if not target_parts:
         raise ValueError("target schema path must not be empty")
     _check_schema("target_schema", target_schema)
@@ -50,6 +55,11 @@ def project_property_to_schema_path(
     )
     leaf = target_parts[-1]
     if leaf in properties:
+        if allow_existing_equivalent and properties[leaf] == source_property:
+            _merge_definition_block(projected, source_schema, "$defs")
+            _merge_definition_block(projected, source_schema, "definitions")
+            _check_schema("projected target_schema", projected)
+            return projected
         raise ValueError(f"schema path {'.'.join(target_parts)!r} already exists")
     properties[leaf] = deepcopy(source_property)