From 40b0f28b71b87c145b17371aac7f713e40782918 Mon Sep 17 00:00:00 2001 From: lda Date: Mon, 31 Aug 2026 00:32:52 +0700 Subject: [PATCH] fix: preserve validation errors and subgraph pins --- src/wf_api/artifacts.py | 3 +- src/wf_artifacts/__init__.py | 3 +- src/wf_artifacts/factory.py | 34 ++++++++++++++++--- tests/artifacts/test_factory.py | 28 ++++++++++++++++ tests/wf_api/test_artifact_api.py | 54 +++++++++++++++++++++++++++++++ 5 files changed, 115 insertions(+), 7 deletions(-) diff --git a/src/wf_api/artifacts.py b/src/wf_api/artifacts.py index ad37d09b..64f383d2 100644 --- a/src/wf_api/artifacts.py +++ b/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( { diff --git a/src/wf_artifacts/__init__.py b/src/wf_artifacts/__init__.py index d3b79a16..7e9d98b6 100644 --- a/src/wf_artifacts/__init__.py +++ b/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", diff --git a/src/wf_artifacts/factory.py b/src/wf_artifacts/factory.py index 0ca0e67e..b298dd51 100644 --- a/src/wf_artifacts/factory.py +++ b/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") diff --git a/tests/artifacts/test_factory.py b/tests/artifacts/test_factory.py index 23055702..8e3fac7e 100644 --- a/tests/artifacts/test_factory.py +++ b/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") diff --git a/tests/wf_api/test_artifact_api.py b/tests/wf_api/test_artifact_api.py index 5ab4108c..f7aadb1f 100644 --- a/tests/wf_api/test_artifact_api.py +++ b/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,