fix: preserve validation errors and subgraph pins

src/wf_api/artifacts.py

@@ -15,6 +15,7 @@ from wf_artifacts import (
     ArtifactKind,
     RequiredCapability,
     WorkflowArtifact,
+    WorkflowPlanValidationError,
     artifact_catalog_entry,
 )
 from wf_artifacts import (
@@ -263,7 +264,7 @@ class WorkflowArtifactApi:
             return _PROJECT_VALIDATE_ARTIFACT(
                 _invalid_artifact_plan_payload(_diagnostic_from_validation_error(exc))
             )
-        except ValueError as exc:
+        except WorkflowPlanValidationError as exc:
             return _PROJECT_VALIDATE_ARTIFACT(
                 _invalid_artifact_plan_payload(
                     {

src/wf_artifacts/__init__.py

@@ -21,7 +21,7 @@ from .drafts import (
     patch_workflow_draft,
     validate_workflow_draft,
 )
-from .factory import create_workflow_artifact_from_plan
+from .factory import WorkflowPlanValidationError, create_workflow_artifact_from_plan
 from .models import (
     ArtifactKind,
     AvailableCapability,
@@ -78,6 +78,7 @@ __all__ = [
     "WorkflowArtifact",
     "WorkflowArtifactCatalogEntry",
     "WorkflowArtifactStore",
+    "WorkflowPlanValidationError",
     "WorkflowCapabilityRef",
     "WorkflowDeployment",
     "WorkflowDraftWorkspace",

src/wf_artifacts/factory.py

@@ -3,12 +3,17 @@ from __future__ import annotations
 from collections.abc import Mapping
 
 from wf_core import ReducerRef, Workflow
+from wf_core.models.workflow_refs import WorkflowRef
 from wf_platform import CapabilityRef, NodeSpecInventory, hash_json_schema
 
 from .models import ArtifactKind, JsonObject, RequiredCapability, WorkflowArtifact
 from .references import normalize_plan_node_refs
 
 
+class WorkflowPlanValidationError(ValueError):
+    """Expected structural validation failure while preparing a workflow plan."""
+
+
 def create_workflow_artifact_from_plan(
     *,
     artifact_id: str,
@@ -47,6 +52,7 @@ def create_workflow_artifact_from_plan(
         outcomes=outcomes,
         plan=normalized_plan,
         required_capabilities=list(required.values()),
+        workflow_dependencies=_workflow_dependencies_from_plan(normalized_plan),
         created_from_catalog_version=created_from_catalog_version,
     )
 
@@ -54,7 +60,9 @@ def create_workflow_artifact_from_plan(
 def _required_object_field(plan: JsonObject, field_name: str) -> JsonObject:
     value = plan.get(field_name)
     if not isinstance(value, dict):
-        raise ValueError(f"workflow plan is missing object field {field_name!r}")
+        raise WorkflowPlanValidationError(
+            f"workflow plan is missing object field {field_name!r}"
+        )
     return value
 
 
@@ -62,11 +70,11 @@ def _validate_workflow_plan(plan: JsonObject) -> None:
     try:
         workflow = Workflow.model_validate(plan)
     except Exception as exc:
-        raise ValueError(f"invalid workflow plan: {exc}") from exc
+        raise WorkflowPlanValidationError(f"invalid workflow plan: {exc}") from exc
 
     node_ids = {node.id for node in workflow.nodes}
     if workflow.start not in node_ids:
-        raise ValueError(
+        raise WorkflowPlanValidationError(
             f"invalid workflow plan: start node {workflow.start!r} does not exist"
         )
 
@@ -76,15 +84,31 @@ def _validate_workflow_plan(plan: JsonObject) -> None:
     edge_sources = set(node_ids)
     for edge in workflow.edges:
         if edge.from_ not in edge_sources:
-            raise ValueError(
+            raise WorkflowPlanValidationError(
                 f"invalid workflow plan: edge source {edge.from_!r} does not exist"
             )
         if edge.to not in node_ids and edge.to != "__end__":
-            raise ValueError(
+            raise WorkflowPlanValidationError(
                 f"invalid workflow plan: edge destination {edge.to!r} does not exist"
             )
 
 
+def _workflow_dependencies_from_plan(plan: JsonObject) -> dict[str, int]:
+    """Collect immutable artifact pins from native saved-subgraph steps."""
+    nodes = plan.get("nodes")
+    if not isinstance(nodes, list):
+        return {}
+
+    dependencies: dict[str, int] = {}
+    for node in nodes:
+        if not isinstance(node, dict) or node.get("type") != "subgraph":
+            continue
+        workflow_ref = WorkflowRef.model_validate(node.get("workflow"))
+        if workflow_ref.artifact_id is not None and workflow_ref.version is not None:
+            dependencies[workflow_ref.artifact_id] = workflow_ref.version
+    return dependencies
+
+
 def _required_reducers_from_plan(plan: JsonObject) -> dict[str, RequiredCapability]:
     """Infer reducer dependencies from declared state fields in one plan."""
     state_schema = plan.get("state_schema")

tests/artifacts/test_factory.py

@@ -111,6 +111,34 @@ def test_create_workflow_artifact_from_plan_rewrites_bound_node_specs() -> None:
     assert str(required.observed_concrete_source) == "demo.personal"
 
 
+def test_create_workflow_artifact_from_plan_derives_saved_workflow_dependencies() -> (
+    None
+):
+    plan = _plan()
+    plan["nodes"] = [
+        {
+            "id": "child",
+            "type": "subgraph",
+            "workflow": {"artifact_id": "child_workflow", "version": 7},
+            "input_schema": {"type": "object"},
+            "output_schema": {"type": "object"},
+            "outcomes": ["ok"],
+        }
+    ]
+    plan["start"] = "child"
+    plan["edges"] = [{"from": "child", "outcome": "ok", "to": "__end__"}]
+
+    artifact = create_workflow_artifact_from_plan(
+        artifact_id="parent",
+        version=1,
+        title="Parent",
+        plan=plan,
+        outcomes=("done",),
+    )
+
+    assert artifact.workflow_dependencies == {"child_workflow": 7}
+
+
 def test_create_workflow_artifact_from_plan_snapshots_observed_node_spec() -> None:
     plan = _plan()
     _set_first_node_ref(plan, "demo.personal.echo_tool")

tests/wf_api/test_artifact_api.py

@@ -236,6 +236,60 @@ async def test_validate_artifact_plan_projects_invalid_plan_diagnostic(
     }
 
 
+@pytest.mark.asyncio
+async def test_validate_artifact_plan_propagates_unexpected_value_error(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "artifacts_unexpected")
+    api, _service = _artifact_api(artifact_store)
+
+    def raise_programming_error(_context: object) -> dict[str, object]:
+        raise ValueError("unexpected preparation defect")
+
+    monkeypatch.setattr(
+        "wf_api.artifacts.observed_node_specs",
+        raise_programming_error,
+    )
+
+    with pytest.raises(ValueError, match="unexpected preparation defect"):
+        await api.validate_artifact_plan(
+            plan=_echo_artifact().plan,
+            outcomes=("completed",),
+            source_bindings={},
+        )
+
+
+@pytest.mark.asyncio
+async def test_validate_artifact_plan_derives_saved_workflow_dependencies(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "artifacts_dependency")
+    api, _service = _artifact_api(artifact_store)
+    plan = _echo_artifact().plan
+    plan["start"] = "child"
+    plan["nodes"] = [
+        {
+            "id": "child",
+            "type": "subgraph",
+            "workflow": {"artifact_id": "child_workflow", "version": 7},
+            "input_schema": {"type": "object"},
+            "output_schema": {"type": "object"},
+            "outcomes": ["completed"],
+        }
+    ]
+    plan["edges"] = [{"from": "child", "outcome": "completed", "to": "__end__"}]
+
+    result = await api.validate_artifact_plan(
+        plan=plan,
+        outcomes=("completed",),
+        source_bindings={},
+    )
+
+    assert result["status"] == "valid"
+    assert result["workflow_dependencies"] == {"child_workflow": 7}
+
+
 @pytest.mark.asyncio
 async def test_create_artifact_from_workspace_suggests_exact_available_source_binding(
     tmp_path: Path,